import asyncio import os import shlex from typing import Any import httpx GITEA_URL = os.getenv("GITEA_URL", "http://127.0.0.1:3000").rstrip("/") GITEA_TOKEN = os.getenv("GITEA_TOKEN", "") REVIEW_DIFF_MAX_BYTES = 64 * 1024 REVIEW_DIFF_MAX_LINES = 400 def _auth() -> dict[str, str]: headers: dict[str, str] = {"Accept": "application/json"} if GITEA_TOKEN: headers["Authorization"] = f"token {GITEA_TOKEN}" return headers async def fetch(path: str) -> Any: async with httpx.AsyncClient(base_url=GITEA_URL, timeout=10) as client: r = await client.get(f"/api/v1/{path}", headers=_auth()) r.raise_for_status() return r.json() async def fetch_text(path: str, max_bytes: int) -> tuple[str, bool]: chunks: list[bytes] = [] size = 0 truncated = False async with httpx.AsyncClient(base_url=GITEA_URL, timeout=10) as client: async with client.stream( "GET", f"/api/v1/{path}", headers={**_auth(), "Accept": "text/plain"} ) as response: response.raise_for_status() async for chunk in response.aiter_bytes(): remaining = max_bytes - size if len(chunk) > remaining: chunks.append(chunk[:remaining]) truncated = True break chunks.append(chunk) size += len(chunk) return b"".join(chunks).decode("utf-8", errors="replace"), truncated def _diff_previews(diff: str, stream_truncated: bool) -> dict[str, dict]: previews: dict[str, dict] = {} current: dict | None = None in_hunk = False remaining = REVIEW_DIFF_MAX_LINES for line in diff.splitlines(): if line.startswith("diff --git "): try: target = shlex.split(line)[3] filename = target[2:] if target.startswith("b/") else target except (IndexError, ValueError): current = None continue current = { "diff_lines": [], "diff_available": False, "diff_binary": False, "diff_truncated": stream_truncated, } previews[filename] = current in_hunk = False continue if current is None: continue if line.startswith("Binary files ") or line == "GIT binary patch": current["diff_binary"] = True in_hunk = False continue if line.startswith("@@"): in_hunk = True if in_hunk and not line.startswith("\\ No newline at end of file"): if remaining: current["diff_lines"].append(line) current["diff_available"] = True remaining -= 1 else: current["diff_truncated"] = True return previews async def current_user() -> dict: return await fetch("user") async def repos() -> list[dict]: return await fetch("user/repos?limit=50") async def issues() -> list[dict]: return await fetch( "repos/issues/search?state=open&assigned=true&type=issues&limit=50" ) async def pull_requests() -> list[dict]: assigned = await fetch( "repos/issues/search?state=open&assigned=true&type=pulls&limit=50" ) review_requested = await fetch( "repos/issues/search?state=open&review_requested=true&type=pulls&limit=50" ) merged: dict[int, dict] = {} for reason, pulls in ( ("assigned_to_me", assigned or []), ("review_requested", review_requested or []), ): for pull in pulls: identity = pull.get("id") if identity not in merged: merged[identity] = {**pull, "work_reasons": []} merged[identity]["work_reasons"].append(reason) return list(merged.values()) async def is_requested_review(repository: str, number: int) -> bool: pulls = await fetch( "repos/issues/search?state=open&review_requested=true&type=pulls&limit=50" ) return any( isinstance(pull, dict) and pull.get("number") == number and isinstance(pull.get("repository"), dict) and pull["repository"].get("full_name") == repository for pull in (pulls or []) ) async def pull_review_detail(repository: str, number: int) -> dict: base = f"repos/{repository}/pulls/{number}" pull = await fetch(base) if not isinstance(pull, dict): raise ValueError("Gitea pull request response was not an object") head_value = pull.get("head") head: dict = head_value if isinstance(head_value, dict) else {} sha_value = head.get("sha") sha = sha_value if isinstance(sha_value, str) else "" files, status, reviews, diff_result = await asyncio.gather( fetch(f"{base}/files"), fetch(f"repos/{repository}/commits/{sha}/status"), fetch(f"{base}/reviews"), fetch_text( f"repos/{repository}/pulls/{number}.diff", REVIEW_DIFF_MAX_BYTES ), ) diff, diff_truncated = diff_result previews = _diff_previews(diff, diff_truncated) user_value = pull.get("user") user: dict = user_value if isinstance(user_value, dict) else {} normalized_files = [ { "filename": file["filename"], "status": file.get("status") or "changed", "additions": file.get("additions") or 0, "deletions": file.get("deletions") or 0, **previews.get( file["filename"], { "diff_lines": [], "diff_available": False, "diff_binary": False, "diff_truncated": diff_truncated, }, ), } for file in (files if isinstance(files, list) else [])[:100] if isinstance(file, dict) and isinstance(file.get("filename"), str) ] normalized_reviews = [ { "user": { "login": ( review["user"].get("login", "") if isinstance(review.get("user"), dict) else "" ) }, "state": review.get("state") or "COMMENT", "body": review.get("body") or "", } for review in (reviews if isinstance(reviews, list) else [])[:50] if isinstance(review, dict) ] return { "repository": repository, "number": number, "title": pull.get("title", ""), "body": pull.get("body") or "", "url": pull.get("html_url", ""), "author": user.get("login", ""), "head_sha": sha, "ci_state": status.get("state", "unknown") if isinstance(status, dict) else "unknown", "files": normalized_files, "reviews": normalized_reviews, } async def activity_events() -> list[dict]: user = await current_user() events = await fetch(f"users/{user['login']}/activities/feeds?limit=20") if events is None: events = [] elif not isinstance(events, list): raise ValueError("Gitea activity feed response was not a list") return [ { "type": ( event.get("op_type") if isinstance(event.get("op_type"), str) and event.get("op_type") else "activity" ), "actor": ( event.get("act_user") if isinstance(event.get("act_user"), dict) else {} ), "repo": ( event.get("repo") if isinstance(event.get("repo"), dict) else {} ), "created_at": ( event.get("created") if isinstance(event.get("created"), str) else "" ), } for event in (events or []) if isinstance(event, dict) ]