charging-cabinet/tools/db_schema_diff.py

#!/usr/bin/env python3
"""PMS 数据库结构漂移检查（只读）

为什么需要这个工具
------------------
"测试库与生产库结构不一致" 这个坑本项目已经栽了 **5 次**：
BRPOP 卡死、`iccid` NULL、`SCHEMA_SQL` 前向引用、测试环境导出 500、
以及生产 `/api/charge-orders` 一有订单就 500 —— 每一次都是靠人肉发现。
`CREATE TABLE IF NOT EXISTS` 对已存在的表**不改任何列定义**，
`add_columns_if_not_exists` 又只看"列在不在"、不看列的定义，
所以结构漂移会**长期潜伏**，直到某个接口在生产上第一次被调用才炸。

本工具把"比对"变成一条命令，两种口径都支持：

1. **跨库比对**（默认）：`pms_dev`(测试) vs `pms`(生产) 的**全部列**，
   维度 = 类型 / 可空性 / 默认值 / EXTRA(`ON UPDATE`) / 注释。
2. **对 DDL 比对**（`--vs-ddl`）：把库与 `software/api-server/src/db/migrate.rs`
   里的权威清单 `LEGACY_COLUMN_TARGETS` 逐列比对 —— 这才是"漂移"的最终定义：
   代码期望什么，库里就得是什么。

另外无论哪种模式都会做一次 **TIMESTAMP 巡检**：本项目所有时间列在 Rust 侧都是
`NaiveDateTime`/`Option<NaiveDateTime>`，而 sqlx 的 `NaiveDateTime` **只与 `DATETIME` 兼容**
（`TIMESTAMP` 只与 `DateTime<Utc>` 兼容）⇒ 任何一个 `TIMESTAMP` 列都会让对应接口 500。

用法
----
  需先运行 `pwsh tools/dev-tunnel.ps1` 建立 SSH 隧道（默认连 127.0.0.1:13306）；
  `--host/--port` 优先于环境变量 `PMS_MYSQL_HOST/PMS_MYSQL_PORT`。

  python tools/db_schema_diff.py                     # 默认：pms_dev 与 pms 全列比对
  python tools/db_schema_diff.py --db pms            # 被检查库换成 pms（参照库自动取另一个）
  python tools/db_schema_diff.py --reference-db pms  # 显式指定参照库
  python tools/db_schema_diff.py --vs-ddl            # 只与 migrate.rs 的权威清单比对
  python tools/db_schema_diff.py --db pms --vs-ddl   # 生产库 vs DDL
  python tools/db_schema_diff.py --quiet             # 只打印漂移项与结论

退出码
------
  0 = 无漂移 ；1 = 发现漂移（可接进 CI / 部署前置检查）；2 = 参数或连接错误

只发 SELECT，绝不写库。
"""

import argparse
import os
import re
import sys


def die(message, code=2):
    """参数/配置/连接类错误 → 退出码 2（与"发现漂移"的 1 区分开）"""
    print(message, file=sys.stderr)
    sys.exit(code)


try:
    import pymysql
except ImportError:  # pragma: no cover - 环境缺依赖时给出可执行的提示
    die("缺少依赖：pip install pymysql")

# ---- 默认连接参数 ----------------------------------------------------------
# 默认指向 SSH 隧道（tools/dev-tunnel.ps1），不再使用已移除的 ZeroTier/nginx 转发。
# 优先级：命令行 --host/--port > 环境变量 PMS_MYSQL_HOST/PMS_MYSQL_PORT > 下面的默认值。
DEFAULT_HOST = os.environ.get("PMS_MYSQL_HOST", "127.0.0.1")
DEFAULT_PORT = int(os.environ.get("PMS_MYSQL_PORT", "13306"))
DEFAULT_USER = "root"
DEFAULT_PASSWORD = "Hbhyg731024@"

# 仓库内权威 DDL 的来源文件（相对本脚本所在目录）
DEFAULT_MIGRATE_RS = os.path.join(
    os.path.dirname(os.path.abspath(__file__)), "..", "software", "api-server", "src", "db", "migrate.rs"
)
# 权威清单常量名（migrate.rs 里的 `const LEGACY_COLUMN_TARGETS: &[(&str, &str, &str)]`）
TARGETS_CONST = "LEGACY_COLUMN_TARGETS"

# 比对维度（顺序即输出顺序）
DIMENSIONS = ("COLUMN_TYPE", "IS_NULLABLE", "COLUMN_DEFAULT", "EXTRA", "COLUMN_COMMENT")


def connect(args, database):
    """建立只读连接（本工具只执行 SELECT）"""
    return pymysql.connect(
        host=args.host,
        port=args.port,
        user=args.user,
        password=args.password,
        database=database,
        charset="utf8mb4",
    )


def fetch_all_columns(cur, db):
    """取某库全部基础表的列定义：{(表, 列): (类型, 可空, 默认, EXTRA, 注释)}"""
    cur.execute(
        """
        SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE,
               IFNULL(CAST(COLUMN_DEFAULT AS CHAR), '<NULL>'),
               IFNULL(CAST(EXTRA AS CHAR), ''),
               IFNULL(CAST(COLUMN_COMMENT AS CHAR), '')
        FROM information_schema.COLUMNS
        WHERE TABLE_SCHEMA = %s
          AND TABLE_NAME IN (SELECT TABLE_NAME FROM information_schema.TABLES
                             WHERE TABLE_SCHEMA = %s AND TABLE_TYPE = 'BASE TABLE')
        ORDER BY TABLE_NAME, ORDINAL_POSITION
        """,
        (db, db),
    )
    return {(r[0], r[1]): tuple(r[2:]) for r in cur.fetchall()}


def scan_timestamp_columns(snapshot):
    """TIMESTAMP 巡检：返回 [(表, 列, 类型)] —— 非空即为漂移（会导致接口 500）"""
    return sorted(
        (table, column, values[0])
        for (table, column), values in snapshot.items()
        if values[0].startswith("timestamp")
    )


def parse_targets(migrate_rs_path):
    """从 migrate.rs 解析权威清单 LEGACY_COLUMN_TARGETS

    只认 `("表", "列", "目标定义"),` 这种字面量行；解析失败会明确报错而不是静默返回空，
    避免"清单没解析出来 -> 报无漂移"这种最危险的假阴性。
    """
    try:
        with open(migrate_rs_path, "r", encoding="utf-8") as f:
            source = f.read()
    except OSError as e:
        die("无法读取 migrate.rs（%s）：%s" % (migrate_rs_path, e))

    start = source.find("const %s" % TARGETS_CONST)
    if start < 0:
        die("在 %s 里找不到 const %s" % (migrate_rs_path, TARGETS_CONST))
    end = source.find("];", start)
    if end < 0:
        die("const %s 的定义没有正常结束（找不到结尾的 `];`）" % TARGETS_CONST)
    block = source[start:end]

    targets = []
    # 逐条匹配 ("table", "column", "target")；目标定义里可能带单引号注释，用非贪婪到行尾的引号+逗号收尾
    for m in re.finditer(
        r'\(\s*"([^"]+)"\s*,\s*"([^"]+)"\s*,\s*"((?:[^"\\]|\\.)*)"\s*,?\s*\)', block
    ):
        table, column, target = m.group(1), m.group(2), m.group(3)
        targets.append((table, column, target))
    if not targets:
        die("const %s 里一条目标都没解析出来（清单格式变了？）" % TARGETS_CONST)
    return targets


def expect_from_target(target):
    """把目标定义原文解析成判定字段 —— 与 migrate.rs 里 ColumnSpec::from_target 同口径

    - 类型：第一个词，去掉长度（`VARCHAR(50)` -> `VARCHAR`，information_schema 的 DATA_TYPE 不带长度）
    - 可空性：不含 `NOT NULL` 即可空
    - 默认值：`DEFAULT <token>` 的 token，无该子句为空串
    - ON UPDATE：是否含 `ON UPDATE CURRENT_TIMESTAMP`
    - 注释：`COMMENT '<文本>'` 的文本，无该子句为空串
    """
    first = target.split()[0]
    data_type = first.split("(")[0]
    nullable = "NOT NULL" not in target
    m = re.search(r"DEFAULT\s+(\S+)", target)
    default_value = m.group(1) if m else ""
    on_update = "ON UPDATE CURRENT_TIMESTAMP" in target
    m = re.search(r"COMMENT\s+'([^']*)'", target)
    comment = m.group(1) if m else ""
    return data_type, nullable, default_value, on_update, comment


def diff_against_ddl(cur, db, targets):
    """库 vs DDL 权威清单：返回 [(表, 列, 维度, 库里, 期望)]"""
    drifts = []
    for table, column, target in targets:
        cur.execute(
            """
            SELECT IFNULL(CAST(COLUMN_TYPE AS CHAR), ''), IFNULL(CAST(IS_NULLABLE AS CHAR), ''),
                   IFNULL(CAST(COLUMN_DEFAULT AS CHAR), '<NULL>'), IFNULL(CAST(EXTRA AS CHAR), ''),
                   IFNULL(CAST(COLUMN_COMMENT AS CHAR), '')
            FROM information_schema.COLUMNS
            WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s AND COLUMN_NAME = %s
            """,
            (db, table, column),
        )
        row = cur.fetchone()
        if row is None:
            drifts.append((table, column, "存在性", "列不存在", "目标定义 %s" % target))
            continue
        column_type, is_nullable, default, extra, comment = row
        e_type, e_nullable, e_default, e_on_update, e_comment = expect_from_target(target)

        # ⚠️ 无默认值时 COLUMN_DEFAULT 是 SQL NULL（本查询用 '<NULL>' 占位），
        # 而"目标定义里没有 DEFAULT 子句"的期望值是空串 —— 两者语义相同，必须先归一化，
        # 否则会报出假漂移（migrate.rs 里对应的是 `unwrap_or_default()`）。
        if default == "<NULL>":
            default = ""

        # DATA_TYPE 的口径：information_schema 的 COLUMN_TYPE 带长度（bigint(20)/datetime/varchar(50)）
        actual_type = column_type.split("(")[0]
        if actual_type.lower() != e_type.lower():
            drifts.append((table, column, "DATA_TYPE", actual_type, e_type))
        if (is_nullable == "YES") != e_nullable:
            drifts.append((table, column, "IS_NULLABLE", is_nullable, "YES" if e_nullable else "NO"))
        if default.lower() != e_default.lower():
            drifts.append((table, column, "COLUMN_DEFAULT", default, e_default or "<NULL>"))
        has_on_update = "on update current_timestamp" in extra.lower()
        if has_on_update != e_on_update:
            drifts.append((table, column, "ON UPDATE", "有" if has_on_update else "无",
                           "需要" if e_on_update else "不需要"))
        if e_comment and e_comment not in comment:
            drifts.append((table, column, "COLUMN_COMMENT", comment or "<空>", e_comment))
    return drifts


def compare_databases(cur_a, db_a, cur_b, db_b, quiet):
    """跨库全列比对：返回 [(表, 列, 维度, 被检查库的值, 参照库的值)]"""
    snap_a = fetch_all_columns(cur_a, db_a)
    snap_b = fetch_all_columns(cur_b, db_b)
    drifts = []
    for key in sorted(set(snap_a) | set(snap_b)):
        a, b = snap_a.get(key), snap_b.get(key)
        if a == b:
            continue
        for idx, dim in enumerate(DIMENSIONS):
            va = a[idx] if a else "<列不存在>"
            vb = b[idx] if b else "<列不存在>"
            if va != vb:
                drifts.append((key[0], key[1], dim, va, vb))
    if not quiet:
        print("库 %s: %d 列 / 库 %s: %d 列" % (db_a, len(snap_a), db_b, len(snap_b)))
    return drifts, snap_a


def main():
    ap = argparse.ArgumentParser(
        description="PMS 数据库结构漂移检查（只读）：跨库比对 或 与 migrate.rs 权威 DDL 比对",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="退出码：0=无漂移；1=发现漂移；2=参数/连接错误",
    )
    ap.add_argument("--db", default="pms_dev", choices=["pms", "pms_dev"],
                    help="被检查的库（默认 pms_dev）")
    ap.add_argument("--reference-db", choices=["pms", "pms_dev"],
                    help="跨库比对的参照库（默认取另一个：--db pms_dev 时参照 pms）")
    ap.add_argument("--vs-ddl", action="store_true",
                    help="改为与 migrate.rs 的 LEGACY_COLUMN_TARGETS 权威清单比对（不连参照库）")
    ap.add_argument("--migrate-rs", default=DEFAULT_MIGRATE_RS, help="migrate.rs 路径")
    ap.add_argument("--host", default=DEFAULT_HOST)
    ap.add_argument("--port", type=int, default=DEFAULT_PORT)
    ap.add_argument("--user", default=DEFAULT_USER)
    ap.add_argument("--password", default=DEFAULT_PASSWORD)
    ap.add_argument("--quiet", action="store_true", help="只打印漂移项与结论")
    args = ap.parse_args()

    reference_db = args.reference_db or ("pms" if args.db == "pms_dev" else "pms_dev")
    if args.vs_ddl and reference_db == args.db:
        die("--vs-ddl 模式不需要参照库，请勿把 --reference-db 设成与 --db 相同")

    drifts = []
    try:
        if args.vs_ddl:
            targets = parse_targets(args.migrate_rs)
            if not args.quiet:
                print("权威清单：%s 的 %s（%d 条）" % (os.path.normpath(args.migrate_rs),
                                                   TARGETS_CONST, len(targets)))
            conn = connect(args, args.db)
            cur = conn.cursor()
            snapshot = fetch_all_columns(cur, args.db)
            drifts = diff_against_ddl(cur, args.db, targets)
            conn.close()
        else:
            conn_a = connect(args, args.db)
            cur_a = conn_a.cursor()
            conn_b = connect(args, reference_db)
            cur_b = conn_b.cursor()
            drifts, snapshot = compare_databases(cur_a, args.db, cur_b, reference_db, args.quiet)
            conn_a.close()
            conn_b.close()
    except pymysql.MySQLError as e:
        # 连不上时必须是 exit 2（连接错误），绝不能落成"无漂移"的 exit 0
        die("数据库连接/查询失败：%s\n"
            "连接被拒：请先运行 `pwsh tools/dev-tunnel.ps1` 建立 SSH 隧道"
            "（或设置 PMS_MYSQL_HOST/PMS_MYSQL_PORT 指向其它可达地址）" % e)

    # ---- TIMESTAMP 巡检（两种模式都做）----
    ts_columns = scan_timestamp_columns(snapshot)
    if ts_columns:
        drifts.extend((t, c, "TIMESTAMP", typ,
                       "必须是 DATETIME（sqlx 的 NaiveDateTime 只与 DATETIME 兼容）")
                      for t, c, typ in ts_columns)

    mode = "库 %s vs migrate.rs DDL 权威清单" % args.db if args.vs_ddl \
        else "库 %s vs 库 %s" % (args.db, reference_db)
    print("=" * 100)
    print("结构漂移检查：%s" % mode)
    print("=" * 100)
    if drifts:
        print("❌ 发现 %d 处漂移：" % len(drifts))
        print("  %-16s %-22s %-16s %-42s %s" % ("表", "列", "维度", "实际", "期望"))
        for table, column, dim, actual, expected in drifts:
            print("  %-16s %-22s %-16s %-42s %s" % (table, column, dim, actual, expected))
        print()
        print("修复：重启新版 api-server（启动迁移会按 migrate.rs 的清单自动对齐），")
        print("      或按上表手工 ALTER；对齐逻辑见 software/api-server/src/db/migrate.rs")
        return 1

    print("✅ 无漂移")
    if not args.vs_ddl:
        print("   （全库列定义 + 0 个 TIMESTAMP 列）")
    print("   被检查库 %s：%d 列，TIMESTAMP 列 0 个" % (args.db, len(snapshot)))
    return 0


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