Gitea-Tools/webui/queue_loader.py

"""Load live Gitea PR/issue queue state for the web UI dashboard (#429)."""

from __future__ import annotations

import re
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Callable
from urllib.parse import urlparse

from gitea_auth import api_fetch_page, api_get_all, get_auth_header, repo_api_url

from webui.project_registry import ProjectRecord, load_registry

_CLOSES_RE = re.compile(r"(?:closes|fixes|resolves)\s+#(\d+)", re.I)
_ISSUE_REF_RE = re.compile(r"#(\d+)")
_STALE_DAYS = 14


@dataclass(frozen=True)
class PaginationMeta:
    page: int
    per_page: int
    returned_count: int
    has_more: bool
    is_final_page: bool
    inventory_complete: bool
    pages_fetched: int


@dataclass(frozen=True)
class QueueItem:
    number: int
    title: str
    badges: tuple[str, ...]
    extra: dict[str, str]


@dataclass(frozen=True)
class QueueSnapshot:
    project_id: str
    repo_label: str
    prs: tuple[QueueItem, ...]
    issues: tuple[QueueItem, ...]
    pr_pagination: PaginationMeta | None
    issue_pagination: PaginationMeta | None
    fetch_error: str | None = None


def _host_from_url(remote_host: str) -> str:
    parsed = urlparse(remote_host.strip())
    return parsed.netloc or remote_host.strip().rstrip("/")


def _extract_linked_issue(title: str | None, body: str | None = None) -> int | None:
    for text in (title, body):
        if not text:
            continue
        match = _CLOSES_RE.search(text)
        if match:
            return int(match.group(1))
    if body:
        refs = _ISSUE_REF_RE.findall(body)
        if refs:
            return int(refs[0])
    return None


def _is_stale(updated_at: str | None) -> bool:
    if not updated_at:
        return False
    try:
        normalized = updated_at.replace("Z", "+00:00")
        parsed = datetime.fromisoformat(normalized)
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=timezone.utc)
        age = datetime.now(timezone.utc) - parsed.astimezone(timezone.utc)
        return age.days >= _STALE_DAYS
    except ValueError:
        return False


def _classify_pr(
    pr: dict,
    *,
    issue_claimed: set[int],
    issue_to_prs: dict[int, list[int]],
) -> tuple[str, ...]:
    badges: list[str] = []
    mergeable = pr.get("mergeable")
    if mergeable is False:
        badges.append("blocked")
    linked = _extract_linked_issue(pr.get("title"), pr.get("body"))
    if linked is not None:
        if linked in issue_claimed:
            badges.append("claimed")
        if len(issue_to_prs.get(linked, [])) > 1:
            badges.append("duplicate")
    if _is_stale(pr.get("updated_at")):
        badges.append("stale")
    if mergeable is True and "blocked" not in badges:
        badges.append("in-review")
    if not badges:
        badges.append("open")
    return tuple(dict.fromkeys(badges))


def _classify_issue(
    issue: dict,
    *,
    linked_prs: list[int],
) -> tuple[str, ...]:
    badges: list[str] = []
    labels = [lb.get("name", "") for lb in issue.get("labels", [])]
    if "status:in-progress" in labels:
        badges.append("claimed")
    if len(linked_prs) > 1:
        badges.append("duplicate")
    elif linked_prs and "claimed" not in badges:
        badges.append("in-review")
    if _is_stale(issue.get("updated_at")):
        badges.append("stale")
    if not badges:
        badges.append("open")
    return tuple(dict.fromkeys(badges))


def _format_pr_item(pr: dict, badges: tuple[str, ...]) -> QueueItem:
    head = pr.get("head") or {}
    base = pr.get("base") or {}
    mergeable = pr.get("mergeable")
    merge_label = (
        "mergeable" if mergeable is True else "conflicted" if mergeable is False else "unknown"
    )
    linked = _extract_linked_issue(pr.get("title"), pr.get("body"))
    return QueueItem(
        number=int(pr["number"]),
        title=str(pr.get("title") or ""),
        badges=badges,
        extra={
            "branch": f"{head.get('ref', '?')} → {base.get('ref', '?')}",
            "head_sha": str(head.get("sha") or "")[:12],
            "mergeable": merge_label,
            "linked_issue": str(linked) if linked is not None else "",
        },
    )


def _format_issue_item(issue: dict, badges: tuple[str, ...]) -> QueueItem:
    labels = ", ".join(lb.get("name", "") for lb in issue.get("labels", []))
    assignee = (issue.get("assignee") or {}).get("login", "")
    return QueueItem(
        number=int(issue["number"]),
        title=str(issue.get("title") or ""),
        badges=badges,
        extra={
            "labels": labels or "—",
            "assignee": assignee or "unassigned",
            "state": str(issue.get("state") or ""),
        },
    )


def _pagination_from_pages(
    *,
    per_page: int,
    pages_fetched: int,
    returned_count: int,
    is_final_page: bool,
) -> PaginationMeta:
    return PaginationMeta(
        page=1,
        per_page=per_page,
        returned_count=returned_count,
        has_more=not is_final_page,
        is_final_page=is_final_page,
        inventory_complete=is_final_page,
        pages_fetched=pages_fetched,
    )


def _fetch_prs(
    host: str,
    org: str,
    repo: str,
    auth: str,
    *,
    per_page: int = 50,
) -> tuple[list[dict], PaginationMeta]:
    url = f"{repo_api_url(host, org, repo)}/pulls?state=open"
    all_raw: list[dict] = []
    pages_fetched = 0
    is_final = False
    page = 1
    while pages_fetched < 20:
        raw_page, meta = api_fetch_page(url, auth, page=page, limit=per_page)
        pages_fetched += 1
        all_raw.extend(raw_page)
        is_final = bool(meta["is_final_page"])
        if is_final:
            break
        page += 1
    pagination = _pagination_from_pages(
        per_page=per_page,
        pages_fetched=pages_fetched,
        returned_count=len(all_raw),
        is_final_page=is_final,
    )
    return all_raw, pagination


def _fetch_issues(
    host: str,
    org: str,
    repo: str,
    auth: str,
    *,
    per_page: int = 50,
) -> tuple[list[dict], PaginationMeta]:
    url = f"{repo_api_url(host, org, repo)}/issues?state=open&type=issues"
    all_raw: list[dict] = []
    page = 1
    pages_fetched = 0
    is_final = False
    while pages_fetched < 20:
        raw_page, meta = api_fetch_page(url, auth, page=page, limit=per_page)
        pages_fetched += 1
        all_raw.extend(raw_page)
        is_final = bool(meta["is_final_page"])
        if is_final:
            break
        page += 1
    pagination = _pagination_from_pages(
        per_page=per_page,
        pages_fetched=pages_fetched,
        returned_count=len(all_raw),
        is_final_page=is_final,
    )
    return all_raw, pagination


def load_queue_snapshot(
    project_id: str | None = None,
    *,
    fetch_prs: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
    fetch_issues: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
) -> QueueSnapshot:
    """Load open PR and issue queues for a registry project (default: first entry)."""
    registry = load_registry()
    project: ProjectRecord | None = None
    if project_id:
        for entry in registry.projects:
            if entry.id == project_id:
                project = entry
                break
    else:
        project = registry.projects[0] if registry.projects else None

    if project is None:
        return QueueSnapshot(
            project_id=project_id or "",
            repo_label="",
            prs=(),
            issues=(),
            pr_pagination=None,
            issue_pagination=None,
            fetch_error="project not found in registry",
        )

    host = _host_from_url(project.remote_host)
    pr_fetch = fetch_prs or _fetch_prs
    issue_fetch = fetch_issues or _fetch_issues
    using_live_fetch = fetch_prs is None or fetch_issues is None
    auth = get_auth_header(host) if using_live_fetch else "test-auth"
    if using_live_fetch and not auth:
        return QueueSnapshot(
            project_id=project.id,
            repo_label=f"{project.gitea_owner}/{project.repo_name}",
            prs=(),
            issues=(),
            pr_pagination=None,
            issue_pagination=None,
            fetch_error=(
                f"Gitea credentials unavailable for {host}; "
                "queue cannot be loaded (fail closed — not showing empty queue)"
            ),
        )

    try:
        raw_prs, pr_pagination = pr_fetch(
            host, project.gitea_owner, project.repo_name, auth
        )
        raw_issues, issue_pagination = issue_fetch(
            host, project.gitea_owner, project.repo_name, auth
        )
    except Exception as exc:  # noqa: BLE001 — surface operator-visible fetch errors
        return QueueSnapshot(
            project_id=project.id,
            repo_label=f"{project.gitea_owner}/{project.repo_name}",
            prs=(),
            issues=(),
            pr_pagination=None,
            issue_pagination=None,
            fetch_error=f"Gitea fetch failed: {exc}",
        )

    issue_to_prs: dict[int, list[int]] = {}
    for pr in raw_prs:
        linked = _extract_linked_issue(pr.get("title"), pr.get("body"))
        if linked is not None:
            issue_to_prs.setdefault(linked, []).append(int(pr["number"]))

    issue_claimed = {
        int(i["number"])
        for i in raw_issues
        if any(lb.get("name") == "status:in-progress" for lb in i.get("labels", []))
    }

    pr_items = tuple(
        _format_pr_item(
            pr,
            _classify_pr(
                pr,
                issue_claimed=issue_claimed,
                issue_to_prs=issue_to_prs,
            ),
        )
        for pr in sorted(raw_prs, key=lambda p: int(p["number"]), reverse=True)
    )
    issue_items = tuple(
        _format_issue_item(
            issue,
            _classify_issue(
                issue,
                linked_prs=issue_to_prs.get(int(issue["number"]), []),
            ),
        )
        for issue in sorted(raw_issues, key=lambda i: int(i["number"]), reverse=True)
    )

    return QueueSnapshot(
        project_id=project.id,
        repo_label=f"{project.gitea_owner}/{project.repo_name}",
        prs=pr_items,
        issues=issue_items,
        pr_pagination=pr_pagination,
        issue_pagination=issue_pagination,
    )


def snapshot_to_dict(snapshot: QueueSnapshot) -> dict[str, Any]:
    """JSON-serializable export for /api/queue."""

    def _page(meta: PaginationMeta | None) -> dict[str, Any] | None:
        if meta is None:
            return None
        return {
            "page": meta.page,
            "per_page": meta.per_page,
            "returned_count": meta.returned_count,
            "has_more": meta.has_more,
            "is_final_page": meta.is_final_page,
            "inventory_complete": meta.inventory_complete,
            "pages_fetched": meta.pages_fetched,
        }

    def _item(item: QueueItem) -> dict[str, Any]:
        return {
            "number": item.number,
            "title": item.title,
            "badges": list(item.badges),
            **item.extra,
        }

    return {
        "project_id": snapshot.project_id,
        "repo": snapshot.repo_label,
        "fetch_error": snapshot.fetch_error,
        "prs": [_item(p) for p in snapshot.prs],
        "issues": [_item(i) for i in snapshot.issues],
        "pagination": {
            "prs": _page(snapshot.pr_pagination),
            "issues": _page(snapshot.issue_pagination),
        },
    }