commit 877cbd5e931bd114249945d9537592554ae83c83 Author: H5N3RG Date: Fri Mar 6 13:57:31 2026 +0000 gitea_connector.py hinzugefügt diff --git a/gitea_connector.py b/gitea_connector.py new file mode 100644 index 0000000..e68d03d --- /dev/null +++ b/gitea_connector.py @@ -0,0 +1,195 @@ +""" +title: Gitea Connector +author: H5N3RG +version: 0.1 +license: MIT +description: Allows Open WebUI LLMs to read and edit Gitea repositories. +""" + +import requests +import base64 +from typing import Optional +from pydantic import BaseModel + + +class GiteaAPI: + def __init__(self, base_url: str, token: str): + self.base_url = base_url.rstrip("/") + self.headers = { + "Authorization": f"token {token}", + "Content-Type": "application/json", + } + + def get(self, endpoint): + r = requests.get(f"{self.base_url}{endpoint}", headers=self.headers) + r.raise_for_status() + return r.json() + + def put(self, endpoint, data): + r = requests.put(f"{self.base_url}{endpoint}", headers=self.headers, json=data) + r.raise_for_status() + return r.json() + + def post(self, endpoint, data): + r = requests.post(f"{self.base_url}{endpoint}", headers=self.headers, json=data) + r.raise_for_status() + return r.json() + + +class Tools: + + class Valves(BaseModel): + + GITEA_URL: str = "https://git.example.com/api/v1" + API_TOKEN: str = "" + + DEFAULT_OWNER: str = "user" + DEFAULT_REPO: str = "repo" + DEFAULT_BRANCH: str = "main" + + 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 + + def __init__(self): + self.valves = self.Valves() + + self.api = GiteaAPI(self.valves.GITEA_URL, self.valves.API_TOKEN) + + # -------------------------------------------------- + # LIST FILES + # -------------------------------------------------- + + def list_repository_tree(self, path: str = ""): + """ + List files in the repository. + """ + + if not self.valves.ENABLE_READ: + return "Reading repository disabled." + + owner = self.valves.DEFAULT_OWNER + repo = self.valves.DEFAULT_REPO + branch = self.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] + + # -------------------------------------------------- + # READ FILE + # -------------------------------------------------- + + def read_file(self, path: str): + """ + Read a file from the repository. + """ + + if not self.valves.ENABLE_READ: + return "Reading disabled." + + owner = self.valves.DEFAULT_OWNER + repo = self.valves.DEFAULT_REPO + branch = self.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 + + # -------------------------------------------------- + # CREATE OR UPDATE FILE + # -------------------------------------------------- + + def create_or_update_file( + self, path: str, content: str, commit_message: str, branch: Optional[str] = None + ): + """ + Create or update a file in the repository. + """ + + if not self.valves.ENABLE_WRITE: + return "Write access disabled." + + if not path.startswith(self.valves.ALLOWED_PATH_PREFIX): + return "Path not allowed." + + if len(content) > self.valves.MAX_FILE_SIZE_KB * 1024: + return "File too large." + + owner = self.valves.DEFAULT_OWNER + repo = self.valves.DEFAULT_REPO + branch = branch or self.valves.DEFAULT_BRANCH + + endpoint = f"/repos/{owner}/{repo}/contents/{path}" + + # check if file exists + sha = None + try: + existing = self.api.get(f"{endpoint}?ref={branch}") + sha = existing["sha"] + except: + pass + + encoded = base64.b64encode(content.encode()).decode() + + payload = {"message": commit_message, "content": encoded, "branch": branch} + + if sha: + payload["sha"] = sha + + return self.api.put(endpoint, payload) + + # -------------------------------------------------- + # CREATE BRANCH + # -------------------------------------------------- + + def create_branch(self, new_branch: str, from_branch: Optional[str] = None): + """ + Create a new branch. + """ + + if not self.valves.ENABLE_BRANCH: + return "Branch creation disabled." + + owner = self.valves.DEFAULT_OWNER + repo = self.valves.DEFAULT_REPO + + from_branch = from_branch or self.valves.DEFAULT_BRANCH + + endpoint = f"/repos/{owner}/{repo}/branches" + + payload = {"new_branch_name": new_branch, "old_branch_name": from_branch} + + return self.api.post(endpoint, payload) + + # -------------------------------------------------- + # CREATE PR + # -------------------------------------------------- + + def create_pull_request(self, head: str, base: str, title: str, body: str = ""): + """ + Create a pull request. + """ + + if not self.valves.ENABLE_PR: + return "PR creation disabled." + + owner = self.valves.DEFAULT_OWNER + repo = self.valves.DEFAULT_REPO + + endpoint = f"/repos/{owner}/{repo}/pulls" + + payload = {"title": title, "head": head, "base": base, "body": body} + + return self.api.post(endpoint, payload)