cma-management/backend/scripts/migrate_higher_better_key_entity_20260923.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""全局键拆账套 — `kpi_alert_higher_better` 一次性搬运（数据迁移，非 DDL）

任务：`dama-entity-isolation-sweep-20260923` §一①（DAMA 独立回查新增漏项，任总 2026-09-23 批）

背景
----
`system_configs.config_key` 为 **unique 单行键** ⇒ `kpi_alert_higher_better` 全库只有一行，
被所有账套共用。消费点（拆键前，读的都是那唯一一行）：

    backend/app/utils/deviation_engine.py   get_higher_better_codes()   L32/L37-38
                                             build_deviation_alert()      L305
    backend/app/api/budget.py               GET  /alert-direction        L1087
                                             PUT  /alert-direction        L1114-1125

任一套账写一次，**方向判定（实际低于预算才告警 vs 实际超预算才告警）全平台翻转**。

治本（方案 A，零 DDL）：键名带账套后缀 → `kpi_alert_higher_better.<entity_id>`
    · 读：只读本账套键；本账套未配置 → `DEFAULT_HIGHER_BETTER`（内置默认表，兜底），
      **不回退旧全局键**（回退即跨账套借用）。
    · 写：`PUT /alert-direction` 只写会话账套键，body 里的 entity_id 不信。

本脚本 = 旧值一次性搬运。

⚠️ 生产实况（2026-09-23 只读核对）：**库内无 `kpi_alert_higher_better` 行**
   ⇒ `--check` 报 `legacy=None`，`--apply` 走 `skip`（无可搬运值），零写入。
   即：拆键后各账套行为与拆键前**逐字一致**（都走内置默认表），不存在「其余账套留空
   导致行为退化」的问题。若将来出现旧全局行（例如先于本单部署的 PUT 写入过），
   本脚本可直接把它落到 `--target-entity`。

用法
----
    # 只读预检（默认动作，零写入）
    python3 scripts/migrate_higher_better_key_entity_20260923.py --check

    # 写库（**需审批 / 生产写**；--apply 才写）
    python3 scripts/migrate_higher_better_key_entity_20260923.py --apply --target-entity 1

    # 回退（删新键；旧全局键原样保留）
    python3 scripts/migrate_higher_better_key_entity_20260923.py --down

退出码
------
    0 成功 ｜ 1 执行失败 ｜ 2 用法/保护拒绝（库名不符、目标账套不存在）

⚠️ 本脚本写的是**生产业务库**：不得在开发级会话直接 `--apply`。
   依据《生产资产派单纪律》：碰生产资产一律走派单 + 审批；`--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

KEY = "kpi_alert_higher_better"            # 旧全局键（保留不删）
PROD_DB_NAME = "cma"                       # 生产库名（保护口径）
DESC = "越高越好型KPI编码列表（实际低于预算才告警）— 账套 {eid}"


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 _entity_evidence(conn):
    """逐账套只读证据：KPI 定义数 / 预算行数（供「原值该落哪家」人工确认）"""
    out = {}
    try:
        kpi_rows = conn.execute(text(
            "SELECT entity_id, COUNT(*) FROM kpi_definitions GROUP BY entity_id ORDER BY entity_id"
        )).fetchall()
        bud_rows = conn.execute(text(
            "SELECT kd.entity_id, COUNT(*) FROM budget_plans bp "
            "JOIN kpi_definitions kd ON kd.id = bp.kpi_id GROUP BY kd.entity_id ORDER BY kd.entity_id"
        )).fetchall()
        kpi = {int(r[0]) for r in kpi_rows if r[0] is not None}
        bud = {int(r[0]): int(r[1]) for r in bud_rows if r[0] is not None}
        for eid in sorted(kpi | set(bud)):
            out[eid] = {"budget_plans": bud.get(eid, 0)}
    except Exception as e:  # 表缺失/权限 → 只报告，不阻断
        out["_error"] = str(e)[:300]
    return out


def _snapshot(conn):
    snap = {
        "db": _detect_db_name(conn),
        "legacy_key": KEY,
        "legacy": None,
        "new_keys": {},
        "entities": [],
        "entity_evidence": _entity_evidence(conn),
        "suggested_target_entity": None,
    }
    row = conn.execute(text(
        "SELECT config_value, description, updated_at FROM system_configs WHERE config_key = :k"
    ), {"k": KEY}).fetchone()
    if row:
        snap["legacy"] = {"config_value": row[0], "description": row[1], "updated_at": str(row[2])}

    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": KEY + ".%"}).fetchall():
        snap["new_keys"][r[0]] = r[1]

    # 「原值落哪家」：旧全局值唯一时，若只有一家账套是 active 生产账套 → 建议它；
    # 多 active 账套（e1 酣客 + e2 博海）⇒ 不给自动建议，需人工裁定（写进报告）。
    active = [e["id"] for e in snap["entities"] if e["status"] == "active"]
    if snap["legacy"] and len(active) == 1:
        snap["suggested_target_entity"] = active[0]
    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"旧全局键 {KEY} : {snap['legacy']}")
    print(f"已存在新键: {snap['new_keys']}")
    print("逐账套只读证据（预算行数）:")
    for eid, info in snap["entity_evidence"].items():
        print(f"  entity {eid}: {info}")
    print(f"建议 --target-entity : {snap['suggested_target_entity']}"
          f"（None = 无旧值 或 多 active 账套需人工裁定）")
    print("【只读预检，零写入】")
    return 0


def _apply(conn, target: int, force: bool):
    new_key = f"{KEY}.{target}"
    row = conn.execute(text(
        "SELECT config_value FROM system_configs WHERE config_key = :k"
    ), {"k": KEY}).fetchone()
    if not row:
        return {"key": new_key, "action": "skip", "reason": "旧全局键不存在（无可搬运值）"}

    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 exist:
        conn.execute(text(
            "UPDATE system_configs SET config_value = :v, description = :d WHERE config_key = :k"
        ), {"v": row[0], "d": DESC.format(eid=target), "k": new_key})
        return {"key": new_key, "action": "updated", "value": row[0]}
    conn.execute(text(
        "INSERT INTO system_configs (config_key, config_value, description) VALUES (:k, :v, :d)"
    ), {"k": new_key, "v": row[0], "d": DESC.format(eid=target)})
    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()
    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 = [_apply(conn, args.target_entity, args.force)]
    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()
    with eng.begin() as conn:
        db = _guard_db(args, conn)
        r = conn.execute(text("DELETE FROM system_configs WHERE config_key LIKE :p"),
                         {"p": KEY + ".%"})
        removed = [f"system_configs({KEY}.*) deleted={r.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="全局键拆账套 · 一次性搬运（kpi_alert_higher_better）")
    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())