AiDOPWarehouse/doc/db/check_backup_import_coverage.py

#!/usr/bin/env python3
"""Compare backup INSERT coverage vs 165 schema."""

import re
from collections import Counter
from pathlib import Path

import pymysql

SQL_PATH = Path(
    r"D:\Projects\Ai-DOP\阿里云数据库备份\aidopdev_20260608_053527.sql\aidopdev_20260608_053527.sql"
)
DB = dict(
    host="123.60.180.165",
    port=3306,
    user="aidopremote",
    password="1234567890aiDOP#",
    database="aidopdev",
    charset="utf8mb4",
)

INSERT_RE = re.compile(
    r"^INSERT\s+(?:IGNORE\s+)?INTO\s+[`']?([^`'\s(]+)[`']?\s+VALUES\s",
    re.IGNORECASE,
)


def norm(name: str) -> str:
    return name.strip("`'\"").lower()


def main() -> None:
    conn = pymysql.connect(**DB)
    cur = conn.cursor()
    cur.execute("SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE()")
    existing = {norm(r[0]) for r in cur.fetchall()}

    tables: Counter[str] = Counter()
    for line in SQL_PATH.open("r", encoding="utf-8", errors="replace"):
        m = INSERT_RE.match(line.strip())
        if m:
            tables[norm(m.group(1))] += 1

    missing = sorted(t for t in tables if t not in existing)
    present = sorted(t for t in tables if t in existing)

    print("=== Backup import coverage ===")
    print(f"INSERT tables in backup: {len(tables)}")
    print(f"Table exists on 165: {len(present)} ({sum(tables[t] for t in present)} INSERT stmts)")
    print(f"Table missing on 165: {len(missing)} ({sum(tables[t] for t in missing)} INSERT stmts)")

    if missing:
        print("\nMissing tables (backup has data, 165 has no table):")
        for t in missing:
            print(f"  - {t} ({tables[t]} INSERT)")

    # tables in backup but 0 rows check not needed - just coverage

    conn.close()


if __name__ == "__main__":
    main()