319 lines
11 KiB
Python
319 lines
11 KiB
Python
"""
|
||
title: Gitea Connector
|
||
author: H5N3RG
|
||
version: 0.3
|
||
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, Field
|
||
|
||
|
||
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 = Field(
|
||
default="", description="Gitea API URL, z.B. https://git.example.com/api/v1"
|
||
)
|
||
API_TOKEN: str = Field(default="", description="Gitea API Token")
|
||
|
||
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")
|
||
|
||
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"
|
||
)
|
||
|
||
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()
|
||
|
||
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 = "", __user__: dict = {}):
|
||
"""
|
||
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 valves.ENABLE_READ:
|
||
return "❌ Lesen ist deaktiviert."
|
||
|
||
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}"
|
||
|
||
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, __user__: dict = {}):
|
||
"""
|
||
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 valves.ENABLE_READ:
|
||
return "❌ Lesen ist deaktiviert."
|
||
|
||
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}"
|
||
|
||
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,
|
||
__user__: dict = {},
|
||
):
|
||
"""
|
||
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 valves.ENABLE_WRITE:
|
||
return "❌ Schreiben ist deaktiviert."
|
||
|
||
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) > valves.MAX_FILE_SIZE_KB * 1024:
|
||
return f"❌ Datei zu groß (max. {valves.MAX_FILE_SIZE_KB} KB)."
|
||
|
||
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)
|
||
|
||
try:
|
||
sha = None
|
||
try:
|
||
existing = api.get(f"{endpoint}?ref={target_branch}")
|
||
sha = existing["sha"]
|
||
except Exception:
|
||
pass
|
||
|
||
encoded = base64.b64encode(content.encode()).decode()
|
||
payload = {
|
||
"message": commit_message,
|
||
"content": encoded,
|
||
"branch": target_branch,
|
||
}
|
||
if sha:
|
||
payload["sha"] = sha
|
||
|
||
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, __user__: dict = {}
|
||
):
|
||
"""
|
||
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 valves.ENABLE_BRANCH:
|
||
return "❌ Branch-Erstellung ist deaktiviert."
|
||
|
||
err = self._check_config(valves)
|
||
if err:
|
||
return err
|
||
|
||
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}
|
||
|
||
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 = "", __user__: dict = {}
|
||
):
|
||
"""
|
||
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 valves.ENABLE_PR:
|
||
return "❌ Pull-Request-Erstellung ist deaktiviert."
|
||
|
||
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}
|
||
|
||
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}"
|