#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "psycopg2-binary",
# ]
# ///
"""
Data Comparison Audit Tool for EWAH to DLT Migration

Compares data loaded by EWAH and DLT connectors to validate migration accuracy.

Comparison levels:
1. Basic Stats: Row count, column count
2. Schema Comparison: Column names (excluding metadata columns)
3. Date Range Detection: Find overlapping periods
4. Record Comparison: INTERSECT/EXCEPT on matching data

Usage:
    uv run scripts/audit_compare.py --connector facebook --tables ads_insights,campaigns
    uv run scripts/audit_compare.py --connector facebook --tables ads_insights --cursor-column date_start
"""

import argparse
import sys
import tomllib
from dataclasses import dataclass
from pathlib import Path
from typing import Optional

import psycopg2
from psycopg2.extras import RealDictCursor


# Metadata column prefixes to exclude from comparison
METADATA_PREFIXES = ("_dlt_", "_ewah_", "_sdc_", "_airbyte_")

# Common timestamp column names for auto-detection (in priority order)
TIMESTAMP_COLUMNS = [
    "updated_at",
    "modified_at",
    "created_at",
    "date_start",
    "date",
    "timestamp",
    "created",
    "updated",
    "modified",
    "last_modified",
    "sync_time",
]

# Common primary key column names for fallback
PRIMARY_KEY_COLUMNS = ["id", "uuid", "pk", "key"]


@dataclass
class DatabaseConfig:
    """Database connection configuration."""

    host: str
    port: int
    database: str
    username: str
    password: str

    def connect(self):
        """Create a database connection."""
        return psycopg2.connect(
            host=self.host,
            port=self.port,
            database=self.database,
            user=self.username,
            password=self.password,
        )


@dataclass
class ColumnTypeInfo:
    """Column type information for comparison."""

    column_name: str
    ewah_type: Optional[str]
    dlt_type: Optional[str]

    @property
    def types_match(self) -> bool:
        if self.ewah_type is None or self.dlt_type is None:
            return False
        # Normalize types for comparison
        ewah_normalized = self._normalize_type(self.ewah_type)
        dlt_normalized = self._normalize_type(self.dlt_type)
        return ewah_normalized == dlt_normalized

    @staticmethod
    def _normalize_type(t: str) -> str:
        """Normalize type names for comparison."""
        t = t.lower()
        # Group equivalent types
        if t in ("character varying", "varchar", "text"):
            return "text"
        if t in ("integer", "int", "int4", "bigint", "int8", "smallint"):
            return "integer"
        if t in ("numeric", "decimal", "double precision", "real", "float"):
            return "numeric"
        if t in ("boolean", "bool"):
            return "boolean"
        if t in ("timestamp", "timestamp without time zone", "timestamp with time zone"):
            return "timestamp"
        return t


@dataclass
class ComparisonResult:
    """Result of a table comparison."""

    table_name: str
    ewah_table: str
    dlt_table: str
    ewah_row_count: int
    dlt_row_count: int
    ewah_columns: list[str]
    dlt_columns: list[str]
    common_columns: list[str]
    ewah_only_columns: list[str]
    dlt_only_columns: list[str]
    column_types: list[ColumnTypeInfo]
    ewah_types: dict[str, str]
    dlt_types: dict[str, str]
    cursor_column: Optional[str]
    primary_key: Optional[str]
    date_range: Optional[tuple[str, str]]
    matching_records: int
    ewah_only_records: int
    dlt_only_records: int
    match_percentage: float

    @property
    def type_mismatches(self) -> list[ColumnTypeInfo]:
        """Return columns with type mismatches."""
        return [ct for ct in self.column_types if not ct.types_match]

    @property
    def has_discrepancies(self) -> bool:
        """Check if there are any discrepancies."""
        return (
            self.ewah_only_columns
            or self.dlt_only_columns
            or self.type_mismatches
            or self.match_percentage < 100.0
        )


@dataclass
class AuditConfig:
    """Audit configuration with database and schema settings."""

    db_config: DatabaseConfig
    ewah_schema: str
    dlt_schema: str


def load_secrets(connector_path: Path, connector_name: str) -> AuditConfig:
    """
    Load database credentials and schema configuration from the connector's secrets.toml.

    Returns:
        AuditConfig with database connection and schema names
    """
    secrets_path = connector_path / ".dlt" / "secrets.toml"
    if not secrets_path.exists():
        raise FileNotFoundError(
            f"secrets.toml not found at {secrets_path}\n"
            f"Copy secrets_example.toml to secrets.toml and fill in credentials."
        )

    with open(secrets_path, "rb") as f:
        secrets = tomllib.load(f)

    # Database credentials (single database for both EWAH and DLT schemas)
    db_creds = secrets.get("destination", {}).get("postgres", {}).get("credentials", {})
    if not db_creds:
        raise ValueError("Missing [destination.postgres.credentials] in secrets.toml")

    db_config = DatabaseConfig(
        host=db_creds.get("host", "localhost"),
        port=int(db_creds.get("port", 5432)),
        database=db_creds.get("database", "postgres"),
        username=db_creds.get("username"),
        password=db_creds.get("password"),
    )

    # Schema configuration (optional - defaults to ewah_<connector> and dlt_<connector>)
    audit_config = secrets.get("sources", {}).get("audit", {})
    ewah_schema = audit_config.get("ewah_schema", f"ewah_{connector_name}")
    dlt_schema = audit_config.get("dlt_schema", f"dlt_{connector_name}")

    return AuditConfig(
        db_config=db_config,
        ewah_schema=ewah_schema,
        dlt_schema=dlt_schema,
    )


def get_row_count(conn, schema: str, table: str) -> int:
    """Get row count for a table."""
    with conn.cursor() as cur:
        cur.execute(
            f'SELECT COUNT(*) FROM "{schema}"."{table}"'
        )
        return cur.fetchone()[0]


def get_columns(conn, schema: str, table: str) -> list[str]:
    """Get column names for a table, excluding metadata columns."""
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT column_name
            FROM information_schema.columns
            WHERE table_schema = %s AND table_name = %s
            ORDER BY ordinal_position
            """,
            (schema, table),
        )
        all_columns = [row[0] for row in cur.fetchall()]

    # Filter out metadata columns
    return [
        col for col in all_columns
        if not any(col.lower().startswith(prefix) for prefix in METADATA_PREFIXES)
    ]


def get_column_types(conn, schema: str, table: str) -> dict[str, str]:
    """Get column names and their data types for a table, excluding metadata columns."""
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT column_name, data_type
            FROM information_schema.columns
            WHERE table_schema = %s AND table_name = %s
            ORDER BY ordinal_position
            """,
            (schema, table),
        )
        all_columns = {row[0]: row[1] for row in cur.fetchall()}

    # Filter out metadata columns
    return {
        col: dtype for col, dtype in all_columns.items()
        if not any(col.lower().startswith(prefix) for prefix in METADATA_PREFIXES)
    }


def detect_cursor_column(columns: list[str]) -> Optional[str]:
    """Auto-detect a suitable cursor column for date filtering."""
    columns_lower = {col.lower(): col for col in columns}

    # Try timestamp columns first
    for ts_col in TIMESTAMP_COLUMNS:
        if ts_col in columns_lower:
            return columns_lower[ts_col]

    # Fallback to primary key columns
    for pk_col in PRIMARY_KEY_COLUMNS:
        if pk_col in columns_lower:
            return columns_lower[pk_col]

    return None


def detect_primary_key(columns: list[str]) -> Optional[str]:
    """Auto-detect a primary key column for ordering results using name patterns.

    This is a fallback when database connection is not available.
    Looks for common primary key patterns like *_id, id, uuid, etc.
    """
    columns_lower = {col.lower(): col for col in columns}

    # Try exact matches from PRIMARY_KEY_COLUMNS first
    for pk_col in PRIMARY_KEY_COLUMNS:
        if pk_col in columns_lower:
            return columns_lower[pk_col]

    # Then look for any columns ending with _id (common pattern)
    for col_lower, col_original in columns_lower.items():
        if col_lower.endswith("_id") and col_lower != "id":
            return col_original

    return None


def detect_primary_key_by_cardinality(
    conn, schema: str, table: str, columns: list[str]
) -> Optional[str]:
    """Auto-detect primary key by finding the ID column with highest cardinality.

    Queries a sample of 100 rows and counts distinct values per column.
    Among columns that look like IDs (ending in _id, or named id/uuid/key),
    returns the one with highest cardinality.
    """
    if not columns:
        return None

    # Build query to count distinct values for each column
    distinct_counts = []
    for col in columns:
        distinct_counts.append(f'COUNT(DISTINCT "{col}") AS "{col}"')

    count_list = ", ".join(distinct_counts)

    with conn.cursor() as cur:
        # Sample 100 rows to keep the query fast
        cur.execute(
            f'SELECT {count_list} FROM (SELECT * FROM "{schema}"."{table}" LIMIT 100) AS sample'
        )
        result = cur.fetchone()

        if not result:
            return None

        # Build cardinality map
        cardinality = {col: result[i] for i, col in enumerate(columns)}

        # Filter to ID-like columns only (more reliable for ordering)
        id_columns = [
            col for col in columns
            if col.lower().endswith("_id")
            or col.lower() in ("id", "uuid", "pk", "key")
        ]

        if id_columns:
            # Among ID columns, pick the one with highest cardinality
            best_id = max(id_columns, key=lambda c: cardinality[c])
            return best_id

        # If no ID columns, return highest cardinality overall
        return max(columns, key=lambda c: cardinality[c])


def get_date_range(conn, schema: str, table: str, cursor_column: str) -> tuple[str, str]:
    """Get min and max values for a cursor column."""
    with conn.cursor() as cur:
        cur.execute(
            f'SELECT MIN("{cursor_column}")::text, MAX("{cursor_column}")::text '
            f'FROM "{schema}"."{table}"'
        )
        result = cur.fetchone()
        return (result[0], result[1])


def find_overlapping_range(
    ewah_range: tuple[str, str], dlt_range: tuple[str, str]
) -> Optional[tuple[str, str]]:
    """Find the overlapping date range between EWAH and DLT data."""
    if ewah_range[0] is None or dlt_range[0] is None:
        return None

    # Take the later start and earlier end
    start = max(ewah_range[0], dlt_range[0])
    end = min(ewah_range[1], dlt_range[1])

    if start <= end:
        return (start, end)
    return None


def _build_normalized_cast(col: str, source_type: str, target_type: str) -> str:
    """
    Build a SQL cast expression that normalizes a column to text via a common type.

    When EWAH stores a value as text but DLT stores it as a proper type (e.g.,
    timestamptz, bigint), casting both to ::text produces different representations.
    This function casts the text column through the target type first, so both sides
    produce identical text output.

    Args:
        col: Column name
        source_type: The column's actual database type (from information_schema)
        target_type: The other side's database type to normalize towards

    Returns:
        SQL expression like '"col"::timestamptz::text' or '"col"::text'
    """
    source_norm = ColumnTypeInfo._normalize_type(source_type)
    target_norm = ColumnTypeInfo._normalize_type(target_type)

    # If types already match after normalization, simple cast
    # Exception: integer types should go through numeric for consistent formatting
    if source_norm == target_norm:
        if source_norm == "integer":
            return f'"{col}"::numeric::text'
        return f'"{col}"::text'

    # EWAH text -> DLT has a proper type: cast text through the target type
    if source_norm == "text" and target_norm in ("timestamp", "integer", "numeric", "boolean"):
        # For integer types, cast through numeric first to handle decimal strings like "0.5"
        if target_norm == "integer":
            return f'"{col}"::numeric::text'
        pg_type = target_type  # use the actual pg type, not normalized
        return f'"{col}"::{pg_type}::text'

    # Fallback: just cast to text
    return f'"{col}"::text'


def compare_records(
    conn,
    ewah_schema: str,
    dlt_schema: str,
    ewah_table: str,
    dlt_table: str,
    columns: list[str],
    cursor_column: Optional[str] = None,
    date_range: Optional[tuple[str, str]] = None,
    ewah_types: Optional[dict[str, str]] = None,
    dlt_types: Optional[dict[str, str]] = None,
) -> tuple[int, int, int]:
    """
    Compare records between EWAH and DLT tables using INTERSECT/EXCEPT.

    Uses a single database connection with different schemas. When type info is
    provided, normalizes mismatched columns (e.g., EWAH text vs DLT timestamptz)
    by casting through a common type before comparing.

    Returns:
        Tuple of (matching_count, ewah_only_count, dlt_only_count)
    """
    # Build column lists with type-aware casting
    if ewah_types and dlt_types:
        ewah_col_list = ", ".join(
            _build_normalized_cast(col, ewah_types.get(col, "text"), dlt_types.get(col, "text"))
            for col in columns
        )
        dlt_col_list = ", ".join(
            _build_normalized_cast(col, dlt_types.get(col, "text"), ewah_types.get(col, "text"))
            for col in columns
        )
    else:
        ewah_col_list = ", ".join(f'"{col}"::text' for col in columns)
        dlt_col_list = ewah_col_list

    # Build WHERE clause for date filtering
    where_clause = ""
    if cursor_column and date_range:
        where_clause = (
            f' WHERE "{cursor_column}" >= \'{date_range[0]}\' '
            f'AND "{cursor_column}" <= \'{date_range[1]}\''
        )

    # Get EWAH data
    ewah_query = f'SELECT {ewah_col_list} FROM "{ewah_schema}"."{ewah_table}"{where_clause}'
    with conn.cursor() as cur:
        cur.execute(ewah_query)
        ewah_data = set(cur.fetchall())

    # Get DLT data
    dlt_query = f'SELECT {dlt_col_list} FROM "{dlt_schema}"."{dlt_table}"{where_clause}'
    with conn.cursor() as cur:
        cur.execute(dlt_query)
        dlt_data = set(cur.fetchall())

    # Calculate set operations
    matching = len(ewah_data & dlt_data)
    ewah_only = len(ewah_data - dlt_data)
    dlt_only = len(dlt_data - ewah_data)

    return matching, ewah_only, dlt_only


def generate_audit_query(
    ewah_schema: str,
    dlt_schema: str,
    ewah_table: str,
    dlt_table: str,
    columns: list[str],
    cursor_column: Optional[str] = None,
    primary_key: Optional[str] = None,
    date_range: Optional[tuple[str, str]] = None,
    ewah_types: Optional[dict[str, str]] = None,
    dlt_types: Optional[dict[str, str]] = None,
) -> str:
    """Generate SQL query for manual audit comparison.

    Generates a single query that compares data across two schemas in the same database.
    The detailed query orders by primary_key so matching record pairs appear together.
    When type info is provided, normalizes mismatched columns through a common type.
    """
    # Build type-aware column lists
    if ewah_types and dlt_types:
        ewah_col_list = ",\n        ".join(
            f'{_build_normalized_cast(col, ewah_types.get(col, "text"), dlt_types.get(col, "text"))} AS "{col}"'
            for col in columns
        )
        dlt_col_list = ",\n        ".join(
            f'{_build_normalized_cast(col, dlt_types.get(col, "text"), ewah_types.get(col, "text"))} AS "{col}"'
            for col in columns
        )
    else:
        ewah_col_list = ",\n        ".join(f'"{col}"::text AS "{col}"' for col in columns)
        dlt_col_list = ewah_col_list

    where_clause = ""
    if cursor_column and date_range:
        where_clause = (
            f'\n    WHERE "{cursor_column}" >= \'{date_range[0]}\' '
            f'AND "{cursor_column}" <= \'{date_range[1]}\''
        )

    # Build ORDER BY clause - use primary key if available, otherwise first column
    order_column = primary_key if primary_key else columns[0] if columns else None
    order_clause = f'ORDER BY "{order_column}", in_ewah DESC' if order_column else ""

    query = f'''-- Audit comparison query: {ewah_schema}.{ewah_table} vs {dlt_schema}.{dlt_table}
-- Run this query against the postgres database to compare EWAH and DLT data
-- Generated by audit_compare.py

WITH ewah_data AS (
    SELECT
        {ewah_col_list}
    FROM "{ewah_schema}"."{ewah_table}"{where_clause}
),

dlt_data AS (
    SELECT
        {dlt_col_list}
    FROM "{dlt_schema}"."{dlt_table}"{where_clause}
),

matching AS (
    SELECT * FROM ewah_data
    INTERSECT
    SELECT * FROM dlt_data
),

ewah_only AS (
    SELECT * FROM ewah_data
    EXCEPT
    SELECT * FROM dlt_data
),

dlt_only AS (
    SELECT * FROM dlt_data
    EXCEPT
    SELECT * FROM ewah_data
),

all_records AS (
    SELECT *, true AS in_ewah, true AS in_dlt FROM matching
    UNION ALL
    SELECT *, true AS in_ewah, false AS in_dlt FROM ewah_only
    UNION ALL
    SELECT *, false AS in_ewah, true AS in_dlt FROM dlt_only
),

summary AS (
    SELECT
        in_ewah,
        in_dlt,
        COUNT(*) AS count,
        ROUND(100.0 * COUNT(*) / NULLIF(SUM(COUNT(*)) OVER (), 0), 2) AS percent_of_total
    FROM all_records
    GROUP BY in_ewah, in_dlt
    ORDER BY in_ewah DESC, in_dlt DESC
)

-- Show summary (should be 100% matching ideally)
-- SELECT * FROM summary;

-- Detailed mismatches ordered by primary key (pairs of EWAH/DLT records together)
SELECT * FROM all_records WHERE NOT (in_ewah AND in_dlt)
{order_clause}
LIMIT 100;
'''
    return query


def compare_table(
    conn,
    ewah_schema: str,
    dlt_schema: str,
    ewah_table: str,
    dlt_table: str,
    cursor_column: Optional[str] = None,
) -> ComparisonResult:
    """Run full comparison for a single table using a single database connection."""
    print(f"\n{'='*60}")
    print(f"Comparing: {ewah_schema}.{ewah_table} (EWAH) vs {dlt_schema}.{dlt_table} (DLT)")
    print("=" * 60)

    # Basic stats
    print("\n[1/5] Basic Stats...")
    try:
        ewah_row_count = get_row_count(conn, ewah_schema, ewah_table)
    except psycopg2.errors.UndefinedTable:
        print(f"  ERROR: Table {ewah_schema}.{ewah_table} not found")
        raise

    try:
        dlt_row_count = get_row_count(conn, dlt_schema, dlt_table)
    except psycopg2.errors.UndefinedTable:
        print(f"  ERROR: Table {dlt_schema}.{dlt_table} not found")
        raise

    print(f"  EWAH rows: {ewah_row_count:,}")
    print(f"  DLT rows:  {dlt_row_count:,}")

    # Schema comparison
    print("\n[2/5] Schema Comparison...")
    ewah_columns = get_columns(conn, ewah_schema, ewah_table)
    dlt_columns = get_columns(conn, dlt_schema, dlt_table)
    ewah_types = get_column_types(conn, ewah_schema, ewah_table)
    dlt_types = get_column_types(conn, dlt_schema, dlt_table)

    ewah_col_set = set(ewah_columns)
    dlt_col_set = set(dlt_columns)

    common_columns = sorted(ewah_col_set & dlt_col_set)
    ewah_only_columns = sorted(ewah_col_set - dlt_col_set)
    dlt_only_columns = sorted(dlt_col_set - ewah_col_set)

    # Collect column type information for common columns
    column_types = [
        ColumnTypeInfo(
            column_name=col,
            ewah_type=ewah_types.get(col),
            dlt_type=dlt_types.get(col),
        )
        for col in common_columns
    ]
    type_mismatches = [ct for ct in column_types if not ct.types_match]

    print(f"  EWAH columns: {len(ewah_columns)}")
    print(f"  DLT columns:  {len(dlt_columns)}")
    print(f"  Common columns: {len(common_columns)}")

    if ewah_only_columns:
        print(f"  EWAH-only columns: {ewah_only_columns}")
    if dlt_only_columns:
        print(f"  DLT-only columns: {dlt_only_columns}")
    if type_mismatches:
        print(f"  Type mismatches: {len(type_mismatches)}")
        for tm in type_mismatches:
            print(f"    - {tm.column_name}: {tm.ewah_type} (EWAH) vs {tm.dlt_type} (DLT)")

    # Date range detection
    print("\n[3/5] Date Range Detection...")
    detected_cursor = cursor_column or detect_cursor_column(common_columns)

    date_range = None
    if detected_cursor:
        print(f"  Using cursor column: {detected_cursor}")
        ewah_range = get_date_range(conn, ewah_schema, ewah_table, detected_cursor)
        dlt_range = get_date_range(conn, dlt_schema, dlt_table, detected_cursor)
        print(f"  EWAH range: {ewah_range[0]} to {ewah_range[1]}")
        print(f"  DLT range:  {dlt_range[0]} to {dlt_range[1]}")

        date_range = find_overlapping_range(ewah_range, dlt_range)
        if date_range:
            print(f"  Overlapping range: {date_range[0]} to {date_range[1]}")
        else:
            print("  WARNING: No overlapping date range found!")
    else:
        print("  No cursor column detected - comparing all records")

    # Primary key detection (for ordering results)
    print("\n[4/5] Primary Key Detection...")
    detected_primary_key = detect_primary_key_by_cardinality(
        conn, ewah_schema, ewah_table, common_columns
    )
    if detected_primary_key:
        print(f"  Using primary key (highest cardinality): {detected_primary_key}")
    else:
        # Fallback to pattern-based detection
        detected_primary_key = detect_primary_key(common_columns)
        if detected_primary_key:
            print(f"  Using primary key (pattern match): {detected_primary_key}")
        else:
            print("  No primary key detected - results will use first column for ordering")

    # Record comparison
    print("\n[5/5] Record Comparison...")
    if not common_columns:
        print("  ERROR: No common columns to compare!")
        matching, ewah_only, dlt_only = 0, 0, 0
    else:
        matching, ewah_only, dlt_only = compare_records(
            conn,
            ewah_schema,
            dlt_schema,
            ewah_table,
            dlt_table,
            common_columns,
            detected_cursor,
            date_range,
            ewah_types,
            dlt_types,
        )

    total = matching + ewah_only + dlt_only
    match_pct = (matching / total * 100) if total > 0 else 0

    print(f"  Matching records: {matching:,}")
    print(f"  EWAH-only records: {ewah_only:,}")
    print(f"  DLT-only records: {dlt_only:,}")
    print(f"  Match percentage: {match_pct:.2f}%")

    # Use DLT table name as the primary identifier
    display_name = f"{ewah_table}:{dlt_table}" if ewah_table != dlt_table else dlt_table

    return ComparisonResult(
        table_name=display_name,
        ewah_table=ewah_table,
        dlt_table=dlt_table,
        ewah_row_count=ewah_row_count,
        dlt_row_count=dlt_row_count,
        ewah_columns=ewah_columns,
        dlt_columns=dlt_columns,
        common_columns=common_columns,
        ewah_only_columns=ewah_only_columns,
        dlt_only_columns=dlt_only_columns,
        column_types=column_types,
        ewah_types=ewah_types,
        dlt_types=dlt_types,
        cursor_column=detected_cursor,
        primary_key=detected_primary_key,
        date_range=date_range,
        matching_records=matching,
        ewah_only_records=ewah_only,
        dlt_only_records=dlt_only,
        match_percentage=match_pct,
    )


def save_audit_queries(
    connector: str,
    ewah_schema: str,
    dlt_schema: str,
    results: list[ComparisonResult],
    migration_docs_path: Path,
) -> Path:
    """Save generated audit queries to migration-docs/<connector>/audit_queries.sql."""
    output_dir = migration_docs_path / connector
    output_dir.mkdir(parents=True, exist_ok=True)

    output_file = output_dir / "audit_queries.sql"

    queries = []
    for result in results:
        query = generate_audit_query(
            ewah_schema=ewah_schema,
            dlt_schema=dlt_schema,
            ewah_table=result.ewah_table,
            dlt_table=result.dlt_table,
            columns=result.common_columns,
            cursor_column=result.cursor_column,
            primary_key=result.primary_key,
            date_range=result.date_range,
            ewah_types=result.ewah_types,
            dlt_types=result.dlt_types,
        )
        queries.append(query)

    with open(output_file, "w") as f:
        f.write("\n\n".join(queries))

    return output_file


def generate_discrepancy_report(
    connector: str,
    ewah_schema: str,
    dlt_schema: str,
    results: list[ComparisonResult],
) -> str:
    """Generate a markdown discrepancy report."""
    from datetime import datetime

    lines = [
        f"# Data Validation Discrepancy Report: {connector}",
        "",
        f"**Generated:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
        f"**EWAH Schema:** `{ewah_schema}`",
        f"**DLT Schema:** `{dlt_schema}`",
        "",
    ]

    # Overall summary
    total_tables = len(results)
    passed_tables = sum(1 for r in results if r.match_percentage == 100.0)
    tables_with_discrepancies = sum(1 for r in results if r.has_discrepancies)

    lines.extend([
        "## Summary",
        "",
        f"| Metric | Value |",
        f"|--------|-------|",
        f"| Tables compared | {total_tables} |",
        f"| Tables passed (100% match) | {passed_tables} |",
        f"| Tables with discrepancies | {tables_with_discrepancies} |",
        "",
    ])

    # Per-table details
    for result in results:
        lines.extend([
            f"## Table: `{result.table_name}`",
            "",
            f"| Property | EWAH | DLT |",
            f"|----------|------|-----|",
            f"| Table name | `{result.ewah_table}` | `{result.dlt_table}` |",
            f"| Row count | {result.ewah_row_count:,} | {result.dlt_row_count:,} |",
            f"| Column count | {len(result.ewah_columns)} | {len(result.dlt_columns)} |",
            f"| Common columns | {len(result.common_columns)} | {len(result.common_columns)} |",
            "",
        ])

        # Record comparison
        lines.extend([
            "### Record Comparison",
            "",
            f"| Metric | Count | Percentage |",
            f"|--------|-------|------------|",
            f"| Matching records | {result.matching_records:,} | {result.match_percentage:.2f}% |",
            f"| EWAH-only records | {result.ewah_only_records:,} | - |",
            f"| DLT-only records | {result.dlt_only_records:,} | - |",
            "",
        ])

        if result.date_range:
            lines.extend([
                f"**Comparison date range:** `{result.date_range[0]}` to `{result.date_range[1]}`",
                "",
            ])

        # Column discrepancies
        if result.ewah_only_columns or result.dlt_only_columns:
            lines.extend([
                "### Column Mismatches",
                "",
            ])

            if result.ewah_only_columns:
                lines.append(f"**EWAH-only columns:** `{'`, `'.join(result.ewah_only_columns)}`")
                lines.append("")

            if result.dlt_only_columns:
                lines.append(f"**DLT-only columns:** `{'`, `'.join(result.dlt_only_columns)}`")
                lines.append("")

        # Type mismatches
        type_mismatches = result.type_mismatches
        if type_mismatches:
            lines.extend([
                "### Data Type Mismatches",
                "",
                "| Column | EWAH Type | DLT Type | Action Required |",
                "|--------|-----------|----------|-----------------|",
            ])

            for tm in type_mismatches:
                action = "Cast in dbt staging model"
                lines.append(f"| `{tm.column_name}` | `{tm.ewah_type}` | `{tm.dlt_type}` | {action} |")

            lines.append("")

        # Status
        status = "PASS" if result.match_percentage == 100.0 and not result.has_discrepancies else "NEEDS ATTENTION"
        lines.extend([
            f"**Status:** {status}",
            "",
            "---",
            "",
        ])

    return "\n".join(lines)


def save_discrepancy_report(
    connector: str,
    ewah_schema: str,
    dlt_schema: str,
    results: list[ComparisonResult],
    migration_docs_path: Path,
) -> Path:
    """Save discrepancy report to migration-docs/<connector>/DISCREPANCY_REPORT.md."""
    output_dir = migration_docs_path / connector
    output_dir.mkdir(parents=True, exist_ok=True)

    output_file = output_dir / "DISCREPANCY_REPORT.md"

    report = generate_discrepancy_report(connector, ewah_schema, dlt_schema, results)

    with open(output_file, "w") as f:
        f.write(report)

    return output_file


def print_summary(results: list[ComparisonResult]) -> None:
    """Print summary of all table comparisons."""
    print("\n" + "=" * 60)
    print("SUMMARY")
    print("=" * 60)

    all_passed = True
    for r in results:
        status = "PASS" if r.match_percentage == 100.0 else "FAIL"
        if status == "FAIL":
            all_passed = False
        print(
            f"  {r.table_name}: {status} "
            f"({r.match_percentage:.2f}% match, "
            f"{r.ewah_row_count:,} EWAH / {r.dlt_row_count:,} DLT rows)"
        )

    print("\n" + ("ALL TABLES PASSED" if all_passed else "SOME TABLES FAILED"))


def main():
    parser = argparse.ArgumentParser(
        description="Compare data between EWAH and DLT loaded tables"
    )
    parser.add_argument(
        "--connector",
        required=True,
        help="Connector name (e.g., facebook, pipedrive)",
    )
    parser.add_argument(
        "--tables",
        required=True,
        help="Comma-separated list of tables to compare (format: dlt_table or ewah_table:dlt_table)",
    )
    parser.add_argument(
        "--cursor-column",
        help="Column to use for date range filtering (auto-detected if not specified)",
    )
    parser.add_argument(
        "--schema",
        help="Schema name (defaults to connector name)",
    )
    parser.add_argument(
        "--ewah-schema",
        help="EWAH schema name (defaults to --schema)",
    )

    args = parser.parse_args()

    # Resolve paths
    script_dir = Path(__file__).parent
    workspace_root = script_dir.parent
    dlt_connectors_path = workspace_root / "dlt-connectors" / "connectors" / args.connector
    migration_docs_path = workspace_root / "migration-docs"

    # Parse table mappings (format: "ewah_table:dlt_table" or just "table")
    table_mappings = []
    for t in args.tables.split(","):
        t = t.strip()
        if ":" in t:
            ewah_table, dlt_table = t.split(":", 1)
            table_mappings.append((ewah_table.strip(), dlt_table.strip()))
        else:
            table_mappings.append((t, t))

    # Load configuration
    try:
        audit_config = load_secrets(dlt_connectors_path, args.connector)
    except (FileNotFoundError, ValueError) as e:
        print(f"\nERROR: {e}")
        sys.exit(1)

    # Use schema from config if not overridden by CLI args
    ewah_schema = args.ewah_schema or audit_config.ewah_schema
    dlt_schema = args.schema or audit_config.dlt_schema

    print(f"Connector: {args.connector}")
    print(f"Database: {audit_config.db_config.database} @ {audit_config.db_config.host}:{audit_config.db_config.port}")
    print(f"EWAH schema: {ewah_schema}")
    print(f"DLT schema: {dlt_schema}")
    print(f"Table mappings: {table_mappings}")

    # Connect to database
    try:
        conn = audit_config.db_config.connect()
    except psycopg2.Error as e:
        print(f"\nERROR: Failed to connect to database: {e}")
        sys.exit(1)

    # Run comparisons
    results = []
    for ewah_table, dlt_table in table_mappings:
        try:
            result = compare_table(
                conn,
                ewah_schema,
                dlt_schema,
                ewah_table,
                dlt_table,
                args.cursor_column,
            )
            results.append(result)
        except psycopg2.errors.UndefinedTable:
            continue
        except Exception as e:
            print(f"\nERROR comparing {ewah_table}:{dlt_table}: {e}")
            continue

    # Close connection
    conn.close()

    if not results:
        print("\nNo tables were successfully compared.")
        sys.exit(1)

    # Save audit queries
    queries_file = save_audit_queries(args.connector, ewah_schema, dlt_schema, results, migration_docs_path)
    print(f"\nAudit queries saved to: {queries_file}")

    # Save discrepancy report
    report_file = save_discrepancy_report(
        args.connector, ewah_schema, dlt_schema, results, migration_docs_path
    )
    print(f"Discrepancy report saved to: {report_file}")

    # Print summary
    print_summary(results)

    # Exit with error if any table failed
    if any(r.match_percentage < 100.0 for r in results):
        sys.exit(1)


if __name__ == "__main__":
    main()