cma-management/backend/scripts/migrate_cpa_snapshot_20261004.py

#!/usr/bin/env python3
"""CPA 客户获利性明细快照表 —— 独立特权 DDL 步（幂等 / 可回滚 / 带 journal）

任务：finance-20261003-cpa-customer-profitability-mvp-fullstack（P0-a）

为什么是独立脚本而不是 ORM 模型
------------------------------
运行库账号 `cma_api` 只有 SELECT + 表级 DML 白名单（**无 DDL**），建表须走特权通道
（本机 unix socket + root，或专用迁移账号）。同时 `app/models/__init__.py` 此刻有**他单在途
改动**（CMA-KPI-20261003 KPI 字典分层治理），本单不碰该共享文件 ⇒ 新表以独立迁移脚本落地，
运行时以 raw SQL 访问（与既有 `zxt_masterbill` 同一形态）。

用法
----
    python3 scripts/migrate_cpa_snapshot_20261004.py --check        # 只读预检（默认）
    python3 scripts/migrate_cpa_snapshot_20261004.py --apply        # 幂等建表 + 写 journal
    python3 scripts/migrate_cpa_snapshot_20261004.py --status       # 历史 journal
    python3 scripts/migrate_cpa_snapshot_20261004.py --rollback latest
    python3 scripts/migrate_cpa_snapshot_20261004.py --show-create

选项
----
    --db-url URL    显式连接串（离线演练用 `sqlite:///<scratch>.db`）
    --db-name NAME  目标库名（默认 env CMA_MIGRATE_DB_NAME → CMA_DB_NAME → cma）
    --journal-dir   默认 env CPA_MIGRATE_JOURNAL_DIR 或 backend/scripts/backups/db-migrate-journal/
    --force         回滚时允许删除**有数据**的表（默认拒绝，exit 3）
    --json          机器可读输出

连接凭据（口令只从环境读，禁止硬编码/回显）
    1) CMA_MIGRATE_DB_USER + CMA_MIGRATE_DB_PASS（专用迁移账号，TCP）
    2) 未设 → 本机 unix socket + root（socket 认证）
    ⚠️ 应用运行账号（backend/.env 的 CMA_DB_USER = cma_api）不得被本脚本使用：无 DDL 权限。

幂等契约
    首跑 --apply：created=1；二跑 --apply：created=0 且零副作用（不 ALTER / 不 DROP / 不动数据行）。

退出码
    0 成功 ｜ 1 失败（fail-closed）｜ 2 用法错误 ｜ 3 回滚被拒（目标表有数据且未 --force）
"""
import argparse
import datetime as _dt
import json
import os
import subprocess
import sys

TABLE = "customer_profitability_snapshot"

MYSQL_DDL = f"""
CREATE TABLE IF NOT EXISTS `{TABLE}` (
  `id` INT NOT NULL AUTO_INCREMENT,
  `entity_id` INT NOT NULL COMMENT '企业ID（账套，多租户隔离）',
  `period` VARCHAR(7) NOT NULL COMMENT '期间 YYYY-MM（自然月，来源 zxt_masterbill.bill_date）',
  `unit_id` INT NOT NULL COMMENT '客户/往来单位ID（zxt_masterbill.unit_id；库内无名称源）',
  `revenue` DECIMAL(18,4) NOT NULL DEFAULT 0.0000 COMMENT '客户收入（万元；Σsum_money，含税原值口径）',
  `cogs_alloc` DECIMAL(18,4) NULL COMMENT '客户销货成本(分摊)（万元）——P1 fail-closed，未裁定口径前恒 NULL',
  `chan_sub` DECIMAL(18,4) NULL COMMENT '客户渠道补贴（万元）——P1 待打标台账，恒 NULL',
  `contrib` DECIMAL(18,4) NULL COMMENT '客户贡献毛利（万元）——P1，恒 NULL',
  `contrib_rate` DECIMAL(10,4) NULL COMMENT '客户贡献毛利率（%）——P1，恒 NULL',
  `rank_no` INT NULL COMMENT '期内收入排名（1=最高）',
  `calib_status` VARCHAR(30) NOT NULL DEFAULT 'revenue_only' COMMENT '口径状态: revenue_only / full(P1)',
  `fail_closed_reason` VARCHAR(500) NULL COMMENT '未落值列的原因（fail-closed 留痕，禁静默 0）',
  `created_at` DATETIME DEFAULT CURRENT_TIMESTAMP,
  `updated_at` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uq_cpa_snap_entity_period_unit` (`entity_id`, `period`, `unit_id`),
  KEY `ix_cpa_snap_entity_period_rank` (`entity_id`, `period`, `rank_no`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='客户获利性明细快照（CPA P0-a；每客户每期一行，幂等 upsert）'
"""

SQLITE_DDL = f"""
CREATE TABLE IF NOT EXISTS {TABLE} (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  entity_id INTEGER NOT NULL,
  period TEXT NOT NULL,
  unit_id INTEGER NOT NULL,
  revenue NUMERIC NOT NULL DEFAULT 0,
  cogs_alloc NUMERIC,
  chan_sub NUMERIC,
  contrib NUMERIC,
  contrib_rate NUMERIC,
  rank_no INTEGER,
  calib_status TEXT NOT NULL DEFAULT 'revenue_only',
  fail_closed_reason TEXT,
  created_at TEXT DEFAULT CURRENT_TIMESTAMP,
  updated_at TEXT DEFAULT CURRENT_TIMESTAMP,
  CONSTRAINT uq_cpa_snap_entity_period_unit UNIQUE (entity_id, period, unit_id)
)
"""


def _now():
    return _dt.datetime.now().strftime("%Y%m%d-%H%M%S")


def _journal_dir(arg):
    if arg:
        return arg
    env = os.environ.get("CPA_MIGRATE_JOURNAL_DIR")
    if env:
        return env
    here = os.path.dirname(os.path.abspath(__file__))
    d = os.path.join(here, "backups", "db-migrate-journal")
    os.makedirs(d, exist_ok=True)
    return d


def _resolve_url(args):
    if args.db_url:
        return args.db_url
    if os.environ.get("CMA_MIGRATE_DB_URL"):
        return os.environ["CMA_MIGRATE_DB_URL"]
    user = os.environ.get("CMA_MIGRATE_DB_USER")
    pwd = os.environ.get("CMA_MIGRATE_DB_PASS", "")
    if user:
        host = os.environ.get("CMA_MIGRATE_DB_HOST", "127.0.0.1")
        port = os.environ.get("CMA_MIGRATE_DB_PORT", "3306")
        name = args.db_name or os.environ.get("CMA_MIGRATE_DB_NAME") or os.environ.get("CMA_DB_NAME") or "cma"
        from urllib.parse import quote_plus
        return f"mysql+pymysql://{user}:{quote_plus(pwd)}@{host}:{port}/{name}?charset=utf8mb4"
    name = args.db_name or os.environ.get("CMA_DB_NAME") or "cma"
    return f"mysql+pymysql://root@localhost/{name}?unix_socket=/var/run/mysqld/mysqld.sock&charset=utf8mb4"


def _engine(args):
    from sqlalchemy import create_engine
    url = _resolve_url(args)
    kw = {}
    if url.startswith("sqlite"):
        kw["connect_args"] = {"check_same_thread": False}
    return create_engine(url, **kw), url


def _table_exists(conn, url):
    from sqlalchemy import text
    if url.startswith("sqlite"):
        r = conn.execute(text("SELECT name FROM sqlite_master WHERE type='table' AND name=:t"), {"t": TABLE}).fetchall()
    else:
        r = conn.execute(text("SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=:t"),
                         {"t": TABLE}).fetchall()
    return bool(r)


def _rowcount(conn):
    from sqlalchemy import text
    try:
        return int(conn.execute(text(f"SELECT COUNT(*) FROM {TABLE}")).scalar() or 0)
    except Exception:
        return None


def main():
    ap = argparse.ArgumentParser()
    g = ap.add_mutually_exclusive_group()
    g.add_argument("--check", action="store_true")
    g.add_argument("--apply", action="store_true")
    g.add_argument("--status", action="store_true")
    g.add_argument("--rollback", nargs="?", const="latest")
    g.add_argument("--show-create", action="store_true")
    ap.add_argument("--db-url")
    ap.add_argument("--db-name")
    ap.add_argument("--journal-dir")
    ap.add_argument("--force", action="store_true")
    ap.add_argument("--json", action="store_true")
    args = ap.parse_args()

    engine, url = _engine(args)
    redacted = url.split("//")[0] + "//***"
    dialect = "sqlite" if url.startswith("sqlite") else "mysql"
    out = {"dialect": dialect, "table": TABLE, "url": redacted}

    if args.status:
        d = _journal_dir(args.journal_dir)
        files = sorted(f for f in os.listdir(d) if f.startswith("cpa-snapshot-"))
        out["journals"] = files
        print(json.dumps(out, ensure_ascii=False, indent=2) if args.json else "\n".join(files) or "(no journal)")
        return 0

    from sqlalchemy import text
    with engine.connect() as conn:
        exists = _table_exists(conn, url)
        rows = _rowcount(conn) if exists else None
        out.update(existing=exists, rowcount=rows)

        if args.show_create:
            if not exists:
                out["show_create"] = None
            else:
                if dialect == "sqlite":
                    out["show_create"] = conn.execute(text(
                        "SELECT sql FROM sqlite_master WHERE type='table' AND name=:t"), {"t": TABLE}).scalar()
                else:
                    out["show_create"] = conn.execute(text(f"SHOW CREATE TABLE `{TABLE}`")).fetchone()[1]
            print(json.dumps(out, ensure_ascii=False, indent=2))
            return 0

        if args.rollback is not None:
            if not exists:
                out["action"] = "rollback-noop"; print(json.dumps(out, ensure_ascii=False)); return 0
            if rows and not args.force:
                out["action"] = "rollback-refused"; out["why"] = f"表内有 {rows} 行，未加 --force"
                print(json.dumps(out, ensure_ascii=False)); return 3
            jdir = _journal_dir(args.journal_dir)
            jf = os.path.join(jdir, f"cpa-snapshot-rollback-{_now()}.json")
            with open(jf, "w") as fh:
                json.dump({"table": TABLE, "rows_before": rows, "ts": _now()}, fh, ensure_ascii=False)
            with engine.begin() as c2:
                c2.execute(text(f"DROP TABLE {TABLE}" if dialect == "sqlite" else f"DROP TABLE `{TABLE}`"))
            out["action"] = "rolled-back"; out["journal"] = jf
            print(json.dumps(out, ensure_ascii=False, indent=2))
            return 0

        if not args.apply:
            # --check：只读预检
            out["action"] = "check"
            out["would_create"] = not exists
            if exists:
                cols = conn.execute(text(
                    "SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=:t ORDER BY ORDINAL_POSITION"
                ), {"t": TABLE}).fetchall() if dialect == "mysql" else conn.execute(
                    text(f"PRAGMA table_info({TABLE})")).fetchall()
                out["columns"] = [c[0] for c in cols] if dialect == "mysql" else [c[1] for c in cols]
            print(json.dumps(out, ensure_ascii=False, indent=2) if args.json else
                  f"[check] table={TABLE} existing={exists} rowcount={rows} would_create={not exists}")
            return 0

        # --apply：幂等建表
        ddl = SQLITE_DDL if dialect == "sqlite" else MYSQL_DDL
        jdir = _journal_dir(args.journal_dir)
        created_new = 0
        if not exists:
            with engine.begin() as c2:
                c2.execute(text(ddl))
            created_new = 1
        with engine.connect() as conn3:
            assert _table_exists(conn3, url), "建表后自检失败"
        jf = os.path.join(jdir, f"cpa-snapshot-{_now()}{'' if created_new else '-noop'}.json")
        with open(jf, "w") as fh:
            json.dump({"table": TABLE, "created": created_new, "ddl": ddl.strip(), "ts": _now()}, fh, ensure_ascii=False, indent=2)
        out.update(action="applied", created_new=created_new, journal=jf,
                   rowcount_after=_rowcount(conn3))
        print(json.dumps(out, ensure_ascii=False, indent=2))
        return 0


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