#!/usr/bin/env python3
"""Preview or idempotently apply Harness Coding labels to one Gitea repo."""

from __future__ import annotations

import argparse
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request
from typing import Any


LABELS: tuple[dict[str, Any], ...] = (
    {
        "name": "kind/task",
        "color": "0052CC",
        "description": "Harness Coding task",
        "exclusive": False,
    },
    {
        "name": "type/docs",
        "color": "5319E7",
        "description": "Documentation change",
        "exclusive": True,
    },
    {
        "name": "type/code",
        "color": "1D76DB",
        "description": "Code or automation change",
        "exclusive": True,
    },
    {
        "name": "status/todo",
        "color": "C5DEF5",
        "description": "Ready to claim",
        "exclusive": True,
    },
    {
        "name": "status/waiting",
        "color": "D4C5F9",
        "description": "Not claimable; waiting on dependency or external condition",
        "exclusive": True,
    },
    {
        "name": "status/doing",
        "color": "FBCA04",
        "description": "Claimed and in progress",
        "exclusive": True,
    },
    {
        "name": "status/blocked",
        "color": "D93F0B",
        "description": "Blocked; claim retained",
        "exclusive": True,
    },
    {
        "name": "status/review",
        "color": "BFD4F2",
        "description": "Pull request under review",
        "exclusive": True,
    },
    {
        "name": "status/done",
        "color": "0E8A16",
        "description": "Merged and completed",
        "exclusive": True,
    },
    {
        "name": "priority/p0",
        "color": "B60205",
        "description": "Highest priority",
        "exclusive": True,
    },
    {
        "name": "priority/p1",
        "color": "D93F0B",
        "description": "High priority",
        "exclusive": True,
    },
    {
        "name": "priority/p2",
        "color": "FBCA04",
        "description": "Normal priority",
        "exclusive": True,
    },
)


class ApiError(RuntimeError):
    def __init__(self, status: int, reason: str) -> None:
        super().__init__(f"Gitea API 返回 HTTP {status}：{reason}")
        self.status = status


class NoRedirect(urllib.request.HTTPRedirectHandler):
    """Never forward the Authorization header to a redirected origin."""

    def redirect_request(self, *args: Any, **kwargs: Any) -> None:
        return None


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="读取远端差异，并可幂等创建或校正 Harness Coding Gitea 标签。"
    )
    parser.add_argument(
        "--repo",
        default=os.environ.get("GITEA_REPOSITORY"),
        help="目标 owner/repo；也可设置 GITEA_REPOSITORY。",
    )
    parser.add_argument(
        "--apply",
        action="store_true",
        help="应用预览中的 create/update；省略时只读远端并打印差异。",
    )
    return parser.parse_args()


def validate_config(url: str, token: str, repo: str) -> tuple[str, str, str]:
    parsed = urllib.parse.urlsplit(url.strip())
    if parsed.scheme not in {"http", "https"} or not parsed.hostname:
        raise ValueError("GITEA_URL 必须是 http(s) 实例根地址。")
    if parsed.username or parsed.password or parsed.query or parsed.fragment:
        raise ValueError("GITEA_URL 不得包含凭据、query 或 fragment。")
    path = parsed.path.rstrip("/")
    if path.lower().endswith("/api/v1"):
        raise ValueError("GITEA_URL 不得包含 /api/v1。")
    if parsed.scheme == "http" and os.environ.get("GITEA_ALLOW_INSECURE_HTTP") != "1":
        raise ValueError("HTTP 需要显式设置 GITEA_ALLOW_INSECURE_HTTP=1。")
    if not token:
        raise ValueError("缺少 GITEA_TOKEN。")
    parts = repo.split("/")
    if len(parts) != 2 or not all(parts):
        raise ValueError("--repo 必须使用 owner/repo 格式。")
    root = urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, path, "", ""))
    return root.rstrip("/"), parts[0], parts[1]


class GiteaClient:
    def __init__(self, root: str, owner: str, repo: str, token: str) -> None:
        owner_q = urllib.parse.quote(owner, safe="")
        repo_q = urllib.parse.quote(repo, safe="")
        self.base = f"{root}/api/v1/repos/{owner_q}/{repo_q}"
        self.token = token
        proxy_handler = (
            urllib.request.ProxyHandler({})
            if os.environ.get("GITEA_DIRECT") == "1"
            else urllib.request.ProxyHandler()
        )
        self.opener = urllib.request.build_opener(proxy_handler, NoRedirect())

    def request(
        self, method: str, path: str, payload: dict[str, Any] | None = None
    ) -> Any:
        data = None if payload is None else json.dumps(payload).encode("utf-8")
        request = urllib.request.Request(
            self.base + path,
            data=data,
            method=method,
            headers={
                "Accept": "application/json",
                "Authorization": f"token {self.token}",
                "Content-Type": "application/json",
            },
        )
        try:
            with self.opener.open(request, timeout=30) as response:
                body = response.read()
                return json.loads(body.decode("utf-8")) if body else None
        except urllib.error.HTTPError as exc:
            raise ApiError(exc.code, exc.reason) from None
        except urllib.error.URLError as exc:
            raise RuntimeError(f"连接 Gitea 失败：{exc.reason}") from None

    def list_labels(self) -> dict[str, dict[str, Any]]:
        result: dict[str, dict[str, Any]] = {}
        seen_pages: set[tuple[str, ...]] = set()
        page = 1
        while True:
            labels = self.request("GET", f"/labels?limit=50&page={page}")
            if not isinstance(labels, list):
                raise RuntimeError("Gitea labels 响应格式异常。")
            if not labels:
                return result
            if not all(isinstance(label, dict) for label in labels):
                raise RuntimeError("Gitea labels 响应包含非对象条目。")
            signature = tuple(str(label.get("id") or label.get("name")) for label in labels)
            if signature in seen_pages or page > 1000:
                raise RuntimeError("Gitea labels 分页重复，已停止以避免无限读取。")
            seen_pages.add(signature)
            for label in labels:
                if isinstance(label, dict) and isinstance(label.get("name"), str):
                    result[label["name"]] = label
            page += 1


def normalize_color(value: Any) -> str:
    return str(value or "").lstrip("#").upper()


def needs_update(current: dict[str, Any], desired: dict[str, Any]) -> bool:
    return (
        normalize_color(current.get("color")) != desired["color"]
        or str(current.get("description") or "") != desired["description"]
        or bool(current.get("exclusive")) != desired["exclusive"]
    )


def build_plan(
    existing: dict[str, dict[str, Any]],
) -> list[tuple[str, dict[str, Any], dict[str, Any] | None]]:
    plan = []
    for desired in LABELS:
        current = existing.get(desired["name"])
        if current is None:
            action = "create"
        elif needs_update(current, desired):
            action = "update"
        else:
            action = "unchanged"
        plan.append((action, desired, current))
    return plan


def show_plan(repo: str, plan: list[tuple[str, dict[str, Any], Any]]) -> None:
    print(f"目标仓库：{repo}")
    for action, desired, _ in plan:
        scope = "exclusive" if desired["exclusive"] else "normal"
        print(f"- {action:9} {desired['name']} #{desired['color']} {scope}")
    counts = {name: sum(action == name for action, _, _ in plan) for name in (
        "create",
        "update",
        "unchanged",
    )}
    print(
        "计划汇总："
        f"创建 {counts['create']}，更新 {counts['update']}，未变化 {counts['unchanged']}。"
    )


def apply_plan(
    client: GiteaClient,
    plan: list[tuple[str, dict[str, Any], dict[str, Any] | None]],
) -> None:
    created = updated = unchanged = 0
    for action, desired, current in plan:
        if action == "unchanged":
            unchanged += 1
            continue
        if action == "update":
            label_id = None if current is None else current.get("id")
            if not isinstance(label_id, int):
                raise RuntimeError(f"标签 {desired['name']} 缺少数字 id。")
            client.request("PATCH", f"/labels/{label_id}", dict(desired))
            updated += 1
            continue
        try:
            client.request("POST", "/labels", dict(desired))
            created += 1
        except ApiError as exc:
            if exc.status != 422:
                raise
            latest = client.list_labels().get(desired["name"])
            if latest is None:
                raise
            if needs_update(latest, desired):
                label_id = latest.get("id")
                if not isinstance(label_id, int):
                    raise RuntimeError(f"标签 {desired['name']} 缺少数字 id。")
                client.request("PATCH", f"/labels/{label_id}", dict(desired))
                updated += 1
            else:
                unchanged += 1
    print(f"标签同步完成：创建 {created}，更新 {updated}，未变化 {unchanged}。")


def main() -> int:
    args = parse_args()
    if not args.repo:
        print("ERROR: 需要 --repo owner/repo 或 GITEA_REPOSITORY。", file=sys.stderr)
        return 2
    url = os.environ.get("GITEA_URL", "")
    token = os.environ.get("GITEA_TOKEN", "")
    try:
        root, owner, name = validate_config(url, token, args.repo)
        client = GiteaClient(root, owner, name, token)
        plan = build_plan(client.list_labels())
        show_plan(args.repo, plan)
        if not args.apply:
            print("dry-run：未写入；追加 --apply 才会应用上述 create/update。")
            return 0
        apply_plan(client, plan)
        return 0
    except (ValueError, ApiError, RuntimeError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())