"""
engine/normalize.py — declarative normalization-step interpreter.

Implements every `normalize` step from recon_source_template_schema.json.
These replace ALL the manual SQL preprocessing from the legacy workflow
(LPAD approval codes, RIGHT(card,4), quote/comma stripping, unit renames,
date reformatting).

Steps operate on pandas Series. String steps keep missing values missing
(pd.NA) rather than materialising the literal "nan".
"""

import pandas as pd

__all__ = ["apply_steps", "apply_derive"]


def apply_steps(series: pd.Series, steps: list, context: dict) -> pd.Series:
    """Apply normalization steps in order. `context` supplies tenant bindings
    (currently `location_map` for value_map steps with map_ref)."""
    out = series
    for step in steps or []:
        name = step["step"]
        fn = _STEPS.get(name)
        if fn is None:
            raise ValueError(f"Unknown normalization step: '{name}'")
        out = fn(out, step, context)
    return out


def apply_derive(df: pd.DataFrame, derive: dict, context: dict) -> pd.Series:
    """Compute a derived field from other canonical fields."""
    op = derive["op"]
    sources = derive["from"]
    missing = [f for f in sources if f not in df.columns]
    if missing:
        raise ValueError(f"Derived field references missing canonical fields: {missing}")

    if op == "coalesce":
        # First non-missing, non-empty, non-zero value across the source fields.
        result = pd.Series(pd.NA, index=df.index, dtype="object")
        for field in sources:
            candidate = df[field]
            usable = candidate.notna() & ~candidate.astype(str).str.strip().isin(["", "0", "0.0", "nan"])
            result = result.where(~(result.isna() & usable), candidate)
        return result

    if op == "sum":
        total = pd.Series(0.0, index=df.index)
        for field in sources:
            total = total + pd.to_numeric(df[field], errors="coerce").fillna(0)
        return total

    if op == "concat":
        sep = derive.get("separator", "")
        parts = [_to_str(df[f]).fillna("") for f in sources]
        result = parts[0]
        for part in parts[1:]:
            result = result.str.cat(part, sep=sep)
        return result

    raise ValueError(f"Unknown derive op: '{op}'")


# ── step implementations ─────────────────────────────────────────────────────

def _to_str(series: pd.Series) -> pd.Series:
    """Cast to pandas string dtype, strip whitespace, drop Excel's trailing
    '.0' on numeric-typed cells (mirrors legacy normalise_str)."""
    return (
        series.astype("string")
        .str.strip()
        .str.replace(r"\.0$", "", regex=True)
    )


def _trim(series, _step, _ctx):
    return _to_str(series)


def _upper(series, _step, _ctx):
    return _to_str(series).str.upper()


def _lower(series, _step, _ctx):
    return _to_str(series).str.lower()


def _last4(series, _step, _ctx):
    # VPAs (contain '@') pass through untouched — same rule as the legacy
    # last4() helper, so one canonical field can hold cards or VPAs.
    s = _to_str(series)
    return s.where(
        s.isna() | s.str.contains("@", na=False),
        s.str[-4:],
    )


def _lpad(series, step, _ctx):
    length = step["length"]
    fill = step.get("fill", "0")
    s = _to_str(series)
    return s.where(s.isna(), s.str.rjust(length, fill))


def _strip_chars(series, step, _ctx):
    s = _to_str(series)
    for token in step.get("chars", []):
        s = s.str.replace(token, "", regex=False)
    return s


def _to_number(series, _step, _ctx):
    return pd.to_numeric(series, errors="coerce")


def _date_parse(series, step, _ctx):
    s = _to_str(series)
    result = pd.Series(pd.NaT, index=s.index, dtype="datetime64[ns]")
    for fmt in step.get("formats", []):
        pending = result.isna()
        if not pending.any():
            break
        parsed = pd.to_datetime(s[pending], format=fmt, errors="coerce")
        result.loc[pending] = parsed
    if result.isna().any():
        # Last resort for stragglers (mixed formats within one file).
        pending = result.isna()
        result.loc[pending] = pd.to_datetime(s[pending], errors="coerce", dayfirst=True)
    return result


def _value_map(series, step, context):
    if step.get("map_ref") == "location_map":
        mapping = context.get("location_map") or {}
    else:
        mapping = step.get("map") or {}
    if not mapping:
        return series
    return series.replace(mapping)


_STEPS = {
    "trim": _trim,
    "upper": _upper,
    "lower": _lower,
    "last4": _last4,
    "lpad": _lpad,
    "strip_chars": _strip_chars,
    "to_number": _to_number,
    "date_parse": _date_parse,
    "value_map": _value_map,
}