195 lines
7.4 KiB
Python
195 lines
7.4 KiB
Python
"""约束与映射一致性审计：文档基线 ↔ MySQL 实际结构 ↔ SQLAlchemy ORM。
|
||
|
||
只读，不修改数据库，也不写入任何文件。补足 audit_schema.py（只比表名/列存在）
|
||
与 schema_fingerprint.py（只对字段做指纹）无法覆盖的盲区：
|
||
|
||
1. docs/00-新数据库基线设计.md 声明的唯一键 vs information_schema.STATISTICS；
|
||
2. ORM 列 vs 库列。ORM 多出的列会让运行期 SQL 直接报错；库多出的 NOT NULL 列
|
||
会让写入静默失败。
|
||
3. 任一不一致以非零退出码暴露，便于纳入提交前检查链。
|
||
|
||
凭据来自 .env 的 MYSQL_DSN，不在脚本内硬编码口令。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from sqlalchemy import create_engine, text
|
||
from sqlalchemy.engine import Connection
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
sys.path.insert(0, str(ROOT))
|
||
|
||
from app.core.config import get_settings # noqa: E402
|
||
from app.model import ( # noqa: E402,F401
|
||
advisor_product,
|
||
audit,
|
||
configuration,
|
||
conversation,
|
||
fund,
|
||
investment_goal,
|
||
memory,
|
||
platform,
|
||
session,
|
||
)
|
||
from app.model.base import Base # noqa: E402
|
||
|
||
BASELINE_DOC = ROOT / "docs" / "00-新数据库基线设计.md"
|
||
UNIQUE_KEY = "唯一键"
|
||
COMBINED_FLAG = "联合唯一键"
|
||
COMBINED_KEY_RE = re.compile(UNIQUE_KEY + r"[^()]*\(([^)]+)\)")
|
||
SKIP_FIELDS = {"字段", "表名"}
|
||
|
||
|
||
def normalize(columns: Any) -> tuple[str, ...]:
|
||
"""列顺序不影响唯一性语义，排序后再比较以避免顺序噪音。"""
|
||
return tuple(dict.fromkeys(sorted(str(column) for column in columns)))
|
||
|
||
|
||
def document_unique_keys() -> dict[str, set[tuple[str, ...]]]:
|
||
"""解析基线文档中每个 `#### \\`表名\\`` 小节声明的唯一键。"""
|
||
document = BASELINE_DOC.read_text(encoding="utf-8")
|
||
heads = list(re.finditer(r"(?m)^#### `([^`]+)`", document))
|
||
expected: dict[str, set[tuple[str, ...]]] = {}
|
||
for index, match in enumerate(heads):
|
||
name = match.group(1)
|
||
following = re.search(r"(?m)^#{3,5} ", document[match.end():])
|
||
end = match.end() + following.start() if following else len(document)
|
||
keys: set[tuple[str, ...]] = set()
|
||
combined_flag_columns: list[str] = []
|
||
for line in document[match.end():end].splitlines():
|
||
if not line.startswith("|"):
|
||
continue
|
||
cells = [cell.strip() for cell in line.strip().strip("|").split("|")]
|
||
if len(cells) < 3:
|
||
continue
|
||
field, rule = cells[0].strip("`"), cells[2]
|
||
if not field or field in SKIP_FIELDS or set(field) <= {"-", " "}:
|
||
continue
|
||
matched = False
|
||
for raw in COMBINED_KEY_RE.findall(rule):
|
||
columns = tuple(
|
||
dict.fromkeys(cell.strip().strip("`") for cell in raw.split(",") if cell.strip())
|
||
)
|
||
if columns:
|
||
keys.add(normalize(columns))
|
||
matched = True
|
||
if COMBINED_FLAG in rule:
|
||
combined_flag_columns.append(field)
|
||
elif UNIQUE_KEY in rule and not matched:
|
||
keys.add((field,))
|
||
if len(set(combined_flag_columns)) > 1:
|
||
keys.add(normalize(combined_flag_columns))
|
||
if keys:
|
||
expected[name] = keys
|
||
return expected
|
||
|
||
|
||
def database_unique_keys(connection: Connection) -> dict[str, set[tuple[str, ...]]]:
|
||
rows = connection.execute(
|
||
text(
|
||
"""
|
||
SELECT TABLE_NAME, INDEX_NAME, COLUMN_NAME, SEQ_IN_INDEX
|
||
FROM information_schema.STATISTICS
|
||
WHERE TABLE_SCHEMA = DATABASE() AND NON_UNIQUE = 0 AND INDEX_NAME <> 'PRIMARY'
|
||
ORDER BY TABLE_NAME, INDEX_NAME, SEQ_IN_INDEX
|
||
"""
|
||
)
|
||
).all()
|
||
grouped: dict[tuple[str, str], list[str]] = {}
|
||
for table, index, column, _sequence in rows:
|
||
grouped.setdefault((table, index), []).append(column)
|
||
result: dict[str, set[tuple[str, ...]]] = {}
|
||
for (table, _index), columns in grouped.items():
|
||
result.setdefault(table, set()).add(normalize(columns))
|
||
return result
|
||
|
||
|
||
def orm_columns() -> dict[str, set[str]]:
|
||
return {
|
||
table.name: {column.name for column in table.columns}
|
||
for table in Base.metadata.sorted_tables
|
||
}
|
||
|
||
|
||
def database_columns(
|
||
connection: Connection,
|
||
) -> tuple[dict[str, set[str]], dict[str, set[str]]]:
|
||
"""返回 (普通列, 生成列)。生成列由 MySQL 计算，不要求 ORM 映射。"""
|
||
rows = connection.execute(
|
||
text(
|
||
"""
|
||
SELECT TABLE_NAME, COLUMN_NAME, GENERATION_EXPRESSION
|
||
FROM information_schema.COLUMNS
|
||
WHERE TABLE_SCHEMA = DATABASE()
|
||
"""
|
||
)
|
||
).all()
|
||
plain: dict[str, set[str]] = {}
|
||
generated: dict[str, set[str]] = {}
|
||
for table, column, expression in rows:
|
||
target = generated if str(expression or "").strip() else plain
|
||
target.setdefault(table, set()).add(column)
|
||
return plain, generated
|
||
|
||
|
||
def main() -> int:
|
||
settings = get_settings()
|
||
engine = create_engine(settings.mysql_dsn.replace("mysql+asyncmy", "mysql+pymysql"))
|
||
constraint_problems: list[str] = []
|
||
mapping_problems: list[str] = []
|
||
generated_columns: list[str] = []
|
||
with engine.connect() as connection:
|
||
expected = document_unique_keys()
|
||
actual = database_unique_keys(connection)
|
||
for table in sorted(expected):
|
||
existing = actual.get(table, set())
|
||
for key in sorted(expected[table] - existing):
|
||
constraint_problems.append(f"[{table}] MISSING in DB : UNIQUE ({', '.join(key)})")
|
||
for key in sorted(existing - expected[table]):
|
||
constraint_problems.append(f"[{table}] EXTRA in DB : UNIQUE ({', '.join(key)})")
|
||
|
||
orm = orm_columns()
|
||
database, generated = database_columns(connection)
|
||
for table in sorted(orm):
|
||
if table not in database:
|
||
mapping_problems.append(f"[{table}] ORM table does not exist in database")
|
||
continue
|
||
for column in sorted(orm[table] - database[table]):
|
||
mapping_problems.append(f"[{table}] ORM column not in DB : {column}")
|
||
for column in sorted(database[table] - orm[table]):
|
||
if column in generated.get(table, set()):
|
||
generated_columns.append(f"[{table}].{column}")
|
||
continue
|
||
mapping_problems.append(f"[{table}] DB column not mapped : {column}")
|
||
engine.dispose()
|
||
|
||
print(f"constraint check: {len(expected)} documented tables compared")
|
||
if constraint_problems:
|
||
print("--- unique key mismatches ---")
|
||
for problem in constraint_problems:
|
||
print(" " + problem)
|
||
if mapping_problems:
|
||
print("--- orm/database mapping mismatches ---")
|
||
for problem in mapping_problems:
|
||
print(" " + problem)
|
||
if generated_columns:
|
||
print(f"note: {len(generated_columns)} generated column(s) intentionally not mapped: "
|
||
+ ", ".join(generated_columns))
|
||
|
||
total = len(constraint_problems) + len(mapping_problems)
|
||
if total:
|
||
print(f"\nFAILED: {total} mismatch(es) "
|
||
f"(constraints={len(constraint_problems)}, mapping={len(mapping_problems)})")
|
||
return 1
|
||
print("\nPASSED: unique keys and ORM mappings match the baseline")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|