#!/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.` · 读：只读本账套键；本账套未配置 → `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())