#!/usr/bin/env python3
"""Validate generated BigQuery SQL against the Manual Invoice read-only policy."""

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any, Optional

from runtime_config import load_runtime_config
from semantic_model import is_table_allowed, load_semantic_model


def _function_name(function: Any) -> str:
    from sqlglot import exp

    if isinstance(function, exp.Anonymous):
        return str(function.name).upper()
    try:
        return str(function.sql_name()).upper()
    except (AttributeError, TypeError):
        return type(function).__name__.upper()


def _table_id(table: Any) -> Optional[str]:
    if not table.name:
        return None
    parts = [str(table.catalog or ""), str(table.db or ""), str(table.name or "")]
    if not all(parts):
        return None
    return ".".join(parts)


def validate_sql(
    sql: str,
    semantic_model_path: Optional[Path] = None,
    start: Optional[Path] = None,
) -> dict[str, Any]:
    try:
        import sqlglot
        from sqlglot import exp
        from sqlglot.errors import ParseError
    except ImportError as exc:
        raise RuntimeError("sqlglot is required; run bootstrap.py.") from exc

    model = load_semantic_model(semantic_model_path, start=start)
    policy = model.get("policy", {})
    errors: list[str] = []
    warnings: list[str] = []
    referenced_tables: list[str] = []

    try:
        statements = sqlglot.parse(sql, read="bigquery", error_level="RAISE")
    except ParseError as exc:
        return {
            "valid": False,
            "errors": [f"BigQuery SQL parse failed: {exc}"],
            "warnings": [],
            "referenced_tables": [],
            "semantic_model_path": model["_path"],
        }

    statements = [statement for statement in statements if statement is not None]
    if len(statements) != 1:
        errors.append("SQL must contain exactly one statement.")
    if not statements:
        errors.append("SQL is empty.")
        return {
            "valid": False,
            "errors": errors,
            "warnings": warnings,
            "referenced_tables": referenced_tables,
            "semantic_model_path": model["_path"],
        }

    tree = statements[0]
    if not isinstance(tree, (exp.Query, exp.Subquery)):
        errors.append(
            f"Only read-only SELECT/WITH/UNION queries are allowed; found {tree.key.upper()}."
        )

    forbidden_functions = {
        str(name).upper() for name in policy.get("forbidden_functions", [])
    }
    found_forbidden = sorted(
        {
            name
            for function in tree.find_all(exp.Func)
            if (name := _function_name(function)) in forbidden_functions
        }
    )
    for name in found_forbidden:
        errors.append(f"Function {name} is forbidden by the semantic policy.")

    cte_aliases = {
        str(cte.alias_or_name).lower()
        for cte in tree.find_all(exp.CTE)
        if cte.alias_or_name
    }
    require_fully_qualified = bool(policy.get("require_fully_qualified_tables", True))
    seen_tables: set[str] = set()
    for table in tree.find_all(exp.Table):
        if not table.name:
            continue
        if not table.catalog and not table.db and str(table.name).lower() in cte_aliases:
            continue
        table_id = _table_id(table)
        if table_id is None:
            rendered = table.sql(dialect="bigquery")
            if require_fully_qualified:
                errors.append(
                    f"Physical table {rendered!r} must be fully qualified as project.dataset.table."
                )
            continue
        if table_id not in seen_tables:
            referenced_tables.append(table_id)
            seen_tables.add(table_id)
        if not is_table_allowed(model, table_id):
            errors.append(f"Table {table_id} is outside the configured allowlist.")

    projection_stars = [
        star for star in tree.find_all(exp.Star) if not isinstance(star.parent, exp.Count)
    ]
    if policy.get("warn_on_select_star", True) and projection_stars:
        warnings.append(
            "SELECT * was found. Use explicit columns for reviewed or frozen billing queries."
        )

    return {
        "valid": not errors,
        "errors": errors,
        "warnings": warnings,
        "referenced_tables": referenced_tables,
        "semantic_model_path": model["_path"],
    }


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Validate read-only BigQuery SQL.")
    source = parser.add_mutually_exclusive_group(required=True)
    source.add_argument("--sql-file", type=Path)
    source.add_argument("--sql")
    parser.add_argument("--semantic-model", type=Path)
    parser.add_argument("--output", type=Path)
    return parser.parse_args()


def main() -> int:
    load_runtime_config()
    args = parse_args()
    sql = args.sql if args.sql is not None else args.sql_file.read_text(encoding="utf-8")
    result = validate_sql(sql, args.semantic_model)
    rendered = json.dumps(result, indent=2, ensure_ascii=False) + "\n"
    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(rendered, encoding="utf-8")
    print(rendered, end="")
    return 0 if result["valid"] else 2


if __name__ == "__main__":
    raise SystemExit(main())