cma-management/backend/scripts/migrate_cash_balance_key_entity_20260922.py

#!/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.<entity_id>
    role_permissions.alignment_config    → alignment_config.<entity_id>

代码侧（读写）已改为**只认带账套后缀的新键**（方案 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())