You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
136 lines
7.4 KiB
136 lines
7.4 KiB
"""只读结构与完整性盘点。依赖 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)
|
|
|