cma-management/backend/scripts/db_migrate.py

#!/usr/bin/env python3
"""CMA 数据库结构迁移（特权 DDL 步）—— 与应用启动 create_all 剥离

任务：cma-create-all-minpriv-rootfix-20260916（P2）

背景（2026-09-16 01:08 发布中断事故）
--------------------------------------
应用启动时 `app/database.py::init_db()` 调 `Base.metadata.create_all()`（需要 CREATE 权限），
而运行库账号 `cma_api` 只有 SELECT + 表级 DML 白名单（**无 DDL**）。新代码引入新模型
（`tax_filing_log` / `treasury_*`）→ 启动即 `CREATE command denied` → 崩溃重启循环。

本脚本 = 部署链里的**独立、幂等、可回滚**的特权迁移步，在应用重启**之前**执行：

    deploy.sh [4b]  →  backend/scripts/db_migrate.py --apply  →  systemctl restart cma-backend

于是：DDL 不再随应用 boot 发生；运行库账号可保持最小权限（无需 CREATE）。

用法
----
    python3 scripts/db_migrate.py --check                 # 只读预检（默认动作，零写入）
    python3 scripts/db_migrate.py --apply                 # 幂等建表（缺什么建什么）+ 写 journal
    python3 scripts/db_migrate.py --status                # 列出历史 journal
    python3 scripts/db_migrate.py --rollback latest       # 回滚最近一次 --apply 新建的表
    python3 scripts/db_migrate.py --rollback <journal.json>

选项
----
    --json          机器可读输出（部署链/CI 断言用）
    --force         回滚时允许删除**有数据**的表（默认拒绝，exit 3）
    --no-dump       回滚时不 mysqldump（默认先备份到 journal 目录，dump 失败即拒绝回滚）
    --socket PATH   unix socket 路径（默认 /var/run/mysqld/mysqld.sock）
    --db-url URL    显式连接串（覆盖下方全部推导；不含口令时应配合环境变量）
    --db-name NAME  目标库名（默认 env CMA_MIGRATE_DB_NAME → CMA_DB_NAME → cma）

连接凭据（按优先级；口令只从环境读，禁止硬编码/回显）
----------------------------------------------------
    1) CMA_MIGRATE_DB_USER + CMA_MIGRATE_DB_PASS（**专用迁移账号**，TCP）——推荐用于非本机场景
    2) 未设 CMA_MIGRATE_DB_USER → 本机 unix socket + root（socket 认证，无需口令）
    ⚠️ 应用运行账号（backend/.env 的 CMA_DB_USER，= cma_api）**不得**被本脚本使用：
       迁移必须走在有 DDL 权限的账号上，本脚本会显式拒绝。

幂等契约
--------
    首跑 --apply：created_new = 缺失表数（本次实测：0 —— 库已由应急 create_all 补齐）
    二跑 --apply：created_new = 0，零副作用
    已存在的表**不** ALTER、**不** DROP、**不**动数据行（列级差异只报告，见下）

退出码
------
    0 成功 ｜ 1 迁移失败 / 迁移后自检不一致 / 连接失败（fail-closed，部署链据此中止）
    2 用法错误 ｜ 3 回滚被拒（目标表有数据且未 --force）
"""
import argparse
import fcntl
import json
import os
import random
import socket
import string
import subprocess
import sys
import tempfile
from datetime import datetime

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

from sqlalchemy import create_engine, inspect, text  # noqa: E402
from sqlalchemy.engine.url import make_url  # noqa: E402

HERE = os.path.dirname(os.path.abspath(__file__))
BACKEND_DIR = os.path.dirname(HERE)
DEFAULT_JOURNAL_DIR = os.path.join(HERE, "backups", "db-migrate-journal")
DEFAULT_SOCKET = "/var/run/mysqld/mysqld.sock"
DEFAULT_DB = "cma"
# 应用运行账号：绝不允许用于本迁移步（无 DDL 权限 = 事故根因）
FORBIDDEN_RUN_ACCOUNTS = {"cma_api"}


# ────────────────────────────────────────────────────────────────
# 连接（特权通道）
# ────────────────────────────────────────────────────────────────
def _resolve_db_name(args) -> str:
    return (
        args.db_name
        or os.getenv("CMA_MIGRATE_DB_NAME")
        or os.getenv("CMA_DB_NAME")
        or DEFAULT_DB
    )


def build_engine(args):
    """构造特权连接。返回 (engine, 描述串)。描述串不含口令。"""
    db_name = _resolve_db_name(args)
    if args.db_url:
        url = make_url(args.db_url)
        if url.database is None:
            url = url.set(database=db_name)
        return create_engine(url, pool_pre_ping=True), f"db-url({url.drivername}@{url.host or 'socket'}/{url.database})"

    user = os.getenv("CMA_MIGRATE_DB_USER", "").strip()
    if user:
        if user in FORBIDDEN_RUN_ACCOUNTS:
            raise SystemExit(
                f"[FAIL] CMA_MIGRATE_DB_USER={user} 是**应用运行账号**（无 DDL 权限）——\n"
                f"       迁移步必须使用有 DDL 权限的专用账号（或本机 root socket）。已拒绝执行。"
            )
        password = os.getenv("CMA_MIGRATE_DB_PASS", "")
        host = os.getenv("CMA_MIGRATE_DB_HOST", "127.0.0.1")
        port = int(os.getenv("CMA_MIGRATE_DB_PORT", "3306"))
        url = (
            f"mysql+pymysql://{user}:{password}@{host}:{port}/{db_name}"
            f"?charset=utf8mb4"
        )
        return create_engine(url, pool_pre_ping=True), f"tcp({user}@{host}:{port}/{db_name})"

    # 默认：本机 unix socket + root（socket 认证，无口令落盘）
    sock = args.socket or os.getenv("CMA_MIGRATE_DB_SOCKET") or DEFAULT_SOCKET
    if not os.path.exists(sock):
        raise SystemExit(
            f"[FAIL] 未设 CMA_MIGRATE_DB_USER，且 unix socket 不存在：{sock}\n"
            f"       请提供专用迁移账号（CMA_MIGRATE_DB_USER/PASS）或用 --socket 指定路径。"
        )
    url = f"mysql+pymysql://root@localhost/{db_name}?unix_socket={sock}&charset=utf8mb4"
    return create_engine(url, pool_pre_ping=True), f"socket(root@{sock}/{db_name})"


# ────────────────────────────────────────────────────────────────
# schema 比对（ORM 期望 vs 库内实况）
# ────────────────────────────────────────────────────────────────
def _import_models():
    import app.models  # noqa: F401  触发全部模型注册
    from app.database import Base
    return Base


def schema_diff(engine):
    """返回 (missing_tables, missing_columns{表:[列]})。只读。"""
    Base = _import_models()
    insp = inspect(engine)
    existing = set(insp.get_table_names())
    missing_tables, missing_columns = [], {}
    for table in Base.metadata.sorted_tables:
        if table.name not in existing:
            missing_tables.append(table.name)
            continue
        have = {c["name"] for c in insp.get_columns(table.name)}
        miss = [c.name for c in table.columns if c.name not in have]
        if miss:
            missing_columns[table.name] = miss
    return missing_tables, missing_columns


# ────────────────────────────────────────────────────────────────
# journal（迁移留痕：谁/何时/建了什么 → 回滚依据）
# ────────────────────────────────────────────────────────────────
def journal_dir() -> str:
    d = os.getenv("CMA_MIGRATE_JOURNAL_DIR") or DEFAULT_JOURNAL_DIR
    os.makedirs(d, exist_ok=True)
    return d


def _git_head() -> str:
    try:
        return subprocess.run(
            ["git", "-C", os.path.dirname(BACKEND_DIR), "rev-parse", "--short", "HEAD"],
            capture_output=True, text=True, timeout=10,
        ).stdout.strip() or "unknown"
    except Exception:
        return "unknown"


def write_journal(payload: dict) -> str:
    path = os.path.join(
        journal_dir(), f"{datetime.now().strftime('%Y%m%dT%H%M%S')}-{payload.get('status', 'na')}.json"
    )
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(payload, fh, ensure_ascii=False, indent=2)
    return path

def write_anchor_pointer(payload: dict, journal_path: str) -> str:
    """原子写 journal/last-anchor.json —— 授权面收敛步（grant_surface_sync.py）消费的锚（R1）

    键：journal（本次 journal 文件名，供 R1-a 同源自证）/ status / created_tables / created_new /
    anchor_csv / finished_at / git_head / db_name / actor / carried_over / carried_from /
    written_at / converged_at
    R1-b：已存在指针且其 status=applied 而未收敛（无 converged_at）时，本次 created_tables =
    旧集合 ∪ 本次 created，并记 carried_over / carried_from；无旧指针或旧指针已收敛时 = 本次
    加性写：不改 journal、不改任何既有键语义；失败由调用方按告警处理（收敛步会据此 rc=2 中止）
    """
    d = journal_dir()
    path = os.path.join(d, "last-anchor.json")
    prev = None
    try:
        with open(path, encoding="utf-8") as fh:
            prev = json.load(fh)
    except Exception:
        prev = None
    created = [t for t in (payload.get("created_tables") or [])]
    carried_over, carried_from = False, None
    if (isinstance(prev, dict) and prev.get("status") == "applied"
            and not prev.get("converged_at") and (prev.get("created_tables") or [])):
        old = list(prev.get("created_tables") or [])
        created = sorted(set(old) | set(created))
        carried_from = os.path.basename(str(prev.get("journal") or "")) or None
        carried_over = True
    doc = {
        "pointer": "last-anchor.json",
        "journal": os.path.basename(journal_path),
        "status": payload.get("status"),
        "created_tables": created,
        "created_new": payload.get("created_new"),
        "anchor_csv": ",".join(created),
        "carried_over": carried_over,
        "carried_from": carried_from,
        "finished_at": payload.get("finished_at"),
        "git_head": payload.get("git_head"),
        "db_name": payload.get("db_name"),
        "actor": payload.get("actor"),
        "written_at": datetime.now().isoformat(timespec="seconds"),
        "converged_at": None,
    }
    tmp = "%s.tmp.%d" % (path, os.getpid())
    with open(tmp, "w", encoding="utf-8") as fh:
        json.dump(doc, fh, ensure_ascii=False, indent=2)
        fh.write("\n")
    try:
        ref = path if os.path.exists(path) else journal_path
        if os.path.exists(ref):
            st = os.stat(ref)
            os.chmod(tmp, st.st_mode & 0o777)
            os.chown(tmp, st.st_uid, st.st_gid)
    except Exception:
        pass
    os.replace(tmp, path)
    return path


def list_journals():
    d = journal_dir()
    return sorted(
        (os.path.join(d, f) for f in os.listdir(d) if f.endswith(".json")),
        reverse=True,
    )


def resolve_journal(ref: str) -> str:
    if ref in ("latest", ""):
        items = [p for p in list_journals() if "-applied" in os.path.basename(p)]
        if not items:
            raise SystemExit("[FAIL] 没有可回滚的 journal（无 --apply 记录）")
        return items[0]
    if os.path.exists(ref):
        return ref
    cand = os.path.join(journal_dir(), os.path.basename(ref))
    if os.path.exists(cand):
        return cand
    raise SystemExit(f"[FAIL] journal 不存在：{ref}")


# ────────────────────────────────────────────────────────────────
# 动作
# ────────────────────────────────────────────────────────────────
def cmd_check(args, engine, conn_desc) -> int:
    missing_tables, missing_cols = schema_diff(engine)
    out = {
        "action": "check",
        "connection": conn_desc,
        "missing_tables": missing_tables,
        "missing_table_count": len(missing_tables),
        "missing_columns": missing_cols,
    }
    if args.json:
        print(json.dumps(out, ensure_ascii=False))
    else:
        print(f"[check] 连接: {conn_desc}")
        print(f"[check] 缺失表 {len(missing_tables)} 张: {missing_tables if missing_tables else '（无）'}")
        if missing_cols:
            print(f"[check] 列级差异 {len(missing_cols)} 张表: {missing_cols}")
            print("        ⚠️ create_all 不会补列 → 需要独立的 ALTER 迁移（见 journal/回报）")
        else:
            print("[check] 列级差异：无")
    # 列级差异属"存在但结构过期"：部署链上以告警呈现，不阻断（阻断判据 = 缺表未补齐）
    return 0


def run_structural() -> dict:
    """结构增量（补列/索引/种子/授权）——create_all 覆盖不到。
    复用 `treasury_migrate.py --apply`（幂等；其自身走特权连接）。失败不中断部署，只告警+留痕。"""
    script = os.path.join(HERE, "treasury_migrate.py")
    if not os.path.exists(script):
        return {"status": "skipped", "reason": "treasury_migrate.py 不存在"}
    try:
        proc = subprocess.run([sys.executable, script, "--apply"],
                              capture_output=True, text=True, timeout=300)
    except Exception as exc:
        return {"status": "error", "reason": f"{type(exc).__name__}: {exc}"}
    return {
        "status": "ok" if proc.returncode == 0 else "failed",
        "rc": proc.returncode,
        "stdout_tail": (proc.stdout or "").strip()[-600:],
        "stderr_tail": (proc.stderr or "").strip()[-400:],
    }


def cmd_apply(args, engine, conn_desc) -> int:
    Base = _import_models()
    before_missing, before_cols = schema_diff(engine)
    started = datetime.now().isoformat(timespec="seconds")
    payload = {
        "action": "apply",
        "status": "running",
        "started_at": started,
        "finished_at": None,
        "connection": conn_desc,
        "git_head": _git_head(),
        "actor": f"{os.getenv('USER', 'unknown')}@{socket.gethostname()}",
        "db_name": _resolve_db_name(args),
        "missing_tables_before": before_missing,
        "missing_columns_before": before_cols,
        "created_tables": [],
        "created_new": 0,
        "error": None,
    }
    print(f"[apply] 连接: {conn_desc}")
    print(f"[apply] 迁移前缺失表 {len(before_missing)} 张: {before_missing if before_missing else '（无）'}")
    try:
        if before_missing:
            Base.metadata.create_all(bind=engine)  # checkfirst=True → 已存在的表零动作
        # 结构增量（补列/索引/种子/授权）：create_all 不补列 → 复用 treasury_migrate（幂等，特权连接）
        structural = run_structural()
        after_missing, after_cols = schema_diff(engine)
        created = [t for t in before_missing if t not in after_missing]
        payload.update(
            status="applied",
            finished_at=datetime.now().isoformat(timespec="seconds"),
            created_tables=created,
            created_new=len(created),
            missing_tables_after=after_missing,
            structural=structural,
        )
        path = write_journal(payload)
        try:
            write_anchor_pointer(payload, path)
        except Exception as exc:  # 加性：指针写失败不阻断迁移，由收敛步 fail-closed 兜住
            print(f"[apply] [warn] 锚指针写入失败（收敛步会据此 rc=2 中止部署）："
                  f"{type(exc).__name__}: {exc}")
    except Exception as exc:  # DDL 无事务：失败时记录已建对象，供 --rollback 恢复
        try:
            after_missing, _ = schema_diff(engine)
        except Exception:
            after_missing = before_missing
        created = [t for t in before_missing if t not in after_missing]
        payload.update(
            status="failed",
            finished_at=datetime.now().isoformat(timespec="seconds"),
            created_tables=created,
            created_new=len(created),
            error=f"{type(exc).__name__}: {exc}",
        )
        path = write_journal(payload)
        try:
            write_anchor_pointer(payload, path)
        except Exception as exc:  # 加性：失败路径同样留锚（status=failed），收敛步只消费 applied
            print(f"[apply] [warn] 锚指针写入失败（失败路径）：{type(exc).__name__}: {exc}")
        print(f"[FAIL] 迁移失败：{payload['error']}")
        print(f"       已建对象 {created}（journal: {path}）")
        print(f"       回滚：python3 scripts/db_migrate.py --rollback {os.path.basename(path)}")
        return 1

    ok = not payload["missing_tables_after"]
    print(f"[apply] 新建 {payload['created_new']} 张表: {created if created else '（无 → 幂等空跑）'}")
    print(f"[apply] 迁移后缺失表 {len(payload['missing_tables_after'])} 张: "
          f"{payload['missing_tables_after'] if payload['missing_tables_after'] else '（无）'}")
    print(f"[apply] 结构增量(treasury_migrate) status={structural.get('status')} rc={structural.get('rc')}")
    if after_cols:
        print(f"[apply] ⚠️ 列级差异仍存（结构步未覆盖）: {after_cols}")
    print(f"[apply] journal: {path}")
    if not ok:
        print("[FAIL] 迁移后自检不一致（仍有缺失表）→ 部署链应中止")
        return 1
    return 0


def _dump_tables(tables, db_name) -> str:
    """mysqldump 备份（凭据走 600 临时 defaults 文件，不进 argv/日志）。返回文件路径。"""
    fd, cfg = tempfile.mkstemp(prefix="cma-migrate-dump-", suffix=".cnf")
    os.close(fd)
    os.chmod(cfg, 0o600)
    try:
        lines = ["[client]"]
        if os.getenv("CMA_MIGRATE_DB_USER"):
            lines += [
                f"user={os.getenv('CMA_MIGRATE_DB_USER')}",
                f"password={os.getenv('CMA_MIGRATE_DB_PASS', '')}",
                f"host={os.getenv('CMA_MIGRATE_DB_HOST', '127.0.0.1')}",
                f"port={os.getenv('CMA_MIGRATE_DB_PORT', '3306')}",
            ]
        else:
            lines += ["user=root", f"socket={os.getenv('CMA_MIGRATE_DB_SOCKET', DEFAULT_SOCKET)}"]
        with open(cfg, "w", encoding="utf-8") as fh:
            fh.write("\n".join(lines) + "\n")
        out = os.path.join(
            journal_dir(), f"dump-{datetime.now().strftime('%Y%m%dT%H%M%S')}.sql"
        )
        cmd = ["mysqldump", f"--defaults-extra-file={cfg}", "--no-tablespaces", "--skip-add-locks", db_name, *tables]
        with open(out, "wb") as fh:
            subprocess.run(cmd, stdout=fh, stderr=subprocess.PIPE, check=True)
        return out
    finally:
        os.unlink(cfg)


def cmd_rollback(args, engine, conn_desc) -> int:
    path = resolve_journal(args.rollback)
    with open(path, encoding="utf-8") as fh:
        payload = json.load(fh)
    tables = payload.get("created_tables") or []
    print(f"[rollback] journal: {path}")
    print(f"[rollback] 该次迁移新建的表: {tables if tables else '（无）'}")
    if not tables:
        print("[rollback] 无新建对象 → 无需回滚（空操作）")
        return 0

    db_name = payload.get("db_name") or _resolve_db_name(args)
    if not args.no_dump:
        try:
            dump_path = _dump_tables(tables, db_name)
            print(f"[rollback] 已备份: {dump_path}")
        except Exception as exc:
            print(f"[FAIL] 备份失败 → 拒绝回滚（fail-closed）：{type(exc).__name__}: {exc}")
            return 1

    with engine.connect() as conn:
        rows = {}
        for t in tables:
            try:
                rows[t] = conn.execute(text(f"SELECT COUNT(*) FROM `{t}`")).scalar()
            except Exception:
                rows[t] = 0
        non_empty = {t: n for t, n in rows.items() if n}
        if non_empty and not args.force:
            print(f"[REFUSED] 以下表已有数据，未加 --force 拒绝删除：{non_empty}")
            return 3
        for t in tables:
            conn.execute(text(f"DROP TABLE IF EXISTS `{t}`"))
        conn.commit()
    print(f"[rollback] 已删除 {len(tables)} 张表: {tables}")
    payload["rolled_back_at"] = datetime.now().isoformat(timespec="seconds")
    payload["rolled_back_by"] = f"{os.getenv('USER', 'unknown')}@{socket.gethostname()}"
    payload["rollback_dump"] = None if args.no_dump else "见同目录 dump-*.sql"
    base = os.path.basename(path)
    archived = os.path.join(journal_dir(), base.replace("-applied.json", "-rolled_back.json"))
    with open(archived, "w", encoding="utf-8") as fh:
        json.dump(payload, fh, ensure_ascii=False, indent=2)
    print(f"[rollback] 记录: {archived}")
    return 0


def cmd_status(args) -> int:
    items = list_journals()
    if not items:
        print("[status] 无 journal 记录")
        return 0
    for p in items[:20]:
        try:
            with open(p, encoding="utf-8") as fh:
                j = json.load(fh)
            print(
                f"{os.path.basename(p)}  status={j.get('status')}  created_new={j.get('created_new')}  "
                f"tables={len(j.get('created_tables') or [])}  git={j.get('git_head')}  by={j.get('actor')}"
            )
        except Exception as exc:
            print(f"{os.path.basename(p)}  [unreadable] {exc}")
    return 0


# ────────────────────────────────────────────────────────────────
def main() -> int:
    ap = argparse.ArgumentParser(
        description="CMA 特权数据库迁移步（与应用启动 create_all 剥离）",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    g = ap.add_mutually_exclusive_group()
    g.add_argument("--check", action="store_true", help="只读预检（默认）")
    g.add_argument("--apply", action="store_true", help="幂等建表 + journal")
    g.add_argument("--rollback", metavar="JOURNAL", help="回滚某次 --apply 新建的表（latest 或文件名）")
    g.add_argument("--status", action="store_true", help="列出 journal 历史")
    ap.add_argument("--json", action="store_true", help="机器可读输出")
    ap.add_argument("--force", action="store_true", help="回滚允许删有数据的表")
    ap.add_argument("--no-dump", action="store_true", help="回滚前不 dump")
    ap.add_argument("--socket", default=None, help=f"unix socket 路径（默认 {DEFAULT_SOCKET}）")
    ap.add_argument("--db-url", default=None, help="显式连接串（覆盖推导）")
    ap.add_argument("--db-name", default=None, help="目标库名")
    args = ap.parse_args()

    if not (args.apply or args.rollback or args.status):
        args.check = True  # 默认只读预检

    if args.status:
        return cmd_status(args)

    engine, conn_desc = build_engine(args)

    # 迁移互斥：防并发会话/手工操作同时跑 DDL
    lock_path = os.path.join(journal_dir(), ".db-migrate.lock")
    lock_fh = open(lock_path, "a+")
    try:
        fcntl.flock(lock_fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        print("[FAIL] 另一个迁移正在运行（锁被占用）→ 本次中止")
        return 1

    if args.rollback:
        return cmd_rollback(args, engine, conn_desc)
    if args.apply:
        return cmd_apply(args, engine, conn_desc)
    return cmd_check(args, engine, conn_desc)


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