358 lines
15 KiB
Python
Executable File
358 lines
15 KiB
Python
Executable File
"""
|
|
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
|