"""约束与映射一致性审计：文档基线 ↔ 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 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())