group_fqcd_jr/tools/schema_fingerprint.py

"""Read-only structural fingerprints; no credentials or row contents in the report."""
import hashlib
import json
import sys
from pathlib import Path

from sqlalchemy import create_engine, text

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from app.core.config import get_settings


def fingerprint():
    engine = create_engine(get_settings().mysql_dsn.replace("mysql+asyncmy", "mysql+pymysql"))
    result = {}
    with engine.connect() as connection:
        names = connection.execute(text("""
            SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE()
        """)).scalars().all()
        for name in sorted(names):
            if name == "alembic_version":
                continue
            columns = connection.execute(text("""
                SELECT COLUMN_NAME,COLUMN_TYPE,IS_NULLABLE,COLUMN_DEFAULT,EXTRA,
                       CHARACTER_SET_NAME,COLLATION_NAME,GENERATION_EXPRESSION
                FROM information_schema.COLUMNS
                WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=:name ORDER BY ORDINAL_POSITION
            """), {"name": name}).all()
            # 索引、唯一键与外键同样属于结构契约：只对字段做指纹无法发现约束漂移。
            indexes = connection.execute(text("""
                SELECT INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,INDEX_TYPE
                FROM information_schema.STATISTICS
                WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=:name
                ORDER BY INDEX_NAME, SEQ_IN_INDEX
            """), {"name": name}).all()
            foreign_keys = connection.execute(text("""
                SELECT 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.CONSTRAINT_NAME=k.CONSTRAINT_NAME
                WHERE k.TABLE_SCHEMA=DATABASE() AND k.TABLE_NAME=:name
                  AND k.REFERENCED_TABLE_NAME IS NOT NULL
                ORDER BY k.CONSTRAINT_NAME, k.ORDINAL_POSITION
            """), {"name": name}).all()
            canonical = json.dumps({
                "columns": [list(row) for row in columns],
                "indexes": [list(row) for row in indexes],
                "foreign_keys": [list(row) for row in foreign_keys],
            }, ensure_ascii=False, default=str)
            result[name] = hashlib.sha256(canonical.encode()).hexdigest()
    engine.dispose()
    return result


if __name__ == "__main__":
    report = fingerprint()
    if "--out" in sys.argv:
        # 直接由 Python 写 UTF-8，避免 shell 重定向按 UTF-16 落盘导致后续读取失败。
        target = Path(sys.argv[sys.argv.index("--out") + 1])
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8")
        print(f"fingerprint written: {target} ({len(report)} tables)")
    else:
        print(json.dumps(report, indent=2, ensure_ascii=False))