ai-resume/corpus/sources/gitea.py

"""Gitea API client — the only network dependency in the pipeline.

`docs/phase-4-corpus.md` §1: Gitea is reached over HTTPS, and the vault and the
graph arrive through the same API, so there is no cross-machine copy and no SSH
from the runtime host. This client is standard library only (`urllib`), it never
prints or stores the token, and every call names only the path it requested.

Read-only by construction: there is no method here that writes, creates, deletes
or changes anything in Gitea. The token only needs read scope.
"""

from __future__ import annotations

import http.client
import json
import time
import urllib.error
import urllib.parse
import urllib.request


class GiteaError(RuntimeError):
    """A Gitea call failed. The message never contains the token."""


class GiteaNotFound(GiteaError):
    """The requested repository, branch or file does not exist."""


class GiteaClient:
    def __init__(self, base: str, token: str, timeout: float = 30.0, attempts: int = 3) -> None:
        if not base:
            raise GiteaError("a Gitea base URL is required (GITEA_URL or the guide's base)")
        if not token:
            raise GiteaError(
                "a Gitea token is required. It is read from the environment variable "
                "named by the platform guide (GITEA_TOKEN) and never from this repository."
            )
        self.base = base.rstrip("/")
        self._token = token
        self.timeout = timeout
        self.attempts = max(1, attempts)
        self.calls = 0

    # -- transport --------------------------------------------------------

    def _request(self, path: str, expect_bytes: bool = False):
        """One GET with retry/backoff. The only place that touches the network."""
        url = f"{self.base}/api/v1{path}"
        last_error = None
        for attempt in range(self.attempts):
            request = urllib.request.Request(
                url,
                method="GET",
                headers={
                    "Authorization": f"token {self._token}",
                    "Accept": "application/vnd.gitea+json, application/json",
                    "User-Agent": "ai-resume-corpus/1",
                },
            )
            try:
                self.calls += 1
                with urllib.request.urlopen(request, timeout=self.timeout) as response:
                    raw = response.read()
                if expect_bytes:
                    return raw
                return json.loads(raw.decode("utf-8")) if raw.strip() else {}
            except urllib.error.HTTPError as exc:
                detail = exc.read().decode("utf-8", "replace")[:200]
                exc.close()
                if exc.code == 404:
                    raise GiteaNotFound(f"GET {path}: not found") from None
                if exc.code == 401:
                    raise GiteaError(
                        f"GET {path}: authentication failed. Check the token named by the "
                        "platform guide; it is not printed here."
                    ) from None
                if exc.code == 403:
                    raise GiteaError(f"GET {path}: forbidden (the token may lack read scope)") from None
                if exc.code != 429 and exc.code < 500:
                    raise GiteaError(f"GET {path}: HTTP {exc.code} {detail}") from None
                last_error = f"HTTP {exc.code}"
            except (urllib.error.URLError, TimeoutError, http.client.HTTPException, ConnectionError) as exc:
                last_error = f"cannot reach {self.base}: {getattr(exc, 'reason', exc)}"
            if attempt + 1 < self.attempts:
                time.sleep(0.5 * (2 ** attempt))
        raise GiteaError(f"GET {path} failed after {self.attempts} attempts: {last_error}")

    # -- repositories -----------------------------------------------------

    def repos(self, limit: int = 50) -> list:
        """Every repository the token can read, sorted by full name."""
        out = []
        page = 1
        while True:
            response = self._request(f"/repos/search?limit={limit}&page={page}")
            rows = response.get("data") or []
            out.extend(rows)
            if len(rows) < limit:
                break
            page += 1
            if page > 200:  # a guard, not a limit anyone should reach
                break
        return sorted(out, key=lambda repo: repo.get("full_name", ""))

    def repo(self, full_name: str) -> dict:
        return self._request(f"/repos/{_repo_path(full_name)}")

    # -- contents ---------------------------------------------------------

    def tree(self, full_name: str, ref: str) -> list:
        """Recursive blob/tree listing. Returns every entry, following pages."""
        entries = []
        page = 1
        while True:
            response = self._request(
                f"/repos/{_repo_path(full_name)}/git/trees/{urllib.parse.quote(ref, safe='')}"
                f"?recursive=true&per_page=1000&page={page}"
            )
            rows = response.get("tree") or []
            entries.extend(rows)
            if not rows or not response.get("truncated") or len(rows) < 1000:
                break
            page += 1
            if page > 50:
                break
        return entries

    def raw(self, full_name: str, ref: str, path: str) -> bytes:
        quoted = urllib.parse.quote(path, safe="/")
        return self._request(
            f"/repos/{_repo_path(full_name)}/raw/{quoted}?ref={urllib.parse.quote(ref, safe='')}",
            expect_bytes=True,
        )

    def commits(self, full_name: str, ref: str, limit: int = 20) -> list:
        return self._request(
            f"/repos/{_repo_path(full_name)}/commits?sha={urllib.parse.quote(ref, safe='')}"
            f"&limit={int(limit)}&page=1"
        )

    def languages(self, full_name: str) -> dict:
        return self._request(f"/repos/{_repo_path(full_name)}/languages")

    # -- presentation -----------------------------------------------------

    def describe(self) -> dict:
        return {"base": self.base, "token": "set", "calls": self.calls}


def file_url(base: str, full_name: str, ref: str, path: str) -> str:
    """The URL a reader can open: the file at its branch, in Gitea."""
    return (
        f"{base.rstrip('/')}/{full_name}/src/branch/{urllib.parse.quote(ref, safe='')}/"
        f"{urllib.parse.quote(path, safe='/')}"
    )


def repo_url(base: str, full_name: str) -> str:
    return f"{base.rstrip('/')}/{full_name}"


def _repo_path(full_name: str) -> str:
    owner, _, name = full_name.partition("/")
    return f"{urllib.parse.quote(owner, safe='')}/{urllib.parse.quote(name, safe='')}"