verify.py

#!/usr/bin/env python3
"""Independent fidelity verifier: re-parses the raw dump separately from the
converter and checks the SQLite output for schema + data faithfulness."""
import re
import sys
import sqlite3

if len(sys.argv) != 3:
    sys.exit("usage: verify.py <dump.sql> <database.db>")

dump_path, db_path = sys.argv[1], sys.argv[2]
text = open(dump_path, encoding='utf-8').read()
lines = text.split('\n')
conn = sqlite3.connect(db_path)
cur = conn.cursor()
problems = []
notes = []


# --- 1. table set parity -------------------------------------------------
dump_tables = set(re.findall(r'^CREATE TABLE (?:public\.)?"?([^"\s(]+)', text, re.M))
sql_tables = {r[0] for r in cur.execute(
    "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")}
missing = dump_tables - sql_tables
extra = sql_tables - dump_tables
if missing:
    problems.append(f"tables missing in sqlite: {missing}")
if extra:
    problems.append(f"unexpected tables in sqlite: {extra}")
notes.append(f"tables: dump={len(dump_tables)} sqlite={len(sql_tables)}")

# --- 2. column parity from COPY headers ----------------------------------
copy_cols = {}
for m in re.finditer(r'^COPY (?:public\.)?"?([^"\s(]+)"?\s*\(([^)]*)\) FROM stdin;',
                     text, re.M):
    t = m.group(1)
    cols = [c.strip().strip('"') for c in m.group(2).split(',')]
    copy_cols[t] = cols
col_mismatch = 0
for t, cols in copy_cols.items():
    sql_cols = [r[1] for r in cur.execute(f'PRAGMA table_info("{t}")')]
    if set(cols) - set(sql_cols):
        problems.append(f"{t}: columns missing in sqlite: {set(cols)-set(sql_cols)}")
        col_mismatch += 1
notes.append(f"checked column presence for {len(copy_cols)} populated tables; "
             f"mismatches={col_mismatch}")

# --- 3. JSON validity for json-typed columns ----------------------------
# find json columns from CREATE TABLE blocks
json_cols = {}
for blk in re.finditer(r'CREATE TABLE (?:public\.)?"?([^"\s(]+)"?\s*\((.*?)\n\);',
                       text, re.S):
    t = blk.group(1)
    for cl in blk.group(2).split('\n'):
        cl = cl.strip().rstrip(',')
        mm = re.match(r'"?([^"\s]+)"?\s+(.*)', cl)
        if mm and re.match(r'jsonb?\b', mm.group(2), re.I):
            json_cols.setdefault(t, []).append(mm.group(1))
bad_json = 0
json_checked = 0
for t, cols in json_cols.items():
    for c in cols:
        rows = cur.execute(
            f'SELECT COUNT(*) FROM "{t}" WHERE "{c}" IS NOT NULL '
            f'AND json_valid("{c}")=0').fetchone()[0]
        total = cur.execute(
            f'SELECT COUNT(*) FROM "{t}" WHERE "{c}" IS NOT NULL').fetchone()[0]
        json_checked += total
        if rows:
            problems.append(f"{t}.{c}: {rows} invalid JSON values")
            bad_json += rows
notes.append(f"json values validated={json_checked}; invalid={bad_json}")

# --- 4. boolean columns only 0/1/NULL ------------------------------------
bool_cols = {}
for blk in re.finditer(r'CREATE TABLE (?:public\.)?"?([^"\s(]+)"?\s*\((.*?)\n\);',
                       text, re.S):
    t = blk.group(1)
    for cl in blk.group(2).split('\n'):
        cl = cl.strip().rstrip(',')
        mm = re.match(r'"?([^"\s]+)"?\s+(.*)', cl)
        if mm and re.match(r'boolean\b', mm.group(2), re.I):
            bool_cols.setdefault(t, []).append(mm.group(1))
bad_bool = 0
for t, cols in bool_cols.items():
    for c in cols:
        bad = cur.execute(
            f'SELECT COUNT(*) FROM "{t}" WHERE "{c}" IS NOT NULL '
            f'AND "{c}" NOT IN (0,1)').fetchone()[0]
        if bad:
            problems.append(f"{t}.{c}: {bad} non-0/1 boolean values")
            bad_bool += bad
notes.append(f"boolean columns checked={sum(len(v) for v in bool_cols.values())}; "
             f"out-of-range={bad_bool}")

# --- 5. PK integrity: no NULL pk, no duplicate pk ------------------------
pk_problems = 0
for t in sql_tables:
    info = cur.execute(f'PRAGMA table_info("{t}")').fetchall()
    pks = [r[1] for r in info if r[5]]  # pk flag
    if not pks:
        continue
    cols = ', '.join(f'"{p}"' for p in pks)
    nulls = cur.execute(
        f'SELECT COUNT(*) FROM "{t}" WHERE ' +
        ' OR '.join(f'"{p}" IS NULL' for p in pks)).fetchone()[0]
    total = cur.execute(f'SELECT COUNT(*) FROM "{t}"').fetchone()[0]
    distinct = cur.execute(f'SELECT COUNT(*) FROM (SELECT DISTINCT {cols} FROM "{t}")').fetchone()[0]
    if nulls:
        problems.append(f"{t}: {nulls} NULL primary-key values")
        pk_problems += 1
    if distinct != total:
        problems.append(f"{t}: PK not unique ({total} rows, {distinct} distinct)")
        pk_problems += 1
notes.append(f"PK integrity checked; problems={pk_problems}")

# --- 6. FK target existence (informational) ------------------------------
orphan_report = []
for t in sql_tables:
    for fk in cur.execute(f'PRAGMA foreign_key_list("{t}")').fetchall():
        _, _, reftab, frm, to, *_ = fk
        try:
            orphans = cur.execute(
                f'SELECT COUNT(*) FROM "{t}" a WHERE a."{frm}" IS NOT NULL '
                f'AND NOT EXISTS (SELECT 1 FROM "{reftab}" b WHERE b."{to}"=a."{frm}")'
            ).fetchone()[0]
            if orphans:
                orphan_report.append(f"{t}.{frm}->{reftab}.{to}: {orphans} orphan refs")
        except sqlite3.OperationalError:
            pass

print(f"=== VERIFY {db_path} ===")
for nline in notes:
    print("  note:", nline)
if orphan_report:
    print("  FK orphans (pre-existing in source data; FK enforcement is OFF by default):")
    for o in orphan_report:
        print("     -", o)
if problems:
    print("  PROBLEMS:")
    for p in problems:
        print("     !!", p)
    sys.exit(1)
else:
    print("  RESULT: all schema + data fidelity checks PASSED")