Загрузить файлы в «src»

2026-06-04 16:45:30 +00:00
parent e654afeb2d
commit 9ceab971b7
5 changed files with 623 additions and 0 deletions
@@ -0,0 +1,112 @@
+"""
+Gitea REST API клиент — fallback когда Gitea MCP недоступен.
+Используется напрямую из агента если uvx/gitea-mcp не установлен.
+
+REST-клиент для Gitea — альтернатива когда uvx gitea-mcp не установлен. Умеет:
+ensure_repo(repo) — создать репозиторий если не существует
+upload_file(file_path, content, commit_message) — залить файл, при обновлении сам находит SHA
+get_file(file_path) — прочитать файл из репозитория
+list_files(repo, path) — список файлов в папке
+"""
+
+import base64
+import os
+from typing import Optional
+
+import requests
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+class GiteaRestClient:
+    def __init__(self):
+        self.base_url = os.getenv("GITEA_URL", "https://git.brojs.ru").rstrip("/")
+        self.token = os.getenv("GITEA_TOKEN")
+        self.username = os.getenv("GITEA_USERNAME")
+        self.repo = os.getenv("GITEA_REPO", "dz")
+        self.api = f"{self.base_url}/api/v1"
+        self.session = requests.Session()
+        self.session.headers.update({
+            "Authorization": f"token {self.token}",
+            "Content-Type": "application/json",
+        })
+
+    def ensure_repo(self, repo: str | None = None) -> bool:
+        """Создаёт репозиторий если не существует. Возвращает True при успехе."""
+        target = repo or self.repo
+        resp = self.session.get(f"{self.api}/repos/{self.username}/{target}")
+        if resp.status_code == 200:
+            return True
+        resp = self.session.post(
+            f"{self.api}/user/repos",
+            json={
+                "name": target,
+                "description": "Решения домашних заданий",
+                "private": False,
+                "auto_init": True,
+                "default_branch": "main",
+            },
+        )
+        if resp.ok:
+            print(f"[GITEA] 🆕 Создан репозиторий: {target}")
+            return True
+        print(f"[GITEA] ❌ Ошибка создания репо: {resp.status_code} {resp.text[:200]}")
+        return False
+
+    def upload_file(self, file_path: str, content: str, commit_message: str, repo: str | None = None) -> str:
+        """
+        Загружает файл в репозиторий.
+        Возвращает URL файла на Gitea.
+        """
+        target = repo or self.repo
+        self.ensure_repo(target)
+
+        url = f"{self.api}/repos/{self.username}/{target}/contents/{file_path}"
+        content_b64 = base64.b64encode(content.encode("utf-8")).decode("utf-8")
+
+        # Проверяем sha (для обновления)
+        existing = self.session.get(url, params={"ref": "main"})
+        payload = {
+            "message": commit_message,
+            "content": content_b64,
+            "branch": "main",
+        }
+        if existing.status_code == 200:
+            payload["sha"] = existing.json().get("sha")
+
+        resp = self.session.put(url, json=payload)
+        resp.raise_for_status()
+
+        file_url = f"{self.base_url}/{self.username}/{target}/src/branch/main/{file_path}"
+        print(f"[GITEA] ✅ Загружен: {file_url}")
+        return file_url
+
+    def get_file(self, file_path: str, repo: str | None = None) -> str | None:
+        """Читает содержимое файла из репозитория. Возвращает None если файл не найден."""
+        target = repo or self.repo
+        url = f"{self.api}/repos/{self.username}/{target}/contents/{file_path}"
+        resp = self.session.get(url, params={"ref": "main"})
+        if resp.status_code != 200:
+            return None
+        content_b64 = resp.json().get("content", "")
+        return base64.b64decode(content_b64).decode("utf-8")
+
+    def list_files(self, repo: str | None = None, path: str = "") -> list[str]:
+        """Возвращает список имён файлов в корне репозитория."""
+        target = repo or self.repo
+        url = f"{self.api}/repos/{self.username}/{target}/contents/{path}"
+        resp = self.session.get(url, params={"ref": "main"})
+        if resp.status_code != 200:
+            return []
+        items = resp.json()
+        if not isinstance(items, list):
+            return []
+        return [item["name"] for item in items if item.get("type") == "file"]
+
+    def upload_solution(self, assignment_id: str, title: str, code: str, language: str = "py") -> str:
+        """Удобный метод: залить решение и получить ссылку."""
+        safe = "".join(c if c.isalnum() or c in "-_" else "_" for c in title)[:40]
+        file_path = f"solutions/{assignment_id}_{safe}/solution.{language}"
+        commit_msg = f"Решение задания: {title}"
+        return self.upload_file(file_path, code, commit_msg)