RepoRadar/utils/git_api.py

"""
utils/git_api.py — async client for self-hosted / hosted git servers.

Supports three provider families behind one normalised interface so the cog and
embeds never touch the upstream schemas directly:

  * Gitea / Forgejo  (REST API v1)   — typical self-hosted "git server"
  * GitLab           (REST API v4)
  * GitHub           (REST API v3 / api.github.com)

Responsibilities:
  * Single shared aiohttp.ClientSession (started on bot boot, closed on shutdown).
  * Provider auto-detection (probe /api/v1/version then /api/v4/version).
  * Repository listing for a specific user/org OR server-wide, plus a single repo.
  * Normalises every payload into the simple `Repo` dataclass.

The `created_at` timestamp is what drives "newest projects only" — the monitor
baselines existing repos on first sight, then only drops repos created after
that baseline, in chronological order.
"""

from __future__ import annotations

import asyncio
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any
from urllib.parse import quote

import aiohttp

from config import settings

log = logging.getLogger("reporadar.git_api")

# Provider identifiers.
GITEA = "gitea"
GITLAB = "gitlab"
GITHUB = "github"


def _parse_dt(raw: str | None) -> datetime | None:
    """Parse an ISO8601 timestamp from any provider into an aware UTC datetime."""
    if not raw:
        return None
    cleaned = raw.replace("Z", "+00:00")
    for fmt in ("%Y-%m-%dT%H:%M:%S.%f%z", "%Y-%m-%dT%H:%M:%S%z"):
        try:
            return datetime.strptime(cleaned, fmt)
        except ValueError:
            continue
    try:
        parsed = datetime.fromisoformat(cleaned)
        return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
    except ValueError:
        log.debug("Could not parse git datetime: %s", raw)
        return None


@dataclass
class Repo:
    """A normalised view of a single repository across providers."""

    # Stable, provider-unique key used for de-duplication + vote tallies.
    key: str
    name: str
    full_name: str
    owner: str
    description: str
    url: str
    created_at: datetime | None
    updated_at: datetime | None
    stars: int = 0
    forks: int = 0
    language: str = ""
    topics: list[str] = field(default_factory=list)
    default_branch: str = ""
    private: bool = False
    archived: bool = False

    @property
    def created_sort_key(self) -> str:
        """Sortable ISO string; empty timestamps sort first (oldest)."""
        return self.created_at.isoformat() if self.created_at else ""


class GitClient:
    """Async, provider-agnostic git server client (cache-light, retry-aware)."""

    def __init__(self) -> None:
        self._session: aiohttp.ClientSession | None = None
        self.provider: str = settings.git_provider
        self.base_url: str = settings.git_base_url
        self._detected = False

    # ── lifecycle ────────────────────────────────────────────
    async def start(self) -> None:
        if self._session is None or self._session.closed:
            headers = {"User-Agent": settings.user_agent, "Accept": "application/json"}
            timeout = aiohttp.ClientTimeout(total=settings.http_timeout)
            self._session = aiohttp.ClientSession(headers=headers, timeout=timeout)
        await self._detect_provider()
        log.info(
            "Git client ready (provider=%s, base=%s, auth=%s)",
            self.provider or "unconfigured",
            self.base_url or "<none>",
            bool(settings.git_token),
        )

    async def close(self) -> None:
        if self._session and not self._session.closed:
            await self._session.close()

    def _require_session(self) -> aiohttp.ClientSession:
        if self._session is None or self._session.closed:
            raise RuntimeError("GitClient session not started — call start() first.")
        return self._session

    @property
    def configured(self) -> bool:
        """True when we have enough config to talk to a git server."""
        if self.provider == GITHUB:
            return True  # api.github.com default works without a base URL
        return bool(self.base_url)

    # ── auth headers per provider ────────────────────────────
    def _auth_headers(self) -> dict[str, str]:
        token = settings.git_token
        if not token:
            return {}
        if self.provider == GITLAB:
            return {"PRIVATE-TOKEN": token}
        if self.provider == GITHUB:
            return {"Authorization": f"Bearer {token}"}
        # Gitea / Forgejo
        return {"Authorization": f"token {token}"}

    def _api_root(self) -> str:
        if self.provider == GITHUB:
            return (self.base_url or "https://api.github.com").rstrip("/")
        if self.provider == GITLAB:
            return f"{self.base_url}/api/v4"
        return f"{self.base_url}/api/v1"  # gitea / forgejo

    # ── low-level GET with retry/backoff ─────────────────────
    async def _get(
        self, url: str, params: dict[str, Any] | None = None
    ) -> tuple[Any | None, dict[str, str]]:
        """GET returning (json, response_headers). None json on failure."""
        session = self._require_session()
        headers = self._auth_headers()
        delay = 2.0
        for attempt in range(1, settings.request_retries + 1):
            try:
                async with session.get(url, params=params, headers=headers) as resp:
                    if resp.status == 200:
                        return await resp.json(), dict(resp.headers)
                    if resp.status in (403, 429, 500, 502, 503):
                        log.warning(
                            "GET %s → %s (attempt %s/%s), backoff %.1fs",
                            url, resp.status, attempt, settings.request_retries, delay,
                        )
                        await asyncio.sleep(delay)
                        delay *= 2
                        continue
                    log.error("GET %s → unexpected status %s", url, resp.status)
                    return None, {}
            except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
                log.warning("GET %s failed (%s/%s): %s", url, attempt, settings.request_retries, exc)
                await asyncio.sleep(delay)
                delay *= 2
        log.error("GET %s exhausted retries", url)
        return None, {}

    # ── provider detection ───────────────────────────────────
    async def _detect_provider(self) -> None:
        if self._detected:
            return
        self._detected = True

        if self.provider and self.provider != "auto":
            return  # explicit provider — trust the operator
        if not self.base_url:
            self.provider = GITHUB
            return
        host = self.base_url.lower()
        if "github" in host:
            self.provider = GITHUB
            return
        if "gitlab" in host:
            self.provider = GITLAB
            return

        # Probe Gitea/Forgejo, then GitLab.
        gitea_ver, _ = await self._get(f"{self.base_url}/api/v1/version")
        if isinstance(gitea_ver, dict) and gitea_ver.get("version"):
            self.provider = GITEA
            return
        gitlab_ver, _ = await self._get(f"{self.base_url}/api/v4/version")
        if isinstance(gitlab_ver, dict) and gitlab_ver.get("version"):
            self.provider = GITLAB
            return
        log.warning("Could not auto-detect git provider for %s; defaulting to gitea", self.base_url)
        self.provider = GITEA

    # ── normalisation per provider ───────────────────────────
    def _norm_gitea(self, r: dict[str, Any]) -> Repo:
        owner = (r.get("owner") or {}).get("login") or (r.get("owner") or {}).get("username") or ""
        full = r.get("full_name") or f"{owner}/{r.get('name', '')}"
        return Repo(
            key=f"{self.provider}:{full}",
            name=r.get("name", ""),
            full_name=full,
            owner=owner,
            description=r.get("description") or "",
            url=r.get("html_url") or "",
            created_at=_parse_dt(r.get("created_at")),
            updated_at=_parse_dt(r.get("updated_at")),
            stars=int(r.get("stars_count", 0) or 0),
            forks=int(r.get("forks_count", 0) or 0),
            language=r.get("language") or "",
            topics=list(r.get("topics") or []),
            default_branch=r.get("default_branch") or "",
            private=bool(r.get("private", False)),
            archived=bool(r.get("archived", False)),
        )

    def _norm_gitlab(self, r: dict[str, Any]) -> Repo:
        ns = r.get("namespace") or {}
        owner = ns.get("path") or ns.get("name") or ""
        full = r.get("path_with_namespace") or f"{owner}/{r.get('path', '')}"
        return Repo(
            key=f"{self.provider}:{r.get('id', full)}",
            name=r.get("name") or r.get("path", ""),
            full_name=full,
            owner=owner,
            description=r.get("description") or "",
            url=r.get("web_url") or "",
            created_at=_parse_dt(r.get("created_at")),
            updated_at=_parse_dt(r.get("last_activity_at")),
            stars=int(r.get("star_count", 0) or 0),
            forks=int(r.get("forks_count", 0) or 0),
            language="",
            topics=list(r.get("topics") or r.get("tag_list") or []),
            default_branch=r.get("default_branch") or "",
            private=(r.get("visibility") == "private"),
            archived=bool(r.get("archived", False)),
        )

    def _norm_github(self, r: dict[str, Any]) -> Repo:
        owner = (r.get("owner") or {}).get("login") or ""
        full = r.get("full_name") or f"{owner}/{r.get('name', '')}"
        return Repo(
            key=f"{self.provider}:{full}",
            name=r.get("name", ""),
            full_name=full,
            owner=owner,
            description=r.get("description") or "",
            url=r.get("html_url") or "",
            created_at=_parse_dt(r.get("created_at")),
            updated_at=_parse_dt(r.get("pushed_at") or r.get("updated_at")),
            stars=int(r.get("stargazers_count", 0) or 0),
            forks=int(r.get("forks_count", 0) or 0),
            language=r.get("language") or "",
            topics=list(r.get("topics") or []),
            default_branch=r.get("default_branch") or "",
            private=bool(r.get("private", False)),
            archived=bool(r.get("archived", False)),
        )

    def _normalize(self, raw: dict[str, Any]) -> Repo:
        if self.provider == GITLAB:
            return self._norm_gitlab(raw)
        if self.provider == GITHUB:
            return self._norm_github(raw)
        return self._norm_gitea(raw)

    # ── public listing API ───────────────────────────────────
    async def list_repos_for_owner(self, owner: str, limit: int = 100) -> list[Repo]:
        """Repos belonging to a user or organisation (newest first upstream)."""
        await self._detect_provider()
        root = self._api_root()
        out: list[Repo] = []

        if self.provider == GITLAB:
            # Try group projects first, fall back to user projects.
            enc = quote(owner, safe="")
            for path in (f"/groups/{enc}/projects", f"/users/{enc}/projects"):
                data, _ = await self._get(
                    f"{root}{path}",
                    {"per_page": min(limit, 100), "order_by": "created_at", "sort": "desc",
                     "include_subgroups": "true"},
                )
                if isinstance(data, list) and data:
                    out = [self._normalize(x) for x in data]
                    break
        elif self.provider == GITHUB:
            # Works for both users and orgs via the generic /users endpoint.
            data, _ = await self._get(
                f"{root}/users/{quote(owner, safe='')}/repos",
                {"per_page": min(limit, 100), "sort": "created", "direction": "desc"},
            )
            if isinstance(data, list):
                out = [self._normalize(x) for x in data]
        else:  # gitea / forgejo — /users/{u}/repos covers orgs too on most builds
            for path in (f"/users/{quote(owner, safe='')}/repos",
                         f"/orgs/{quote(owner, safe='')}/repos"):
                data, _ = await self._get(f"{root}{path}", {"limit": min(limit, 50)})
                if isinstance(data, list) and data:
                    out = [self._normalize(x) for x in data]
                    break

        return out

    async def list_all_repos(self, limit: int = 100) -> list[Repo]:
        """Server-wide repository discovery (most recent first)."""
        await self._detect_provider()
        root = self._api_root()

        if self.provider == GITLAB:
            data, _ = await self._get(
                f"{root}/projects",
                {"per_page": min(limit, 100), "order_by": "created_at", "sort": "desc",
                 "simple": "true"},
            )
            return [self._normalize(x) for x in data] if isinstance(data, list) else []

        if self.provider == GITHUB:
            # api.github.com has no "all repos" concept; use the public timeline.
            data, _ = await self._get(f"{root}/repositories", {"per_page": min(limit, 100)})
            repos = [self._normalize(x) for x in data] if isinstance(data, list) else []
            repos.sort(key=lambda r: r.created_sort_key, reverse=True)
            return repos[:limit]

        # Gitea / Forgejo search endpoint sorted by newest.
        data, _ = await self._get(
            f"{root}/repos/search",
            {"limit": min(limit, 50), "sort": "created", "order": "desc"},
        )
        items = (data or {}).get("data") if isinstance(data, dict) else None
        return [self._normalize(x) for x in items] if isinstance(items, list) else []

    async def get_repo(self, owner: str, name: str) -> Repo | None:
        """Fetch a single repository by owner + name."""
        await self._detect_provider()
        root = self._api_root()
        if self.provider == GITLAB:
            enc = quote(f"{owner}/{name}", safe="")
            data, _ = await self._get(f"{root}/projects/{enc}")
        elif self.provider == GITHUB:
            data, _ = await self._get(f"{root}/repos/{quote(owner, safe='')}/{quote(name, safe='')}")
        else:
            data, _ = await self._get(
                f"{root}/repos/{quote(owner, safe='')}/{quote(name, safe='')}"
            )
        return self._normalize(data) if isinstance(data, dict) and data else None