"""
engine/loader.py — template-driven file loading.

Turns (base64 file content + source template definition) into a DataFrame of
canonical fields. The original file columns are kept alongside so persisted
payloads (recon_matches, D8) can show exactly what was received; canonical
fields win on name collision.
"""

import io
import re
import base64

import pandas as pd

from .normalize import apply_steps, apply_derive

DEFAULT_ROW_LIMIT = 200_000  # D7 hard guard
_AUTO_SCAN_ROWS = 30


class LoaderError(ValueError):
    """Raised for template/file mismatches with a message safe to surface in the UI."""


def load_source(content_b64: str, template: dict, context: dict) -> pd.DataFrame:
    """Parse + normalize one uploaded file according to its source template."""
    raw = base64.b64decode(content_b64)
    df = _read_raw(raw, template)
    df = _drop_skip_rows(df, template)

    row_limit = template.get("row_limit", DEFAULT_ROW_LIMIT)
    if len(df) > row_limit:
        raise LoaderError(
            f"File for template '{template['name']}' has {len(df)} rows, "
            f"exceeding the limit of {row_limit}"
        )

    out = df.copy()
    canonical_fields = []

    # Regular (alias-mapped) fields first, derived fields second — derives
    # reference canonical names produced in the first pass.
    plain = [f for f in template["fields"] if "derive" not in f]
    derived = [f for f in template["fields"] if "derive" in f]

    for field in plain:
        source_col = _resolve_alias(df, field["aliases"])
        if source_col is None:
            if field.get("required", True):
                raise LoaderError(
                    f"Template '{template['name']}': cannot find column for "
                    f"'{field['canonical']}'. Tried aliases {field['aliases']}. "
                    f"File columns: {list(df.columns)}"
                )
            out[field["canonical"]] = pd.Series(pd.NA, index=df.index, dtype="object")
            canonical_fields.append(field["canonical"])
            continue
        out[field["canonical"]] = apply_steps(df[source_col], field.get("normalize"), context)
        canonical_fields.append(field["canonical"])

    for field in derived:
        base = apply_derive(out, field["derive"], context)
        out[field["canonical"]] = apply_steps(base, field.get("normalize"), context)
        canonical_fields.append(field["canonical"])

    out.attrs["canonical_fields"] = canonical_fields
    out.attrs["template_name"] = template["name"]
    return out


# ── private ──────────────────────────────────────────────────────────────────

def _read_raw(raw: bytes, template: dict) -> pd.DataFrame:
    file_format = template["file_format"]

    if file_format == "csv":
        opts = template.get("csv_options", {})
        return pd.read_csv(
            io.BytesIO(raw),
            delimiter=opts.get("delimiter", ","),
            encoding=opts.get("encoding", "utf-8"),
        )

    if file_format == "xlsx":
        sheet = template.get("sheet", 0)
        header_cfg = template.get("header", {})
        header_row = header_cfg.get("row", 0)
        if header_row == "auto":
            return _read_excel_auto_header(raw, sheet, template)
        return pd.read_excel(io.BytesIO(raw), sheet_name=sheet, header=header_row)

    raise LoaderError(f"Unsupported file_format: '{file_format}'")


def _read_excel_auto_header(raw: bytes, sheet, template: dict) -> pd.DataFrame:
    """Find the header row by scanning for the row with the most alias hits —
    bank portals often prepend title/summary rows above the real header."""
    probe = pd.read_excel(io.BytesIO(raw), sheet_name=sheet, header=None, nrows=_AUTO_SCAN_ROWS)

    all_aliases = {
        _norm_header(alias)
        for field in template["fields"]
        for alias in field.get("aliases", [])
    }

    best_row, best_hits = 0, 0
    for i in range(len(probe)):
        cells = {_norm_header(c) for c in probe.iloc[i].tolist()}
        hits = len(cells & all_aliases)
        if hits > best_hits:
            best_row, best_hits = i, hits

    if best_hits == 0:
        raise LoaderError(
            f"Template '{template['name']}': no header row found in the first "
            f"{_AUTO_SCAN_ROWS} rows (no known column aliases present)"
        )
    return pd.read_excel(io.BytesIO(raw), sheet_name=sheet, header=best_row)


def _drop_skip_rows(df: pd.DataFrame, template: dict) -> pd.DataFrame:
    patterns = template.get("header", {}).get("skip_rows_containing", [])
    if not patterns or df.empty:
        return df
    as_str = df.astype(str)
    mask = pd.Series(False, index=df.index)
    for pattern in patterns:
        mask |= as_str.apply(lambda col: col.str.contains(pattern, case=False, na=False, regex=False)).any(axis=1)
    return df[~mask].copy()


def _norm_header(value) -> str:
    """Normalize a header cell / alias for tolerant comparison."""
    return re.sub(r"[\s_]+", " ", str(value)).strip().lower()


def _resolve_alias(df: pd.DataFrame, aliases: list) -> str | None:
    """Exact match first, then whitespace/case/underscore-insensitive."""
    for alias in aliases:
        if alias in df.columns:
            return alias
    normalized = {_norm_header(col): col for col in df.columns}
    for alias in aliases:
        hit = normalized.get(_norm_header(alias))
        if hit is not None:
            return hit
    return None