Database Schema Designer

#!/usr/bin/env python3
"""Parse SQL DDL and generate Mermaid ER diagrams showing table relationships.

Reads CREATE TABLE statements and outputs a Mermaid erDiagram block with:
- All tables and their columns (with PK/FK/UK annotations)
- Relationships inferred from REFERENCES clauses
- Nullable vs required relationship cardinality

Supports PostgreSQL, MySQL, and SQLite DDL syntax.
"""

import argparse
import json
import re
import sys
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple


@dataclass
class Column:
    name: str
    data_type: str
    is_pk: bool = False
    is_fk: bool = False
    is_unique: bool = False
    nullable: bool = True
    references_table: Optional[str] = None
    references_column: Optional[str] = None


@dataclass
class Table:
    name: str
    columns: List[Column] = field(default_factory=list)


@dataclass
class Relationship:
    from_table: str
    to_table: str
    from_column: str
    to_column: str
    nullable: bool
    label: str


def parse_ddl(sql: str) -> Tuple[List[Table], List[Relationship]]:
    """Parse SQL DDL into tables and relationships."""
    sql = re.sub(r'--[^\n]*', '', sql)
    sql = re.sub(r'/\*.*?\*/', '', sql, flags=re.DOTALL)

    tables: List[Table] = []
    relationships: List[Relationship] = []
    table_map: Dict[str, Table] = {}

    table_pattern = re.compile(
        r'CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?'
        r'(?:"?(\w+)"?\.)?'
        r'"?(\w+)"?'
        r'\s*\((.*?)\)\s*;',
        re.IGNORECASE | re.DOTALL
    )

    for match in table_pattern.finditer(sql):
        _schema, tname, body = match.group(1), match.group(2), match.group(3)
        table = Table(name=tname)

        # Track unique constraints at table level
        unique_cols = set()
        unique_match = re.findall(r'UNIQUE\s*\(\s*"?(\w+)"?\s*\)', body, re.IGNORECASE)
        for u in unique_match:
            unique_cols.add(u)

        pk_cols = set()
        pk_match = re.search(r'PRIMARY\s+KEY\s*\(([^)]+)\)', body, re.IGNORECASE)
        if pk_match:
            pk_cols = {c.strip().strip('"') for c in pk_match.group(1).split(',')}

        parts = _split_comma_top_level(body)

        for part in parts:
            part = part.strip()
            upper = part.upper()

            # Skip table-level constraints
            if re.match(r'(PRIMARY\s+KEY|UNIQUE|INDEX|KEY|CONSTRAINT|CHECK|FOREIGN\s+KEY)\s', upper):
                # But extract FOREIGN KEY ... REFERENCES
                fk_match = re.match(
                    r'(?:CONSTRAINT\s+\w+\s+)?FOREIGN\s+KEY\s*\(\s*"?(\w+)"?\s*\)\s*'
                    r'REFERENCES\s+"?(\w+)"?\s*\(\s*"?(\w+)"?\s*\)',
                    part, re.IGNORECASE
                )
                if fk_match:
                    fk_col, ref_table, ref_col = fk_match.group(1), fk_match.group(2), fk_match.group(3)
                    # Mark existing column as FK
                    for col in table.columns:
                        if col.name == fk_col:
                            col.is_fk = True
                            col.references_table = ref_table
                            col.references_column = ref_col
                            relationships.append(Relationship(
                                from_table=tname, to_table=ref_table,
                                from_column=fk_col, to_column=ref_col,
                                nullable=col.nullable,
                                label=fk_col.replace('_id', '').replace('_', ' ')
                            ))
                            break
                continue

            # Column definition
            col_match = re.match(r'"?(\w+)"?\s+(\w[\w\s()]*)', part, re.IGNORECASE)
            if not col_match:
                continue

            cname = col_match.group(1)
            ctype = col_match.group(2).strip().split()[0]  # first word of type

            is_pk = bool(re.search(r'PRIMARY\s+KEY', part, re.IGNORECASE)) or cname in pk_cols
            is_unique = bool(re.search(r'\bUNIQUE\b', part, re.IGNORECASE)) or cname in unique_cols
            nullable = not bool(re.search(r'NOT\s+NULL', part, re.IGNORECASE)) and not is_pk

            ref_match = re.search(
                r'REFERENCES\s+"?(\w+)"?\s*(?:\(\s*"?(\w+)"?\s*\))?',
                part, re.IGNORECASE
            )
            ref_table = ref_match.group(1) if ref_match else None
            ref_col = ref_match.group(2) if ref_match and ref_match.group(2) else 'id'

            col = Column(
                name=cname, data_type=_normalize_type(ctype),
                is_pk=is_pk, is_fk=bool(ref_table), is_unique=is_unique,
                nullable=nullable, references_table=ref_table, references_column=ref_col
            )
            table.columns.append(col)

            if ref_table:
                relationships.append(Relationship(
                    from_table=tname, to_table=ref_table,
                    from_column=cname, to_column=ref_col,
                    nullable=nullable,
                    label=cname.replace('_id', '').replace('_', ' ')
                ))

        tables.append(table)
        table_map[tname] = table

    return tables, relationships


def _split_comma_top_level(s: str) -> List[str]:
    """Split on commas not inside parentheses."""
    parts = []
    depth = 0
    current = []
    for ch in s:
        if ch == '(':
            depth += 1
        elif ch == ')':
            depth -= 1
        elif ch == ',' and depth == 0:
            parts.append(''.join(current))
            current = []
            continue
        current.append(ch)
    if current:
        parts.append(''.join(current))
    return parts


def _normalize_type(t: str) -> str:
    """Normalize SQL types to short display names."""
    mapping = {
        'CHARACTER': 'varchar', 'VARCHAR': 'varchar', 'TEXT': 'text',
        'INTEGER': 'int', 'INT': 'int', 'BIGINT': 'bigint',
        'SERIAL': 'serial', 'BIGSERIAL': 'bigserial',
        'BOOLEAN': 'bool', 'BOOL': 'bool',
        'TIMESTAMP': 'timestamp', 'TIMESTAMPTZ': 'timestamptz',
        'DATE': 'date', 'TIME': 'time',
        'NUMERIC': 'numeric', 'DECIMAL': 'decimal',
        'FLOAT': 'float', 'DOUBLE': 'double',
        'REAL': 'real', 'UUID': 'uuid', 'JSONB': 'jsonb', 'JSON': 'json',
    }
    return mapping.get(t.upper(), t.lower())


def _table_display_name(name: str) -> str:
    """Convert snake_case to PascalCase for Mermaid display."""
    return ''.join(word.capitalize() for word in name.split('_'))


def generate_mermaid(tables: List[Table], relationships: List[Relationship]) -> str:
    """Generate Mermaid ER diagram syntax."""
    lines = ["erDiagram"]

    # Deduplicate relationships by (from_table, to_table, from_column)
    seen = set()
    for rel in relationships:
        key = (rel.from_table, rel.to_table, rel.from_column)
        if key in seen:
            continue
        seen.add(key)

        from_name = _table_display_name(rel.from_table)
        to_name = _table_display_name(rel.to_table)

        # Determine cardinality:
        # parent (to_table) has one, child (from_table) has many
        # nullable FK = zero-or-more, non-nullable = one-or-more
        if rel.nullable:
            # to_table ||--o{ from_table : "label"
            connector = "||--o{"
        else:
            connector = "||--|{"

        lines.append(f"    {to_name} {connector} {from_name} : \"{rel.label}\"")

    lines.append("")

    # Table definitions
    for table in tables:
        display = _table_display_name(table.name)
        lines.append(f"    {display} {{")
        for col in table.columns:
            annotations = []
            if col.is_pk:
                annotations.append("PK")
            if col.is_fk:
                annotations.append("FK")
            if col.is_unique and not col.is_pk:
                annotations.append("UK")

            ann_str = ",".join(annotations)
            if ann_str:
                lines.append(f"        {col.data_type} {col.name} {ann_str}")
            else:
                lines.append(f"        {col.data_type} {col.name}")
        lines.append("    }")

    return '\n'.join(lines)


def generate_summary(tables: List[Table], relationships: List[Relationship]) -> Dict:
    """Generate a structured summary of the schema."""
    return {
        "table_count": len(tables),
        "relationship_count": len(relationships),
        "tables": [
            {
                "name": t.name,
                "column_count": len(t.columns),
                "columns": [
                    {
                        "name": c.name,
                        "type": c.data_type,
                        "pk": c.is_pk,
                        "fk": c.is_fk,
                        "unique": c.is_unique,
                        "nullable": c.nullable,
                        **({"references": f"{c.references_table}.{c.references_column}"} if c.references_table else {})
                    }
                    for c in t.columns
                ]
            }
            for t in tables
        ],
        "relationships": [
            {
                "from": f"{r.from_table}.{r.from_column}",
                "to": f"{r.to_table}.{r.to_column}",
                "nullable": r.nullable
            }
            for r in relationships
        ]
    }


def format_human(tables: List[Table], relationships: List[Relationship], mermaid: str) -> str:
    """Format output for human reading."""
    lines = []
    lines.append("ERD Generator Report")
    lines.append("=" * 50)
    lines.append(f"Tables: {len(tables)}")
    lines.append(f"Relationships: {len(relationships)}")
    lines.append("")
    lines.append("Mermaid ER Diagram:")
    lines.append("-" * 50)
    lines.append(f"```mermaid")
    lines.append(mermaid)
    lines.append("```")
    lines.append("")
    lines.append("Table Summary:")
    lines.append("-" * 50)
    for t in tables:
        pk_cols = [c.name for c in t.columns if c.is_pk]
        fk_cols = [c.name for c in t.columns if c.is_fk]
        lines.append(f"  {t.name} ({len(t.columns)} columns)")
        if pk_cols:
            lines.append(f"    PK: {', '.join(pk_cols)}")
        if fk_cols:
            lines.append(f"    FK: {', '.join(fk_cols)}")
    return '\n'.join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Parse SQL DDL and generate Mermaid ER diagrams showing table relationships."
    )
    parser.add_argument("file", help="Path to SQL DDL file")
    parser.add_argument("--json", action="store_true", dest="json_output",
                        help="Output results as JSON (includes Mermaid diagram and structured summary)")
    parser.add_argument("-o", "--output", help="Write Mermaid diagram to file (raw .mmd)")
    args = parser.parse_args()

    try:
        with open(args.file, 'r') as f:
            sql = f.read()
    except FileNotFoundError:
        print(f"Error: File not found: {args.file}", file=sys.stderr)
        sys.exit(2)
    except IOError as e:
        print(f"Error reading file: {e}", file=sys.stderr)
        sys.exit(2)

    tables, relationships = parse_ddl(sql)

    if not tables:
        print("Warning: No CREATE TABLE statements found.", file=sys.stderr)
        sys.exit(0)

    mermaid = generate_mermaid(tables, relationships)

    if args.output:
        try:
            with open(args.output, 'w') as f:
                f.write(mermaid + '\n')
            print(f"Mermaid diagram written to {args.output}")
        except IOError as e:
            print(f"Error writing output file: {e}", file=sys.stderr)
            sys.exit(2)

    if args.json_output:
        summary = generate_summary(tables, relationships)
        summary["mermaid"] = mermaid
        print(json.dumps(summary, indent=2))
    elif not args.output:
        print(format_human(tables, relationships, mermaid))
    elif args.output and not args.json_output:
        # Already wrote file, print summary
        print(f"Tables: {len(tables)}, Relationships: {len(relationships)}")

    sys.exit(0)


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""Compare two SQL schema files and generate migration SQL (ALTER statements).

Performs a structural diff between an 'old' and 'new' SQL DDL schema and produces:
- ALTER TABLE ADD COLUMN for new columns
- ALTER TABLE DROP COLUMN for removed columns
- ALTER TABLE ALTER COLUMN for type changes
- CREATE TABLE for new tables
- DROP TABLE for removed tables
- CREATE INDEX / DROP INDEX for index changes
- Rollback SQL for reversing the migration

Supports PostgreSQL syntax. Designed for review before execution.
"""

import argparse
import json
import re
import sys
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple, Set


@dataclass
class Column:
    name: str
    data_type: str
    nullable: bool
    is_pk: bool
    default: Optional[str]
    references: Optional[str]
    full_definition: str  # original DDL fragment for comparison


@dataclass
class Index:
    name: str
    table: str
    columns: List[str]
    unique: bool
    where_clause: Optional[str]
    original: str


@dataclass
class Table:
    name: str
    columns: Dict[str, Column] = field(default_factory=dict)
    pk_columns: List[str] = field(default_factory=list)
    original_ddl: str = ""


@dataclass
class Migration:
    description: str
    up_sql: str
    down_sql: str
    risk: str  # "low", "medium", "high"


def parse_schema(sql: str) -> Tuple[Dict[str, Table], Dict[str, Index]]:
    """Parse SQL DDL into tables and indexes."""
    sql = re.sub(r'--[^\n]*', '', sql)
    sql = re.sub(r'/\*.*?\*/', '', sql, flags=re.DOTALL)

    tables: Dict[str, Table] = {}
    indexes: Dict[str, Index] = {}

    # Parse CREATE TABLE
    table_pattern = re.compile(
        r'(CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?'
        r'(?:"?(\w+)"?\.)?'
        r'"?(\w+)"?'
        r'\s*\((.*?)\)\s*;)',
        re.IGNORECASE | re.DOTALL
    )

    for match in table_pattern.finditer(sql):
        full_ddl = match.group(1)
        tname = match.group(3)
        body = match.group(4)
        table = Table(name=tname, original_ddl=full_ddl)

        pk_cols = set()
        pk_match = re.search(r'PRIMARY\s+KEY\s*\(([^)]+)\)', body, re.IGNORECASE)
        if pk_match:
            pk_cols = {c.strip().strip('"') for c in pk_match.group(1).split(',')}

        parts = _split_comma_top_level(body)

        for part in parts:
            part = part.strip()
            upper = part.upper()

            if re.match(r'(PRIMARY\s+KEY|UNIQUE|INDEX|KEY|CONSTRAINT|CHECK|FOREIGN\s+KEY)\s', upper):
                continue

            col_match = re.match(r'"?(\w+)"?\s+(.+)', part, re.IGNORECASE)
            if not col_match:
                continue

            cname = col_match.group(1)
            rest = col_match.group(2).strip()

            # Extract type (first token, possibly with parenthesized args)
            type_match = re.match(r'(\w+(?:\s*\([^)]*\))?)', rest)
            ctype = type_match.group(1).upper() if type_match else rest.split()[0].upper()

            is_pk = bool(re.search(r'PRIMARY\s+KEY', part, re.IGNORECASE)) or cname in pk_cols
            nullable = not bool(re.search(r'NOT\s+NULL', part, re.IGNORECASE)) and not is_pk
            default_match = re.search(r'DEFAULT\s+((?:\'[^\']*\'|\S+))', part, re.IGNORECASE)
            ref_match = re.search(r'REFERENCES\s+"?(\w+)"?', part, re.IGNORECASE)

            col = Column(
                name=cname,
                data_type=ctype,
                nullable=nullable,
                is_pk=is_pk,
                default=default_match.group(1) if default_match else None,
                references=ref_match.group(1) if ref_match else None,
                full_definition=rest,
            )
            table.columns[cname] = col
            if is_pk:
                table.pk_columns.append(cname)

        tables[tname] = table

    # Parse CREATE INDEX
    idx_pattern = re.compile(
        r'(CREATE\s+(UNIQUE\s+)?INDEX\s+(?:CONCURRENTLY\s+)?(?:IF\s+NOT\s+EXISTS\s+)?'
        r'"?(\w+)"?\s+ON\s+"?(\w+)"?\s*(?:USING\s+\w+\s*)?\(([^)]+)\)'
        r'(?:\s+WHERE\s+(.+?))?)\s*;',
        re.IGNORECASE | re.DOTALL
    )
    for match in idx_pattern.finditer(sql):
        original = match.group(1) + ';'
        unique = bool(match.group(2))
        idx_name = match.group(3)
        tname = match.group(4)
        cols = [c.strip().strip('"') for c in match.group(5).split(',')]
        where = match.group(6).strip() if match.group(6) else None

        indexes[idx_name] = Index(
            name=idx_name, table=tname, columns=cols,
            unique=unique, where_clause=where, original=original
        )

    return tables, indexes


def _split_comma_top_level(s: str) -> List[str]:
    parts = []
    depth = 0
    current = []
    for ch in s:
        if ch == '(':
            depth += 1
        elif ch == ')':
            depth -= 1
        elif ch == ',' and depth == 0:
            parts.append(''.join(current))
            current = []
            continue
        current.append(ch)
    if current:
        parts.append(''.join(current))
    return parts


def diff_schemas(
    old_tables: Dict[str, Table], old_indexes: Dict[str, Index],
    new_tables: Dict[str, Table], new_indexes: Dict[str, Index]
) -> List[Migration]:
    """Compare two schemas and generate migrations."""
    migrations: List[Migration] = []

    old_names = set(old_tables.keys())
    new_names = set(new_tables.keys())

    # New tables
    for tname in sorted(new_names - old_names):
        table = new_tables[tname]
        migrations.append(Migration(
            description=f"Create table '{tname}'",
            up_sql=table.original_ddl,
            down_sql=f"DROP TABLE IF EXISTS {tname};",
            risk="low"
        ))

    # Dropped tables
    for tname in sorted(old_names - new_names):
        table = old_tables[tname]
        migrations.append(Migration(
            description=f"Drop table '{tname}'",
            up_sql=f"DROP TABLE IF EXISTS {tname};",
            down_sql=table.original_ddl,
            risk="high"
        ))

    # Modified tables
    for tname in sorted(old_names & new_names):
        old_t = old_tables[tname]
        new_t = new_tables[tname]
        old_cols = set(old_t.columns.keys())
        new_cols = set(new_t.columns.keys())

        # Added columns
        for cname in sorted(new_cols - old_cols):
            col = new_t.columns[cname]
            parts = [f"ALTER TABLE {tname} ADD COLUMN {cname} {col.data_type}"]
            if not col.nullable:
                if col.default:
                    parts.append(f"NOT NULL DEFAULT {col.default}")
                else:
                    # Suggest safe pattern for NOT NULL without default
                    parts = [
                        f"-- Phase 1: Add nullable column",
                        f"ALTER TABLE {tname} ADD COLUMN {cname} {col.data_type};",
                        f"-- Phase 2: Backfill data (adjust value as needed)",
                        f"-- UPDATE {tname} SET {cname} = <default_value> WHERE {cname} IS NULL;",
                        f"-- Phase 3: Set NOT NULL after backfill",
                        f"-- ALTER TABLE {tname} ALTER COLUMN {cname} SET NOT NULL;",
                    ]
                    migrations.append(Migration(
                        description=f"Add column '{cname}' to '{tname}' (NOT NULL, needs backfill)",
                        up_sql='\n'.join(parts),
                        down_sql=f"ALTER TABLE {tname} DROP COLUMN IF EXISTS {cname};",
                        risk="medium"
                    ))
                    continue
            else:
                if col.default:
                    parts.append(f"DEFAULT {col.default}")
            if col.references:
                parts.append(f"REFERENCES {col.references}")

            up_sql = ' '.join(parts) + ';'
            migrations.append(Migration(
                description=f"Add column '{cname}' to '{tname}'",
                up_sql=up_sql,
                down_sql=f"ALTER TABLE {tname} DROP COLUMN IF EXISTS {cname};",
                risk="low"
            ))

        # Removed columns
        for cname in sorted(old_cols - new_cols):
            col = old_t.columns[cname]
            rebuild_parts = [f"{cname} {col.data_type}"]
            if not col.nullable:
                rebuild_parts.append("NOT NULL")
            if col.default:
                rebuild_parts.append(f"DEFAULT {col.default}")

            migrations.append(Migration(
                description=f"Drop column '{cname}' from '{tname}'",
                up_sql=f"ALTER TABLE {tname} DROP COLUMN IF EXISTS {cname};",
                down_sql=f"ALTER TABLE {tname} ADD COLUMN {' '.join(rebuild_parts)};",
                risk="high"
            ))

        # Modified columns (type or nullability changes)
        for cname in sorted(old_cols & new_cols):
            old_c = old_t.columns[cname]
            new_c = new_t.columns[cname]

            stmts_up = []
            stmts_down = []

            if old_c.data_type != new_c.data_type:
                stmts_up.append(
                    f"ALTER TABLE {tname} ALTER COLUMN {cname} TYPE {new_c.data_type} "
                    f"USING {cname}::{new_c.data_type};"
                )
                stmts_down.append(
                    f"ALTER TABLE {tname} ALTER COLUMN {cname} TYPE {old_c.data_type} "
                    f"USING {cname}::{old_c.data_type};"
                )

            if old_c.nullable and not new_c.nullable:
                stmts_up.append(f"ALTER TABLE {tname} ALTER COLUMN {cname} SET NOT NULL;")
                stmts_down.append(f"ALTER TABLE {tname} ALTER COLUMN {cname} DROP NOT NULL;")
            elif not old_c.nullable and new_c.nullable:
                stmts_up.append(f"ALTER TABLE {tname} ALTER COLUMN {cname} DROP NOT NULL;")
                stmts_down.append(f"ALTER TABLE {tname} ALTER COLUMN {cname} SET NOT NULL;")

            if old_c.default != new_c.default:
                if new_c.default:
                    stmts_up.append(f"ALTER TABLE {tname} ALTER COLUMN {cname} SET DEFAULT {new_c.default};")
                else:
                    stmts_up.append(f"ALTER TABLE {tname} ALTER COLUMN {cname} DROP DEFAULT;")
                if old_c.default:
                    stmts_down.append(f"ALTER TABLE {tname} ALTER COLUMN {cname} SET DEFAULT {old_c.default};")
                else:
                    stmts_down.append(f"ALTER TABLE {tname} ALTER COLUMN {cname} DROP DEFAULT;")

            if stmts_up:
                risk = "medium" if any('TYPE' in s for s in stmts_up) else "low"
                migrations.append(Migration(
                    description=f"Alter column '{cname}' in '{tname}'",
                    up_sql='\n'.join(stmts_up),
                    down_sql='\n'.join(stmts_down),
                    risk=risk
                ))

    # Index changes
    old_idx_names = set(old_indexes.keys())
    new_idx_names = set(new_indexes.keys())

    for iname in sorted(new_idx_names - old_idx_names):
        idx = new_indexes[iname]
        migrations.append(Migration(
            description=f"Create index '{iname}' on '{idx.table}'",
            up_sql=idx.original,
            down_sql=f"DROP INDEX IF EXISTS {iname};",
            risk="low"
        ))

    for iname in sorted(old_idx_names - new_idx_names):
        idx = old_indexes[iname]
        migrations.append(Migration(
            description=f"Drop index '{iname}' from '{idx.table}'",
            up_sql=f"DROP INDEX IF EXISTS {iname};",
            down_sql=idx.original,
            risk="medium"
        ))

    return migrations


def format_sql(migrations: List[Migration], include_rollback: bool = True) -> str:
    """Format migrations as executable SQL."""
    if not migrations:
        return "-- No schema differences found."

    lines = []
    lines.append("-- Migration Script")
    lines.append("-- Generated by migration_diffr.py")
    lines.append(f"-- Changes: {len(migrations)}")
    lines.append("")

    # Group by risk
    high = [m for m in migrations if m.risk == "high"]
    medium = [m for m in migrations if m.risk == "medium"]
    low = [m for m in migrations if m.risk == "low"]

    if high:
        lines.append("-- ============================================")
        lines.append("-- HIGH RISK (review carefully, may lose data)")
        lines.append("-- ============================================")
        for m in high:
            lines.append(f"")
            lines.append(f"-- {m.description}")
            lines.append(m.up_sql)

    if medium:
        lines.append("")
        lines.append("-- ============================================")
        lines.append("-- MEDIUM RISK (may require backfill or lock)")
        lines.append("-- ============================================")
        for m in medium:
            lines.append(f"")
            lines.append(f"-- {m.description}")
            lines.append(m.up_sql)

    if low:
        lines.append("")
        lines.append("-- ============================================")
        lines.append("-- LOW RISK (safe, additive changes)")
        lines.append("-- ============================================")
        for m in low:
            lines.append(f"")
            lines.append(f"-- {m.description}")
            lines.append(m.up_sql)

    if include_rollback:
        lines.append("")
        lines.append("")
        lines.append("-- ============================================")
        lines.append("-- ROLLBACK SCRIPT")
        lines.append("-- ============================================")
        for m in reversed(migrations):
            lines.append(f"")
            lines.append(f"-- Rollback: {m.description}")
            lines.append(m.down_sql)

    return '\n'.join(lines)


def format_human(migrations: List[Migration]) -> str:
    """Format a human-readable summary."""
    lines = []
    lines.append("Migration Diff Report")
    lines.append("=" * 50)
    lines.append(f"Total changes: {len(migrations)}")

    risk_counts = {"high": 0, "medium": 0, "low": 0}
    for m in migrations:
        risk_counts[m.risk] += 1

    lines.append(f"Risk breakdown: {risk_counts['high']} high, {risk_counts['medium']} medium, {risk_counts['low']} low")
    lines.append("")

    if not migrations:
        lines.append("Schemas are identical. No migration needed.")
        return '\n'.join(lines)

    for i, m in enumerate(migrations, 1):
        risk_label = {"high": "[HIGH]  ", "medium": "[MEDIUM]", "low": "[LOW]   "}
        lines.append(f"  {i}. {risk_label[m.risk]} {m.description}")

    lines.append("")
    lines.append("SQL Migration:")
    lines.append("-" * 50)
    lines.append(format_sql(migrations))

    return '\n'.join(lines)


def format_json(migrations: List[Migration]) -> str:
    """Format as JSON."""
    risk_counts = {"high": 0, "medium": 0, "low": 0}
    for m in migrations:
        risk_counts[m.risk] += 1

    return json.dumps({
        "total_changes": len(migrations),
        "risk_summary": risk_counts,
        "migrations": [
            {
                "description": m.description,
                "risk": m.risk,
                "up": m.up_sql,
                "down": m.down_sql,
            }
            for m in migrations
        ],
        "up_sql": format_sql(migrations, include_rollback=False),
        "down_sql": format_sql(
            migrations, include_rollback=False
        ).replace("Migration Script", "Rollback Script") if migrations else "",
    }, indent=2)


def main():
    parser = argparse.ArgumentParser(
        description="Compare two SQL schema files and generate migration ALTER statements."
    )
    parser.add_argument("old_schema", help="Path to the current/old SQL schema file")
    parser.add_argument("new_schema", help="Path to the target/new SQL schema file")
    parser.add_argument("--json", action="store_true", dest="json_output",
                        help="Output results as JSON")
    parser.add_argument("--no-rollback", action="store_true",
                        help="Omit rollback SQL from output")
    parser.add_argument("-o", "--output", help="Write migration SQL to file")
    args = parser.parse_args()

    try:
        with open(args.old_schema, 'r') as f:
            old_sql = f.read()
    except FileNotFoundError:
        print(f"Error: File not found: {args.old_schema}", file=sys.stderr)
        sys.exit(2)
    except IOError as e:
        print(f"Error reading file: {e}", file=sys.stderr)
        sys.exit(2)

    try:
        with open(args.new_schema, 'r') as f:
            new_sql = f.read()
    except FileNotFoundError:
        print(f"Error: File not found: {args.new_schema}", file=sys.stderr)
        sys.exit(2)
    except IOError as e:
        print(f"Error reading file: {e}", file=sys.stderr)
        sys.exit(2)

    old_tables, old_indexes = parse_schema(old_sql)
    new_tables, new_indexes = parse_schema(new_sql)

    migrations = diff_schemas(old_tables, old_indexes, new_tables, new_indexes)

    if args.output:
        try:
            sql_out = format_sql(migrations, include_rollback=not args.no_rollback)
            with open(args.output, 'w') as f:
                f.write(sql_out + '\n')
            print(f"Migration SQL written to {args.output}")
        except IOError as e:
            print(f"Error writing output file: {e}", file=sys.stderr)
            sys.exit(2)

    if args.json_output:
        print(format_json(migrations))
    elif not args.output:
        print(format_human(migrations))

    # Exit code: 0 = no changes, 1 = changes found (useful in CI)
    sys.exit(0 if not migrations else 1)


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""Validate SQL DDL schemas for normalization violations, missing indexes,
naming conventions, and common anti-patterns.

Parses CREATE TABLE statements and checks for:
- Missing indexes on foreign key columns
- Missing timestamp columns (created_at, updated_at)
- Naming convention violations (snake_case enforcement)
- Missing primary keys
- Soft-delete columns without partial indexes
- Sequential integer PKs exposed (suggests CUID2/UUIDv7)
- Missing NOT NULL on foreign keys
- Tables without any indexes
"""

import argparse
import json
import re
import sys
from dataclasses import dataclass, field
from enum import Enum
from typing import List, Dict, Optional, Tuple


class Severity(str, Enum):
    ERROR = "error"
    WARNING = "warning"
    INFO = "info"


@dataclass
class Finding:
    table: str
    severity: str
    rule: str
    message: str
    suggestion: str


@dataclass
class Column:
    name: str
    data_type: str
    nullable: bool
    is_pk: bool
    default: Optional[str]
    references: Optional[str]  # referenced table


@dataclass
class Index:
    name: str
    columns: List[str]
    unique: bool
    where_clause: Optional[str]


@dataclass
class Table:
    name: str
    columns: Dict[str, Column] = field(default_factory=dict)
    indexes: List[Index] = field(default_factory=list)
    primary_key: Optional[List[str]] = None


def parse_ddl(sql: str) -> List[Table]:
    """Parse SQL DDL into structured Table objects."""
    sql = re.sub(r'--[^\n]*', '', sql)
    sql = re.sub(r'/\*.*?\*/', '', sql, flags=re.DOTALL)
    tables: List[Table] = []
    table_map: Dict[str, Table] = {}

    # Extract CREATE TABLE blocks
    table_pattern = re.compile(
        r'CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?'
        r'(?:"?(\w+)"?\.)?'   # optional schema
        r'"?(\w+)"?'          # table name
        r'\s*\((.*?)\)\s*;',
        re.IGNORECASE | re.DOTALL
    )

    for match in table_pattern.finditer(sql):
        _schema, tname, body = match.group(1), match.group(2), match.group(3)
        table = Table(name=tname)

        # Split body on commas, respecting parentheses depth
        parts = _split_comma_top_level(body)

        for part in parts:
            part = part.strip()
            upper = part.upper()

            # Table-level PRIMARY KEY
            pk_match = re.match(r'PRIMARY\s+KEY\s*\(([^)]+)\)', part, re.IGNORECASE)
            if pk_match:
                table.primary_key = [c.strip().strip('"') for c in pk_match.group(1).split(',')]
                continue

            # Table-level UNIQUE or INDEX (inline)
            if re.match(r'(UNIQUE|INDEX|KEY|CONSTRAINT)\s', upper):
                continue

            # Column definition
            col_match = re.match(
                r'"?(\w+)"?\s+(\w[\w\s()]*?)(?:\s+(NOT\s+NULL|NULL|PRIMARY\s+KEY|DEFAULT\s+.+?|REFERENCES\s+\w+(?:\s*\([^)]*\))?))*\s*$',
                part, re.IGNORECASE
            )
            if col_match:
                cname = col_match.group(1)
                ctype = col_match.group(2).strip()

                is_pk = bool(re.search(r'PRIMARY\s+KEY', part, re.IGNORECASE))
                nullable = not bool(re.search(r'NOT\s+NULL', part, re.IGNORECASE)) and not is_pk
                default_match = re.search(r'DEFAULT\s+(\S+)', part, re.IGNORECASE)
                ref_match = re.search(r'REFERENCES\s+"?(\w+)"?', part, re.IGNORECASE)

                col = Column(
                    name=cname,
                    data_type=ctype,
                    nullable=nullable,
                    is_pk=is_pk,
                    default=default_match.group(1) if default_match else None,
                    references=ref_match.group(1) if ref_match else None,
                )
                table.columns[cname] = col
                if is_pk:
                    table.primary_key = [cname]

        tables.append(table)
        table_map[tname] = table

    # Extract CREATE INDEX statements
    idx_pattern = re.compile(
        r'CREATE\s+(UNIQUE\s+)?INDEX\s+(?:CONCURRENTLY\s+)?(?:IF\s+NOT\s+EXISTS\s+)?'
        r'"?(\w+)"?\s+ON\s+"?(\w+)"?\s*(?:USING\s+\w+\s*)?\(([^)]+)\)'
        r'(?:\s+WHERE\s+(.+?))?;',
        re.IGNORECASE | re.DOTALL
    )
    for match in idx_pattern.finditer(sql):
        unique = bool(match.group(1))
        idx_name = match.group(2)
        tname = match.group(3)
        cols = [c.strip().strip('"') for c in match.group(4).split(',')]
        where = match.group(5).strip() if match.group(5) else None

        idx = Index(name=idx_name, columns=cols, unique=unique, where_clause=where)
        if tname in table_map:
            table_map[tname].indexes.append(idx)

    return tables


def _split_comma_top_level(s: str) -> List[str]:
    """Split on commas that are not inside parentheses."""
    parts = []
    depth = 0
    current = []
    for ch in s:
        if ch == '(':
            depth += 1
        elif ch == ')':
            depth -= 1
        elif ch == ',' and depth == 0:
            parts.append(''.join(current))
            current = []
            continue
        current.append(ch)
    if current:
        parts.append(''.join(current))
    return parts


def validate(tables: List[Table]) -> List[Finding]:
    """Run all validation rules against parsed tables."""
    findings: List[Finding] = []

    for table in tables:
        _check_naming(table, findings)
        _check_primary_key(table, findings)
        _check_timestamps(table, findings)
        _check_fk_indexes(table, findings)
        _check_fk_nullable(table, findings)
        _check_soft_delete_index(table, findings)
        _check_sequential_pk(table, findings)
        _check_no_indexes(table, findings)

    return findings


def _check_naming(table: Table, findings: List[Finding]):
    """Enforce snake_case naming for tables and columns."""
    snake = re.compile(r'^[a-z][a-z0-9]*(_[a-z0-9]+)*$')
    if not snake.match(table.name):
        findings.append(Finding(
            table=table.name, severity=Severity.WARNING, rule="naming_convention",
            message=f"Table name '{table.name}' is not snake_case.",
            suggestion=f"Rename to '{_to_snake(table.name)}'."
        ))
    for cname in table.columns:
        if not snake.match(cname):
            findings.append(Finding(
                table=table.name, severity=Severity.WARNING, rule="naming_convention",
                message=f"Column '{cname}' is not snake_case.",
                suggestion=f"Rename to '{_to_snake(cname)}'."
            ))


def _to_snake(name: str) -> str:
    s = re.sub(r'([A-Z]+)([A-Z][a-z])', r'\1_\2', name)
    s = re.sub(r'([a-z0-9])([A-Z])', r'\1_\2', s)
    return s.lower().replace(' ', '_').replace('-', '_')


def _check_primary_key(table: Table, findings: List[Finding]):
    if not table.primary_key:
        findings.append(Finding(
            table=table.name, severity=Severity.ERROR, rule="missing_primary_key",
            message=f"Table '{table.name}' has no PRIMARY KEY defined.",
            suggestion="Add a PRIMARY KEY column (e.g., id TEXT PRIMARY KEY or id UUID PRIMARY KEY)."
        ))


def _check_timestamps(table: Table, findings: List[Finding]):
    cols = set(table.columns.keys())
    if 'created_at' not in cols:
        findings.append(Finding(
            table=table.name, severity=Severity.WARNING, rule="missing_created_at",
            message=f"Table '{table.name}' is missing a 'created_at' column.",
            suggestion="Add: created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()"
        ))
    if 'updated_at' not in cols:
        findings.append(Finding(
            table=table.name, severity=Severity.WARNING, rule="missing_updated_at",
            message=f"Table '{table.name}' is missing an 'updated_at' column.",
            suggestion="Add: updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()"
        ))


def _check_fk_indexes(table: Table, findings: List[Finding]):
    indexed_cols = set()
    for idx in table.indexes:
        if idx.columns:
            indexed_cols.add(idx.columns[0])

    for cname, col in table.columns.items():
        if col.references and cname not in indexed_cols and not col.is_pk:
            findings.append(Finding(
                table=table.name, severity=Severity.ERROR, rule="missing_fk_index",
                message=f"Foreign key column '{cname}' has no index.",
                suggestion=f"CREATE INDEX idx_{table.name}_{cname} ON {table.name} ({cname});"
            ))


def _check_fk_nullable(table: Table, findings: List[Finding]):
    for cname, col in table.columns.items():
        if col.references and col.nullable:
            findings.append(Finding(
                table=table.name, severity=Severity.INFO, rule="nullable_fk",
                message=f"Foreign key column '{cname}' is nullable.",
                suggestion="Consider adding NOT NULL if the relationship is mandatory."
            ))


def _check_soft_delete_index(table: Table, findings: List[Finding]):
    if 'deleted_at' not in table.columns:
        return
    has_partial = any(
        idx.where_clause and 'deleted_at' in idx.where_clause.lower()
        for idx in table.indexes
    )
    if not has_partial:
        findings.append(Finding(
            table=table.name, severity=Severity.WARNING, rule="soft_delete_no_partial_index",
            message=f"Table '{table.name}' has soft deletes but no partial index filtering deleted rows.",
            suggestion="Add a partial index: WHERE deleted_at IS NULL on frequently queried columns."
        ))


def _check_sequential_pk(table: Table, findings: List[Finding]):
    if not table.primary_key:
        return
    for pk_col_name in table.primary_key:
        col = table.columns.get(pk_col_name)
        if col and col.data_type.upper() in ('SERIAL', 'BIGSERIAL', 'INT', 'INTEGER', 'BIGINT'):
            if col.is_pk:
                findings.append(Finding(
                    table=table.name, severity=Severity.INFO, rule="sequential_pk",
                    message=f"Primary key '{pk_col_name}' uses sequential integer type '{col.data_type}'.",
                    suggestion="Consider CUID2 or UUIDv7 for non-guessable, sortable IDs (especially if exposed in URLs)."
                ))


def _check_no_indexes(table: Table, findings: List[Finding]):
    if not table.indexes and len(table.columns) > 2:
        findings.append(Finding(
            table=table.name, severity=Severity.WARNING, rule="no_indexes",
            message=f"Table '{table.name}' has no indexes besides the primary key.",
            suggestion="Add indexes on columns used in WHERE clauses, JOINs, and ORDER BY."
        ))


def format_human(findings: List[Finding], tables: List[Table]) -> str:
    """Format findings for human-readable output."""
    lines = []
    lines.append(f"Schema Validation Report")
    lines.append(f"{'=' * 50}")
    lines.append(f"Tables analyzed: {len(tables)}")

    error_count = sum(1 for f in findings if f.severity == Severity.ERROR)
    warn_count = sum(1 for f in findings if f.severity == Severity.WARNING)
    info_count = sum(1 for f in findings if f.severity == Severity.INFO)

    lines.append(f"Findings: {error_count} errors, {warn_count} warnings, {info_count} info")
    lines.append("")

    if not findings:
        lines.append("No issues found. Schema looks good!")
        return '\n'.join(lines)

    severity_order = {Severity.ERROR: 0, Severity.WARNING: 1, Severity.INFO: 2}
    sorted_findings = sorted(findings, key=lambda f: (severity_order.get(f.severity, 3), f.table))

    icons = {Severity.ERROR: "[ERROR]", Severity.WARNING: "[WARN] ", Severity.INFO: "[INFO] "}

    for f in sorted_findings:
        icon = icons.get(f.severity, "       ")
        lines.append(f"{icon} {f.table}: {f.message}")
        lines.append(f"         -> {f.suggestion}")
        lines.append("")

    return '\n'.join(lines)


def format_json(findings: List[Finding], tables: List[Table]) -> str:
    """Format findings as JSON."""
    return json.dumps({
        "tables_analyzed": len(tables),
        "summary": {
            "errors": sum(1 for f in findings if f.severity == Severity.ERROR),
            "warnings": sum(1 for f in findings if f.severity == Severity.WARNING),
            "info": sum(1 for f in findings if f.severity == Severity.INFO),
        },
        "findings": [
            {
                "table": f.table,
                "severity": f.severity,
                "rule": f.rule,
                "message": f.message,
                "suggestion": f.suggestion,
            }
            for f in findings
        ]
    }, indent=2)


def main():
    parser = argparse.ArgumentParser(
        description="Validate SQL DDL schemas for normalization violations, missing indexes, and naming conventions."
    )
    parser.add_argument("file", help="Path to SQL DDL file to validate")
    parser.add_argument("--json", action="store_true", dest="json_output",
                        help="Output results as JSON")
    parser.add_argument("--strict", action="store_true",
                        help="Exit with code 1 on any warning (not just errors)")
    args = parser.parse_args()

    try:
        with open(args.file, 'r') as f:
            sql = f.read()
    except FileNotFoundError:
        print(f"Error: File not found: {args.file}", file=sys.stderr)
        sys.exit(2)
    except IOError as e:
        print(f"Error reading file: {e}", file=sys.stderr)
        sys.exit(2)

    tables = parse_ddl(sql)
    if not tables:
        print("Warning: No CREATE TABLE statements found in the file.", file=sys.stderr)
        sys.exit(0)

    findings = validate(tables)

    if args.json_output:
        print(format_json(findings, tables))
    else:
        print(format_human(findings, tables))

    error_count = sum(1 for f in findings if f.severity == Severity.ERROR)
    warn_count = sum(1 for f in findings if f.severity == Severity.WARNING)

    if error_count > 0:
        sys.exit(1)
    if args.strict and warn_count > 0:
        sys.exit(1)
    sys.exit(0)


if __name__ == '__main__':
    main()