227 lines
6.9 KiB
Python
227 lines
6.9 KiB
Python
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 list_commits(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
sha: str | None = None,
|
|
path: str | None = None,
|
|
limit: int = 20,
|
|
page: int = 1,
|
|
) -> list[dict[str, Any]]:
|
|
query = {"limit": str(limit), "page": str(page)}
|
|
if sha:
|
|
query["sha"] = sha
|
|
if path:
|
|
query["path"] = path
|
|
result = self._request(
|
|
"GET",
|
|
f"/repos/{_quote(owner)}/{_quote(repo)}/commits",
|
|
query=query,
|
|
)
|
|
return result if isinstance(result, list) else []
|
|
|
|
def get_commit(self, owner: str, repo: str, sha: str) -> dict[str, Any]:
|
|
result = self._request(
|
|
"GET",
|
|
f"/repos/{_quote(owner)}/{_quote(repo)}/git/commits/{_quote(sha)}",
|
|
)
|
|
return result if isinstance(result, dict) 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] + "..."
|