From 494fd21a996d08a8dba00749bc314074634def30 Mon Sep 17 00:00:00 2001 From: H5N3RG Date: Fri, 6 Mar 2026 14:22:23 +0000 Subject: [PATCH] gitea_connector.py aktualisiert --- gitea_connector.py | 289 +++++++++++++++++++++++++++++++-------------- 1 file changed, 203 insertions(+), 86 deletions(-) diff --git a/gitea_connector.py b/gitea_connector.py index 4e9251f..5818729 100644 --- a/gitea_connector.py +++ b/gitea_connector.py @@ -1,7 +1,7 @@ """ title: Gitea Connector author: H5N3RG -version: 0.2 +version: 0.3 license: MIT description: Allows Open WebUI LLMs to read and edit Gitea repositories. """ @@ -9,7 +9,7 @@ description: Allows Open WebUI LLMs to read and edit Gitea repositories. import requests import base64 from typing import Optional -from pydantic import BaseModel +from pydantic import BaseModel, Field class GiteaAPI: @@ -39,163 +39,280 @@ class GiteaAPI: class Tools: class Valves(BaseModel): + GITEA_URL: str = Field( + default="", description="Gitea API URL, z.B. https://git.example.com/api/v1" + ) + API_TOKEN: str = Field(default="", description="Gitea API Token") - GITEA_URL: str = "https://git.example.com/api/v1" - API_TOKEN: str = "" + DEFAULT_OWNER: str = Field( + default="", description="Standard-Repository-Owner/Organisation" + ) + DEFAULT_REPO: str = Field(default="", description="Standard-Repository-Name") + DEFAULT_BRANCH: str = Field(default="main", description="Standard-Branch") - DEFAULT_OWNER: str = "user" - DEFAULT_REPO: str = "repo" - DEFAULT_BRANCH: str = "main" + ENABLE_READ: bool = Field(default=True, description="Lesen erlauben") + ENABLE_WRITE: bool = Field(default=False, description="Schreiben erlauben") + ENABLE_BRANCH: bool = Field( + default=True, description="Branch-Erstellung erlauben" + ) + ENABLE_PR: bool = Field( + default=True, description="Pull-Request-Erstellung erlauben" + ) - ENABLE_READ: bool = True - ENABLE_WRITE: bool = False - ENABLE_BRANCH: bool = True - ENABLE_PR: bool = True - - ALLOWED_PATH_PREFIX: str = "docs/" - - MAX_FILE_SIZE_KB: int = 200 + ALLOWED_PATH_PREFIX: str = Field( + default="", + description="Erlaubtes Pfad-Präfix für Schreibzugriff (leer = alle Pfade erlaubt)", + ) + MAX_FILE_SIZE_KB: int = Field( + default=200, description="Maximale Dateigröße in KB" + ) def __init__(self): self.valves = self.Valves() - # FIX: GiteaAPI wird NICHT mehr hier instanziiert. - # Stattdessen wird self.api als Property definiert, - # damit die Valve-Werte immer aktuell sind wenn OWUI sie injiziert. - @property - def api(self) -> GiteaAPI: - """Lazy property: baut GiteaAPI immer frisch mit den aktuellen Valve-Werten.""" - return GiteaAPI(self.valves.GITEA_URL, self.valves.API_TOKEN) + def _get_valves(self, __user__: dict) -> Valves: + """ + Holt die Valves aus __user__ falls von OWUI injiziert, + sonst Fallback auf self.valves. + Dies ist der Workaround für den OWUI-Bug bei dem Valves + nicht korrekt in self.valves landen. + """ + user_valves = __user__.get("valves") if __user__ else None + if user_valves: + try: + return ( + self.Valves(**user_valves) + if isinstance(user_valves, dict) + else user_valves + ) + except Exception: + pass + return self.valves + + def _get_api(self, valves: Valves) -> GiteaAPI: + return GiteaAPI(valves.GITEA_URL, valves.API_TOKEN) + + def _check_config(self, valves: Valves) -> Optional[str]: + if not valves.GITEA_URL: + return "❌ Bitte GITEA_URL in den Valve-Einstellungen eintragen." + if not valves.API_TOKEN: + return "❌ Bitte API_TOKEN in den Valve-Einstellungen eintragen." + return None # -------------------------------------------------- # LIST FILES # -------------------------------------------------- - def list_repository_tree(self, path: str = ""): + def list_repository_tree(self, path: str = "", __user__: dict = {}): """ - List files in the repository. + List files and directories in the Gitea repository. + + :param path: Optional subdirectory path to list. Leave empty for root. + :return: List of files and directories with name, path and type. """ + valves = self._get_valves(__user__) - if not self.valves.ENABLE_READ: - return "Reading repository disabled." + if not valves.ENABLE_READ: + return "❌ Lesen ist deaktiviert." - owner = self.valves.DEFAULT_OWNER - repo = self.valves.DEFAULT_REPO - branch = self.valves.DEFAULT_BRANCH + err = self._check_config(valves) + if err: + return err + + owner = valves.DEFAULT_OWNER + repo = valves.DEFAULT_REPO + branch = valves.DEFAULT_BRANCH endpoint = f"/repos/{owner}/{repo}/contents/{path}?ref={branch}" - data = self.api.get(endpoint) - - return [{"name": f["name"], "path": f["path"], "type": f["type"]} for f in data] + try: + data = self._get_api(valves).get(endpoint) + return [ + {"name": f["name"], "path": f["path"], "type": f["type"]} for f in data + ] + except requests.HTTPError as e: + return f"❌ API-Fehler: HTTP {e.response.status_code} – {e.response.text[:300]}" + except Exception as e: + return f"❌ Fehler: {type(e).__name__}: {e}" # -------------------------------------------------- # READ FILE # -------------------------------------------------- - def read_file(self, path: str): + def read_file(self, path: str, __user__: dict = {}): """ - Read a file from the repository. + Read the content of a file from the Gitea repository. + + :param path: Path to the file within the repository. + :return: The decoded file content as a string. """ + valves = self._get_valves(__user__) - if not self.valves.ENABLE_READ: - return "Reading disabled." + if not valves.ENABLE_READ: + return "❌ Lesen ist deaktiviert." - owner = self.valves.DEFAULT_OWNER - repo = self.valves.DEFAULT_REPO - branch = self.valves.DEFAULT_BRANCH + err = self._check_config(valves) + if err: + return err + + owner = valves.DEFAULT_OWNER + repo = valves.DEFAULT_REPO + branch = valves.DEFAULT_BRANCH endpoint = f"/repos/{owner}/{repo}/contents/{path}?ref={branch}" - data = self.api.get(endpoint) - - content = base64.b64decode(data["content"]).decode("utf-8") - - return content + try: + data = self._get_api(valves).get(endpoint) + return base64.b64decode(data["content"]).decode("utf-8") + except requests.HTTPError as e: + return f"❌ API-Fehler: HTTP {e.response.status_code} – {e.response.text[:300]}" + except Exception as e: + return f"❌ Fehler: {type(e).__name__}: {e}" # -------------------------------------------------- # CREATE OR UPDATE FILE # -------------------------------------------------- def create_or_update_file( - self, path: str, content: str, commit_message: str, branch: Optional[str] = None + self, + path: str, + content: str, + commit_message: str, + branch: Optional[str] = None, + __user__: dict = {}, ): """ - Create or update a file in the repository. + Create or update a file in the Gitea repository. + + :param path: Path of the file to create or update. + :param content: The new file content as a plain string. + :param commit_message: Git commit message. + :param branch: Target branch. Uses DEFAULT_BRANCH if omitted. + :return: Confirmation of the operation. """ + valves = self._get_valves(__user__) - if not self.valves.ENABLE_WRITE: - return "Write access disabled." + if not valves.ENABLE_WRITE: + return "❌ Schreiben ist deaktiviert." - if not path.startswith(self.valves.ALLOWED_PATH_PREFIX): - return "Path not allowed." + if valves.ALLOWED_PATH_PREFIX and not path.startswith( + valves.ALLOWED_PATH_PREFIX + ): + return f"❌ Pfad nicht erlaubt. Erlaubtes Präfix: '{valves.ALLOWED_PATH_PREFIX}'" - if len(content) > self.valves.MAX_FILE_SIZE_KB * 1024: - return "File too large." + if len(content) > valves.MAX_FILE_SIZE_KB * 1024: + return f"❌ Datei zu groß (max. {valves.MAX_FILE_SIZE_KB} KB)." - owner = self.valves.DEFAULT_OWNER - repo = self.valves.DEFAULT_REPO - branch = branch or self.valves.DEFAULT_BRANCH + err = self._check_config(valves) + if err: + return err + + owner = valves.DEFAULT_OWNER + repo = valves.DEFAULT_REPO + target_branch = branch or valves.DEFAULT_BRANCH endpoint = f"/repos/{owner}/{repo}/contents/{path}" + api = self._get_api(valves) - # check if file exists - sha = None try: - existing = self.api.get(f"{endpoint}?ref={branch}") - sha = existing["sha"] - except: - pass + sha = None + try: + existing = api.get(f"{endpoint}?ref={target_branch}") + sha = existing["sha"] + except Exception: + pass - encoded = base64.b64encode(content.encode()).decode() + encoded = base64.b64encode(content.encode()).decode() + payload = { + "message": commit_message, + "content": encoded, + "branch": target_branch, + } + if sha: + payload["sha"] = sha - payload = {"message": commit_message, "content": encoded, "branch": branch} - - if sha: - payload["sha"] = sha - - return self.api.put(endpoint, payload) + api.put(endpoint, payload) + action = "aktualisiert" if sha else "erstellt" + return f"✅ Datei '{path}' erfolgreich {action} (Branch: {target_branch})." + except requests.HTTPError as e: + return f"❌ API-Fehler: HTTP {e.response.status_code} – {e.response.text[:300]}" + except Exception as e: + return f"❌ Fehler: {type(e).__name__}: {e}" # -------------------------------------------------- # CREATE BRANCH # -------------------------------------------------- - def create_branch(self, new_branch: str, from_branch: Optional[str] = None): + def create_branch( + self, new_branch: str, from_branch: Optional[str] = None, __user__: dict = {} + ): """ - Create a new branch. + Create a new branch in the Gitea repository. + + :param new_branch: Name of the new branch to create. + :param from_branch: Source branch. Uses DEFAULT_BRANCH if omitted. + :return: Confirmation of the operation. """ + valves = self._get_valves(__user__) - if not self.valves.ENABLE_BRANCH: - return "Branch creation disabled." + if not valves.ENABLE_BRANCH: + return "❌ Branch-Erstellung ist deaktiviert." - owner = self.valves.DEFAULT_OWNER - repo = self.valves.DEFAULT_REPO + err = self._check_config(valves) + if err: + return err - from_branch = from_branch or self.valves.DEFAULT_BRANCH + owner = valves.DEFAULT_OWNER + repo = valves.DEFAULT_REPO + source = from_branch or valves.DEFAULT_BRANCH endpoint = f"/repos/{owner}/{repo}/branches" + payload = {"new_branch_name": new_branch, "old_branch_name": source} - payload = {"new_branch_name": new_branch, "old_branch_name": from_branch} - - return self.api.post(endpoint, payload) + try: + self._get_api(valves).post(endpoint, payload) + return f"✅ Branch '{new_branch}' erfolgreich aus '{source}' erstellt." + except requests.HTTPError as e: + return f"❌ API-Fehler: HTTP {e.response.status_code} – {e.response.text[:300]}" + except Exception as e: + return f"❌ Fehler: {type(e).__name__}: {e}" # -------------------------------------------------- # CREATE PR # -------------------------------------------------- - def create_pull_request(self, head: str, base: str, title: str, body: str = ""): + def create_pull_request( + self, head: str, base: str, title: str, body: str = "", __user__: dict = {} + ): """ - Create a pull request. + Create a pull request in the Gitea repository. + + :param head: Source branch (the branch with your changes). + :param base: Target branch (e.g. 'main'). + :param title: Title of the pull request. + :param body: Optional description for the pull request. + :return: Confirmation of the operation. """ + valves = self._get_valves(__user__) - if not self.valves.ENABLE_PR: - return "PR creation disabled." + if not valves.ENABLE_PR: + return "❌ Pull-Request-Erstellung ist deaktiviert." - owner = self.valves.DEFAULT_OWNER - repo = self.valves.DEFAULT_REPO + err = self._check_config(valves) + if err: + return err + + owner = valves.DEFAULT_OWNER + repo = valves.DEFAULT_REPO endpoint = f"/repos/{owner}/{repo}/pulls" - payload = {"title": title, "head": head, "base": base, "body": body} - return self.api.post(endpoint, payload) \ No newline at end of file + try: + result = self._get_api(valves).post(endpoint, payload) + pr_id = result.get("number", "?") + return f"✅ Pull Request #{pr_id} '{title}' erfolgreich erstellt ({head} → {base})." + except requests.HTTPError as e: + return f"❌ API-Fehler: HTTP {e.response.status_code} – {e.response.text[:300]}" + except Exception as e: + return f"❌ Fehler: {type(e).__name__}: {e}"