"""只读结构与完整性盘点。依赖 PyMySQL、PyYAML；不输出连接信息或业务正文。""" import argparse import hashlib import json import re from datetime import datetime, timezone from pathlib import Path from urllib.parse import urlsplit def migration_manifest(directory): result = [] versions = set() for file in directory.glob("V*__*.sql"): match = re.fullmatch(r"V(\d+)__(.+)\.sql", file.name) if not match: raise ValueError("迁移文件名不符合整数版本约定") version = int(match[1]) if version in versions: raise ValueError("迁移版本重复：V" + str(version)) versions.add(version) # 统一换行，避免 Windows/Linux 检出方式产生假差异。 content = file.read_text(encoding="utf-8-sig").replace("\r\n", "\n") result.append({"version": version, "file": file.name, "sha256": hashlib.sha256(content.encode("utf-8")).hexdigest()}) return sorted(result, key=lambda item: item["version"]) def compare_snapshots(previous, current): """只比较结构和迁移指纹，不把统计变化误认为结构漂移。""" differences = [] for key in ("tables", "columns", "indexes", "foreign_keys", "migrations"): before = previous.get(key, []) after = current.get(key, []) if before != after: differences.append(key) return differences def collect(connection, migrations): statements = { "tables": "SELECT TABLE_NAME,ENGINE,TABLE_COLLATION FROM information_schema.TABLES " "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_TYPE='BASE TABLE' ORDER BY TABLE_NAME", "columns": "SELECT TABLE_NAME,COLUMN_NAME,COLUMN_TYPE,IS_NULLABLE,COLUMN_DEFAULT,EXTRA," "CHARACTER_SET_NAME,COLLATION_NAME FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA=DATABASE() ORDER BY TABLE_NAME,ORDINAL_POSITION", "indexes": "SELECT TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,SUB_PART " "FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() " "ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX", "foreign_keys": "SELECT k.TABLE_NAME,k.CONSTRAINT_NAME,k.COLUMN_NAME,k.REFERENCED_TABLE_NAME," "k.REFERENCED_COLUMN_NAME,r.DELETE_RULE,r.UPDATE_RULE " "FROM information_schema.KEY_COLUMN_USAGE k JOIN information_schema.REFERENTIAL_CONSTRAINTS r " "ON r.CONSTRAINT_SCHEMA=k.CONSTRAINT_SCHEMA AND r.TABLE_NAME=k.TABLE_NAME " "AND r.CONSTRAINT_NAME=k.CONSTRAINT_NAME WHERE k.TABLE_SCHEMA=DATABASE() " "ORDER BY k.TABLE_NAME,k.CONSTRAINT_NAME,k.ORDINAL_POSITION", } checks = { "orphan_moment_comments": "SELECT COUNT(*) n FROM moment_comment c LEFT JOIN moment m " "ON m.id=c.moment_id WHERE m.id IS NULL", "orphan_video_progress": "SELECT COUNT(*) n FROM video_play_record v LEFT JOIN moment m " "ON m.id=v.video_id WHERE m.id IS NULL", "orphan_comment_likes": "SELECT COUNT(*) n FROM moment_comment_like l LEFT JOIN moment_comment c " "ON c.id=l.comment_id LEFT JOIN moment m ON m.id=c.moment_id " "WHERE c.id IS NULL OR m.id IS NULL", } result = {"captured_at_utc": datetime.now(timezone.utc).isoformat(), "migrations": migrations, "migration_execution": "UNCONFIRMED"} with connection.cursor() as cursor: cursor.execute("START TRANSACTION READ ONLY") cursor.execute("SELECT VERSION() version, DATABASE() database_name") result["database"] = cursor.fetchone() for name, sql in statements.items(): cursor.execute(sql) result[name] = cursor.fetchall() result["integrity"] = {} for name, sql in checks.items(): cursor.execute(sql) result["integrity"][name] = cursor.fetchone()["n"] # 只有真实历史表存在才读取；不创建 baseline、不 repair、不执行任何版本脚本。 if any(t["TABLE_NAME"] == "flyway_schema_history" for t in result["tables"]): cursor.execute("SELECT installed_rank,version,script,checksum,success FROM flyway_schema_history " "ORDER BY installed_rank") result["flyway_history"] = cursor.fetchall() connection.rollback() return result def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--config", type=Path, required=True) parser.add_argument("--output", type=Path, required=True, help="本地受限目录，勿提交或对外分发") parser.add_argument("--compare", type=Path, help="之前人工验收的结构快照") args = parser.parse_args() import pymysql import yaml module = Path(__file__).resolve().parents[1] manifest = migration_manifest(module / "src/main/resources/db/migration") config = yaml.safe_load(args.config.read_text(encoding="utf-8"))["spring"]["datasource"] endpoint = urlsplit(config["url"].removeprefix("jdbc:")) if endpoint.scheme != "mysql" or not endpoint.hostname or not endpoint.path.strip("/"): raise ValueError("必须显式配置 MySQL 数据库") if any("${" in str(config.get(k, "")) for k in ("url", "username", "password")): raise ValueError("配置含未解析的环境变量，请使用本地已解析配置") connection = pymysql.connect(host=endpoint.hostname, port=endpoint.port or 3306, database=endpoint.path.strip("/"), user=config["username"], password=str(config["password"]), charset="utf8mb4", connect_timeout=10, read_timeout=30, cursorclass=pymysql.cursors.DictCursor) try: snapshot = collect(connection, manifest) finally: connection.close() # round trip 统一驱动返回的 Decimal 等类型，保证后续比较一致。 snapshot = json.loads(json.dumps(snapshot, ensure_ascii=False, default=str)) if args.compare: previous = json.loads(args.compare.read_text(encoding="utf-8")) if previous.get("database", {}).get("database_name") != snapshot["database"]["database_name"]: raise ValueError("禁止将不同数据库名的快照当成同一迁移基线") snapshot["changed_sections"] = compare_snapshots(previous, snapshot) args.output.parent.mkdir(parents=True, exist_ok=True) # 不覆盖已验收证据；每次使用新的文件名。 with args.output.open("x", encoding="utf-8") as file: json.dump(snapshot, file, ensure_ascii=False, indent=2) print(json.dumps({"tables": len(snapshot["tables"]), "integrity": snapshot["integrity"], "changed_sections": snapshot.get("changed_sections"), "migration_execution": "UNCONFIRMED"}, ensure_ascii=False)) return 2 if snapshot.get("changed_sections") or any(snapshot["integrity"].values()) else 0 if __name__ == "__main__": try: raise SystemExit(main()) except Exception as error: # 驱动异常可能含主机、用户名或配置内容，不将异常原文写入终端/日志。 print("数据库盘点失败：" + type(error).__name__ + "；请检查本地配置、权限、依赖和输出路径。") raise SystemExit(1)