Add Gitea integration for Telegram work mode

app/gitea_client.py

@@ -0,0 +1,198 @@
+from __future__ import annotations
+
+import json
+import os
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from typing import Any
+
+
+class GiteaError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True)
+class GiteaConfig:
+    base_url: str
+    token: str
+    username: str | None = None
+    email: str | None = None
+    default_owner: str | None = None
+    default_repo: str | None = None
+
+    @classmethod
+    def from_env(cls) -> "GiteaConfig":
+        base_url = os.getenv("GITEA_BASE_URL", "").strip().rstrip("/")
+        token = os.getenv("GITEA_TOKEN", "").strip()
+        if not base_url:
+            raise GiteaError("GITEA_BASE_URL is required")
+        if not token:
+            raise GiteaError("GITEA_TOKEN is required")
+        return cls(
+            base_url=base_url,
+            token=token,
+            username=os.getenv("GITEA_USERNAME", "").strip() or None,
+            email=os.getenv("GITEA_EMAIL", "").strip() or None,
+            default_owner=os.getenv("GITEA_DEFAULT_OWNER", "").strip() or None,
+            default_repo=os.getenv("GITEA_DEFAULT_REPO", "").strip() or None,
+        )
+
+
+class GiteaClient:
+    def __init__(self, config: GiteaConfig) -> None:
+        self.config = config
+
+    @classmethod
+    def from_env(cls) -> "GiteaClient":
+        return cls(GiteaConfig.from_env())
+
+    def list_repositories(self, limit: int = 30, page: int = 1) -> list[dict[str, Any]]:
+        result = self._request(
+            "GET",
+            "/user/repos",
+            query={"limit": str(limit), "page": str(page)},
+        )
+        return result if isinstance(result, list) else []
+
+    def get_repository(self, owner: str, repo: str) -> dict[str, Any]:
+        result = self._request("GET", f"/repos/{_quote(owner)}/{_quote(repo)}")
+        return result if isinstance(result, dict) else {}
+
+    def list_issues(
+        self,
+        owner: str,
+        repo: str,
+        state: str = "open",
+        limit: int = 20,
+        page: int = 1,
+    ) -> list[dict[str, Any]]:
+        result = self._request(
+            "GET",
+            f"/repos/{_quote(owner)}/{_quote(repo)}/issues",
+            query={"state": state, "limit": str(limit), "page": str(page)},
+        )
+        return result if isinstance(result, list) else []
+
+    def create_issue(
+        self,
+        owner: str,
+        repo: str,
+        title: str,
+        body: str = "",
+    ) -> dict[str, Any]:
+        result = self._request(
+            "POST",
+            f"/repos/{_quote(owner)}/{_quote(repo)}/issues",
+            body={"title": title, "body": body},
+        )
+        return result if isinstance(result, dict) else {}
+
+    def comment_issue(
+        self,
+        owner: str,
+        repo: str,
+        issue_index: int,
+        body: str,
+    ) -> dict[str, Any]:
+        result = self._request(
+            "POST",
+            f"/repos/{_quote(owner)}/{_quote(repo)}/issues/{issue_index}/comments",
+            body={"body": body},
+        )
+        return result if isinstance(result, dict) else {}
+
+    def list_pull_requests(
+        self,
+        owner: str,
+        repo: str,
+        state: str = "open",
+        limit: int = 20,
+        page: int = 1,
+    ) -> list[dict[str, Any]]:
+        result = self._request(
+            "GET",
+            f"/repos/{_quote(owner)}/{_quote(repo)}/pulls",
+            query={"state": state, "limit": str(limit), "page": str(page)},
+        )
+        return result if isinstance(result, list) else []
+
+    def create_pull_request(
+        self,
+        owner: str,
+        repo: str,
+        head: str,
+        base: str,
+        title: str,
+        body: str = "",
+    ) -> dict[str, Any]:
+        result = self._request(
+            "POST",
+            f"/repos/{_quote(owner)}/{_quote(repo)}/pulls",
+            body={"head": head, "base": base, "title": title, "body": body},
+        )
+        return result if isinstance(result, dict) else {}
+
+    def list_branches(self, owner: str, repo: str, limit: int = 30, page: int = 1) -> list[dict[str, Any]]:
+        result = self._request(
+            "GET",
+            f"/repos/{_quote(owner)}/{_quote(repo)}/branches",
+            query={"limit": str(limit), "page": str(page)},
+        )
+        return result if isinstance(result, list) else []
+
+    def _request(
+        self,
+        method: str,
+        path: str,
+        body: dict[str, Any] | None = None,
+        query: dict[str, str] | None = None,
+    ) -> Any:
+        url = self._url(path, query)
+        data = None
+        headers = {
+            "Accept": "application/json",
+            "Authorization": f"token {self.config.token}",
+            "User-Agent": "telegram-codex-bridge",
+        }
+        if body is not None:
+            data = json.dumps(body).encode("utf-8")
+            headers["Content-Type"] = "application/json"
+
+        request = urllib.request.Request(url, data=data, headers=headers, method=method)
+        try:
+            with urllib.request.urlopen(request, timeout=30) as response:
+                raw = response.read().decode("utf-8", errors="replace")
+        except urllib.error.HTTPError as exc:
+            detail = exc.read().decode("utf-8", errors="replace")
+            raise GiteaError(f"Gitea API failed ({exc.code}): {_short_error(detail)}") from exc
+        except urllib.error.URLError as exc:
+            raise GiteaError(f"Gitea API connection failed: {exc.reason}") from exc
+
+        if not raw:
+            return None
+        try:
+            return json.loads(raw)
+        except json.JSONDecodeError as exc:
+            raise GiteaError(f"Gitea API returned invalid JSON: {exc}") from exc
+
+    def _url(self, path: str, query: dict[str, str] | None = None) -> str:
+        base = self.config.base_url
+        if not base.endswith("/api/v1"):
+            base = base + "/api/v1"
+        url = base + path
+        if query:
+            url += "?" + urllib.parse.urlencode(query)
+        return url
+
+
+def _quote(value: str) -> str:
+    return urllib.parse.quote(value, safe="")
+
+
+def _short_error(text: str, limit: int = 600) -> str:
+    text = text.strip()
+    if len(text) <= limit:
+        return text
+    return text[:limit] + "..."