From 33b8105513aa4d4ce46111159d27cf7e9505d732 Mon Sep 17 00:00:00 2001 From: host Date: Mon, 3 Aug 2026 13:55:20 +0300 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=20Gitea=20Manager=20Tool=20=D0=B4=D0=BB=D1=8F=20Open=20W?= =?UTF-8?q?ebUI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- gitea_manager.py | 122 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 122 insertions(+) create mode 100644 gitea_manager.py diff --git a/gitea_manager.py b/gitea_manager.py new file mode 100644 index 0000000..9c0fced --- /dev/null +++ b/gitea_manager.py @@ -0,0 +1,122 @@ +""" +title: Gitea Manager +author: Claude +version: 1.0.0 +description: Инструмент для управления Gitea из Open WebUI — создание и удаление репозиториев, организаций, а также создание, чтение, обновление, удаление файлов и просмотр веток через Gitea API. +requirements: requests +""" +import base64 +from typing import Optional + +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.") + + 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/{org}/repos", json={"name": name, "description": description, "private": private, "auto_init": auto_init}) + + def list_repositories(self) -> str: + """Получить список репозиториев текущего пользователя.""" + return self._request("GET", "/user/repos") + + def get_repository_info(self, owner: str, repo: str) -> str: + """Получить информацию о репозитории.""" + return self._request("GET", f"/repos/{owner}/{repo}") + + def delete_repository(self, owner: str, repo: str) -> str: + """Удалить репозиторий. Операция необратима.""" + return self._request("DELETE", f"/repos/{owner}/{repo}") + + 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/{owner}/{repo}/contents/{path}", 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/{owner}/{repo}/contents/{file_path}", params=params) + try: + data = __import__("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/{owner}/{repo}/contents/{file_path}", 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/{owner}/{repo}/contents/{file_path}", 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/{owner}/{repo}/contents/{file_path}", json=payload) + + def list_branches(self, owner: str, repo: str) -> str: + """Получить список веток репозитория.""" + return self._request("GET", f"/repos/{owner}/{repo}/branches")