diff --git a/gitea_manager_proxmox.py b/gitea_manager_proxmox.py new file mode 100644 index 0000000..8750bae --- /dev/null +++ b/gitea_manager_proxmox.py @@ -0,0 +1,336 @@ +""" +title: Gitea Manager — Proxmox +author: Claude +version: 1.3.0 +description: Инструмент для управления второй Gitea (Proxmox, 192.168.31.57:3000) из 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 (Proxmox) и локальными файлами рабочего пространства.""" + + class Valves(BaseModel): + GITEA_URL: str = Field(default="http://192.168.31.57:3000", description="Базовый URL Gitea (Proxmox) без слэша в конце.") + GITEA_TOKEN: str = Field(default="", description="API-токен Gitea. Заполните в настройках инструмента.") + TIMEOUT: int = Field(default=15, description="Таймаут запросов в секундах.") + ENABLED: bool = Field(default=True, description="Включить инструмент Gitea Manager — Proxmox.") + WORKING_DIRECTORY: str = Field(default="/tmp/gitea-manager-proxmox-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 — Proxmox выключен." + if not self.valves.GITEA_TOKEN: + return "API-токен Gitea не задан в настройках инструмента." + return None + + def _request(self, method: str, path: str, **kwargs) -> str: + """Выполнить запрос к Gitea (Proxmox) и вернуть 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 (Proxmox): {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 (Proxmox) в локальную рабочую директорию 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 (Proxmox): {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)) \ No newline at end of file