From acf179a199e9977d7db01152e4334fc74b537801 Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Tue, 6 Oct 2026 17:03:44 +0800 Subject: [PATCH] =?UTF-8?q?fix(subjects):=20migrate=20=E8=84=9A=E6=9C=AC?= =?UTF-8?q?=E5=8F=A3=E5=BE=84=E4=B9=99=20fail-closed=EF=BC=88=E6=8E=92?= =?UTF-8?q?=E9=99=A4=E7=99=BD=E5=90=8D=E5=8D=95=E9=80=90=E7=A0=81=20+=20?= =?UTF-8?q?=E5=8F=8C=E8=BD=A8=E9=94=9A=E7=82=B9=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../migrate_subjects_entity_id_20260923.py | 189 +++++++++++++++--- 1 file changed, 163 insertions(+), 26 deletions(-) diff --git a/backend/scripts/migrate_subjects_entity_id_20260923.py b/backend/scripts/migrate_subjects_entity_id_20260923.py index f7d7b9ec..dd022372 100644 --- a/backend/scripts/migrate_subjects_entity_id_20260923.py +++ b/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 **，与 DDL 分开放行）： - `remark='zxt-sync'` 的 164 行 → `--target-entity`（默认 1）；其余保持 NULL + --backfill 存量回填（**须 --evidence **，与 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 只读预检附口径乙校验（白名单逐码命中 + 锚点双轨读数），零写入 --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 ）") + 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 （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: