gitea_client

  1"""
  2Обёртка над Gitea API.
  3
  4Бот работает через публичный REST API: читает данные PR и диффы,
  5постит ревью. Для приватных репозиториев используется токен из env.
  6"""
  7
  8from __future__ import annotations
  9
 10from typing import Any
 11
 12import httpx
 13from loguru import logger
 14
 15from config import Config
 16
 17
 18class GiteaClient:
 19    """Тонкий клиент для нужных эндпоинтов Gitea API."""
 20
 21    def __init__(self, config: Config):
 22        self._base = config.gitea_url.rstrip("/") + "/api/v1"
 23        self._timeout = httpx.Timeout(60.0)
 24        self._headers = {"Authorization": f"token {config.gitea_token}"}
 25
 26    def get(self, path: str) -> Any:
 27        """GET-запрос к Gitea API с токеном, возвращает JSON."""
 28        with httpx.Client(timeout=self._timeout, headers=self._headers) as client:
 29            resp = client.get(f"{self._base}{path}")
 30            resp.raise_for_status()
 31            return resp.json()
 32
 33    def get_text(self, path: str) -> str:
 34        """GET-запрос к Gitea API с токеном, возвращает сырой текст."""
 35        with httpx.Client(timeout=self._timeout, headers=self._headers) as client:
 36            resp = client.get(f"{self._base}{path}")
 37            resp.raise_for_status()
 38            return resp.text
 39
 40    def get_pr(self, owner: str, repo: str, index: int) -> dict:
 41        """Возвращает данные PR (head/base sha, title, draft...)."""
 42        return self.get(f"/repos/{owner}/{repo}/pulls/{index}")
 43
 44    def get_pr_commits(self, owner: str, repo: str, index: int) -> list[str]:
 45        """Возвращает список sha коммитов PR в порядке следования."""
 46        data = self.get(f"/repos/{owner}/{repo}/pulls/{index}/commits")
 47        return [c["sha"] for c in data]
 48
 49    def get_commit_diff(self, owner: str, repo: str, sha: str) -> str:
 50        """Возвращает unified diff одного коммита."""
 51        return self.get_text(f"/repos/{owner}/{repo}/git/commits/{sha}.diff")
 52
 53    def get_pr_diff(self, owner: str, repo: str, index: int) -> str:
 54        """Возвращает совокупный unified diff PR (vs base-ветки).
 55
 56        В Gitea 1.19 compare API недоступен (404), поэтому используем
 57        pulls/{index}.diff, который отдаёт полный дифф PR за один запрос.
 58        """
 59        return self.get_text(f"/repos/{owner}/{repo}/pulls/{index}.diff")
 60
 61    def get_raw_file(self, owner: str, repo: str, sha: str, path: str) -> str:
 62        """Возвращает содержимое файла на конкретном коммите."""
 63        encoded = path.replace("/", "%2F")
 64        return self.get_text(f"/repos/{owner}/{repo}/raw/{sha}/{encoded}")
 65
 66    def list_reviews(self, owner: str, repo: str, index: int) -> list[dict]:
 67        """Возвращает список ревью PR."""
 68        return self.get(f"/repos/{owner}/{repo}/pulls/{index}/reviews")
 69
 70    def post_review(
 71        self,
 72        owner: str,
 73        repo: str,
 74        index: int,
 75        body: str,
 76        event: str,
 77        commit_id: str,
 78        comments: list[dict] | None = None,
 79    ) -> dict:
 80        """
 81        Постит ревью на PR.
 82
 83        event: 'COMMENT' | 'APPROVED' | 'REQUEST_CHANGES'.
 84        comments: список {'path': ..., 'body': ..., 'new_position': ..., 'old_position': ...}.
 85        """
 86        payload: dict[str, Any] = {
 87            "body": body,
 88            "event": event,
 89            "commit_id": commit_id,
 90        }
 91        if comments:
 92            payload["comments"] = comments
 93        with httpx.Client(timeout=self._timeout, headers=self._headers) as client:
 94            resp = client.post(
 95                f"{self._base}/repos/{owner}/{repo}/pulls/{index}/reviews",
 96                json=payload,
 97            )
 98            if resp.status_code >= 400:
 99                logger.error(
100                    f"Gitea: не удалось создать ревью: {resp.status_code} {resp.text[:300]}"
101                )
102            resp.raise_for_status()
103            return resp.json()