docs: adopt harness coding workflow

scripts/setup_gitea_labels.py

@@ -0,0 +1,297 @@
+#!/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/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())