AiDOPWarehouse/doc/plan/sql/uat-data/run_wp_sd8_compare_reruns.py

#!/usr/bin/env python3
"""Compare two WP-SD8 API rerun evidence files, ignoring IDs and timestamps."""

from __future__ import annotations

import json
from pathlib import Path

EVIDENCE = Path(__file__).resolve().parents[4] / "doc" / "plan" / "UAT留证" / "2026-08-17-智慧诊断正式落地"
IGNORE_KEYS = {
    "generated_at",
    "planId",
    "planNo",
    "actionId",
    "taskId",
    "firstFact",
}


def public_view(value):
    if isinstance(value, dict):
        return {k: public_view(v) for k, v in value.items() if k not in IGNORE_KEYS}
    if isinstance(value, list):
        return [public_view(v) for v in value]
    return value


def module_sig(row: dict) -> dict:
    return {
        "module": row.get("module"),
        "selectionMode": (row.get("global") or {}).get("selectionMode"),
        "metricCode": (row.get("global") or {}).get("metricCode"),
        "evidenceScope": (row.get("global") or {}).get("evidenceScope"),
        "specified_same_as_current": row.get("specified_same_as_current"),
        "kanban_root_present": row.get("kanban_root_present"),
        "loopClosed": row.get("loopClosed"),
        "auto": (row.get("verify") or {}).get("auto") if row.get("verify") else None,
        "createSkipped": (row.get("create") or {}).get("skipped"),
        "createReason": (row.get("create") or {}).get("reason"),
        "actionHttps": [
            step.get("http") if isinstance(step, dict) and "http" in step else step
            for item in (row.get("actionsCompleted") or [])
            for step in ([item] if "http" in (item or {}) else (item or {}).get("steps") or [])
        ],
    }


def load(name: str) -> dict:
    return json.loads((EVIDENCE / name).read_text(encoding="utf-8"))


def compare_tenant(left_name: str, right_name: str) -> dict:
    left = load(left_name)
    right = load(right_name)
    left_mods = {x["module"]: module_sig(x) for x in left.get("modules") or []}
    right_mods = {x["module"]: module_sig(x) for x in right.get("modules") or []}
    keys = sorted(set(left_mods) | set(right_mods))
    diffs = []
    for key in keys:
        if left_mods.get(key) != right_mods.get(key):
            diffs.append({"module": key, "left": left_mods.get(key), "right": right_mods.get(key)})
    return {
        "left": left_name,
        "right": right_name,
        "tenant": left.get("tenant"),
        "same_closed": left.get("closed_count") == right.get("closed_count"),
        "same_blocked": left.get("blocked_count") == right.get("blocked_count"),
        "module_diffs": diffs,
        "consistent": not diffs
        and left.get("closed_count") == right.get("closed_count")
        and left.get("blocked_count") == right.get("blocked_count"),
    }


def main() -> None:
    report = {
        "work_package": "WP-SD8-COMPARE",
        "pairs": [
            compare_tenant("08-rerun1-A.json", "08-rerun2-A.json"),
            compare_tenant("08-rerun1-B.json", "08-rerun2-B.json"),
            compare_tenant("08-rerun1-DEMO.json", "08-rerun2-DEMO.json"),
        ],
    }
    report["all_consistent"] = all(x["consistent"] for x in report["pairs"])
    out = EVIDENCE / "08-rerun-compare.json"
    out.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print(out)
    print(json.dumps({"all_consistent": report["all_consistent"], "tenants": [
        f"{x['tenant']}:consistent={x['consistent']}:diffs={len(x['module_diffs'])}"
        for x in report["pairs"]
    ]}, ensure_ascii=False))


if __name__ == "__main__":
    main()