""" title: Gitea Manager author: Claude version: 1.3.0 description: Инструмент для управления Gitea из Open WebUI — создание и удаление репозиториев, организаций, файлов, веток, загрузка и безопасная распаковка ZIP-архивов. requirements: requests """ import base64 import json import shutil import stat import tempfile import zipfile from pathlib import Path, PurePosixPath from typing import Optional from urllib.parse import quote import requests from pydantic import BaseModel, Field class Tools: """Инструмент управления Gitea и локальными файлами рабочего пространства.""" class Valves(BaseModel): GITEA_URL: str = Field(default="http://192.168.31.100:30008", description="Базовый URL Gitea без слэша в конце.") GITEA_TOKEN: str = Field(default="", description="API-токен Gitea. Заполните в настройках инструмента.") TIMEOUT: int = Field(default=15, description="Таймаут запросов в секундах.") ENABLED: bool = Field(default=True, description="Включить инструмент Gitea Manager.") WORKING_DIRECTORY: str = Field(default="/tmp/gitea-manager-workdir", description="Корневая рабочая директория.") MAX_ARCHIVE_FILES: int = Field(default=10000, description="Максимальное количество элементов ZIP.") MAX_ARCHIVE_SIZE_MB: int = Field(default=2048, description="Максимальный распакованный размер ZIP в МБ.") def __init__(self): self.valves = self.Valves() def _check(self) -> Optional[str]: """Проверить доступность инструмента и наличие токена.""" if not self.valves.ENABLED: return "Инструмент Gitea Manager выключен." if not self.valves.GITEA_TOKEN: return "API-токен Gitea не задан в настройках инструмента." return None def _request(self, method: str, path: str, **kwargs) -> str: """Выполнить запрос к Gitea и вернуть JSON или сообщение об ошибке.""" error = self._check() if error: return error headers = {"Authorization": f"token {self.valves.GITEA_TOKEN}", "Content-Type": "application/json"} try: response = requests.request( method, f"{self.valves.GITEA_URL.rstrip('/')}/api/v1{path}", headers=headers, timeout=self.valves.TIMEOUT, **kwargs, ) if not response.ok: return f"Ошибка {response.status_code}: {response.text}" return response.text or "Операция выполнена успешно." except requests.RequestException as exc: return f"Ошибка соединения с Gitea: {exc}" def create_repository(self, name: str, description: str = "", private: bool = False, auto_init: bool = True) -> str: """Создать репозиторий текущего пользователя.""" return self._request("POST", "/user/repos", json={"name": name, "description": description, "private": private, "auto_init": auto_init}) def create_repository_in_org(self, org: str, name: str, description: str = "", private: bool = False, auto_init: bool = True) -> str: """Создать репозиторий внутри организации.""" return self._request("POST", f"/orgs/{quote(org, safe='')}/repos", json={"name": name, "description": description, "private": private, "auto_init": auto_init}) def list_repositories( self, limit: int = 50, page: int = 1, sort: str = "updated", fields: Optional[str] = "name,private,language,html_url,description,updated_at,default_branch", ) -> str: """ Получить список репозиториев текущего пользователя в компактном виде. По умолчанию возвращает не весь JSON Gitea, а только краткий набор полей — иначе ответ раздувается из-за permissions/url/настроек и обрезается в чате. Если репозиториев больше, чем limit, в конце добавляется подсказка про page+1. Args: limit: максимум репозиториев на странице (1..100). page: номер страницы (начиная с 1). sort: сортировка — updated | created | alpha | size | id. fields: список полей через запятую, которые нужно оставить. Пустая строка — вернуть весь JSON как есть. """ safe_limit = max(1, min(int(limit or 50), 100)) safe_page = max(1, int(page or 1)) allowed_sort = {"updated", "created", "alpha", "size", "id"} safe_sort = sort if sort in allowed_sort else "updated" raw = self._request( "GET", "/user/repos", params={"limit": safe_limit, "page": safe_page, "sort": safe_sort}, ) # Если Gitea вернула ошибку — _request отдаёт строку "Ошибка ...". # Парсить как JSON нечего — возвращаем как есть. if not raw or not raw.lstrip().startswith("["): return raw try: repos = json.loads(raw) except (ValueError, TypeError): return raw if not isinstance(repos, list): return raw # Пустой fields — режим «как раньше», отдаём весь JSON. if not fields: return raw keep = {f.strip() for f in fields.split(",") if f.strip()} compact = [ {k: r.get(k) for k in keep if isinstance(r, dict) and k in r} for r in repos if isinstance(r, dict) ] hint = "" if len(repos) >= safe_limit: hint = ( f"\n\n[Есть ещё репозитории — вызови " f"list_repositories(limit={safe_limit}, page={safe_page + 1})]" ) return json.dumps(compact, ensure_ascii=False, indent=2) + hint def get_repository_info(self, owner: str, repo: str) -> str: """Получить информацию о репозитории.""" return self._request("GET", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}") def delete_repository(self, owner: str, repo: str) -> str: """Удалить репозиторий. Операция необратима.""" return self._request("DELETE", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}") def create_organization(self, org_name: str, description: str = "", visibility: str = "public") -> str: """Создать организацию.""" return self._request("POST", "/orgs", json={"username": org_name, "description": description, "visibility": visibility}) def list_organizations(self) -> str: """Получить организации текущего пользователя.""" return self._request("GET", "/user/orgs") def list_files(self, owner: str, repo: str, path: str = "", branch: str = "") -> str: """Получить список файлов и папок.""" params = {"ref": branch} if branch else {} return self._request("GET", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(path, safe='/')}", params=params) def get_file_content(self, owner: str, repo: str, file_path: str, branch: str = "") -> str: """Получить содержимое файла и его SHA.""" params = {"ref": branch} if branch else {} result = self._request("GET", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(file_path, safe='/')}", params=params) try: data = json.loads(result) content = base64.b64decode(data["content"]).decode("utf-8", errors="replace") return f"SHA: {data['sha']}\n\n{content}" except (ValueError, KeyError, TypeError): return result def create_file(self, owner: str, repo: str, file_path: str, content: str, commit_message: str = "Добавлен файл", branch: str = "") -> str: """Создать файл в репозитории.""" payload = {"content": base64.b64encode(content.encode()).decode(), "message": commit_message} if branch: payload["branch"] = branch return self._request("POST", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(file_path, safe='/')}", json=payload) def update_file(self, owner: str, repo: str, file_path: str, content: str, sha: str, commit_message: str = "Обновление файла", branch: str = "") -> str: """Обновить существующий файл по SHA.""" payload = {"content": base64.b64encode(content.encode()).decode(), "message": commit_message, "sha": sha} if branch: payload["branch"] = branch return self._request("PUT", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(file_path, safe='/')}", json=payload) def delete_file(self, owner: str, repo: str, file_path: str, sha: str, commit_message: str = "Удаление файла", branch: str = "") -> str: """Удалить файл по SHA.""" payload = {"message": commit_message, "sha": sha} if branch: payload["branch"] = branch return self._request("DELETE", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(file_path, safe='/')}", json=payload) def list_branches(self, owner: str, repo: str) -> str: """Получить список веток репозитория.""" return self._request("GET", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/branches") def download_repository_file_to_workspace(self, owner: str, repo: str, file_path: str, destination_name: str = "", branch: str = "", overwrite: bool = False) -> str: """Скачать файл из Gitea в локальную рабочую директорию Open WebUI.""" error = self._check() if error: return error try: root = Path(self.valves.WORKING_DIRECTORY).expanduser().resolve() root.mkdir(parents=True, exist_ok=True) name = destination_name or Path(file_path).name destination = (root / name).resolve() if not self._is_inside(destination, root) or destination == root: return "Ошибка безопасности: destination_name должен быть обычным именем внутри WORKING_DIRECTORY." if destination.exists() and not overwrite: return f"Ошибка: файл уже существует: {destination}. Укажи overwrite=true." params = {"ref": branch} if branch else {} path = f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(file_path, safe='/')}" headers = {"Authorization": f"token {self.valves.GITEA_TOKEN}"} response = requests.get( f"{self.valves.GITEA_URL.rstrip('/')}/api/v1{path}", headers=headers, params=params, timeout=self.valves.TIMEOUT, ) if not response.ok: return f"Ошибка {response.status_code}: {response.text}" data = response.json() encoded = data.get("content", "").replace("\n", "") raw = base64.b64decode(encoded, validate=True) max_size = max(1, self.valves.MAX_ARCHIVE_SIZE_MB) * 1024 * 1024 if len(raw) > max_size: return f"Ошибка безопасности: размер файла превышает лимит {self.valves.MAX_ARCHIVE_SIZE_MB} МБ." temporary = destination.with_name(f".{destination.name}.tmp") temporary.write_bytes(raw) temporary.replace(destination) return json.dumps({"status": "ok", "path": str(destination), "size": len(raw)}, ensure_ascii=False, indent=2) except (OSError, ValueError, KeyError, requests.RequestException) as exc: return f"Ошибка загрузки файла из Gitea: {exc}" def extract_zip(self, zip_path: str, destination_dir: str = "", overwrite: bool = False) -> str: """Безопасно распаковать ZIP в рабочую директорию.""" try: archive = Path(zip_path).expanduser().resolve(strict=True) if not archive.is_file(): return "Ошибка: zip_path должен указывать на обычный файл." working_root = Path(self.valves.WORKING_DIRECTORY).expanduser().resolve() working_root.mkdir(parents=True, exist_ok=True) if destination_dir: destination = Path(destination_dir).expanduser() if not destination.is_absolute(): destination = working_root / destination destination = destination.resolve() else: destination = working_root / archive.stem if not self._is_inside(destination, working_root): return "Ошибка безопасности: destination_dir находится за пределами WORKING_DIRECTORY." if destination.exists() and not overwrite: return f"Ошибка: каталог назначения уже существует: {destination}. Укажи overwrite=true для слияния." max_files = max(1, self.valves.MAX_ARCHIVE_FILES) max_size = max(1, self.valves.MAX_ARCHIVE_SIZE_MB) * 1024 * 1024 total_size = 0 with zipfile.ZipFile(archive, "r") as zip_file: members = zip_file.infolist() if len(members) > max_files: return f"Ошибка безопасности: архив содержит {len(members)} элементов, лимит — {max_files}." for info in members: relative = PurePosixPath(info.filename) if relative.is_absolute() or ".." in relative.parts or (relative.parts and ":" in relative.parts[0]): return f"Ошибка безопасности: запрещённый путь в архиве: {info.filename}" mode = (info.external_attr >> 16) & 0xFFFF if stat.S_ISLNK(mode): return f"Ошибка безопасности: символическая ссылка в архиве запрещена: {info.filename}" total_size += info.file_size if total_size > max_size: return f"Ошибка безопасности: распакованный размер превышает лимит {self.valves.MAX_ARCHIVE_SIZE_MB} МБ." temporary_dir = Path(tempfile.mkdtemp(prefix=".zip-extract-", dir=working_root)) try: for info in members: relative = PurePosixPath(info.filename) target = (temporary_dir / Path(*relative.parts)).resolve() if not self._is_inside(target, temporary_dir): raise ValueError(f"Запрещённый путь в архиве: {info.filename}") if info.is_dir() or info.filename.endswith("/"): target.mkdir(parents=True, exist_ok=True) continue target.parent.mkdir(parents=True, exist_ok=True) with zip_file.open(info, "r") as source, target.open("wb") as output: shutil.copyfileobj(source, output, length=1024 * 1024) if destination.exists(): if not destination.is_dir() or destination.is_symlink(): return f"Ошибка: путь назначения не является безопасным каталогом: {destination}" for item in temporary_dir.iterdir(): target = destination / item.name if target.is_symlink(): target.unlink() if target.exists() and target.is_dir() and item.is_dir(): self._merge_directory(item, target) else: if target.exists() and target.is_dir(): shutil.rmtree(target) elif target.exists(): target.unlink() shutil.move(str(item), str(target)) else: temporary_dir.rename(destination) temporary_dir = None return json.dumps({"status": "ok", "message": "ZIP-архив успешно распакован.", "archive": str(archive), "destination": str(destination), "items": len(members), "uncompressed_size": total_size}, ensure_ascii=False, indent=2) finally: if temporary_dir is not None and temporary_dir.exists(): shutil.rmtree(temporary_dir, ignore_errors=True) except (OSError, ValueError, zipfile.BadZipFile) as exc: return f"Ошибка распаковки ZIP: {exc}" @staticmethod def _is_inside(path: Path, root: Path) -> bool: """Проверить, находится ли путь внутри корневого каталога.""" try: path.relative_to(root) return True except ValueError: return False def _merge_directory(self, source: Path, destination: Path) -> None: """Рекурсивно объединить временный каталог с существующим.""" destination.mkdir(parents=True, exist_ok=True) for item in source.iterdir(): target = destination / item.name if target.is_symlink(): target.unlink() if item.is_dir(): if target.exists() and not target.is_dir(): target.unlink() self._merge_directory(item, target) else: if target.exists(): target.unlink() shutil.move(str(item), str(target))