schema_diff_analyzer.py
import anthropic
import subprocess
import json
from typing import NamedTuple

client = anthropic.Anthropic()


class SchemaDiffResult(NamedTuple):
    raw_diff: str
    safety_assessment: str
    migration_sql: str
    rollback_sql: str


def get_schema_dump(db_url: str) -> str:
    """Dump schema using pg_dump."""
    result = subprocess.run(
        ["pg_dump", "--schema-only", "--no-owner", "--no-acl", db_url],
        capture_output=True, text=True, check=True
    )
    return result.stdout


def analyze_schema_diff(
    old_schema: str,
    new_schema: str,
    table_row_counts: dict[str, int] | None = None
) -> SchemaDiffResult:
    """Use Claude to analyze schema differences and generate migration plan."""

    row_count_context = ""
    if table_row_counts:
        row_count_context = "Table row counts:\n" + "\n".join(
            f"  {table}: {count:,} rows"
            for table, count in table_row_counts.items()
        )

    message = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": f"""Analyze the differences between these two PostgreSQL schemas.

OLD SCHEMA:
{old_schema[:8000]}

NEW SCHEMA:
{new_schema[:8000]}

{row_count_context}

Provide:
1. SAFETY ASSESSMENT: List each change with risk level (LOW/MEDIUM/HIGH) and reason
2. MIGRATION SQL: The ALTER statements to migrate from old to new schema, in correct order
3. ROLLBACK SQL: The statements to undo the migration
4. ONLINE SAFE: Which changes can run without a maintenance window
5. WARNINGS: Any potential breaking changes for application code

Format your response as JSON with keys: safety_assessment, migration_sql, rollback_sql, warnings"""
        }]
    )

    response_text = message.content[0].text

    # Parse JSON response
    try:
        # Claude sometimes wraps in ```json
 if "```json" in response_text:
            json_str = response_text.split("```json")[1].split("```")[0]
        else:
            json_str = response_text

        data = json.loads(json_str)
        return SchemaDiffResult(
            raw_diff=f"Diff between provided schemas",
            safety_assessment=data.get("safety_assessment", ""),
            migration_sql=data.get("migration_sql", ""),
            rollback_sql=data.get("rollback_sql", ""),
        )
    except json.JSONDecodeError:
        # Fall back to raw text
        return SchemaDiffResult(
            raw_diff="Parse error",
            safety_assessment=response_text,
            migration_sql="",
            rollback_sql="",
        )


def check_migration_safety(migration_file: str, db_url: str) -> dict:
    """Check a migration file for unsafe operations."""
    migration_sql = open(migration_file).read()

    # Get approximate row counts for affected tables
    affected_tables = extract_affected_tables(migration_sql)
    row_counts = get_table_row_counts(db_url, affected_tables)

    message = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"""Review this PostgreSQL migration for safety issues.

Migration SQL:
{migration_sql}

Table sizes:
{json.dumps(row_counts, indent=2)}

Identify:
1. Operations that require full table locks (AccessExclusiveLock)
2. Operations that rewrite the table (sequential scan + write)
3. Estimated lock duration for large tables
4. Safer alternatives (e.g., CREATE INDEX CONCURRENTLY, batched updates)
5. Overall risk: LOW / MEDIUM / HIGH

Be specific about PostgreSQL internals."""
        }]
    )

    return {"analysis": message.content[0].text, "row_counts": row_counts}


def extract_affected_tables(sql: str) -> list[str]:
    """Extract table names from SQL statements."""
    import re
    tables = re.findall(
        r'(?:ALTER TABLE|UPDATE|DROP TABLE|CREATE INDEX ON)\s+"?(\w+)"?',
        sql, re.IGNORECASE
    )
    return list(set(tables))


def get_table_row_counts(db_url: str, tables: list[str]) -> dict[str, int]:
    """Get approximate row counts from pg_stat_user_tables."""
    import psycopg2
    counts = {}
    if not tables:
        return counts
    try:
        conn = psycopg2.connect(db_url)
        cur = conn.cursor()
        placeholders = ','.join(['%s'] * len(tables))
        cur.execute(
            f"SELECT relname, n_live_tup FROM pg_stat_user_tables "
            f"WHERE relname IN ({placeholders})",
            tables
        )
        counts = dict(cur.fetchall())
        conn.close()
    except Exception:
        pass
    return counts