From 676b55edb29147730191fc7360aebd8b873e8085 Mon Sep 17 00:00:00 2001 From: host Date: Wed, 23 Sep 2026 11:00:35 +0300 Subject: [PATCH] =?UTF-8?q?=D0=A3=D0=B4=D0=B0=D0=BB=D1=91=D0=BD=20=D1=83?= =?UTF-8?q?=D1=81=D1=82=D0=B0=D1=80=D0=B5=D0=B2=D1=88=D0=B8=D0=B9=20gitea?= =?UTF-8?q?=5Fmanager.py=20(=D0=B7=D0=B0=D0=BC=D0=B5=D1=89=D1=91=D0=BD=20g?= =?UTF-8?q?itea=5Fmanager=5Fmain.py)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- gitea_manager.py | 336 ----------------------------------------------- 1 file changed, 336 deletions(-) delete mode 100644 gitea_manager.py diff --git a/gitea_manager.py b/gitea_manager.py deleted file mode 100644 index 864d562..0000000 --- a/gitea_manager.py +++ /dev/null @@ -1,336 +0,0 @@ -""" -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)) \ No newline at end of file