#!/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`，而 sqlx 的 `NaiveDateTime` **只与 `DATETIME` 兼容** （`TIMESTAMP` 只与 `DateTime` 兼容）⇒ 任何一个 `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), ''), 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，无该子句为空串 - 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), ''), 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（本查询用 '' 占位）， # 而"目标定义里没有 DEFAULT 子句"的期望值是空串 —— 两者语义相同，必须先归一化， # 否则会报出假漂移（migrate.rs 里对应的是 `unwrap_or_default()`）。 if default == "": 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 "")) 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())