#!/usr/bin/env python3 # -*- coding: utf-8 -*- """全局键拆账套 — 一次性搬运（数据迁移，非 DDL） 任务：`dama-workbench-entity-isolation-fullstack-20260922` §三 B1/B3(a) （DAMA 诊断 / 项目Bot batch5 §B 方案段 / 任总 2026-09-22「都批」） 背景 ---- `system_configs.config_key` 与 `role_permissions.key` 均为 **unique 单行键** ⇒ 下列两个「全局键」全库只有一行，被所有账套共用（本次「酣客工作台显示博海数据」 污染的同源结构性缺陷）： 键 处置 ------------------------------- ------------------------------------------ system_configs.cash.current_balance → cash.current_balance. role_permissions.alignment_config → alignment_config. 代码侧（读写）已改为**只认带账套后缀的新键**（方案 A，零 DDL）： backend/app/utils/cash_forecast_engine.py （get_/set_current_cash_balance） backend/app/api/cash.py （POST /balance、网银流水联动） backend/app/api/alignment.py （GET/POST /config、GET /tree） 本脚本 = **旧值一次性搬运**（把原来唯一的那一行值落到「当前生产账套」名下）。 「当前生产账套」判定：默认 `--target-entity 1`（entity 1 = 陕西酣客，生产账套）， `--check` 会打印**逐账套凭证推导的货币资金期末余额**与旧键值的比对，供人工确认 （哪个账套的 1001/1002 期末余额 == 旧键值 ⇒ 该值本来就是它写的）。 口径（须与代码注释一致） ------------------------ · 新键值 = 旧键值（一次性搬运）**只落到 target-entity**；**其余账套留空** ⇒ 留空语义 = 该账套现金基线走 KPI 实际值兜底 / 预测引擎默认值， **不跨账套借用**（正是本次污染的成因）。 · 旧键**原样保留不删**（回退杠杆：回退老代码即可继续读旧键）。 · alignment_config 同理：旧全局值落到 target-entity；其余账套 configured=false。 · 幂等：新键已存在 → 不覆盖（除非 `--force`）；二跑 `--apply` 零副作用。 用法 ---- # 只读预检（默认动作，零写入） python3 scripts/migrate_cash_balance_key_entity_20260922.py --check # 写库（**需审批 / 生产写**；--apply 才写） python3 scripts/migrate_cash_balance_key_entity_20260922.py --apply python3 scripts/migrate_cash_balance_key_entity_20260922.py --apply --target-entity 2 python3 scripts/migrate_cash_balance_key_entity_20260922.py --apply --force # 覆盖已存在的新键 # 回退（删新键 + 把旧全局键写回原值） python3 scripts/migrate_cash_balance_key_entity_20260922.py --down 选项 ---- --target-entity N 旧值落哪个账套（默认 1；--check 会给出证据建议） --force 新键已存在时允许覆盖（默认拒绝，保留既有值） --json 机器可读输出 --allow-db NAME 生产库名保护豁免（默认只允许库名 `cma`） 退出码 ------ 0 成功 ｜ 1 执行失败 ｜ 2 用法/保护拒绝（库名不符、目标账套不存在） ⚠️ 本脚本写的是**生产业务库**：不得在开发级会话直接 `--apply`。 依据《生产资产派单纪律》（skills-shared/operations/prod-asset-dispatch-discipline）： 碰生产资产一律走派单 + 审批；`--check` 同样是生产读，须留单痕。 """ import argparse import json import os import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from sqlalchemy import text # noqa: E402 CASH_KEY = "cash.current_balance" # 旧全局键（保留不删） ALIGN_KEY = "alignment_config" # 旧全局键（保留不删） PROD_DB_NAME = "cma" # 生产库名（保护口径） CASH_DESC = "当前现金余额（万元），资金预测基线 — 账套 {eid}" ALIGN_DESC = "对齐模式配置 — 账套 {eid}（B3(a) 拆账套 2026-09-22）" def _engine(): from app.database import get_engine return get_engine() def _scalar(conn, sql, **params): row = conn.execute(text(sql), params).fetchone() return row[0] if row else None def _detect_db_name(conn) -> str: try: return _scalar(conn, "SELECT DATABASE()") or "" except Exception: return "" def _cash_ledger_by_entity(conn): """逐账套：货币资金(1001/1002, carry_forward=0) 期末净额（元 → 万元） 与 `cash.py::_sync_cash_balance_from_ledger` 同口径（该函数正是写旧键的调用方） ——用于回答「-9.3 这个值原本是哪家写的」。 """ out = {} try: rows = conn.execute(text( "SELECT entity_id, " " ROUND(SUM(COALESCE(debit_amount,0) - COALESCE(credit_amount,0))/10000, 4) AS wan, " " COUNT(*) AS n " "FROM voucher_details " "WHERE carry_forward = 0 AND subject_code IS NOT NULL " " AND (subject_code LIKE '1001%' OR subject_code LIKE '1002%') " "GROUP BY entity_id ORDER BY entity_id" )).fetchall() for r in rows: out[int(r[0])] = {"cash_wan": float(r[1] or 0), "voucher_rows": int(r[2] or 0)} except Exception as e: # 表缺失/权限 → 只报告，不阻断 out["_error"] = str(e)[:300] return out def _snapshot(conn): """只读快照：新旧键现状 + 逐账套凭证推导余额""" snap = { "db": _detect_db_name(conn), "legacy": { "cash.current_balance": None, "alignment_config": None, }, "new_keys": {"cash": {}, "alignment": {}}, "entities": [], "cash_ledger_by_entity": _cash_ledger_by_entity(conn), } row = conn.execute(text( "SELECT config_value, description FROM system_configs WHERE config_key = :k" ), {"k": CASH_KEY}).fetchone() if row: snap["legacy"]["cash.current_balance"] = {"config_value": row[0], "description": row[1]} row = conn.execute(text( "SELECT value FROM role_permissions WHERE `key` = :k" ), {"k": ALIGN_KEY}).fetchone() if row: snap["legacy"]["alignment_config"] = {"value": row[0]} for r in conn.execute(text( "SELECT id, name, short_name, status FROM entities ORDER BY id" )).fetchall(): snap["entities"].append({"id": int(r[0]), "name": r[1], "short_name": r[2], "status": r[3]}) for r in conn.execute(text( "SELECT config_key, config_value FROM system_configs " "WHERE config_key LIKE :p ORDER BY config_key" ), {"p": CASH_KEY + ".%"}).fetchall(): snap["new_keys"]["cash"][r[0]] = r[1] for r in conn.execute(text( "SELECT `key`, value FROM role_permissions WHERE `key` LIKE :p ORDER BY `key`" ), {"p": ALIGN_KEY + ".%"}).fetchall(): snap["new_keys"]["alignment"][r[0]] = r[1] # 建议目标账套（只读推断）：旧值 == 某账套凭证推导余额 suggestion = None legacy_cash = (snap["legacy"]["cash.current_balance"] or {}).get("config_value") if legacy_cash is not None: try: want = float(legacy_cash) for eid, info in snap["cash_ledger_by_entity"].items(): if isinstance(eid, int) and abs(float(info["cash_wan"]) - want) < 1e-6: suggestion = eid break except Exception: suggestion = None snap["suggested_target_entity"] = suggestion return snap def cmd_check(args): eng = _engine() with eng.connect() as conn: snap = _snapshot(conn) if args.json: print(json.dumps(snap, ensure_ascii=False, indent=2)) return 0 print(f"目标库: {snap['db']}") print(f"旧全局键 cash.current_balance : {snap['legacy']['cash.current_balance']}") print(f"旧全局键 alignment_config : {snap['legacy']['alignment_config']}") print(f"已存在新键(cash) : {snap['new_keys']['cash']}") print(f"已存在新键(alignment) : {snap['new_keys']['alignment']}") print("逐账套凭证推导货币资金期末余额（万元）:") for eid, info in snap["cash_ledger_by_entity"].items(): print(f" entity {eid}: {info}") print(f"建议 --target-entity : {snap['suggested_target_entity']}" f"（旧值与他家凭证推导余额一致时给出；None = 无匹配，按默认 1 或人工裁定）") print("【只读预检，零写入】") return 0 def _apply_cash(conn, target: int, force: bool, dry: bool): new_key = f"{CASH_KEY}.{target}" row = conn.execute(text( "SELECT config_value FROM system_configs WHERE config_key = :k" ), {"k": CASH_KEY}).fetchone() if not row: return {"key": new_key, "action": "skip", "reason": "旧全局键不存在（无可搬运值）"} legacy_value = row[0] exist = conn.execute(text( "SELECT config_value FROM system_configs WHERE config_key = :k" ), {"k": new_key}).fetchone() if exist and not force: return {"key": new_key, "action": "noop", "reason": "新键已存在且未 --force（保留既有值，幂等）", "existing_value": exist[0]} if dry: return {"key": new_key, "action": "would-insert-or-update", "value": legacy_value, "existing_value": exist[0] if exist else None} if exist: conn.execute(text( "UPDATE system_configs SET config_value = :v, description = :d WHERE config_key = :k" ), {"v": legacy_value, "d": CASH_DESC.format(eid=target), "k": new_key}) action = "updated" else: conn.execute(text( "INSERT INTO system_configs (config_key, config_value, description) " "VALUES (:k, :v, :d)" ), {"k": new_key, "v": legacy_value, "d": CASH_DESC.format(eid=target)}) action = "inserted" return {"key": new_key, "action": action, "value": legacy_value} def _apply_alignment(conn, target: int, force: bool, dry: bool): new_key = f"{ALIGN_KEY}.{target}" row = conn.execute(text( "SELECT value FROM role_permissions WHERE `key` = :k" ), {"k": ALIGN_KEY}).fetchone() if not row: return {"key": new_key, "action": "skip", "reason": "旧全局键不存在（无可搬运值）"} exist = conn.execute(text( "SELECT value FROM role_permissions WHERE `key` = :k" ), {"k": new_key}).fetchone() if exist and not force: return {"key": new_key, "action": "noop", "reason": "新键已存在且未 --force（保留既有值，幂等）", "existing_value": exist[0]} if dry: return {"key": new_key, "action": "would-insert", "value": row[0]} conn.execute(text( "INSERT INTO role_permissions (`key`, value) VALUES (:k, :v)" ), {"k": new_key, "v": row[0]}) return {"key": new_key, "action": "inserted", "value": row[0]} def _guard_db(args, conn): db = _detect_db_name(conn) if args.allow_db: return db if db != PROD_DB_NAME: print(f"[拒绝] 库名 {db!r} ≠ 生产库 {PROD_DB_NAME!r}；" f"如确为预期，请显式加 --allow-db {db}", file=sys.stderr) sys.exit(2) return db def cmd_apply(args): eng = _engine() results = [] with eng.begin() as conn: db = _guard_db(args, conn) # 目标账套存在性校验（防写出孤儿键） ent = conn.execute(text("SELECT id FROM entities WHERE id = :i"), {"i": args.target_entity}).fetchone() if not ent: print(f"[拒绝] 目标账套 entity_id={args.target_entity} 不存在", file=sys.stderr) sys.exit(2) results.append(_apply_cash(conn, args.target_entity, args.force, dry=False)) results.append(_apply_alignment(conn, args.target_entity, args.force, dry=False)) out = {"mode": "apply", "db": db, "target_entity": args.target_entity, "results": results} print(json.dumps(out, ensure_ascii=False, indent=2) if args.json else "\n".join(f" {r}" for r in results)) return 0 def cmd_down(args): """回退：删除新键（旧全局键原样保留，老代码即可继续读）""" eng = _engine() removed = [] with eng.begin() as conn: db = _guard_db(args, conn) r1 = conn.execute(text( "DELETE FROM system_configs WHERE config_key LIKE :p"), {"p": CASH_KEY + ".%"}) r2 = conn.execute(text( "DELETE FROM role_permissions WHERE `key` LIKE :p"), {"p": ALIGN_KEY + ".%"}) removed = [f"system_configs(+{CASH_KEY}.*) deleted={r1.rowcount}", f"role_permissions({ALIGN_KEY}.*) deleted={r2.rowcount}"] print(json.dumps({"mode": "down", "db": db, "removed": removed}, ensure_ascii=False, indent=2) if args.json else "\n".join(removed)) print("旧全局键未改动 → 代码回退到老版本即可继续读旧键。") return 0 def main(): p = argparse.ArgumentParser(description="全局键拆账套 · 一次性搬运（cash.current_balance / alignment_config）") g = p.add_mutually_exclusive_group() g.add_argument("--check", action="store_true", help="只读预检（默认）") g.add_argument("--apply", action="store_true", help="写库（需审批）") g.add_argument("--down", action="store_true", help="回退：删除新键") p.add_argument("--target-entity", type=int, default=1, help="旧值落到哪个账套（默认 1 = 陕西酣客，生产账套）") p.add_argument("--force", action="store_true", help="允许覆盖已存在的新键") p.add_argument("--json", action="store_true", help="机器可读输出") p.add_argument("--allow-db", default=None, help="生产库名保护豁免（默认仅允许 cma）") args = p.parse_args() if args.apply: return cmd_apply(args) if args.down: return cmd_down(args) return cmd_check(args) if __name__ == "__main__": sys.exit(main())