fix(subjects): migrate 脚本口径乙 fail-closed（排除白名单逐码 + 双轨锚点）

backend/scripts/migrate_subjects_entity_id_20260923.py

@@ -26,8 +26,15 @@
   --check     只读预检（默认）：列/索引现状 + 行数分布 + 唯一键重复检查，零写入
   --apply     DDL：ADD COLUMN entity_id INT NULL + INDEX + UNIQUE(entity_id,subject_code)
               − DROP INDEX uk_subject_code（幂等：已存在的步骤自动跳过）
-  --backfill  存量回填（**须 --evidence <json>**，与 DDL 分开放行）：
-              `remark='zxt-sync'` 的 164 行 → `--target-entity`（默认 1）；其余保持 NULL
+  --backfill  存量回填（**须 --evidence <json>**，与 DDL 分开放行；口径乙 = 2026-09-23 财务裁定）：
+              `remark='zxt-sync'` 的 164 行**剔除排除白名单**后 → 156 行置 `--target-entity`
+              （默认 1）；`remark IS NULL` 的 43 行 ＋ 排除白名单 8 行保持/复位 `NULL`（合计 51）
+              ⇒ 写后 `GROUP BY entity_id` = {1:156, NULL:51}
+              取证件 `exclude_subject_codes`：逐码命中须 = `expect_hits_per_code` 且总和须 =
+              `total_expect`，否则 rc=3（**不做部分回填**）；**锚点双轨**：`by_remark_anchor`
+              （zxt-sync 164 / (null) 43 / total 207）与 `counts`（e1 156 / global_null 51）
+              并存校验，任一不符即 fail-closed rc=3
+  --check --evidence <json>  只读预检附口径乙校验（白名单逐码命中 + 锚点双轨读数），零写入
   --down      回滚：重建 `uk_subject_code`（先断言无同码重复，fail-closed）→ 删新唯一键 → 删列
   --show-create  打印 SHOW CREATE TABLE subjects 原文（验收证据）
 
@@ -101,6 +108,138 @@ def db_name_of(engine) -> str:
         return ""
 
 
+# ============================================================
+# 口径乙（2026-09-23 财务裁定，wecom-finance）
+#   · e1（酣客私有）= remark='zxt-sync' 行 **剔除排除白名单** 后的行数
+#   · global_null   = remark IS NULL 行 ＋ 排除白名单行
+#   · by_remark_anchor 与 counts 两轨并存，须同时与取证件一致（fail-closed）
+# ============================================================
+EXCLUDE_KEY = "exclude_subject_codes"
+ANCHOR_KEY = "by_remark_anchor"
+NULL_REMARK = "(null)"
+
+
+class EvidenceError(Exception):
+    """取证件/锚点校验失败 ⇒ fail-closed（rc=3：零写入、不做部分回填）"""
+
+
+def load_evidence(path) -> dict:
+    """读取证件 JSON 并做结构校验（缺件/缺字段 ⇒ EvidenceError）"""
+    if not path or not os.path.exists(path):
+        raise EvidenceError(f"取证件不存在：{path!r}（--backfill 必须带 --evidence <json>）")
+    try:
+        with open(path, encoding="utf-8") as fh:
+            ev = json.load(fh)
+    except Exception as exc:  # noqa: BLE001
+        raise EvidenceError(f"取证件解析失败：{exc}") from exc
+    missing = [k for k in ("dama_evidence", "finance_confirmation", "counts") if not ev.get(k)]
+    if missing:
+        raise EvidenceError(f"取证件缺字段 {missing}")
+    return ev
+
+
+def exclude_spec(ev: dict) -> dict:
+    """口径乙排除白名单规格（字段名以取证件 JSON 为准）"""
+    spec = ev.get(EXCLUDE_KEY)
+    if not isinstance(spec, dict):
+        raise EvidenceError(f"取证件缺 {EXCLUDE_KEY} 段（无法确定白名单，拒绝执行）")
+    codes = [str(c).strip() for c in (spec.get("codes") or [])]
+    per = spec.get("expect_hits_per_code")
+    total = spec.get("total_expect")
+    if not codes:
+        raise EvidenceError(f"{EXCLUDE_KEY}.codes 为空，拒绝执行")
+    if not spec.get("fail_closed"):
+        raise EvidenceError(f"{EXCLUDE_KEY}.fail_closed 缺失（非 fail-closed 形态，拒绝执行）")
+    if len(set(codes)) != len(codes):
+        raise EvidenceError(f"{EXCLUDE_KEY}.codes 存在重复：{codes}")
+    if per != 1:
+        raise EvidenceError(f"expect_hits_per_code={per!r} != 1，拒绝执行")
+    if total is None or int(total) != len(codes):
+        raise EvidenceError(f"total_expect={total!r} != 白名单码数 {len(codes)}，拒绝执行")
+    return {"codes": codes, "expect_hits_per_code": 1, "total_expect": int(total)}
+
+
+def probe_exclude_hits(engine, codes) -> dict:
+    """逐码命中数（只读）：subject_code 精确匹配行数"""
+    hits = {}
+    with engine.connect() as conn:
+        for code in codes:
+            hits[code] = int(conn.execute(
+                text(f"SELECT COUNT(*) FROM {TABLE} WHERE subject_code = :c"), {"c": code}).scalar())
+    return hits
+
+
+def caliber_b_readings(engine, codes) -> dict:
+    """现网口径乙读数（只读）：by_remark 实读 ＋ e1/global_null 推算 ＋ 白名单逐码命中"""
+    st = state(engine)
+    by_remark = dict(st["counts"].get("by_remark") or {})
+    live_remark = {"zxt-sync": int(by_remark.get("zxt-sync", 0)),
+                   NULL_REMARK: int(by_remark.get(NULL_REMARK, 0)),
+                   "total": int(st["counts"].get("total") or 0)}
+    hits = probe_exclude_hits(engine, codes)
+    hit_total = sum(hits.values())
+    live_counts = {"e1": live_remark["zxt-sync"] - hit_total,
+                   "global_null": live_remark[NULL_REMARK] + hit_total}
+    return {"state": st, "by_remark": live_remark, "counts": live_counts,
+            "exclude_hits": hits, "exclude_total": hit_total}
+
+
+def verify_caliber_b(engine, ev, spec) -> dict:
+    """双轨锚点校验：白名单逐码命中 ＋ by_remark ＋ counts，三者全部须与取证件一致"""
+    rd = caliber_b_readings(engine, spec["codes"])
+    hits = rd["exclude_hits"]
+    bad = {c: n for c, n in hits.items() if n != spec["expect_hits_per_code"]}
+    if bad:
+        raise EvidenceError(f"排除白名单逐码命中 != {spec['expect_hits_per_code']}：{bad}（拒绝执行）")
+    if rd["exclude_total"] != spec["total_expect"]:
+        raise EvidenceError(f"排除白名单命中总数 {rd['exclude_total']} != total_expect "
+                            f"{spec['total_expect']}（拒绝执行，不做部分回填）")
+    exp_anchor = ev.get(ANCHOR_KEY)
+    if not isinstance(exp_anchor, dict):
+        raise EvidenceError(f"取证件缺 {ANCHOR_KEY}（锚点双轨的 remark 轨缺失，拒绝执行）")
+    counts = ev.get("counts") or {}
+    pairs = [("by_remark_anchor.zxt-sync", exp_anchor.get("zxt-sync"), rd["by_remark"]["zxt-sync"]),
+             (f"by_remark_anchor.{NULL_REMARK}", exp_anchor.get(NULL_REMARK), rd["by_remark"][NULL_REMARK]),
+             ("by_remark_anchor.total", exp_anchor.get("total"), rd["by_remark"]["total"]),
+             ("counts.e1", counts.get("e1"), rd["counts"]["e1"]),
+             ("counts.global_null", counts.get("global_null"), rd["counts"]["global_null"])]
+    mismatch = [(k, e, l) for k, e, l in pairs if e is None or int(e) != int(l)]
+    if mismatch:
+        raise EvidenceError("锚点双轨与现网不符：" + "；".join(
+            f"{k}: evidence={e} vs live={l}" for k, e, l in mismatch))
+    if not rd["state"]["has_column"]:
+        raise EvidenceError("entity_id 列未落地，先跑 --apply（本单不得重跑 DDL）")
+    rd["spec"] = spec
+    rd["evidence_by_remark"] = {k: exp_anchor.get(k) for k in ("zxt-sync", NULL_REMARK, "total")}
+    rd["evidence_counts"] = {k: counts.get(k) for k in ("e1", "global_null")}
+    return rd
+
+
+def print_caliber_b(rd: dict, evidence_path) -> None:
+    spec = rd.get("spec") or {}
+    print(f"== 口径乙校验（evidence: {evidence_path}）==")
+    print(f"  排除白名单逐码命中：{rd['exclude_hits']}（合计 {rd['exclude_total']}；"
+          f"expect_hits_per_code={spec.get('expect_hits_per_code')} / "
+          f"total_expect={spec.get('total_expect')}）")
+    print(f"  锚点轨 A by_remark（现网）：{rd['by_remark']} ｜ 取证件：{rd['evidence_by_remark']}")
+    print(f"  锚点轨 B counts（现网）：{rd['counts']} ｜ 取证件：{rd['evidence_counts']}")
+
+
+def do_precheck_evidence(engine, evidence_path) -> int:
+    """--check 附 --evidence 的只读口径乙校验（零写入）：通过 rc=0，任一不符 rc=3"""
+    try:
+        ev = load_evidence(evidence_path)
+        spec = exclude_spec(ev)
+        rd = verify_caliber_b(engine, ev, spec)
+    except EvidenceError as exc:
+        print(f"!!! {exc}")
+        print("!!! fail-closed：只读预检未通过（零写入）")
+        return 3
+    print_caliber_b(rd, evidence_path)
+    print("  [OK] 白名单命中 = total_expect 且锚点双轨一致 ⇒ --backfill 前置判据满足")
+    return 0
+
+
 def state(engine) -> dict:
     insp = inspect(engine)
     tables = set(insp.get_table_names())
@@ -181,34 +320,28 @@ def do_apply(engine) -> int:
 
 
 def do_backfill(engine, evidence_path: str, target: int) -> int:
-    if not evidence_path or not os.path.exists(evidence_path):
-        print("!!! --backfill 必须提供 --evidence <json>（DAMA 只读取证 + 财务确认件锚点），拒绝执行")
-        return 3
-    ev = json.load(open(evidence_path, encoding="utf-8"))
-    need = ("dama_evidence", "finance_confirmation", "counts")
-    missing = [k for k in need if not ev.get(k)]
-    if missing:
-        print(f"!!! 取证件缺字段 {missing}，拒绝执行（fail-closed）")
-        return 3
-    st = state(engine)
-    if not st["has_column"]:
-        print("!!! entity_id 列未落地，先跑 --apply")
-        return 3
-    by_remark = dict(st["counts"]["by_remark"])
-    live_private = int(by_remark.get("zxt-sync", 0))
-    live_global = int(by_remark.get("(null)", 0))
-    exp = ev["counts"]
-    if int(exp.get("e1", -1)) != live_private or int(exp.get("global_null", -1)) != live_global:
-        print(f"!!! 取证件行数锚点与现网不符：evidence e1={exp.get('e1')}/global_null="
-              f"{exp.get('global_null')} vs live {live_private}/{live_global}，拒绝执行")
+    """口径乙回填：zxt-sync 剔除排除白名单 → target；其余（含白名单码）保持/复位 NULL"""
+    try:
+        ev = load_evidence(evidence_path)
+        spec = exclude_spec(ev)
+        rd = verify_caliber_b(engine, ev, spec)
+    except EvidenceError as exc:
+        print(f"!!! {exc}")
+        print("!!! fail-closed：拒绝执行 --backfill（事务未开启，零写入、不做部分回填）")
         return 3
+    print_caliber_b(rd, evidence_path)
+    codes = spec["codes"]
+    ph = {f"c{i}": c for i, c in enumerate(codes)}
+    in_clause = ", ".join(f":c{i}" for i in range(len(codes)))
     with engine.begin() as conn:
         n1 = conn.execute(text(
-            f"UPDATE {TABLE} SET {COLUMN} = :e WHERE remark = 'zxt-sync'"), {"e": target}).rowcount
+            f"UPDATE {TABLE} SET {COLUMN} = :e WHERE remark = 'zxt-sync' "
+            f"AND subject_code NOT IN ({in_clause})"), dict(ph, e=target)).rowcount
         n0 = conn.execute(text(
-            f"UPDATE {TABLE} SET {COLUMN} = NULL WHERE remark IS NULL AND {COLUMN} IS NOT NULL")).rowcount
-    print(f"== BACKFILL 完成：remark='zxt-sync' → entity_id={target} 命中 {n1} 行；"
-          f"标准准则科目置 NULL {n0} 行（evidence: {evidence_path}）")
+            f"UPDATE {TABLE} SET {COLUMN} = NULL WHERE {COLUMN} IS NOT NULL "
+            f"AND (remark IS NULL OR subject_code IN ({in_clause}))"), ph).rowcount
+    print(f"== BACKFILL 完成（口径乙）：remark='zxt-sync' 且非排除白名单 → entity_id={target} "
+          f"命中 {n1} 行；标准准则科目 ＋ 排除白名单码保持/复位 NULL {n0} 行")
     print_check(state(engine))
     return 0
 
@@ -282,6 +415,10 @@ def main():
 
     st = state(engine)
     print_check(st)
+    if args.evidence:
+        rc = do_precheck_evidence(engine, args.evidence)
+        if rc != 0:
+            return rc
     if is_prod:
         print("  [注意] 目标是生产库：以上仅为只读预检，未做任何写入。")
     if args.show_create: