from __future__ import annotations

import json
import sys
from pathlib import Path

import click

from felderize.constants import MINIMUM_COMPILER_VERSION
from felderize.install_feldera_sql_compiler import (
    download_compiler,
    ensure_compiler,
    is_supported_version,
    jar_version,
)
from felderize.config import Config
from felderize.models import Status, TranslationResult
from felderize.translator import (
    split_combined_sql,
    translate_spark_to_feldera,
    validate_inputs,
    validate_query,
    validate_schema,
)


def _warn_if_unsupported_compiler(compiler_path: str | None) -> None:
    """Warn when the configured compiler JAR is older than the minimum supported version."""
    if not compiler_path:
        return
    tag = jar_version(compiler_path)
    if tag and not is_supported_version(tag):
        click.echo(
            f"Warning: compiler {tag} is older than the minimum supported "
            f"{MINIMUM_COMPILER_VERSION}; run 'felderize download-compiler' to "
            "update. Translations validated against it may be inaccurate.",
            err=True,
        )


def _prepare_config(compiler: str | None, model: str | None, validate: bool) -> Config:
    """Build the run config from env, apply CLI overrides, and resolve a compiler.

    An explicit ``--compiler`` (or ``FELDERA_COMPILER``) always wins. Otherwise,
    when validating, felderize reuses a compiler JAR cached in ``~/.felderize/``
    or downloads the latest release — unless auto-download is disabled via
    ``FELDERIZE_AUTO_DOWNLOAD=0``. Download progress goes to stderr so ``--json-output``
    stays clean.
    """
    config = Config.from_env()
    if compiler:
        config.feldera_compiler = compiler
    if model:
        config.model = model
    if validate:
        if not config.feldera_compiler:
            resolved = ensure_compiler(
                auto_download=config.auto_download_compiler, logs=sys.stderr
            )
            if resolved is not None:
                config.feldera_compiler = str(resolved)
        _warn_if_unsupported_compiler(config.feldera_compiler)
    return config


def _split_examples(paths: tuple[str, ...]) -> tuple[list[Path], list[Path]]:
    """Split --examples paths into (dirs, files)."""
    dirs, files = [], []
    for p in paths:
        path = Path(p)
        (dirs if path.is_dir() else files).append(path)
    return dirs, files


def _expand_query_paths(paths: tuple[Path, ...]) -> list[Path]:
    """Expand query file arguments: directories become sorted *.sql files within them."""
    result: list[Path] = []
    for path in paths:
        if path.is_dir():
            result.extend(sorted(path.glob("*.sql")))
        else:
            result.append(path)
    return result


def _comment_lines(prefix: str, items: list[str]) -> str:
    """Render items as SQL line comments, collapsing each to a single line so a
    multi-line note can't leak uncommented (invalid) text into the file."""
    out = ""
    for item in items:
        one = " ".join(str(item).split())
        if one:
            out += f"-- {prefix}{one}\n"
    return out


def _read_text(path) -> str:
    """Read a user-supplied SQL file, exiting cleanly on binary/unreadable input."""
    try:
        return Path(path).read_text()
    except (UnicodeDecodeError, OSError) as e:
        click.echo(
            f"Error: cannot read {path} as a UTF-8 text file ({type(e).__name__}) — "
            "expected a SQL file, not binary.",
            err=True,
        )
        sys.exit(1)


def _write_sql_file(result, path: Path) -> None:
    """Write the translated schema + query to a deployable .sql file.

    A header comment block carries the translation's caveats (status, any
    unsupported constructs, and warnings) so the file is self-documenting.
    """
    with Path(path).open("w") as f:
        f.write(f"-- Translated by felderize — status: {result.status.value}\n")
        if result.unsupported:
            f.write(
                "-- Unsupported Spark constructs (emitted as CAST(NULL ...) placeholders):\n"
            )
            f.write(_comment_lines("  - ", result.unsupported))
        if result.warnings:
            f.write("-- Warnings:\n")
            f.write(_comment_lines("  - ", result.warnings))
        f.write("\n")
        if result.feldera_schema:
            f.write("-- Schema --\n" + result.feldera_schema.strip() + "\n\n")
        if result.feldera_query:
            f.write("-- Query --\n" + result.feldera_query.strip() + "\n")


@click.group()
def cli():
    """Feldera SQL translator — converts source SQL dialects to Feldera SQL."""


@cli.command("download-compiler")
@click.option(
    "--version", default=None, help="Release tag (e.g. v0.291.0); default: latest"
)
@click.option(
    "--output-dir",
    type=click.Path(),
    default=None,
    help="Directory to save the JAR (default: ~/.felderize/)",
)
@click.option(
    "--force", is_flag=True, help="Re-download even if the file already exists"
)
def download_compiler_cmd(version: str | None, output_dir: str | None, force: bool):
    """Download the Feldera SQL compiler JAR from GitHub releases.

    After downloading, set FELDERA_COMPILER in your .env to the printed path.
    """
    dest_dir = Path(output_dir) if output_dir else None
    try:
        jar_path = download_compiler(output_dir=dest_dir, version=version, force=force)
    except Exception as e:
        click.echo(f"Error: {e}", err=True)
        sys.exit(1)

    click.echo(f"Compiler ready: {jar_path}")
    click.echo(f"\nAdd to your .env:\n  FELDERA_COMPILER={jar_path}")


@cli.group()
def spark():
    """Translate Spark SQL to Feldera SQL."""


@spark.command()
@click.argument("schema_file", type=click.Path(exists=True))
@click.argument("query_file", type=click.Path(exists=True))
@click.option("--validate", is_flag=True, help="Validate against Feldera instance")
@click.option("--compiler", type=click.Path(), help="Path to Feldera compiler binary")
@click.option("--model", help="LLM model to use (overrides FELDERIZE_MODEL env var)")
@click.option("--json-output", is_flag=True, help="Output as JSON")
@click.option(
    "-o",
    "--output",
    "output",
    type=click.Path(),
    help="Write the translated SQL (schema + views) to this .sql file",
)
@click.option("--no-docs", is_flag=True, help="Disable Feldera doc inclusion in prompt")
@click.option(
    "--rules",
    type=click.Path(exists=True),
    multiple=True,
    help="Extra translation rule .md files (repeatable)",
)
@click.option(
    "--examples",
    type=click.Path(exists=True),
    multiple=True,
    help="Extra example .md files or directories (repeatable)",
)
@click.option(
    "--verbose", is_flag=True, help="Log SQL submitted to validator at each attempt"
)
def translate(
    schema_file: str,
    query_file: str,
    validate: bool,
    compiler: str | None,
    model: str | None,
    json_output: bool,
    output: str | None,
    no_docs: bool,
    rules: tuple[str, ...],
    examples: tuple[str, ...],
    verbose: bool,
):
    """Translate a single Spark SQL schema + query/views pair to Feldera SQL."""
    if not validate:
        click.echo(
            "Warning: running without validation — output SQL is not verified against the Feldera compiler.",
            err=True,
        )
    config = _prepare_config(compiler, model, validate)
    schema_sql = _read_text(schema_file)
    query_sql = _read_text(query_file)

    errors = validate_inputs(schema_sql, query_sql)
    if errors:
        for e in errors:
            click.echo(f"Error: {e}", err=True)
        sys.exit(1)

    ex_dirs, ex_files = _split_examples(examples)
    result = translate_spark_to_feldera(
        schema_sql,
        query_sql,
        config,
        validate=validate,
        include_docs=not no_docs,
        verbose=verbose,
        extra_rules=[Path(r) for r in rules] or None,
        extra_examples_dirs=ex_dirs or None,
        extra_examples_files=ex_files or None,
    )

    if output:
        _write_sql_file(result, Path(output))
        click.echo(f"Wrote {output} (status: {result.status.value})", err=True)
    if json_output:
        click.echo(json.dumps(result.to_dict(), indent=2))
    elif not output:
        _print_result(result)


@spark.command("translate-file")
@click.argument("sql_file", type=click.Path(exists=True))
@click.option("--validate", is_flag=True, help="Validate against Feldera instance")
@click.option("--compiler", type=click.Path(), help="Path to Feldera compiler binary")
@click.option("--model", help="LLM model to use (overrides FELDERIZE_MODEL env var)")
@click.option("--json-output", is_flag=True, help="Output as JSON")
@click.option(
    "-o",
    "--output",
    "output",
    type=click.Path(),
    help="Write the translated SQL (schema + views) to this .sql file",
)
@click.option("--no-docs", is_flag=True, help="Disable Feldera doc inclusion in prompt")
@click.option(
    "--rules",
    type=click.Path(exists=True),
    multiple=True,
    help="Extra translation rule .md files (repeatable)",
)
@click.option(
    "--examples",
    type=click.Path(exists=True),
    multiple=True,
    help="Extra example .md files or directories (repeatable)",
)
@click.option(
    "--verbose", is_flag=True, help="Log SQL submitted to validator at each attempt"
)
def translate_file(
    sql_file: str,
    validate: bool,
    compiler: str | None,
    model: str | None,
    json_output: bool,
    output: str | None,
    no_docs: bool,
    rules: tuple[str, ...],
    examples: tuple[str, ...],
    verbose: bool,
):
    """Translate a single combined Spark SQL file (schema + views) to Feldera SQL."""
    if not validate:
        click.echo(
            "Warning: running without validation — output SQL is not verified against the Feldera compiler.",
            err=True,
        )
    config = _prepare_config(compiler, model, validate)
    combined_sql = _read_text(sql_file)
    schema_sql, query_sql = split_combined_sql(combined_sql)

    errors = validate_inputs(schema_sql, query_sql)
    if errors:
        for e in errors:
            click.echo(f"Error: {e}", err=True)
        sys.exit(1)

    ex_dirs, ex_files = _split_examples(examples)
    result = translate_spark_to_feldera(
        schema_sql,
        query_sql,
        config,
        validate=validate,
        include_docs=not no_docs,
        verbose=verbose,
        extra_rules=[Path(r) for r in rules] or None,
        extra_examples_dirs=ex_dirs or None,
        extra_examples_files=ex_files or None,
    )

    if output:
        _write_sql_file(result, Path(output))
        click.echo(f"Wrote {output} (status: {result.status.value})", err=True)
    if json_output:
        click.echo(json.dumps(result.to_dict(), indent=2))
    elif not output:
        _print_result(result)


@spark.command("translate-batch")
@click.argument("schema_file", type=click.Path(exists=True))
@click.argument(
    "query_files", nargs=-1, required=True, type=click.Path(exists=True, path_type=Path)
)
@click.option("--validate", is_flag=True, help="Validate against Feldera compiler")
@click.option("--compiler", type=click.Path(), help="Path to Feldera compiler binary")
@click.option("--model", help="LLM model to use (overrides FELDERIZE_MODEL env var)")
@click.option(
    "--output-dir",
    type=click.Path(),
    help="Directory to write translated SQL files (one per query)",
)
@click.option("--no-docs", is_flag=True, help="Disable Feldera doc inclusion in prompt")
@click.option(
    "--rules",
    type=click.Path(exists=True),
    multiple=True,
    help="Extra translation rule .md files (repeatable)",
)
@click.option(
    "--examples",
    type=click.Path(exists=True),
    multiple=True,
    help="Extra example .md files or directories (repeatable)",
)
@click.option(
    "--verbose", is_flag=True, help="Log SQL submitted to validator at each attempt"
)
def translate_batch(
    schema_file: str,
    query_files: tuple[Path, ...],
    validate: bool,
    compiler: str | None,
    model: str | None,
    output_dir: str | None,
    no_docs: bool,
    rules: tuple[str, ...],
    examples: tuple[str, ...],
    verbose: bool,
):
    """Translate multiple Spark SQL query files against a shared schema.

    Runs all queries in a single process so doc and example caches stay warm,
    making batch translation faster than calling translate in a shell loop.

    \b
    Usage:
      felderize spark translate-batch schema.sql q1.sql q2.sql q3.sql --validate
      felderize spark translate-batch schema.sql queries/ --validate --output-dir out/
    """
    if not validate:
        click.echo(
            "Warning: running without validation — output SQL is not verified against the Feldera compiler.",
            err=True,
        )
    config = _prepare_config(compiler, model, validate)

    schema_sql = _read_text(schema_file)
    schema_errors = validate_schema(schema_sql)
    if schema_errors:
        for e in schema_errors:
            click.echo(f"Error: {e}", err=True)
        sys.exit(1)

    out_dir = Path(output_dir) if output_dir else None
    if out_dir:
        out_dir.mkdir(parents=True, exist_ok=True)

    ex_dirs, ex_files = _split_examples(examples)
    expanded_queries = _expand_query_paths(query_files)
    if not expanded_queries:
        click.echo("Error: no .sql query files found.", err=True)
        sys.exit(1)
    results: list[tuple[str, TranslationResult]] = []

    for qpath in expanded_queries:
        try:
            query_sql = qpath.read_text()
        except (UnicodeDecodeError, OSError) as e:
            msg = f"cannot read as a UTF-8 text file ({type(e).__name__}) — not a SQL file"
            click.echo(f"Error ({qpath.name}): {msg}", err=True)
            results.append(
                (qpath.stem, TranslationResult(status=Status.ERROR, warnings=[msg]))
            )
            continue

        errors = validate_query(query_sql)
        if errors:
            for e in errors:
                click.echo(f"Error ({qpath.name}): {e}", err=True)
            results.append(
                (qpath.stem, TranslationResult(status=Status.ERROR, warnings=errors))
            )
            continue

        result = translate_spark_to_feldera(
            schema_sql,
            query_sql,
            config,
            validate=validate,
            include_docs=not no_docs,
            verbose=verbose,
            extra_rules=[Path(r) for r in rules] or None,
            extra_examples_dirs=ex_dirs or None,
            extra_examples_files=ex_files or None,
        )

        if out_dir:
            _write_sql_file(result, out_dir / f"{qpath.stem}_feldera.sql")
        else:
            click.echo(f"\n=== {qpath.name} ===")
            _print_result(result)

        click.echo(f"{qpath.stem}: {result.status.value}", err=True)
        results.append((qpath.stem, result))

    n_success = sum(1 for _, r in results if r.status == Status.SUCCESS)
    click.echo(f"\n{n_success}/{len(results)} translated cleanly.", err=True)
    counts: dict[str, int] = {}
    for _, r in results:
        counts[r.status.value] = counts.get(r.status.value, 0) + 1
    breakdown = ", ".join(f"{n} {status}" for status, n in sorted(counts.items()))
    click.echo(f"  Status breakdown: {breakdown}", err=True)


_EXAMPLES_DIR = Path(__file__).resolve().parent / "spark_sql"


@spark.command()
@click.argument("name", required=False)
@click.option(
    "--validate/--no-validate",
    default=True,
    help="Validate against Feldera instance (default: on)",
)
@click.option("--compiler", type=click.Path(), help="Path to Feldera compiler binary")
@click.option("--model", help="LLM model to use (overrides FELDERIZE_MODEL env var)")
@click.option("--json-output", is_flag=True, help="Output as JSON")
@click.option("--no-docs", is_flag=True, help="Disable Feldera doc inclusion in prompt")
@click.option(
    "--verbose", is_flag=True, help="Log SQL submitted to validator at each attempt"
)
def example(
    name: str | None,
    validate: bool,
    compiler: str | None,
    model: str | None,
    json_output: bool,
    no_docs: bool,
    verbose: bool,
):
    """Run a built-in example translation.

    Without NAME, lists available examples. With NAME, translates that example.

    \b
    Usage:
      felderize spark example              # list available examples
      felderize spark example simple       # translate the 'simple' example
    """
    pairs: dict[str, tuple[Path, Path] | Path] = {}
    for schema_file in sorted(_EXAMPLES_DIR.glob("*_schema.sql")):
        example_name = schema_file.name.replace("_schema.sql", "")
        query_file = _EXAMPLES_DIR / f"{example_name}_query.sql"
        if query_file.is_file():
            pairs[example_name] = (schema_file, query_file)
    for combined_file in sorted(_EXAMPLES_DIR.glob("*_combined.sql")):
        example_name = combined_file.name.replace("_combined.sql", "")
        pairs[example_name] = combined_file

    if not name:
        click.echo("Available examples:\n")
        for ex_name, files in pairs.items():
            tag = "[combined]" if isinstance(files, Path) else "[schema+query]"
            click.echo(f"  {ex_name:20s} {tag}")
        click.echo("\nRun one with: felderize spark example <name>")
        return

    if name not in pairs:
        click.echo(f"Unknown example '{name}'. Available: {', '.join(pairs)}", err=True)
        sys.exit(1)

    files = pairs[name]
    if isinstance(files, Path):
        combined_sql = files.read_text()
        schema_sql, query_sql = split_combined_sql(combined_sql)
        click.echo(f"-- Spark SQL ({name}) --", err=True)
        click.echo(combined_sql.strip(), err=True)
    else:
        schema_file, query_file = files
        schema_sql = schema_file.read_text()
        query_sql = query_file.read_text()
        click.echo(f"-- Spark Schema ({name}) --", err=True)
        click.echo(schema_sql.strip(), err=True)
        click.echo(f"\n-- Spark Query ({name}) --", err=True)
        click.echo(query_sql.strip(), err=True)
    click.echo("\nTranslating...\n", err=True)

    if not validate:
        click.echo(
            "Warning: running without validation — output SQL is not verified against the Feldera compiler.",
            err=True,
        )
    config = _prepare_config(compiler, model, validate)
    result = translate_spark_to_feldera(
        schema_sql,
        query_sql,
        config,
        validate=validate,
        include_docs=not no_docs,
        verbose=verbose,
    )

    if json_output:
        click.echo(json.dumps(result.to_dict(), indent=2))
    else:
        _print_result(result)


_CONTACT_MESSAGE = (
    "\n  Some Spark SQL features are not yet supported in Feldera.\n"
    "  Contact us at support@feldera.com to request support for these features or "
    "  consider filing an issue at github.com/feldera/feldera/issue. "
)

_ERROR_CONTACT_MESSAGE = "\n  Contact us at support@feldera.com for assistance."


def _print_result(result: TranslationResult) -> None:
    """Pretty-print a translation result."""
    if result.status == Status.ERROR:
        click.echo("-- Translation Failed --", err=True)
        click.echo(
            "  The translation could not be validated against the Feldera compiler.",
            err=True,
        )
        if result.warnings:
            click.echo("  Errors:", err=True)
            for item in result.warnings:
                click.echo(f"    - {item}", err=True)
        click.echo(_ERROR_CONTACT_MESSAGE, err=True)
        click.echo(f"\nStatus: {result.status.value}", err=True)
        return

    if result.feldera_schema:
        click.echo("-- Schema --")
        click.echo(result.feldera_schema)
        click.echo()

    if result.feldera_query:
        click.echo("-- Query --")
        click.echo(result.feldera_query)
        click.echo()

    if result.explanations:
        click.echo("-- Transformations Applied --", err=True)
        for item in result.explanations:
            click.echo(f"  - {item}", err=True)
        click.echo(err=True)

    if result.unsupported:
        click.echo("-- Unsupported --", err=True)
        for item in result.unsupported:
            click.echo(f"  - {item}", err=True)
        click.echo(_CONTACT_MESSAGE, err=True)
        click.echo(err=True)

    if result.warnings:
        click.echo("-- Warnings --", err=True)
        for item in result.warnings:
            click.echo(f"  - {item}", err=True)

    click.echo(f"\nStatus: {result.status.value}", err=True)