"""DataTools Automated Workflows.

Chain the cleaning tools (text-clean, format-standardize, missing,
column-map, dedup) into a single orchestrated workflow. The pipeline
threads the DataFrame from one step to the next; each step's options
are JSON-serializable so the entire pipeline can be saved, shared, and
re-run on next week's export.

Design tenets
-------------
* **Recommended, not forced.** The recommended order
  (text → format → missing → dedup, with column-map fitting either
  end depending on use case) is encoded in
  :data:`SOFT_DEPENDENCIES`. The runner WARNS on out-of-order
  pipelines but never refuses to execute them — the user owns their
  workflow.
* **Each step is opt-in / opt-out.** ``Step.enabled = False`` skips
  the step without removing it from the saved configuration.
* **Adapters are tiny.** Each tool is wrapped by a small adapter that
  bridges its native ``Options`` / ``Result`` shape to the pipeline's
  uniform ``(df, options_dict) → (new_df, summary)`` contract.

Public API
----------
Types:
    Step, Pipeline, StepResult, PipelineResult

Functions:
    run_pipeline(df, pipeline) -> PipelineResult
    validate_pipeline(pipeline) -> list[str]
    recommended_pipeline(*, include=None, **opts) -> Pipeline

Constants:
    TOOL_ADAPTERS   — name → adapter callable
    TOOL_NAMES      — sorted list of recognised tool names
    SOFT_DEPENDENCIES — list of (earlier, later, reason) tuples
"""

from __future__ import annotations

import json
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any, Callable, Iterable, Optional

import pandas as pd
from loguru import logger

from .errors import (
    ConfigError,
    DataToolsError,
    InputValidationError,
    ensure_choice,
    ensure_dataframe,
)


# ---------------------------------------------------------------------------
# Tool adapters — bridge each tool's native API to the pipeline contract
# ---------------------------------------------------------------------------

def _adapter_text_clean(
    df: pd.DataFrame, options: dict[str, Any],
) -> tuple[pd.DataFrame, dict[str, Any]]:
    from .text_clean import CleanOptions, clean_dataframe
    opts = CleanOptions.from_dict(options) if options else CleanOptions()
    res = clean_dataframe(df, opts)
    return res.cleaned_df, {
        "cells_total": res.cells_total,
        "cells_changed": res.cells_changed,
        "columns_processed": list(res.columns_processed),
    }


def _adapter_format_standardize(
    df: pd.DataFrame, options: dict[str, Any],
) -> tuple[pd.DataFrame, dict[str, Any]]:
    from .format_standardize import StandardizeOptions, standardize_dataframe
    opts = StandardizeOptions.from_dict(options) if options else StandardizeOptions()
    res = standardize_dataframe(df, opts)
    return res.standardized_df, {
        "cells_total": res.cells_total,
        "cells_changed": res.cells_changed,
        "cells_unparseable": res.cells_unparseable,
        "columns_processed": list(res.columns_processed),
    }


def _adapter_missing(
    df: pd.DataFrame, options: dict[str, Any],
) -> tuple[pd.DataFrame, dict[str, Any]]:
    from .missing import MissingOptions, handle_missing
    opts = MissingOptions.from_dict(options) if options else MissingOptions()
    res = handle_missing(df, opts)
    return res.handled_df, {
        "sentinels_standardized": res.sentinels_standardized,
        "cells_filled": res.cells_filled,
        "rows_dropped": res.rows_dropped,
        "columns_dropped": list(res.columns_dropped),
        "columns_processed": list(res.columns_processed),
    }


def _adapter_column_map(
    df: pd.DataFrame, options: dict[str, Any],
) -> tuple[pd.DataFrame, dict[str, Any]]:
    from .column_mapper import MapOptions, map_columns
    opts = MapOptions.from_dict(options) if options else MapOptions()
    res = map_columns(df, opts)
    return res.mapped_df, {
        "columns_renamed": res.columns_renamed,
        "columns_dropped": list(res.columns_dropped),
        "columns_added": list(res.columns_added),
        "coercion_failures": dict(res.coercion_failures),
        "missing_required_targets": list(res.missing_required_targets),
    }


def _adapter_dedup(
    df: pd.DataFrame, options: dict[str, Any],
) -> tuple[pd.DataFrame, dict[str, Any]]:
    from .dedup import deduplicate, SurvivorRule
    from .config import DeduplicationConfig
    options = options or {}
    survivor = options.get("survivor_rule", "first")
    if isinstance(survivor, str):
        try:
            survivor = SurvivorRule(survivor)
        except ValueError as e:
            raise ConfigError(
                f"Unknown survivor_rule {survivor!r}",
                operation="pipeline.dedup",
                cause=e,
                suggestion=f"Valid: {[r.value for r in SurvivorRule]}",
            ) from e

    # Optional explicit strategies via the same JSON shape as
    # DeduplicationConfig: ``[{"columns": [{"column": "phone",
    # "algorithm": "exact", "threshold": 100}, ...]}, ...]``.
    raw_strategies = options.get("strategies")
    explicit_strategies = None
    if raw_strategies:
        cfg = DeduplicationConfig.from_dict({"strategies": raw_strategies})
        explicit_strategies = cfg.to_strategies()

    res = deduplicate(
        df,
        strategies=explicit_strategies,
        survivor_rule=survivor,
        merge=options.get("merge", False),
        preview=False,           # pipeline always commits the dedup output
        date_column=options.get("date_column"),
    )
    final = res.deduplicated_df if res.deduplicated_df is not None else df
    return final, {
        "input_rows": len(df),
        "output_rows": len(final),
        "duplicates_removed": len(df) - len(final),
        "groups": len(res.match_groups) if res.match_groups else 0,
    }


TOOL_ADAPTERS: dict[str, Callable[..., tuple[pd.DataFrame, dict[str, Any]]]] = {
    "text_clean":          _adapter_text_clean,
    "format_standardize":  _adapter_format_standardize,
    "missing":             _adapter_missing,
    "column_map":          _adapter_column_map,
    "dedup":               _adapter_dedup,
}

TOOL_NAMES: list[str] = sorted(TOOL_ADAPTERS)


# ---------------------------------------------------------------------------
# Soft dependencies
# ---------------------------------------------------------------------------

# Pairs of (earlier, later, reason) where running *earlier* before
# *later* is recommended. A reversal triggers a WARNING — never a
# block. The user owns their workflow.
SOFT_DEPENDENCIES: list[tuple[str, str, str]] = [
    (
        "text_clean", "format_standardize",
        "format parsers (phone / currency / date) fail on smart-quote-"
        "contaminated or NBSP-padded input — clean text first",
    ),
    (
        "text_clean", "missing",
        "sentinel detection misses cells padded with NBSP / zero-width "
        "characters — clean text first",
    ),
    (
        "text_clean", "dedup",
        "fuzzy matching treats NBSP-padded values as different — "
        "clean text first",
    ),
    (
        "format_standardize", "missing",
        "numeric imputation needs numeric dtypes; canonical phones / "
        "currencies improve sentinel detection",
    ),
    (
        "format_standardize", "dedup",
        "canonical phones / lowercase emails enable cross-format "
        "duplicate matching",
    ),
    (
        "missing", "dedup",
        "deduping rows with mixed NaN sentinels produces brittle merges "
        "— resolve missing values first",
    ),
]


# ---------------------------------------------------------------------------
# Step / Pipeline / Result dataclasses
# ---------------------------------------------------------------------------

@dataclass
class Step:
    """One step in a pipeline.

    Attributes
    ----------
    tool : Name of the tool to run. Must be a key of :data:`TOOL_ADAPTERS`.
    options : JSON-serializable dict of tool-specific options. Each
        adapter parses this through the tool's ``Options.from_dict``.
    enabled : Skip the step (without removing it) when False.
    name : Optional friendly label for logs / GUI rendering. Defaults
        to the tool name.
    """

    tool: str
    options: dict[str, Any] = field(default_factory=dict)
    enabled: bool = True
    name: Optional[str] = None

    def display_name(self) -> str:
        return self.name or self.tool

    def __post_init__(self) -> None:
        if self.tool not in TOOL_ADAPTERS:
            raise ConfigError(
                f"Unknown tool {self.tool!r}",
                operation="Step.__post_init__",
                suggestion=f"Valid tools: {TOOL_NAMES}",
            )


@dataclass
class Pipeline:
    """An ordered sequence of :class:`Step` records."""

    steps: list[Step] = field(default_factory=list)

    def to_dict(self) -> dict:
        return {"steps": [asdict(s) for s in self.steps]}

    def to_file(self, path: str | Path) -> Path:
        out = Path(path)
        out.write_text(json.dumps(self.to_dict(), indent=2, default=str))
        return out

    @classmethod
    def from_dict(cls, data: dict) -> Pipeline:
        if "steps" not in data:
            raise ConfigError(
                "Pipeline file must contain a 'steps' list",
                operation="Pipeline.from_dict",
                suggestion='Example: {"steps": [{"tool": "text_clean"}, ...]}',
            )
        steps: list[Step] = []
        for raw in data["steps"]:
            if "tool" not in raw:
                raise ConfigError(
                    f"Step is missing 'tool': {raw!r}",
                    operation="Pipeline.from_dict",
                )
            steps.append(Step(
                tool=raw["tool"],
                options=dict(raw.get("options") or {}),
                enabled=bool(raw.get("enabled", True)),
                name=raw.get("name"),
            ))
        return cls(steps=steps)

    @classmethod
    def from_file(cls, path: str | Path) -> Pipeline:
        return cls.from_dict(json.loads(Path(path).read_text()))


@dataclass
class StepResult:
    """One step's outcome."""

    step: Step
    summary: dict[str, Any]
    elapsed_seconds: float
    skipped: bool = False
    error: Optional[str] = None  # rendered exception, not the live one


@dataclass
class PipelineResult:
    """Whole-run outcome."""

    final_df: pd.DataFrame
    step_results: list[StepResult]
    total_elapsed: float
    initial_rows: int
    final_rows: int
    warnings: list[str]


# ---------------------------------------------------------------------------
# Recommended pipeline + validation
# ---------------------------------------------------------------------------

# The single canonical default. Column-map is omitted: include it only
# when the caller needs header alignment (early) or schema enforcement
# (late). Adding it as an "auto" middle step would override the user's
# downstream column lookups without their having asked.
_DEFAULT_ORDER: list[str] = [
    "text_clean",
    "format_standardize",
    "missing",
    "dedup",
]


def recommended_pipeline(
    *,
    include: Optional[Iterable[str]] = None,
    options: Optional[dict[str, dict[str, Any]]] = None,
) -> Pipeline:
    """Build the recommended pipeline.

    Defaults to ``[text_clean, format_standardize, missing, dedup]`` —
    the canonical workflow surfaced in DECISIONS.md and
    ``src.core.pipeline.SOFT_DEPENDENCIES``.

    Parameters
    ----------
    include
        Names of tools to include, in the desired order. When None,
        uses :data:`_DEFAULT_ORDER`. Pass ``["column_map", "text_clean",
        ...]`` to put column-map first (header-alignment use case) or
        ``[..., "column_map"]`` to put it last (schema-enforcement use
        case).
    options
        Optional ``{tool_name: {option_dict}}`` to seed each step. A
        missing entry uses the tool's default options.
    """
    chosen = list(include) if include is not None else list(_DEFAULT_ORDER)
    seed = options or {}
    for t in chosen:
        ensure_choice(
            t, name="tool", choices=TOOL_NAMES,
            function="recommended_pipeline",
        )
    return Pipeline(steps=[
        Step(tool=t, options=dict(seed.get(t) or {}))
        for t in chosen
    ])


def validate_pipeline(pipeline: Pipeline) -> list[str]:
    """Return a list of WARNING strings for soft-dependency violations.

    Empty list = pipeline is in recommended order. Each warning is a
    single human-readable sentence the CLI / GUI can surface verbatim.
    Disabled steps are ignored.
    """
    enabled = [s for s in pipeline.steps if s.enabled]
    positions: dict[str, int] = {}
    duplicates: list[str] = []
    for i, s in enumerate(enabled):
        if s.tool in positions:
            # Multiple steps for the same tool is allowed (a user might
            # text-clean twice with different scopes). Skip the dep
            # check for the duplicate so we don't spam warnings.
            duplicates.append(s.tool)
        else:
            positions[s.tool] = i

    warnings: list[str] = []
    for earlier, later, why in SOFT_DEPENDENCIES:
        if earlier in positions and later in positions:
            if positions[earlier] > positions[later]:
                warnings.append(
                    f"step {later!r} runs BEFORE {earlier!r} — {why}"
                )
    return warnings


# ---------------------------------------------------------------------------
# Execution
# ---------------------------------------------------------------------------

def run_pipeline(
    df: pd.DataFrame,
    pipeline: Pipeline,
    *,
    on_step_complete: Optional[Callable[[StepResult], None]] = None,
    stop_on_error: bool = True,
) -> PipelineResult:
    """Execute *pipeline* against *df*.

    The DataFrame from each step's adapter is passed to the next step;
    the original input is never mutated. Soft-dependency warnings are
    captured up-front and returned via ``PipelineResult.warnings`` so
    the caller can surface them — the run proceeds regardless.

    Parameters
    ----------
    on_step_complete
        Optional ``callable(StepResult)`` fired after each step. Useful
        for live progress in the GUI.
    stop_on_error
        When True (default), the first failing step's exception
        propagates and execution halts. Set False to continue past a
        failing step using the previous step's output (the failed
        step's ``StepResult.error`` holds the rendered exception).
    """
    ensure_dataframe(df, function="run_pipeline")
    if not isinstance(pipeline, Pipeline):
        raise InputValidationError(
            f"Expected Pipeline, got {type(pipeline).__name__}",
            operation="run_pipeline",
        )

    warnings = validate_pipeline(pipeline)
    if warnings:
        for w in warnings:
            logger.warning("pipeline order: {}", w)

    initial_rows = len(df)
    step_results: list[StepResult] = []
    current = df
    t_start = time.perf_counter()

    for step in pipeline.steps:
        if not step.enabled:
            sr = StepResult(
                step=step, summary={}, elapsed_seconds=0.0, skipped=True,
            )
            step_results.append(sr)
            if on_step_complete:
                _safe_call(on_step_complete, sr)
            continue

        adapter = TOOL_ADAPTERS[step.tool]
        s_start = time.perf_counter()
        try:
            new_df, summary = adapter(current, step.options)
        except Exception as e:  # noqa: BLE001 — pipeline owns the error contract
            elapsed = time.perf_counter() - s_start
            err_msg = (
                e.format() if isinstance(e, DataToolsError) else f"{type(e).__name__}: {e}"
            )
            sr = StepResult(
                step=step, summary={}, elapsed_seconds=elapsed,
                error=err_msg,
            )
            step_results.append(sr)
            if on_step_complete:
                _safe_call(on_step_complete, sr)
            if stop_on_error:
                raise
            logger.warning(
                "pipeline step {!r} failed; continuing with previous output",
                step.display_name(),
            )
            continue

        current = new_df
        sr = StepResult(
            step=step, summary=summary,
            elapsed_seconds=time.perf_counter() - s_start,
        )
        step_results.append(sr)
        if on_step_complete:
            _safe_call(on_step_complete, sr)

    return PipelineResult(
        final_df=current,
        step_results=step_results,
        total_elapsed=time.perf_counter() - t_start,
        initial_rows=initial_rows,
        final_rows=len(current),
        warnings=warnings,
    )


def _safe_call(callback: Callable, *args: Any) -> None:
    """Run a user-supplied callback, logging but never propagating errors."""
    try:
        callback(*args)
    except Exception:  # noqa: BLE001 — progress callbacks are advisory
        logger.opt(exception=True).debug("pipeline callback raised; ignoring")
