slaj/_analyze_tables.py

#!/usr/bin/env python3
"""
Parse ROHS SQL file to extract all table definitions (table name, columns, types),
then connect to 安监 database to compare.
"""
import re
import json

def parse_sql_file(filepath):
    """Parse the ROHS SQL file and extract table definitions."""
    with open(filepath, 'r', encoding='utf-8') as f:
        content = f.read()

    # Split by CREATE TABLE statements
    tables = {}

    # Find all CREATE TABLE blocks
    pattern = r'CREATE TABLE "SLAJ_ROHS"\."(\w+)"\s*\((.*?)\);'
    matches = re.finditer(pattern, content, re.DOTALL)

    for match in matches:
        table_name = match.group(1)
        columns_text = match.group(2)

        columns = []
        # Parse each column line
        for line in columns_text.strip().split('\n'):
            line = line.strip().rstrip(',')
            if not line or line.startswith('CONSTRAINT') or line.startswith('PRIMARY KEY') or line.startswith('UNIQUE') or line.startswith('CHECK') or line.startswith('USING'):
                continue

            # Extract column name and type
            # Pattern: "COLUMN_NAME" TYPE ... or COLUMN_NAME TYPE ...
            col_match = re.match(r'"?(\w+)"?\s+([\w\(\)\s,]+?)(?:\s+DEFAULT|\s+NOT\s+NULL|\s+NULL|\s*$|,)', line)
            if col_match:
                col_name = col_match.group(1).upper()
                col_type = col_match.group(2).strip().upper()
                columns.append({'name': col_name, 'type': col_type})

        tables[table_name] = columns

    return tables

def extract_all_table_names(content):
    """Extract all table names from CREATE TABLE statements."""
    pattern = r'CREATE TABLE "SLAJ_ROHS"\."(\w+)"'
    return re.findall(pattern, content)

if __name__ == '__main__':
    filepath = r'd:\Workspaces\slaj\docs\slaj-rohs_tbl-all_without-data_hg.sql'

    with open(filepath, 'r', encoding='utf-8') as f:
        content = f.read()

    tables = parse_sql_file(filepath)

    print(f"Total ROHS tables: {len(tables)}")
    print()

    # Categorize tables
    skip_tables = {t for t in tables if t.startswith('MLOG$_') or t.endswith('_BAK') or t.endswith('BAK') or
                   '_BAK' in t or t.endswith('_20250407') or t.endswith('_0326') or t.endswith('_1125') or
                   t.endswith('_BAK25') or t.endswith('_0731') or 'BAK2025' in t or '20250113' in t or
                   '20250110' in t or '_MW' in t or '_MW_' in t or t.endswith('_MW')}

    real_tables = {k: v for k, v in tables.items() if k not in skip_tables}
    skipped = sorted({t for t in tables if t in skip_tables})

    print(f"Real tables (excluding backups/logs/materialized views): {len(real_tables)}")
    print(f"Skipped tables (backups/logs/MV): {len(skipped)}")
    print()

    print("=== REAL TABLES ===")
    for tname in sorted(real_tables.keys()):
        cols = real_tables[tname]
        print(f"\n{tname} ({len(cols)} columns):")
        for c in cols:
            print(f"  {c['name']}: {c['type']}")

    print("\n=== SKIPPED TABLES ===")
    for t in skipped:
        print(f"  {t} ({len(tables[t])} columns)")

    # Save as JSON for programmatic comparison
    with open(r'd:\Workspaces\slaj\_rohs_tables.json', 'w', encoding='utf-8') as f:
        json.dump(real_tables, f, indent=2, ensure_ascii=False)

    print(f"\nSaved ROHS tables to _rohs_tables.json")