cma-management/backend/scripts/migrate_analysis_results_entity_id_20260923.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""`analysis_results` 账套化 DDL — 2026-09-23

任务：`0-P1-project-20260923-entity-dimension-convergence-fullstack` §三④
（裁定件 `project-entity-dimension-ruling-20260923` §3.2「读侧先止血、写侧随 DDL」）

背景
----
`analysis_results` 只有 2 行（`id=1 period=2026-06 kpi_code=TEST001` /
`id=2 period=2026-07 kpi_code=NULL`，均 2026-07-29 创建），**无账套列**；
`GET/POST/DELETE /api/cma/analysis/result*` 三端点原无 `get_entity_id`
⇒ 双 token 返回体**逐字节相同**（跨账套读，页面 `AnalysisConfidence.vue` 已上线）。

本脚本只做 DDL（加列 + 索引），不做任何数据订正：
  · 存量 2 行**保持 `entity_id = NULL`** —— 归属存疑，按裁定件交 DAMA 取证 / 财务确认后另办；
  · 代码侧口径（`app/api/analysis_results.py`）：`NULL` 行 = 无归属，进 `unattributed` 登记池，
    不参与任何账套过滤；列落地后读侧自动从 `kpi_code JOIN` 止血口径切到列口径。

用法
----
  venv/bin/python3 scripts/migrate_analysis_results_entity_id_20260923.py --check
  CMA_DDL_APPROVED=1 venv/bin/python3 scripts/migrate_analysis_results_entity_id_20260923.py \\
      --apply --allow-prod
  CMA_DDL_APPROVED=1 venv/bin/python3 scripts/migrate_analysis_results_entity_id_20260923.py \\
      --down --allow-prod --yes
  venv/bin/python3 scripts/migrate_analysis_results_entity_id_20260923.py --show-create

生产保护（生产资产派单纪律）：目标库名 == 'cma' 时 `--apply/--down` 一律拒绝；
须 `--allow-prod` **且** `CMA_DDL_APPROVED=1`（= 已拿到 ops 审批单）才放行；
`--check/--show-create` 为只读，不受限但会打标。
"""
from __future__ import annotations

import argparse
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from sqlalchemy import create_engine, inspect, text  # noqa: E402

TABLE = "analysis_results"
COLUMN = "entity_id"
NEW_INDEX = "ix_analysis_results_entity_id"
PROD_DB_NAMES = ("cma",)
DDL_ADD_COLUMN = (
    f"ALTER TABLE {TABLE} ADD COLUMN {COLUMN} INT NULL "
    "COMMENT '企业ID（NULL=历史未归属行，不参与账套过滤）' AFTER id"
)
DDL_DROP_COLUMN = f"ALTER TABLE {TABLE} DROP COLUMN {COLUMN}"


def is_sqlite(engine) -> bool:
    """沙箱演练（SQLite）与生产（MySQL）的索引/列 DDL 方言差异"""
    return engine.dialect.name == "sqlite"


def stmt_add_column(engine) -> str:
    """生产形态 = MySQL（带 COMMENT / AFTER）；沙箱 = SQLite 精简形态（语法不支持二者）"""
    if is_sqlite(engine):
        return f"ALTER TABLE {TABLE} ADD COLUMN {COLUMN} INT NULL"
    return DDL_ADD_COLUMN


def stmt_add_index(engine, name: str, cols: list) -> str:
    cols_s = ", ".join(cols)
    if is_sqlite(engine):
        return f"CREATE INDEX {name} ON {TABLE} ({cols_s})"
    return f"ALTER TABLE {TABLE} ADD INDEX {name} ({cols_s})"


def stmt_drop_index(engine, name: str) -> str:
    return f"DROP INDEX {name}" if is_sqlite(engine) else f"ALTER TABLE {TABLE} DROP INDEX {name}"


def db_name_of(engine) -> str:
    try:
        return engine.url.database or ""
    except Exception:
        return ""


def state(engine) -> dict:
    insp = inspect(engine)
    tables = set(insp.get_table_names())
    out = {"table_exists": TABLE in tables, "has_column": False, "indexes": [], "rows": []}
    if not out["table_exists"]:
        return out
    out["has_column"] = COLUMN in {c["name"] for c in insp.get_columns(TABLE)}
    out["indexes"] = sorted({i["name"] for i in insp.get_indexes(TABLE)})
    with engine.connect() as conn:
        out["total"] = conn.execute(text(f"SELECT COUNT(*) FROM {TABLE}")).scalar()
        if out["has_column"]:
            rows = conn.execute(text(
                f"SELECT id, period, kpi_code, IFNULL({COLUMN},'NULL') FROM {TABLE} ORDER BY id")).fetchall()
        else:
            rows = conn.execute(text(
                f"SELECT id, period, kpi_code, 'NO_COLUMN' FROM {TABLE} ORDER BY id")).fetchall()
        out["rows"] = [[r[0], r[1], r[2], str(r[3])] for r in rows]
    return out


def print_check(st: dict) -> None:
    print("== 只读预检（inspector / SELECT，零写入）==")
    print(f"  table_exists={st['table_exists']} has_column={st['has_column']} total={st.get('total')}")
    print(f"  indexes={st['indexes']}")
    for r in st.get("rows", []):
        print(f"    id={r[0]} period={r[1]} kpi_code={r[2]} entity_id={r[3]}")


def do_apply(engine) -> int:
    st = state(engine)
    if not st["table_exists"]:
        print("!!! 目标库没有 analysis_results 表，拒绝 --apply")
        return 3
    done = []
    with engine.begin() as conn:
        if not st["has_column"]:
            add_col = stmt_add_column(engine)
            conn.execute(text(add_col))
            done.append(add_col)
        cur = state(engine)
        if NEW_INDEX not in cur["indexes"]:
            add_idx = stmt_add_index(engine, NEW_INDEX, [COLUMN])
            conn.execute(text(add_idx))
            done.append(add_idx)
    print("== APPLY 完成，执行语句：")
    for s in done:
        print("   " + s)
    if not done:
        print("   （无语句执行：已是目标形态，幂等）")
    print_check(state(engine))
    return 0


def do_down(engine, yes: bool) -> int:
    st = state(engine)
    if not st["has_column"]:
        print("== DOWN 跳过：entity_id 列不存在（已是回滚态）")
        return 0
    assigned = [r for r in st.get("rows", []) if r[3] not in ("NULL", "NO_COLUMN")]
    if assigned and not yes:
        print(f"!!! 已有 {len(assigned)} 行带账套归属 {assigned}：回滚会丢失归属，须显式 --yes")
        return 3
    with engine.begin() as conn:
        if NEW_INDEX in st["indexes"]:
            conn.execute(text(stmt_drop_index(engine, NEW_INDEX)))
        conn.execute(text(DDL_DROP_COLUMN))
    print("== DOWN 完成：删除 entity_id 索引 + 列")
    print_check(state(engine))
    return 0


def main():
    ap = argparse.ArgumentParser(description="analysis_results 账套化 DDL")
    g = ap.add_mutually_exclusive_group()
    g.add_argument("--check", action="store_true", help="只读预检（默认）")
    g.add_argument("--apply", action="store_true", help="执行 DDL（幂等）")
    g.add_argument("--down", action="store_true", help="回滚：删索引 + 删列")
    ap.add_argument("--show-create", action="store_true", help="打印 SHOW CREATE TABLE")
    ap.add_argument("--url", default=None, help="目标库 SQLAlchemy URL（默认 backend/.env）")
    ap.add_argument("--allow-prod", action="store_true", help="生产库放行开关（须 CMA_DDL_APPROVED=1）")
    ap.add_argument("--yes", action="store_true", help="--down 有账套归属行时的确认开关")
    args = ap.parse_args()

    if args.url:
        engine = create_engine(args.url, echo=False)
    else:
        from app.database import get_engine
        engine = get_engine()

    dbname = db_name_of(engine)
    print(f"目标库: {dbname or '(未知)'} | dialect={engine.dialect.name} | host={engine.url.host or 'socket'}")
    is_prod = dbname in PROD_DB_NAMES

    if (args.apply or args.down) and is_prod:
        if not args.allow_prod or os.getenv("CMA_DDL_APPROVED") != "1":
            print(f"!!! 拒绝执行：目标库是生产库 {dbname}。")
            print("!!! 生产 DDL 须 ops 审批后另派执行（本单边界：不得直连生产写入）。")
            print("!!! 放行方式：--allow-prod 且 export CMA_DDL_APPROVED=1（持有审批单）。")
            return 3

    if args.apply:
        return do_apply(engine)
    if args.down:
        return do_down(engine, args.yes)

    st = state(engine)
    print_check(st)
    if is_prod:
        print("  [注意] 目标是生产库：以上仅为只读预检，未做任何写入。")
    if args.show_create:
        if engine.dialect.name != "mysql":
            print("== SHOW CREATE TABLE 跳过（非 MySQL 目标）==")
        else:
            with engine.connect() as conn:
                print(conn.execute(text(f"SHOW CREATE TABLE {TABLE}")).fetchone()[1])
    return 0


if __name__ == "__main__":
    sys.exit(main())