298 lines
10 KiB
Python
298 lines
10 KiB
Python
#!/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())
|