Добавлен Gitea Manager Tool для Open WebUI

2026-08-03 13:55:20 +03:00
parent 728a3ea57c
commit 33b8105513
1 changed files with 122 additions and 0 deletions
@@ -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")