""" title: Gitea Manager (Proxmox) author: Claude version: 1.4.0 description: Инструмент для управления Gitea на Proxmox (192.168.31.57:3000) из Open WebUI — создание и удаление репозиториев, организаций, файлов, веток, загрузка и безопасная распаковка ZIP-архивов. Все методы имеют суффикс _proxmox, чтобы не путаться с инструментом для основной Gitea. 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 (192.168.31.57:3000) и локальными файлами рабочего пространства. ВАЖНО: это инструмент для Gitea на Proxmox. Для основного сервера Gitea (192.168.31.100) используется отдельный инструмент "Gitea Manager (Main)" с суффиксом _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 на Proxmox. Заполните в настройках инструмента.") 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]: """Проверить доступность инструмента и наличие токена (Gitea на Proxmox).""" if not self.valves.ENABLED: return "Инструмент Gitea Manager (Proxmox) выключен." if not self.valves.GITEA_TOKEN: return "API-токен Gitea на Proxmox не задан в настройках инструмента." return None def _request(self, method: str, path: str, **kwargs) -> str: """Выполнить запрос к Gitea на Proxmox (192.168.31.57) и вернуть 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 (192.168.31.57): {exc}" def create_repository_proxmox(self, name: str, description: str = "", private: bool = False, auto_init: bool = True) -> str: """Создать репозиторий текущего пользователя на Gitea на PROXMOX (192.168.31.57).""" return self._request("POST", "/user/repos", json={"name": name, "description": description, "private": private, "auto_init": auto_init}) def create_repository_in_org_proxmox(self, org: str, name: str, description: str = "", private: bool = False, auto_init: bool = True) -> str: """Создать репозиторий внутри организации на Gitea на PROXMOX (192.168.31.57).""" return self._request("POST", f"/orgs/{quote(org, safe='')}/repos", json={"name": name, "description": description, "private": private, "auto_init": auto_init}) def list_repositories_proxmox( self, limit: int = 50, page: int = 1, sort: str = "updated", fields: Optional[str] = "name,private,language,html_url,description,updated_at,default_branch", ) -> str: """ Получить список репозиториев текущего пользователя на Gitea на PROXMOX (192.168.31.57) в компактном виде. По умолчанию возвращает не весь 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}, ) 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 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_proxmox(limit={safe_limit}, page={safe_page + 1})]" ) return json.dumps(compact, ensure_ascii=False, indent=2) + hint def get_repository_info_proxmox(self, owner: str, repo: str) -> str: """Получить информацию о репозитории на Gitea на PROXMOX (192.168.31.57).""" return self._request("GET", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}") def delete_repository_proxmox(self, owner: str, repo: str) -> str: """Удалить репозиторий на Gitea на PROXMOX (192.168.31.57). Операция необратима.""" return self._request("DELETE", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}") def create_organization_proxmox(self, org_name: str, description: str = "", visibility: str = "public") -> str: """Создать организацию на Gitea на PROXMOX (192.168.31.57).""" return self._request("POST", "/orgs", json={"username": org_name, "description": description, "visibility": visibility}) def list_organizations_proxmox(self) -> str: """Получить организации текущего пользователя на Gitea на PROXMOX (192.168.31.57).""" return self._request("GET", "/user/orgs") def list_files_proxmox(self, owner: str, repo: str, path: str = "", branch: str = "") -> str: """Получить список файлов и папок на Gitea на PROXMOX (192.168.31.57).""" 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_proxmox(self, owner: str, repo: str, file_path: str, branch: str = "") -> str: """Получить содержимое файла и его SHA на Gitea на PROXMOX (192.168.31.57).""" 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_proxmox(self, owner: str, repo: str, file_path: str, content: str, commit_message: str = "Добавлен файл", branch: str = "") -> str: """Создать файл в репозитории на Gitea на PROXMOX (192.168.31.57).""" 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_proxmox(self, owner: str, repo: str, file_path: str, content: str, sha: str, commit_message: str = "Обновление файла", branch: str = "") -> str: """Обновить существующий файл по SHA на Gitea на PROXMOX (192.168.31.57).""" 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_proxmox(self, owner: str, repo: str, file_path: str, sha: str, commit_message: str = "Удаление файла", branch: str = "") -> str: """Удалить файл по SHA на Gitea на PROXMOX (192.168.31.57).""" 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_proxmox(self, owner: str, repo: str) -> str: """Получить список веток репозитория на Gitea на PROXMOX (192.168.31.57).""" return self._request("GET", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/branches") def download_repository_file_to_workspace_proxmox(self, owner: str, repo: str, file_path: str, destination_name: str = "", branch: str = "", overwrite: bool = False) -> str: """Скачать файл из Gitea на PROXMOX (192.168.31.57) в локальную рабочую директорию 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_proxmox(self, zip_path: str, destination_dir: str = "", overwrite: bool = False) -> str: """Безопасно распаковать ZIP в рабочую директорию инструмента Gitea на PROXMOX (192.168.31.57).""" 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))