Gitea-Connector/gitea_connector.py

"""
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)