#!/usr/bin/env python3 """產生 source-control refs reconcile plan 草案。 此工具只讀取既有 redacted snapshot，不呼叫遠端 Git，不 push、不 fetch、不改 remote。 輸出用途是 review / AwoooP mirror / approval candidate，不是 execution plan runner。 """ from __future__ import annotations import argparse import json from pathlib import Path from typing import Any DEFAULT_STILL_FORBIDDEN = [ "push refs", "force push", "delete refs", "create GitHub repo", "change repo visibility", "switch GitHub primary", "disable Gitea", "move secret values", ] DEFAULT_EXECUTION_GATES = [ "Gitea authenticated 或 admin_export server-side repo inventory status=ok", "branch-by-branch SHA diff 已完成", "tag-by-tag SHA diff 已完成", "workflow / webhook / runner / secret 名稱 inventory 已完成", "repo owner / visibility / branch protection / CODEOWNERS 已確認", "rollback plan 與 GitHub primary ADR 已完成", "人工批准只針對單一 repo 生效，不得批次套用到所有 repo", ] def load_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8")) def risk_for_repo(gitea_repo: str) -> str: if gitea_repo == "wooo/awoooi": return "HIGH" return "MEDIUM" def proposed_steps(event: dict[str, Any]) -> list[str]: gitea_repo = str(event.get("gitea_repo", "")) github_repo = str(event.get("github_repo", "")) steps = [ f"針對 `{gitea_repo}` 與 `{github_repo}` 產生 branch-by-branch diff 表。", f"針對 `{gitea_repo}` 與 `{github_repo}` 產生 tag-by-tag diff 表。", "標記每個 diff 的真相來源候選：Gitea、GitHub、人工指定或 deprecated。", "列出 workflow / webhook / runner / secret 名稱差異，只記名稱不記 value。", "產生 dry-run PR / ADR 草案，仍不 push refs。", ] if gitea_repo == "wooo/awoooi": steps.insert( 0, "先確認目前 production deploy 真相來源與 deploy marker 流程，避免主控切換影響發版。", ) return steps def detail_by_repo(path: str | None) -> dict[tuple[str, str], dict[str, Any]]: if not path: return {} payload = load_json(Path(path)) return { (str(item.get("gitea_repo", "")), str(item.get("github_repo", ""))): item for item in payload.get("repos", []) if isinstance(item, dict) } def main_shas_from_detail(detail: dict[str, Any]) -> tuple[str, str]: branch = detail.get("branch_diff", {}) if not isinstance(branch, dict): return "", "" for item in branch.get("sha_mismatch", []): if isinstance(item, dict) and item.get("name") == "main": return str(item.get("gitea_sha", "")), str(item.get("github_sha", "")) for key in ("matching", "only_gitea", "only_github"): for item in branch.get(key, []): if isinstance(item, dict) and item.get("name") == "main": sha = str(item.get("sha", "")) if key == "only_gitea": return sha, "" if key == "only_github": return "", sha return sha, sha return "", "" def build_plan(args: argparse.Namespace) -> dict[str, Any]: events = [load_json(Path(path)) for path in args.source_events] gitea_inventory = load_json(Path(args.gitea_inventory)) details = detail_by_repo(args.ref_detail_diff) inventory_ready = gitea_inventory.get("status") == "ok" inventory_gate_status = "ready" if inventory_ready else "blocked" inventory_reason = ( "Gitea server-side inventory 已完成，可在人工批准後進入更細 diff。" if inventory_ready else "Gitea authenticated / admin_export server-side inventory 尚未完成；本 plan 只能作草案，不可執行 refs sync。" ) plans: list[dict[str, Any]] = [] for event in events: gitea_repo = str(event.get("gitea_repo", "")) github_repo = str(event.get("github_repo", "")) detail = details.get((gitea_repo, github_repo), {}) branch_detail = detail.get("branch_diff", {}) if isinstance(detail, dict) else {} tag_detail = detail.get("tag_diff", {}) if isinstance(detail, dict) else {} detail_gitea_main, detail_github_main = main_shas_from_detail(detail) plans.append( { "gitea_repo": gitea_repo, "github_repo": github_repo, "risk": risk_for_repo(gitea_repo), "source_status": str(event.get("status", "")), "divergence_summary": { "gitea_branch_count": int(branch_detail.get("gitea_count", event.get("branch_count_gitea", 0))), "github_branch_count": int(branch_detail.get("github_count", event.get("branch_count_github", 0))), "gitea_tag_count": int(tag_detail.get("gitea_count", event.get("tag_count_gitea", 0))), "github_tag_count": int(tag_detail.get("github_count", event.get("tag_count_github", 0))), "gitea_main_sha": detail_gitea_main or str(event.get("latest_sha_gitea", "")), "github_main_sha": detail_github_main or str(event.get("latest_sha_github", "")), "blocking_reason": str(detail.get("blocking_reason") or event.get("blocking_reason", "")), }, "proposed_plan_steps": proposed_steps(event), "execution_gates": DEFAULT_EXECUTION_GATES, "allowed_now": [ "更新 read-only evidence", "更新 approval board", "產生 draft reconcile plan", "讓 AwoooP mirror plan 狀態", ], "still_forbidden": DEFAULT_STILL_FORBIDDEN, "evidence_refs": [str(value) for value in event.get("evidence_refs", [])], "awooop_consumption": "approval_candidate", } ) return { "schema_version": "source_control_reconcile_plan_v1", "status": "draft_blocked", "date": args.date, "default_mode": "plan_only", "inventory_gate": { "status": inventory_gate_status, "reason": inventory_reason, "required_before_execution": DEFAULT_EXECUTION_GATES, }, "plan_count": len(plans), "plans": plans, } def short_sha(value: str) -> str: return value[:8] if value else "" def write_markdown(plan: dict[str, Any], path: Path) -> None: gate = plan["inventory_gate"] lines = [ "# Source Control Draft Reconcile Plan", "", "| 項目 | 內容 |", "|------|------|", f"| 日期 | {plan['date']} |", f"| 狀態 | `{plan['status']}` |", f"| 預設模式 | `{plan['default_mode']}` |", f"| inventory gate | `{gate['status']}` |", f"| gate 原因 | {gate['reason']} |", f"| plan count | {plan['plan_count']} |", "", "## 0. 核心結論", "", "這份文件只是 refs reconcile 草案，不是同步腳本，也不授權任何 GitHub primary 切換。AwoooP 可以 mirror 成 approval candidate，但不得執行 board item 或呼叫任何 push / sync 工具。", "", "若已存在 `source_control_ref_truth_classification_v1`，請把它視為本 plan 的人工 review lane 補充：分類結果只協助 repo owner 判定，不授權同步或刪除。", "", "## 1. Repo 差異摘要", "", "| Repo | Risk | Gitea branches | GitHub branches | Gitea tags | GitHub tags | Gitea main | GitHub main |", "|------|------|----------------|-----------------|------------|-------------|------------|-------------|", ] for item in plan["plans"]: diff = item["divergence_summary"] lines.append( "| " + " | ".join( [ f"`{item['gitea_repo']} -> {item['github_repo']}`", f"`{item['risk']}`", f"`{diff['gitea_branch_count']}`", f"`{diff['github_branch_count']}`", f"`{diff['gitea_tag_count']}`", f"`{diff['github_tag_count']}`", f"`{short_sha(diff['gitea_main_sha'])}`", f"`{short_sha(diff['github_main_sha'])}`", ] ) + " |" ) lines.extend(["", "## 2. Draft Plan", ""]) for item in plan["plans"]: diff = item["divergence_summary"] lines.extend( [ f"### {item['gitea_repo']} -> {item['github_repo']}", "", f"- 狀態：`{item['source_status']}`", f"- 阻塞原因：{diff['blocking_reason']}", "- 允許現在做：", ] ) for value in item["allowed_now"]: lines.append(f" - {value}") lines.append("- 草案步驟：") for value in item["proposed_plan_steps"]: lines.append(f" - {value}") lines.append("- 執行前 gate：") for value in item["execution_gates"]: lines.append(f" - {value}") lines.append("- 仍然禁止：") for value in item["still_forbidden"]: lines.append(f" - {value}") lines.append("- Evidence refs：") for value in item["evidence_refs"]: lines.append(f" - `{value}`") lines.append("") lines.extend( [ "## 3. AwoooP 消費方式", "", "1. 只 mirror `source_control_reconcile_plan_v1`。", "2. 只顯示 `draft_blocked` 與 blocking reason。", "3. 可產生 approval candidate，但不得自動批准。", "4. 不得新增 execution action button。", "5. 真相來源分類請讀 `docs/security/SOURCE-CONTROL-REF-TRUTH-CLASSIFICATION.md`，並維持單 repo / 單 ref 人工 gate。", "", ] ) path.write_text("\n".join(lines), encoding="utf-8") def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--date", required=True) parser.add_argument("--gitea-inventory", default="docs/security/gitea-repo-inventory.snapshot.json") parser.add_argument("--ref-detail-diff") parser.add_argument("--source-event", action="append", dest="source_events", required=True) parser.add_argument("--output-json", required=True) parser.add_argument("--output-md", required=True) args = parser.parse_args() plan = build_plan(args) Path(args.output_json).write_text( json.dumps(plan, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) write_markdown(plan, Path(args.output_md)) print(f"OK source-control reconcile plan count={plan['plan_count']}") return 0 if __name__ == "__main__": raise SystemExit(main())