#!/usr/bin/env python3 # -*- coding: utf-8 -*- """`quality_rule_config` 出厂默认值落库（全账套模板行 entity_id=0，is_default=1） 任务：ASK-20260928-05（CMA 财务七规则可配置化 · 最小实现），清单 §8「初始化」条目。 **本脚本是可选的**：读路径三层解析的最后一层就是**代码内置默认** （`app/dq_rule_config.py` 的 SPEC/GLOBAL_SPEC，与清单 §7 建议默认值逐项一致）， 因此**不落库也能上线且行为相同**（零行为差异）。落库的价值只有两个： 1. 让「全账套模板」显式可见（UI 的「来源」列由 `内置默认` 变成 `全账套模板`）； 2. 给「恢复出厂默认 / 当前值 vs 默认值 diff」留一行可审计的锚（`is_default=1`）。 幂等契约 -------- 首跑 --apply：created = 39（缺失项），skipped = 0 二跑 --apply：created = 0，skipped = 39，零副作用（**不覆盖**已存在的行） 已存在且值与默认**不同** → 只报告 diff（`differs`），**不覆盖**（`--force` 才覆盖） 边界 ---- * 生产写：需要在**审批门**放行后执行（`approval-gate`）；不改表结构（DDL 由 `backend/scripts/db_migrate.py --apply` 的部署链特权步负责）。 * 连接：与 db_migrate 同口径 —— `--db-url` > `CMA_MIGRATE_DB_USER/PASS`（TCP）> 本机 root socket； 显式拒绝应用运行账号（`cma_api`）。 * 回滚：`--rollback latest` 只删本次 --apply 记录在 journal 里的行 id（零 `DELETE` 全表）。 用法 ---- python3 scripts/seed_dq_rule_config.py --check # 只读预检（默认动作） python3 scripts/seed_dq_rule_config.py --apply # 落默认行 + 写 journal python3 scripts/seed_dq_rule_config.py --apply --force # 已存在且不同的行也覆盖（慎用） python3 scripts/seed_dq_rule_config.py --rollback latest """ import argparse import json import os import socket import sys from datetime import datetime sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from sqlalchemy import create_engine, text # noqa: E402 HERE = os.path.dirname(os.path.abspath(__file__)) DEFAULT_JOURNAL_DIR = os.path.join(HERE, "backups", "dq-rule-config-seed-journal") DEFAULT_SOCKET = "/var/run/mysqld/mysqld.sock" DEFAULT_DB = "cma" FORBIDDEN_RUN_ACCOUNTS = {"cma_api"} TABLE = "quality_rule_config" def _db_name(args) -> str: return (args.db_name or os.getenv("CMA_MIGRATE_DB_NAME") or os.getenv("CMA_DB_NAME") or DEFAULT_DB) def build_engine(args): db_name = _db_name(args) if args.db_url: url = args.db_url if "/" in args.db_url.rsplit("@", 1)[-1] else args.db_url.rstrip("/") + "/" + db_name return create_engine(url, pool_pre_ping=True), f"db-url({url})" user = os.getenv("CMA_MIGRATE_DB_USER", "").strip() if user: if user in FORBIDDEN_RUN_ACCOUNTS: raise SystemExit(f"[FAIL] {user} 是应用运行账号，不得用于本脚本；请用专用账号或 root socket。") pwd = os.getenv("CMA_MIGRATE_DB_PASS", "") host = os.getenv("CMA_MIGRATE_DB_HOST", "127.0.0.1") port = os.getenv("CMA_MIGRATE_DB_PORT", "3306") url = f"mysql+pymysql://{user}:{pwd}@{host}:{port}/{db_name}?charset=utf8mb4" return create_engine(url, pool_pre_ping=True), f"tcp({user}@{host}:{port}/{db_name})" sock = args.socket or os.getenv("CMA_MIGRATE_DB_SOCKET") or DEFAULT_SOCKET if not os.path.exists(sock): raise SystemExit(f"[FAIL] 未设 CMA_MIGRATE_DB_USER 且 socket 不存在：{sock}") url = f"mysql+pymysql://root@localhost/{db_name}?unix_socket={sock}&charset=utf8mb4" return create_engine(url, pool_pre_ping=True), f"socket(root@{sock}/{db_name})" def _defaults(): """(rule_key, param_key, value, value_type, open_level, desc) 全量默认 —— 单一来源 = app/dq_rule_config。""" from app import dq_rule_config as dqcfg out = [] for rule in dqcfg.RULE_KEYS: for param, spec in dqcfg.SPEC[rule].items(): out.append((rule, param, spec.get("default"), spec.get("type"), spec.get("open_level"), spec.get("desc"))) for param, spec in dqcfg.GLOBAL_SPEC.items(): out.append((dqcfg.GLOBAL_RULE_KEY, param, spec.get("default"), spec.get("type"), spec.get("open_level"), spec.get("desc"))) # 规则级 detail_limit 的出厂值 = 全局 detail_limit（内建默认口径） return out def _journal_dir() -> str: d = os.getenv("CMA_DQ_SEED_JOURNAL_DIR") or DEFAULT_JOURNAL_DIR os.makedirs(d, exist_ok=True) return d def cmd_check(args, engine, conn_desc) -> int: rows = _defaults() missing, same, differs = [], [], [] with engine.connect() as conn: if not conn.dialect.has_table(conn, TABLE): print(f"[check] 连接 {conn_desc} → 表 {TABLE} **不存在**：" f"请先跑 `backend/scripts/db_migrate.py --apply`（D 部署链特权迁移步）。") return 0 for rule, param, value, vtype, level, desc in rows: got = conn.execute(text( f"SELECT param_value FROM {TABLE} WHERE entity_id=0 AND rule_key=:r AND param_key=:p"), {"r": rule, "p": param}).scalar() if got is None: missing.append((rule, param, value)) elif json.loads(got) == value: same.append((rule, param)) else: differs.append((rule, param, json.loads(got), value)) print(f"[check] 连接 {conn_desc}") print(f"[check] 待插入 {len(missing)} / 已一致 {len(same)} / 值不同 {len(differs)}（共 {len(rows)} 项）") for rule, param, value in missing[:10]: print(f" + {rule}.{param} = {value}") for rule, param, have, want in differs: print(f" ! {rule}.{param} 库内={have} 默认={want}（不加 --force 不覆盖）") return 0 def cmd_apply(args, engine, conn_desc) -> int: rows = _defaults() created, skipped, differs, failed = [], [], [], [] with engine.begin() as conn: if not conn.dialect.has_table(conn, TABLE): print(f"[FAIL] 表 {TABLE} 不存在 —— 先跑 db_migrate.py --apply") return 1 for rule, param, value, vtype, level, desc in rows: payload = json.dumps(value, ensure_ascii=False) cur = conn.execute(text( f"SELECT id, param_value FROM {TABLE} WHERE entity_id=0 AND rule_key=:r AND param_key=:p"), {"r": rule, "p": param}).fetchone() if cur is None: res = conn.execute(text( f"INSERT INTO {TABLE} (entity_id, rule_key, param_key, param_value, value_type," f" open_level, is_default, description, updated_by)" f" VALUES (0, :r, :p, :v, :t, :l, 1, :d, 'seed:ASK-20260928-05')"), {"r": rule, "p": param, "v": payload, "t": vtype, "l": level, "d": (desc or "")[:500]}) created.append({"id": int(res.lastrowid), "rule_key": rule, "param_key": param, "value": value}) elif json.loads(cur[1]) == value: skipped.append((rule, param)) elif args.force: conn.execute(text(f"UPDATE {TABLE} SET param_value=:v, is_default=1," f" updated_by='seed:ASK-20260928-05' WHERE id=:i"), {"v": payload, "i": cur[0]}) failed.append({"id": int(cur[0]), "rule_key": rule, "param_key": param, "before": json.loads(cur[1]), "after": value, "forced": True}) else: differs.append((rule, param, json.loads(cur[1]), value)) payload = { "action": "seed_dq_rule_config", "status": "applied", "at": datetime.now().isoformat(timespec="seconds"), "connection": conn_desc, "db_name": _db_name(args), "created": created, "created_count": len(created), "skipped_count": len(skipped), "differs": differs, "forced": failed, "actor": f"{os.getenv('USER', 'unknown')}@{socket.gethostname()}", } # 文件名带微秒：同一秒内连跑两次不会互相覆盖（离线回归实证） path = os.path.join(_journal_dir(), f"{datetime.now().strftime('%Y%m%dT%H%M%S-%f')}-applied.json") with open(path, "w", encoding="utf-8") as fh: json.dump(payload, fh, ensure_ascii=False, indent=2) print(f"[apply] 连接 {conn_desc}") print(f"[apply] 新增 {len(created)} / 跳过(已一致) {len(skipped)} / 值不同未覆盖 {len(differs)}" f" / 强制覆盖 {len(failed)}") for rule, param, have, want in differs: print(f" ! {rule}.{param} 库内={have} ≠ 默认={want}（未覆盖）") print(f"[apply] journal: {path}") return 0 def cmd_rollback(args, engine, conn_desc) -> int: d = _journal_dir() ref = args.rollback if ref in ("latest", ""): cands = sorted(f for f in os.listdir(d) if f.endswith("-applied.json")) if not cands: raise SystemExit("[FAIL] 没有可回滚的 journal") path = os.path.join(d, cands[-1]) else: path = ref if os.path.exists(ref) else os.path.join(d, os.path.basename(ref)) payload = json.load(open(path, encoding="utf-8")) ids = [c["id"] for c in payload.get("created", [])] print(f"[rollback] journal={os.path.basename(path)} 将删除 {len(ids)} 行（仅本次新增的 id）") if not ids: print("[rollback] 无新增行，零动作") return 0 with engine.begin() as conn: res = conn.execute(text(f"DELETE FROM {TABLE} WHERE id IN ({','.join(str(int(i)) for i in ids)})")) print(f"[rollback] deleted={res.rowcount}（连接 {conn_desc}）") return 0 def main() -> int: ap = argparse.ArgumentParser(description="quality_rule_config 出厂默认值落库（ASK-20260928-05）") g = ap.add_mutually_exclusive_group() g.add_argument("--check", action="store_true", help="只读预检（默认动作）") g.add_argument("--apply", action="store_true", help="落默认行 + journal") g.add_argument("--rollback", metavar="REF", nargs="?", const="latest", help="回滚最近一次 --apply 新增的行（REF=latest 或 journal 文件名）") ap.add_argument("--force", action="store_true", help="已存在且值不同时也覆盖（慎用）") ap.add_argument("--json", action="store_true") ap.add_argument("--socket", default=None) ap.add_argument("--db-url", default=None) ap.add_argument("--db-name", default=None) args = ap.parse_args() engine, desc = build_engine(args) if args.apply: return cmd_apply(args, engine, desc) if args.rollback is not None: return cmd_rollback(args, engine, desc) return cmd_check(args, engine, desc) if __name__ == "__main__": sys.exit(main())