15.2 经济事故案例库

# file: sim/ch15_incident_casebook.py
# tier: S
# verify: python3 sim/ch15_incident_casebook.py --config config/
"""经济事故案例库的结构化校验。

事故本身不可复现（reproducibility=X），但案例库的**结构**可以校验：每条案例必须
把「症状 / 根因 / 影响面 / 处置 / 教训」五段写全，分类必须合法，严重度必须在区间内，
分类不得有空档。本脚本读 config/ch15_incidents.json，逐条校验并产出
out/ch15_incidents.json（校验结论 + 分类统计）、out/ch15_incidents.csv（拍平的案例表）。
纯标准库，零依赖。
"""

from __future__ import annotations

import argparse
import copy
import csv
import json
import os

REQUIRED_STR = ["id", "name", "category", "rootCause", "blastRadius", "lesson"]
REQUIRED_LIST = ["symptom", "handling"]
REQUIRED_ALL = REQUIRED_STR + REQUIRED_LIST + ["severity", "reproducibility"]


def validate(cfg: dict) -> tuple[list[str], dict]:
    errs: list[str] = []
    cat_ids = [c["id"] for c in cfg["categories"]]
    cat_set = set(cat_ids)
    cases = cfg["cases"]

    seen: set[str] = set()
    by_cat: dict[str, int] = {c: 0 for c in cat_ids}
    by_sev: dict[int, int] = {}
    for case in cases:
        cid = case.get("id", "<no-id>")
        for f in REQUIRED_ALL:
            if f not in case:
                errs.append(f"MISSING_FIELD {cid}.{f}")
        if cid in seen:
            errs.append(f"DUP_ID {cid}")
        seen.add(cid)

        for f in REQUIRED_STR:
            if not str(case.get(f, "")).strip():
                errs.append(f"EMPTY_STR {cid}.{f}")
        for f in REQUIRED_LIST:
            v = case.get(f)
            if not isinstance(v, list) or len(v) == 0 or not all(str(x).strip() for x in v):
                errs.append(f"EMPTY_LIST {cid}.{f}")

        cat = case.get("category")
        if cat not in cat_set:
            errs.append(f"BAD_CATEGORY {cid}.{cat}")
        else:
            by_cat[cat] += 1

        sev = case.get("severity")
        if not isinstance(sev, int) or not (1 <= sev <= 5):
            errs.append(f"BAD_SEVERITY {cid}.{sev}")
        else:
            by_sev[sev] = by_sev.get(sev, 0) + 1

        if case.get("reproducibility") != "X":
            errs.append(f"BAD_REPRO {cid}.{case.get('reproducibility')}")

    for cat in cat_ids:
        if by_cat[cat] == 0:
            errs.append(f"EMPTY_CATEGORY {cat}")

    stats = {
        "caseCount": len(cases),
        "categoryCount": len(cat_ids),
        "byCategory": by_cat,
        "bySeverity": {str(k): by_sev.get(k, 0) for k in range(1, 6)},
        "avgSeverity": round(sum(c.get("severity", 0) for c in cases) / len(cases), 3) if cases else 0,
        "allX": all(c.get("reproducibility") == "X" for c in cases),
    }
    return errs, stats


def mutation_demo(cfg: dict) -> dict:
    """故意注入五类缺陷，证明校验器确实会报错（不改任何配置文件）。"""
    bad = copy.deepcopy(cfg)
    cases = bad["cases"]
    cases[0]["id"] = cases[1]["id"]                 # 重复 id
    cases[2]["rootCause"] = "   "                   # 空字符串
    cases[3]["handling"] = []                       # 空列表
    cases[4]["category"] = "not_a_category"         # 非法分类
    cases[5]["severity"] = 9                         # 严重度越界
    cases[6]["reproducibility"] = "S"               # 非 X
    errs, _ = validate(bad)
    codes = sorted({e.split()[0] for e in errs})
    return {"injected": ["DUP_ID", "EMPTY_STR", "EMPTY_LIST", "BAD_CATEGORY", "BAD_SEVERITY", "BAD_REPRO"],
            "caught": codes, "rawCount": len(errs)}


def write_csv(cfg: dict, path: str) -> None:
    rows = []
    for c in cfg["cases"]:
        rows.append({
            "id": c["id"], "name": c["name"], "category": c["category"],
            "severity": c["severity"], "reproducibility": c["reproducibility"],
            "symptomCount": len(c["symptom"]), "handlingCount": len(c["handling"]),
            "symptom": " | ".join(c["symptom"]), "rootCause": c["rootCause"],
            "blastRadius": c["blastRadius"], "handling": " | ".join(c["handling"]),
            "lesson": c["lesson"],
        })
    rows.sort(key=lambda r: (-r["severity"], r["category"], r["id"]))
    with open(path, "w", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
        w.writeheader()
        w.writerows(rows)


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--config", default="config/")
    ap.add_argument("--out", default="out/")
    args = ap.parse_args()

    with open(os.path.join(args.config, "ch15_incidents.json"), encoding="utf-8") as f:
        cfg = json.load(f)

    errs, stats = validate(cfg)
    demo = mutation_demo(cfg)
    report = {"version": cfg["version"], "game": cfg["game"], "ok": not errs,
              "errors": errs, "stats": stats, "mutationDemo": demo}

    os.makedirs(args.out, exist_ok=True)
    with open(os.path.join(args.out, "ch15_incidents.json"), "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)
    write_csv(cfg, os.path.join(args.out, "ch15_incidents.csv"))
    print(json.dumps(report, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

ok=True
caseCount=8 categoryCount=5 avgSeverity=4.0 allX=True
byCategory production_leak=2 duplication_bug=2 fx_arbitrage=1 season_reset_error=2 compensation_overissue=1
bySeverity 3=2 4=4 5=2
mutationDemo injected=[DUP_ID, EMPTY_STR, EMPTY_LIST, BAD_CATEGORY, BAD_SEVERITY, BAD_REPRO]
mutationDemo caught=[BAD_CATEGORY, BAD_REPRO, BAD_SEVERITY, DUP_ID, EMPTY_CATEGORY, EMPTY_LIST, EMPTY_STR]