"""Visual pipeline builder — per-step "module" cards + plain-language config panels.

The Automated Workflows page (``9_Pipeline_Runner.py``) used to configure each
step through a raw ``options_json`` text column. This module replaces that with
one **module card** per step: a friendly name + caption, an enable toggle,
reorder/remove controls, and a **Configure** expander that renders that tool's
own controls in plain language (no JSON). Raw JSON survives only as the page's
Advanced import/export surface.

Each config renderer takes the step's current ``options`` dict, renders the
curated controls from the design mockup (``layout-review/09_pipeline_runner.html``),
and returns an updated **JSON-serialisable** options dict — the same shape the
``TOOL_ADAPTERS`` in ``src/core/pipeline.py`` consume via ``Options.from_dict``.

Two hard Streamlit constraints shaped this:
  * No nested expanders — the per-step Configure expander means config renderers
    here must NOT open their own expander, and the page must not wrap the card
    stack in an outer expander.
  * Widget identity must be stable across reorder/remove — every widget key is
    derived from a step's stable ``id``, never its list position.
"""

from __future__ import annotations

from typing import Any, Callable, Optional

import pandas as pd
import streamlit as st

from src.gui.tools_registry import tool_name


# ---------------------------------------------------------------------------
# Adapter-key → registry tool_id bridge
# ---------------------------------------------------------------------------
#
# Pipeline steps are keyed by adapter name (``text_clean``); the tools registry
# and i18n packs are keyed by tool_id (``02_text_cleaner``). The registry has no
# reverse lookup, so we keep the bridge here. ``step_label`` resolves the
# localized friendly name; ``step_caption`` returns a short, plain-English "what
# this step does" line for the card body.

PIPELINE_TOOL_META: dict[str, str] = {
    "text_clean":         "02_text_cleaner",
    "format_standardize": "03_format_standardizer",
    "missing":            "04_missing_handler",
    "column_map":         "05_column_mapper",
    "dedup":              "01_deduplicator",
}

_STEP_CAPTIONS: dict[str, str] = {
    "text_clean":         "Trim spaces, collapse repeats, strip invisible characters.",
    "format_standardize": "Canonicalize phones, dates, currency, names per column.",
    "missing":            "Flag, fill, or drop blank cells (and disguised blanks).",
    "column_map":         "Rename source columns onto your target column names.",
    "dedup":              "Find duplicate rows and keep one survivor per group.",
}


def step_label(tool: str) -> str:
    """Friendly, localized name for a pipeline adapter key (falls back to the key)."""
    tool_id = PIPELINE_TOOL_META.get(tool)
    return tool_name(tool_id) if tool_id else tool


def step_caption(tool: str) -> str:
    return _STEP_CAPTIONS.get(tool, "")


# ---------------------------------------------------------------------------
# Plain-English result phrasing
# ---------------------------------------------------------------------------
#
# Each adapter returns a stats dict (see ``TOOL_ADAPTERS`` in
# ``src/core/pipeline.py``). ``step_phrase`` turns that dict into the one-line
# sentence the mockup shows in the Results table ("312 duplicates removed across
# 147 groups …"); ``step_status`` derives the status pill + an optional inline
# detail line for steps that warn (e.g. unparseable cells) or error.


def _fmt_cols(cols: list) -> str:
    """Join column names for prose: 'name', 'name & city', 'a, b & 2 more'."""
    cols = [str(c) for c in cols]
    if not cols:
        return ""
    if len(cols) == 1:
        return cols[0]
    if len(cols) == 2:
        return f"{cols[0]} & {cols[1]}"
    if len(cols) == 3:
        return f"{cols[0]}, {cols[1]} & {cols[2]}"
    return f"{cols[0]}, {cols[1]} & {len(cols) - 2} more"


def _in_cols(cols: list) -> str:
    label = _fmt_cols(cols)
    return f" in {label}" if label else ""


def _n(count: int, noun: str) -> str:
    """'1 column' / '3 columns' — naive but covers every noun used here."""
    return f"{count:,} {noun}" if count == 1 else f"{count:,} {noun}s"


def step_phrase(tool: str, summary: dict) -> str:
    """A plain-English, one-line summary of what a step did."""
    s = summary or {}

    if tool == "text_clean":
        changed = s.get("cells_changed", 0)
        if not changed:
            return "No changes needed."
        return f"{_n(changed, 'cell')} cleaned{_in_cols(s.get('columns_processed', []))}"

    if tool == "format_standardize":
        changed = s.get("cells_changed", 0)
        bad = s.get("cells_unparseable", 0)
        if not changed and not bad:
            return "Nothing to standardize."
        base = f"{_n(changed, 'cell')} standardized{_in_cols(s.get('columns_processed', []))}"
        return base if not bad else f"{base} ({bad:,} left unchanged)"

    if tool == "missing":
        parts: list[str] = []
        if s.get("cells_filled"):
            parts.append(f"{_n(s['cells_filled'], 'cell')} filled")
        if s.get("rows_dropped"):
            parts.append(f"{_n(s['rows_dropped'], 'row')} dropped")
        if s.get("columns_dropped"):
            parts.append(f"{_n(len(s['columns_dropped']), 'column')} dropped")
        if not parts and s.get("sentinels_standardized"):
            parts.append(f"{_n(s['sentinels_standardized'], 'blank cell')} flagged")
        return ", ".join(parts) if parts else "No missing values to handle."

    if tool == "column_map":
        parts = []
        if s.get("columns_renamed"):
            parts.append(f"{_n(s['columns_renamed'], 'column')} renamed")
        if s.get("columns_added"):
            parts.append(f"{_n(len(s['columns_added']), 'column')} added")
        if s.get("columns_dropped"):
            parts.append(f"{_n(len(s['columns_dropped']), 'column')} dropped")
        return ", ".join(parts) if parts else "Columns already aligned."

    if tool == "dedup":
        removed = s.get("duplicates_removed", 0)
        if not removed:
            return "No duplicates found."
        return (
            f"{_n(removed, 'duplicate')} removed across {_n(s.get('groups', 0), 'group')} "
            f"({s.get('input_rows', 0):,} → {s.get('output_rows', 0):,} rows)"
        )

    return ", ".join(f"{k}: {v}" for k, v in s.items())


def step_status(
    tool: str, summary: dict, *, skipped: bool = False, error: Optional[str] = None,
) -> tuple[str, str, str]:
    """Return ``(pill_label, level, detail)`` for a step result.

    ``level`` is one of ``ok`` / ``warn`` / ``error`` / ``skipped``. ``detail``
    is a longer inline explanation for warn/error rows (else "").
    """
    if error:
        return "✗ error", "error", error.splitlines()[0]
    if skipped:
        return "⏭ skipped", "skipped", ""

    s = summary or {}
    if tool == "format_standardize" and s.get("cells_unparseable"):
        n = s["cells_unparseable"]
        return (
            f"⚠ ok · {n:,} skipped", "warn",
            f"{n:,} values didn't match a known pattern and were left "
            "unchanged. The step still completed — review them in the output "
            "preview if needed.",
        )
    if tool == "column_map":
        fails = s.get("coercion_failures") or {}
        n_fail = sum(fails.values()) if isinstance(fails, dict) else 0
        missing_req = s.get("missing_required_targets") or []
        if missing_req:
            return (
                "⚠ ok · missing targets", "warn",
                "Required target columns had no source match: "
                + ", ".join(map(str, missing_req)) + ".",
            )
        if n_fail:
            return (
                f"⚠ ok · {n_fail:,} not coerced", "warn",
                f"{n_fail:,} values couldn't be coerced to their target type "
                "and were left as-is.",
            )

    return "✓ ok", "ok", ""


# ---------------------------------------------------------------------------
# Per-tool config renderers
# ---------------------------------------------------------------------------
#
# Uniform signature: ``render_<tool>_config(df, options, kp) -> options``.
#   * ``df``      — the uploaded DataFrame (for column lists / type hints).
#   * ``options`` — the step's current options dict (seed widget defaults).
#   * ``kp``      — key prefix, unique per step (``f"{tool}_{id}"``).
# Returns a JSON-serialisable options dict. Renderers must not open expanders.


_CASE_LABELS: list[tuple[str, Optional[str]]] = [
    ("Leave as-is", None),
    ("UPPERCASE", "upper"),
    ("lowercase", "lower"),
    ("Title Case", "title"),
    ("Sentence case", "sentence"),
]


def render_text_clean_config(df: pd.DataFrame, options: dict, kp: str) -> dict:
    trim = st.checkbox(
        "Trim leading & trailing whitespace",
        value=bool(options.get("trim", True)), key=f"{kp}_trim",
    )
    collapse = st.checkbox(
        "Collapse repeated spaces to one",
        value=bool(options.get("collapse_whitespace", True)), key=f"{kp}_collapse",
    )
    fold = st.checkbox(
        "Normalize smart quotes & dashes to plain ASCII",
        value=bool(options.get("fold_smart_chars", True)), key=f"{kp}_fold",
    )
    strip_zw = st.checkbox(
        "Strip zero-width / invisible characters",
        value=bool(options.get("strip_zero_width", True)), key=f"{kp}_zw",
    )

    cur_case = options.get("case")
    case_idx = next((i for i, (_, v) in enumerate(_CASE_LABELS) if v == cur_case), 0)
    case_choice = st.selectbox(
        "Letter case",
        [lbl for lbl, _ in _CASE_LABELS],
        index=case_idx, key=f"{kp}_case",
    )
    case_val = next(v for lbl, v in _CASE_LABELS if lbl == case_choice)

    out: dict[str, Any] = {
        "trim": trim,
        "collapse_whitespace": collapse,
        "fold_smart_chars": fold,
        "strip_zero_width": strip_zw,
    }
    if case_val is not None:
        out["case"] = case_val
    return out


_FORMAT_LABELS: list[tuple[str, Optional[str]]] = [
    ("Leave as-is", None),
    ("Date", "date"),
    ("Phone number", "phone"),
    ("Currency", "currency"),
    ("Name", "name"),
    ("Address", "address"),
    ("Email", "email"),
    ("Boolean (yes/no)", "boolean"),
]


def render_format_standardize_config(df: pd.DataFrame, options: dict, kp: str) -> dict:
    st.caption(
        "Pick a target format for each column. Columns left as “Leave as-is” "
        "are untouched."
    )
    current = dict(options.get("column_types", {}))
    labels = [lbl for lbl, _ in _FORMAT_LABELS]
    column_types: dict[str, str] = {}
    for col in df.columns:
        cur_val = current.get(col)
        idx = next((i for i, (_, v) in enumerate(_FORMAT_LABELS) if v == cur_val), 0)
        choice = st.selectbox(
            str(col), labels, index=idx, key=f"{kp}_fmt__{col}",
        )
        val = next(v for lbl, v in _FORMAT_LABELS if lbl == choice)
        if val is not None:
            column_types[str(col)] = val
    return {"column_types": column_types}


# Plain-language blank-handling choices → core strategy values. "fill" is a UI
# token expanded to numeric median + categorical mode (MissingOptions handles
# the per-dtype split via ``categorical_strategy``).
_MISSING_CHOICES: list[tuple[str, str]] = [
    ("Flag them (mark blanks, change nothing)", "flag"),
    ("Fill them in (numbers → median, text → most common)", "fill"),
    ("Drop rows that have any blank", "drop"),
]


def _missing_mode_from_strategy(strategy: Optional[str]) -> str:
    if strategy in ("drop_row", "drop_col", "drop_both"):
        return "drop"
    if strategy in ("mean", "median", "mode", "constant", "ffill", "bfill", "interpolate"):
        return "fill"
    return "flag"


def render_missing_config(df: pd.DataFrame, options: dict, kp: str) -> dict:
    from src.core.missing import DEFAULT_SENTINELS

    cur_mode = _missing_mode_from_strategy(options.get("strategy"))
    mode_idx = next((i for i, (_, v) in enumerate(_MISSING_CHOICES) if v == cur_mode), 0)
    mode_choice = st.radio(
        "What should happen to blank cells?",
        [lbl for lbl, _ in _MISSING_CHOICES],
        index=mode_idx, key=f"{kp}_strategy",
    )
    mode = next(v for lbl, v in _MISSING_CHOICES if lbl == mode_choice)

    seed_sentinels = options.get("sentinels") or list(DEFAULT_SENTINELS)
    sent_text = st.text_input(
        "Treat these as blank (comma-separated)",
        value=", ".join(seed_sentinels), key=f"{kp}_sentinels",
        help="Matched case-insensitively after stripping whitespace.",
    )
    sentinels = [s.strip() for s in sent_text.split(",") if s.strip()]

    out: dict[str, Any] = {
        "standardize_sentinels": True,
        "sentinels": sentinels,
    }
    if mode == "flag":
        out["strategy"] = "none"
    elif mode == "fill":
        out["strategy"] = "median"
        out["categorical_strategy"] = "mode"
    else:  # drop
        out["strategy"] = "drop_row"
    return out


_UNMAPPED_CHOICES = ["keep", "drop", "error"]


def render_column_map_config(df: pd.DataFrame, options: dict, kp: str) -> dict:
    st.caption(
        "Type the target name each source column should become. Leave a target "
        "blank to keep that column's name unchanged."
    )
    current = dict(options.get("mapping", {}))
    table = pd.DataFrame(
        {
            "source": [str(c) for c in df.columns],
            "target": [current.get(str(c), "") for c in df.columns],
        }
    )
    edited = st.data_editor(
        table,
        width="stretch",
        hide_index=True,
        disabled=["source"],
        column_config={
            "source": st.column_config.TextColumn("Source column"),
            "target": st.column_config.TextColumn("Rename to"),
        },
        key=f"{kp}_mapping",
    )
    mapping = {
        str(r["source"]): str(r["target"]).strip()
        for _, r in edited.iterrows()
        if str(r.get("target") or "").strip()
    }

    c1, c2 = st.columns(2)
    with c1:
        unmapped = st.selectbox(
            "Columns with no rename",
            _UNMAPPED_CHOICES,
            index=_UNMAPPED_CHOICES.index(options.get("unmapped", "keep"))
            if options.get("unmapped") in _UNMAPPED_CHOICES else 0,
            key=f"{kp}_unmapped",
            help="keep: leave them in place · drop: remove them · error: stop the run.",
        )
    with c2:
        coerce = st.checkbox(
            "Coerce values to target types",
            value=bool(options.get("coerce_types", False)), key=f"{kp}_coerce",
        )
    return {"mapping": mapping, "unmapped": unmapped, "coerce_types": coerce}


_SURVIVOR_LABELS: list[tuple[str, str]] = [
    ("Keep the most complete row", "most_complete"),
    ("Keep the first seen", "first"),
    ("Keep the last seen", "last"),
    ("Keep the most recent (by date)", "most_recent"),
]


def render_dedup_config(df: pd.DataFrame, options: dict, kp: str) -> dict:
    cur_rule = options.get("survivor_rule", "first")
    rule_idx = next((i for i, (_, v) in enumerate(_SURVIVOR_LABELS) if v == cur_rule), 0)
    rule_choice = st.selectbox(
        "When rows match, which one survives?",
        [lbl for lbl, _ in _SURVIVOR_LABELS],
        index=rule_idx, key=f"{kp}_survivor",
    )
    survivor_rule = next(v for lbl, v in _SURVIVOR_LABELS if lbl == rule_choice)

    merge = st.checkbox(
        "Merge matched rows (fill each survivor's blanks from its duplicates)",
        value=bool(options.get("merge", False)), key=f"{kp}_merge",
    )

    # Recover the previously-selected match columns from the stored strategies
    # (a single exact-match strategy over the chosen columns).
    prev_cols: list[str] = []
    for strat in options.get("strategies", []) or []:
        for c in strat.get("columns", []):
            if c.get("column"):
                prev_cols.append(c["column"])
    all_cols = [str(c) for c in df.columns]
    match_cols = st.multiselect(
        "Match on these columns",
        all_cols,
        default=[c for c in prev_cols if c in all_cols],
        key=f"{kp}_matchcols",
        help="Rows are duplicates when these columns all match. Leave empty to auto-detect.",
    )

    out: dict[str, Any] = {"survivor_rule": survivor_rule, "merge": merge}
    if match_cols:
        out["strategies"] = [
            {"columns": [
                {"column": c, "algorithm": "exact", "threshold": 100}
                for c in match_cols
            ]}
        ]
    if survivor_rule == "most_recent":
        date_default = options.get("date_column")
        date_idx = all_cols.index(date_default) if date_default in all_cols else 0
        out["date_column"] = st.selectbox(
            "Date column (for most-recent)",
            all_cols, index=date_idx, key=f"{kp}_datecol",
        ) if all_cols else None
    return out


CONFIG_RENDERERS: dict[str, Callable[[pd.DataFrame, dict, str], dict]] = {
    "text_clean":         render_text_clean_config,
    "format_standardize": render_format_standardize_config,
    "missing":            render_missing_config,
    "column_map":         render_column_map_config,
    "dedup":              render_dedup_config,
}


# ---------------------------------------------------------------------------
# Module card
# ---------------------------------------------------------------------------


def render_step_card(
    df: pd.DataFrame, step: dict, idx: int, total: int,
) -> Optional[str]:
    """Render one pipeline step as a module card.

    Mutates ``step`` in place (``enabled`` toggle, ``options`` from the Configure
    panel). Returns an action string (``"up"`` / ``"down"`` / ``"remove"``) when
    the user clicks a reorder/remove control, else ``None`` — the caller applies
    the action to the step list and reruns.
    """
    sid = step["id"]
    kp = f"{step['tool']}_{sid}"
    action: Optional[str] = None

    with st.container(border=True):
        head, toggle, up, down, rm = st.columns([0.66, 0.12, 0.07, 0.07, 0.08])
        with head:
            st.markdown(f"**{idx + 1}. {step_label(step['tool'])}**")
            st.caption(step_caption(step["tool"]))
        with toggle:
            step["enabled"] = st.toggle(
                "On", value=step.get("enabled", True), key=f"{kp}_enabled",
                help="Disabled steps are kept in the pipeline but skipped at run time.",
            )
        with up:
            if st.button("▲", key=f"{kp}_up", disabled=idx == 0,
                         help="Move up", width="stretch"):
                action = "up"
        with down:
            if st.button("▼", key=f"{kp}_down", disabled=idx == total - 1,
                         help="Move down", width="stretch"):
                action = "down"
        with rm:
            if st.button("✕", key=f"{kp}_rm", help="Remove step", width="stretch"):
                action = "remove"

        renderer = CONFIG_RENDERERS.get(step["tool"])
        with st.expander(f"Configure: {step_label(step['tool'])}"):
            if renderer is None:
                st.caption("This step has no options.")
            else:
                step["options"] = renderer(df, step.get("options", {}) or {}, kp)

    return action
