cma-management/backend/scripts/migrate_ext_input_ledgers_20260915.py

#!/usr/bin/env python3
"""外部输入台账 DDL 迁移 — tax_filing_log / hr_monthly_ledger / licence_ledger
（任务 0-P0-project-20260915-cma-eval-m2-ext-input-tables，2026-09-15）

DDL 真源 = scripts/sql/ext_input_ledgers_20260915.sql（单一来源，本脚本读取执行，勿复制 DDL）

动作：
  --check        只读预检（默认）：三表存在性 + 缺列 + 唯一键，零写入
  --apply        执行 CREATE TABLE IF NOT EXISTS（幂等：二次运行新增=0）
  --down         回滚：DROP 三表（表内已有数据须显式 --yes）
  --show-create  打印 SHOW CREATE TABLE 原文（验收证据；非 MySQL 目标自动跳过）
  --url URL      目标库 SQLAlchemy URL 覆盖（默认 app.database.get_engine() = backend/.env）

生产保护（生产资产派单纪律 + 本单「不得直连生产写入」）：
  目标库名 == 'cma' 时，--apply/--down **一律拒绝**；须 --allow-prod 且环境变量
  CMA_DDL_APPROVED=1（= 已拿到审批单）才放行。--check/--show-create 为只读，不受限但会打标。

用法：
  venv/bin/python3 scripts/migrate_ext_input_ledgers_20260915.py --check
  venv/bin/python3 scripts/migrate_ext_input_ledgers_20260915.py --apply \
      --url "mysql+pymysql://root@localhost/cma_scratch_x?unix_socket=/var/run/mysqld/mysqld.sock&charset=utf8mb4"
"""
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

TABLES = ("tax_filing_log", "hr_monthly_ledger", "licence_ledger")
SQL_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "sql",
                        "ext_input_ledgers_20260915.sql")
PROD_DB_NAMES = ("cma",)


def load_statements() -> list:
    """读取 .sql 真源 → 逐条语句（去注释/空行），返回可执行 DDL 列表"""
    with open(SQL_FILE, "r", encoding="utf-8") as f:
        raw = f.read()
    lines = [ln for ln in raw.splitlines() if not ln.strip().startswith("--")]
    cleaned = "\n".join(lines)
    stmts = [s.strip() for s in cleaned.split(";") if s.strip()]
    creates = [s for s in stmts if s.upper().startswith("CREATE TABLE")]
    if len(creates) != len(TABLES):
        raise RuntimeError(f"DDL 真源语句数异常: 期望 {len(TABLES)} 条 CREATE TABLE，"
                           f"实际 {len(creates)} 条（{SQL_FILE}）")
    return creates


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


def describe(engine) -> dict:
    """只读预检：返回 {table: {'exists': bool, 'missing_cols': [...], 'unique_keys': [...]}}"""
    insp = inspect(engine)
    existing_tables = set(insp.get_table_names())
    out = {}
    for t in TABLES:
        info = {"exists": t in existing_tables, "missing_cols": [], "unique_keys": []}
        if info["exists"]:
            cols = {c["name"] for c in insp.get_columns(t)}
            expected = EXPECTED_COLUMNS[t]
            info["missing_cols"] = [c for c in expected if c not in cols]
            info["unique_keys"] = [tuple(uc["column_names"]) for uc in insp.get_unique_constraints(t)]
        out[t] = info
    return out


EXPECTED_COLUMNS = {
    "tax_filing_log": ("id", "entity_id", "period", "tax_type", "legal_deadline", "filed_date",
                       "on_time", "evidence_no", "registered_by", "remark", "created_at", "updated_at"),
    "hr_monthly_ledger": ("id", "entity_id", "period", "headcount_begin", "hires", "separations",
                          "headcount_end", "registered_by", "remark", "created_at", "updated_at"),
    "licence_ledger": ("id", "entity_id", "licence_name", "licence_no", "valid_from", "valid_to",
                       "annual_check_status", "issuer", "owner_role", "attachment_no",
                       "registered_by", "remark", "created_at", "updated_at"),
}


def print_check(state: dict) -> None:
    print("== 只读预检（information_schema / inspector，零写入）==")
    for t in TABLES:
        s = state[t]
        print(f"  {t}: exists={s['exists']} missing_cols={s['missing_cols']} unique_keys={s['unique_keys']}")


def main():
    ap = argparse.ArgumentParser(description="外部输入台账 DDL 迁移")
    g = ap.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="回滚：DROP 三表")
    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} | url_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("!!! 拒绝执行：目标库是生产库 %s。" % dbname)
            print("!!! 生产 DDL 须审批后另派执行（本单边界：不得直连生产写入）。")
            print("!!! 如需放行：--allow-prod 且 export CMA_DDL_APPROVED=1（持有审批单）。")
            return 3

    if args.apply:
        stmts = load_statements()
        with engine.begin() as conn:
            for s in stmts:
                conn.execute(text(s))
        state = describe(engine)
        missing = [t for t in TABLES if not state[t]["exists"]]
        print(f"== APPLY 完成：DDL 真源 {os.path.basename(SQL_FILE)}，语句 {len(stmts)} 条 ==")
        print_check(state)
        print(f"  仍缺表: {missing}（二次运行应为 0 新增表）")
        return 1 if missing else 0

    if args.down:
        counts = {}
        with engine.connect() as conn:
            for t in TABLES:
                if t in set(inspect(engine).get_table_names()):
                    counts[t] = conn.execute(text(f"SELECT COUNT(*) FROM {t}")).scalar()
        if any(counts.values()) and not args.yes:
            print(f"!!! 表内已有数据 {counts}：回滚须显式 --yes（防误删业务数据）")
            return 3
        with engine.begin() as conn:
            for t in TABLES:
                conn.execute(text(f"DROP TABLE IF EXISTS {t}"))
        print(f"== DOWN 完成：DROP {list(TABLES)}（回滚前行数 {counts}）==")
        print_check(describe(engine))
        return 0

    # --check（默认）或 --show-create
    state = describe(engine)
    print_check(state)
    if is_prod:
        print("  [注意] 目标是生产库：以上仅为只读预检，未做任何写入。")
    if args.show_create:
        if engine.dialect.name != "mysql":
            print("== SHOW CREATE TABLE 跳过（非 MySQL 目标）==")
        else:
            with engine.connect() as conn:
                for t in TABLES:
                    if not state[t]["exists"]:
                        print(f"== {t}: 表不存在，无 SHOW CREATE 输出 ==")
                        continue
                    row = conn.execute(text(f"SHOW CREATE TABLE {t}")).fetchone()
                    print(f"== SHOW CREATE TABLE {t} ==")
                    print(row[1])
    return 0


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