import logging
from typing import Optional

import sqlglot
from semantic_models.exceptions import ValidationError
from semantic_models.logging_utils import summarize_for_logging, truncate_text
from semantic_models.schemas import SemanticModelSpec
from semantic_models.services.sql_execution import resolve_table_name
from sqlglot import exp

logger = logging.getLogger(__name__)
SQL_LOG_CHARS = 3000
TABLE_LIST_LOG_ITEMS = 200


def normalize_table_name(raw_table: str, dialect: str) -> tuple[str, str]:
    read_dialect = dialect if dialect and dialect != "generic" else None
    try:
        # Parse as a simple SELECT to extract table expression
        tree = sqlglot.parse_one(f"SELECT * FROM {raw_table}", read=read_dialect)
        table_expr = next(tree.find_all(exp.Table), None)
        if table_expr:
            parts = [p for p in [table_expr.catalog, table_expr.db, table_expr.name] if p]
            full_name = ".".join(parts).lower()
            base_name = table_expr.name.lower() if table_expr.name else ""
            return full_name, base_name
    except Exception:
        pass

    # Fallback: strip quotes manually
    stripped = raw_table.replace('"', "").replace("`", "").replace("[", "").replace("]", "")
    parts = stripped.split(".")
    return stripped.lower(), parts[-1].lower() if parts else ""


def validate_sql(
    sql: str,
    dialect: str = "generic",
    spec: Optional[SemanticModelSpec] = None,
) -> None:
    logger.info(
        "SQL validation started: dialect=%s has_spec=%s sql=%s",
        dialect or "generic",
        spec is not None,
        truncate_text(sql, max_chars=SQL_LOG_CHARS),
    )

    read_dialect = dialect if dialect and dialect != "generic" else None

    try:
        if read_dialect:
            statements = sqlglot.parse(sql, read=read_dialect)
        else:
            statements = sqlglot.parse(sql)
    except Exception as e:
        logger.info("SQL validation failed: %s", truncate_text(f"Invalid SQL syntax: {e}", max_chars=1000))
        raise ValidationError(f"Invalid SQL syntax: {e}")

    if len(statements) != 1:
        logger.info("SQL validation failed: only a single SQL statement is allowed")
        raise ValidationError("Only a single SQL statement is allowed")

    statement = statements[0]
    if statement is None:
        logger.info("SQL validation failed: empty SQL statement")
        raise ValidationError("Empty SQL statement")

    forbidden = (exp.Insert, exp.Update, exp.Delete, exp.Create, exp.Alter, exp.Drop)
    for node_type in forbidden:
        if statement.find(node_type):
            logger.info("SQL validation failed: forbidden operation detected (%s)", node_type.__name__)
            raise ValidationError("Only read-only SELECT queries are allowed")

    # Block SELECT INTO (creates tables)
    if statement.find(exp.Into):
        logger.info("SQL validation failed: SELECT INTO is not allowed")
        raise ValidationError("SELECT INTO is not allowed")

    # Block locking clauses (FOR UPDATE, FOR SHARE, etc.)
    if statement.find(exp.Lock):
        logger.info("SQL validation failed: locking clause is not allowed")
        raise ValidationError("Locking clauses (FOR UPDATE, FOR SHARE) are not allowed")

    root = statement
    if isinstance(root, exp.With):
        root = root.this
    if not isinstance(root, (exp.Select, exp.Union)):
        logger.info("SQL validation failed: query type is %s, expected SELECT/UNION", type(root).__name__)
        raise ValidationError("Only SELECT queries are allowed")

    if spec is not None:
        validate_tables_against_spec(statement, spec, dialect)

    logger.info("SQL validation passed: dialect=%s has_spec=%s", dialect or "generic", spec is not None)


def validate_tables_against_spec(
    statement: exp.Expression,
    spec: SemanticModelSpec,
    dialect: str,
) -> None:
    # Build allowed-table sets from semantic model entities.
    # For security, only tables that are unqualified in the semantic model can be
    # referenced unqualified in SQL. Qualified semantic-model tables must be
    # referenced with matching schema/catalog.
    allowed_full_tables: set[str] = set()
    allowed_unqualified_tables: set[str] = set()
    for entity in spec.entities:
        real_name = resolve_table_name(entity, dialect)
        full_normalized, base_name = normalize_table_name(real_name, dialect)
        allowed_full_tables.add(full_normalized)
        if base_name and "." not in full_normalized:
            allowed_unqualified_tables.add(base_name)

    # Collect CTE names to ignore
    cte_names: set[str] = set()
    for cte in statement.find_all(exp.CTE):
        name = cte.alias_or_name
        if name:
            cte_names.add(name.lower())

    # Find all referenced tables
    unknown_tables: set[str] = set()
    referenced_tables: set[str] = set()
    for table_expr in statement.find_all(exp.Table):
        table_name = table_expr.name
        if not table_name:
            continue

        table_name_lower = table_name.lower()
        referenced_tables.add(table_name)

        # Skip CTE references
        if table_name_lower in cte_names:
            continue

        # Check if table is allowed
        # SECURITY: If the query specifies a schema/catalog, require fully-qualified match
        # to prevent schema bypass attacks (e.g., other_schema.allowed_table)
        is_qualified = bool(table_expr.db or table_expr.catalog)

        if is_qualified:
            # Build fully-qualified name and require exact match
            full_name_parts = []
            if table_expr.catalog:
                full_name_parts.append(table_expr.catalog)
            if table_expr.db:
                full_name_parts.append(table_expr.db)
            full_name_parts.append(table_name)
            full_name = ".".join(full_name_parts).lower()

            if full_name not in allowed_full_tables:
                unknown_tables.add(full_name)
        else:
            # Unqualified table name - only allow if semantic model itself is unqualified
            if table_name_lower not in allowed_unqualified_tables:
                unknown_tables.add(table_name)

    if unknown_tables:
        logger.info(
            "SQL table validation failed: unknown_tables=%s referenced_tables=%s",
            summarize_for_logging(sorted(unknown_tables), max_chars=150, max_items=TABLE_LIST_LOG_ITEMS),
            summarize_for_logging(sorted(referenced_tables), max_chars=150, max_items=TABLE_LIST_LOG_ITEMS),
        )
        raise ValidationError(f"Query references tables not in the semantic model: {sorted(unknown_tables)}")
