471b61d277
1) probe_export.py：任务状态语义**搞反了**（导出成功也会被判失败）
- download_tasks.status 的权威语义是 0 排队 / 1 处理中 / **2 完成** / 3 失败
（db/migrate.rs:225 列注释；写入侧 charge_records.rs::mark_task_completed 写的就是 2）。
- 该脚本此前**等 status == 1（处理中）当成功、把 status == 2（完成）判成"导出失败"**
—— 典型假红。已按权威语义订正，并与另外两处**本来就正确**的口径对齐：
web/src/pages/charge-records/DownloadCenter.tsx::statusText、
test/schema_alignment_e2e.py（等 status in (2,3) 再断言 == 2）。
- 同批修正退出码分类：账号缺 charge_record:export（403）由 exit 1 改为
**exit 2「未验证」** —— 本项要验证的是"降权成 pms 后能否往 exports/ 落盘"，
没权限根本走不到那一步，报 1 是**把夹具缺陷说成产品缺陷**（误报与假绿同样有害）。
- 允许用 PMS_PROBE_PHONE / PMS_PROBE_PASSWORD 换成有权限的账号。
- **C 项结论（由落盘物证直接判定，不依赖该脚本）**：测试环境
exports/充电记录__-.xlsx 5559 B、生产同类产物 5457 B，属主均为 pms:pms；
两环境 exports/uploads 目录均 pms:pms，两 api-server 单元均 User=pms/Group=pms
⇒ 降权进程确实能落盘。已在 README 写明并给出取证依据。
2) deploy-prod.ps1 的退出码（此前失败的探针只打印 ❌ 就继续，脚本仍以 0 退出）
- 消费链路探测失败（exit 1）现在**记入失败清单**并以 exit 1 收尾；
"未验证"（隧道没起）默认仍不计失败（环境问题会误报），
但新增 **-StrictVerify** 开关供 CI/无人值守使用，打开即 exit 1。
- 好处：把这条从"打印了 ❌ 却 exit 0"（仓库第 8 条假绿）里摘出来。
3) 统一退出码 0/1/2 落地到本批脚本，并对每个失败路径实测取证：
- test/tcp_e2e_smoke.py：13/13 通过（含未登录业务报文被静默丢弃）；先复位夹具再跑。
- test/security-hardening/ 新增 verdict.py 判据容器（未跟踪 → 本次入库）；
4 个探针全部按失败→exit 1 改造，并各实测过失败路径。
- 实测结果（2026-09-20 发布后回归）：smoke_prod 4/4 exit 0、
probe_prod 5/5 exit 0、probe_upload 6/6 exit 0。
4) 工具与文档
- tools/db.py 默认库由**生产 pms 改为测试 pms_dev**（可用 PMS_MYSQL_DB 覆盖，
切生产时打印醒目警告）—— "以为在测测试库、实际打生产"的隐患。
- tools/db_schema_diff.py：连不上必须是 exit 2，绝不伪装成"无漂移"的 0。
- 新增 tools/dev-tunnel.ps1（入库，SSH 隧道；口令走 PMS_SSH_PASSWORD 环境变量），
配套 tools/dev-tunnel-run.ps1 自愈启动器**保持 gitignore**（内含口令）。
- deploy/nginx/nginx.conf 镜像随线上同步（移除 stream 转发；与线上逐字节一致）。
- AGENTS.md / README.md：补「踩过的坑」第 7/8 条（掐断上游管道会杀被测进程、
打印 ❌ 却 exit 0 是假绿）、退出码约定、本地开发改 SSH 隧道、
在线判据口径订正、order_timeout 守门补齐、小程序 vue-tsc 门禁与其抓到的两个真 BUG。
发布验收（本次统一发布，按 device→api 顺序）：
- 测试环境与生产环境四服务 active，端口 9000/9001/3000/3001 正确，ERROR 均为 0；
- 生产迁移按判据 grep '更新列注释' 实测对齐 cabinets.status 列注释；
- 结构漂移 python tools/db_schema_diff.py = 0 处漂移 / exit 0（--vs-ddl 两库同样 0）；
- schema_alignment_e2e --env dev 全部通过 exit 0；
- 生产域名 /api/pms/cabinets -> 401、/pms/ -> 200；消费链路探测通过。
316 lines
14 KiB
Python
316 lines
14 KiB
Python
#!/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())
|