87 lines
3.5 KiB
Python
87 lines
3.5 KiB
Python
"""
|
|
Gitea REST API клиент — fallback когда Gitea MCP недоступен.
|
|
Используется напрямую из агента если uvx/gitea-mcp не установлен.
|
|
"""
|
|
|
|
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", "homework-solutions")
|
|
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",
|
|
})
|
|
self.session.verify = False # git.brojs.ru SSL бывает нестабилен
|
|
|
|
def ensure_repo(self) -> bool:
|
|
return self.ensure_repo_named(self.repo)
|
|
|
|
def ensure_repo_named(self, repo_name: str, description: str = "Решение домашнего задания") -> bool:
|
|
"""Создаёт репозиторий с заданным именем если не существует."""
|
|
resp = self.session.get(f"{self.api}/repos/{self.username}/{repo_name}")
|
|
if resp.status_code == 200:
|
|
print(f"[GITEA] 📦 Репо существует: {repo_name}")
|
|
return True
|
|
resp = self.session.post(
|
|
f"{self.api}/user/repos",
|
|
json={
|
|
"name": repo_name,
|
|
"description": description,
|
|
"private": False,
|
|
"auto_init": True,
|
|
"default_branch": "main",
|
|
},
|
|
)
|
|
if resp.ok:
|
|
print(f"[GITEA] 🆕 Создан репозиторий: {repo_name}")
|
|
return True
|
|
print(f"[GITEA] ❌ Ошибка создания репо {repo_name}: {resp.status_code} {resp.text[:200]}")
|
|
return False
|
|
|
|
def upload_file(self, file_path: str, content: str, commit_message: str) -> str:
|
|
"""
|
|
Загружает файл в репозиторий.
|
|
Возвращает URL файла на Gitea.
|
|
"""
|
|
self.ensure_repo()
|
|
|
|
url = f"{self.api}/repos/{self.username}/{self.repo}/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}/{self.repo}/src/branch/main/{file_path}"
|
|
print(f"[GITEA] ✅ Загружен: {file_url}")
|
|
return file_url
|
|
|
|
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) |