"""DataTools Fix Missing Values.

Detects disguised nulls, profiles missingness per column, and applies
imputation or drop strategies with a full audit trail.

Public API
----------
Per-column helpers:
    is_missing_like(value, sentinels) -> bool
    detect_sentinels(series, sentinels) -> dict[str, int]

DataFrame entry points:
    profile_missing(df, options) -> MissingProfile
    handle_missing(df, options) -> MissingResult

Types:
    MissingOptions, MissingProfile, MissingResult, ColumnReport, Strategy

Presets (PRESETS):
    "detect-only"       — only standardize sentinels to NaN, no fill / drop.
    "safe-fill"         — sentinels → NaN, then numeric=median, categorical=mode.
    "drop-incomplete"   — sentinels → NaN, then drop rows with any missing.

Use cases covered
-----------------
1.  Disguised nulls in survey / CRM exports ("N/A", "n/a", "-", "(blank)",
    "TBD", whitespace-only, "?", "null", "NaN").
2.  Per-column profile for QA reports (counts, %, top sentinel hit).
3.  Row-drop with threshold (e.g., drop rows missing >50% of columns).
4.  Column-drop with threshold (e.g., drop columns missing >80%).
5.  Numeric imputation (mean / median / constant), categorical (mode /
    constant), time-series (ffill / bfill).
6.  Per-column overrides — different strategy per column in the same run.

Non-goals
---------
- ML-based imputation (KNN / iterative) — out of scope for v1.
- Group-wise imputation by another column — deferred until a real use case.
"""

from __future__ import annotations

import json
import re
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any, Iterable, Literal, Optional

import numpy as np
import pandas as pd
from loguru import logger
from pandas.api import types as pdtypes

from .errors import ConfigError, InputValidationError, ensure_choice, ensure_dataframe


# ---------------------------------------------------------------------------
# Sentinel detection
# ---------------------------------------------------------------------------

# Default disguised-null sentinels. Matched case-insensitively after a
# strip(). Whitespace-only strings ("", "   ") are always treated as
# missing regardless of this list.
DEFAULT_SENTINELS: tuple[str, ...] = (
    "n/a", "na", "n.a.", "n.a",
    "null", "none", "nil",
    "nan",
    "-", "--", "---",
    "?", "??",
    ".",
    "tbd", "tba",
    "unknown", "unk",
    "(blank)", "(none)", "(empty)", "(null)",
    "#n/a", "#na", "#null!", "#value!",
    "missing",
)

_WHITESPACE_ONLY_RE = re.compile(r"^\s*$")


def is_missing_like(value: Any, sentinels: Iterable[str] = DEFAULT_SENTINELS) -> bool:
    """True when *value* should be treated as missing.

    Catches: real NaN/None, whitespace-only strings, and any string that
    matches a sentinel after case-fold and strip.
    """
    if value is None:
        return True
    # pandas / numpy NaN
    try:
        if isinstance(value, float) and np.isnan(value):
            return True
    except (TypeError, ValueError):
        pass
    if isinstance(value, pd._libs.tslibs.nattype.NaTType):  # type: ignore[attr-defined]
        return True
    if not isinstance(value, str):
        return False
    if _WHITESPACE_ONLY_RE.match(value):
        return True
    needle = value.strip().casefold()
    return needle in {s.casefold() for s in sentinels}


def detect_sentinels(
    series: pd.Series,
    sentinels: Iterable[str] = DEFAULT_SENTINELS,
) -> dict[str, int]:
    """Return ``{sentinel_value: count}`` for sentinels found in *series*.

    Real NaN cells are not counted (they're already missing). Whitespace-
    only strings are bucketed under the literal key ``"(whitespace)"`` so
    callers can surface them distinctly from non-whitespace sentinels.
    """
    counts: dict[str, int] = {}
    needles = {s.casefold(): s for s in sentinels}
    for value in series:
        if value is None or (isinstance(value, float) and pd.isna(value)):
            continue
        if not isinstance(value, str):
            continue
        if _WHITESPACE_ONLY_RE.match(value):
            counts["(whitespace)"] = counts.get("(whitespace)", 0) + 1
            continue
        key = value.strip().casefold()
        if key in needles:
            label = needles[key]
            counts[label] = counts.get(label, 0) + 1
    return counts


# ---------------------------------------------------------------------------
# Strategies / options / results
# ---------------------------------------------------------------------------

Strategy = Literal[
    "none",        # detect-only; do not fill or drop.
    "drop_row",    # drop rows that are missing in any selected column.
    "drop_col",    # drop columns whose missing fraction exceeds threshold.
    "drop_both",   # apply drop_col first, then drop_row on what remains.
    "mean",        # numeric only.
    "median",      # numeric only.
    "mode",        # any dtype.
    "constant",    # fill with options.fill_value.
    "ffill",
    "bfill",
    "interpolate", # linear interpolation, numeric only.
]

_NUMERIC_STRATEGIES: frozenset[str] = frozenset(
    {"mean", "median", "interpolate"},
)
_FILL_STRATEGIES: frozenset[str] = frozenset(
    {"mean", "median", "mode", "constant", "ffill", "bfill", "interpolate"},
)
_DROP_STRATEGIES: frozenset[str] = frozenset(
    {"drop_row", "drop_col", "drop_both"},
)


PRESETS: dict[str, dict[str, Any]] = {
    "detect-only": {
        "standardize_sentinels": True,
        "strategy": "none",
    },
    "safe-fill": {
        "standardize_sentinels": True,
        "strategy": "median",
        "categorical_strategy": "mode",
    },
    "drop-incomplete": {
        "standardize_sentinels": True,
        "strategy": "drop_row",
        # Strict-greater semantics: 0.0 → drop a row as soon as any
        # selected column is missing.
        "row_drop_threshold": 0.0,
    },
}


@dataclass
class MissingOptions:
    """Toggles for missing-value detection and handling.

    Defaults match the ``detect-only`` preset: sentinels are standardized
    to NaN, but no rows are dropped and no values are filled.
    """

    # Detection
    sentinels: list[str] = field(default_factory=lambda: list(DEFAULT_SENTINELS))
    standardize_sentinels: bool = True

    # Strategy applied to all selected columns. ``categorical_strategy``
    # is a fallback used by numeric-only strategies (mean/median/interpolate)
    # when a selected column is non-numeric — rather than crash, fall back
    # to a reasonable categorical strategy.
    strategy: Strategy = "none"
    categorical_strategy: Strategy = "mode"

    # Per-column overrides take precedence over ``strategy`` / preset.
    column_strategies: dict[str, Strategy] = field(default_factory=dict)

    # Constant-fill payload. Either a scalar (applied to every selected
    # column) or a per-column dict for differentiated fills.
    fill_value: Any = None
    column_fill_values: dict[str, Any] = field(default_factory=dict)

    # Drop thresholds (0.0 .. 1.0). A row/column is dropped when its
    # missing fraction is *strictly greater than* the threshold. So:
    #   1.0 (default) — never drop (no fraction exceeds 100%)
    #   0.5           — drop when more than half is missing
    #   0.0           — drop on any missing at all
    row_drop_threshold: float = 1.0
    col_drop_threshold: float = 1.0

    # Scope control
    columns: Optional[list[str]] = None
    skip_columns: list[str] = field(default_factory=list)

    @classmethod
    def from_preset(cls, name: str) -> MissingOptions:
        if name not in PRESETS:
            raise ConfigError(
                f"Unknown preset '{name}'",
                operation="MissingOptions.from_preset",
                suggestion=f"Available: {sorted(PRESETS)}",
            )
        return cls(**PRESETS[name])

    @classmethod
    def from_dict(cls, data: dict) -> MissingOptions:
        known = set(cls.__dataclass_fields__)
        kwargs = {k: v for k, v in data.items() if k in known}
        return cls(**kwargs)

    def to_dict(self) -> dict:
        return asdict(self)

    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_file(cls, path: str | Path) -> MissingOptions:
        return cls.from_dict(json.loads(Path(path).read_text()))

    def validate(self) -> None:
        """Fail fast on incoherent option combinations."""
        choices = (
            "none", "drop_row", "drop_col", "drop_both",
            "mean", "median", "mode", "constant",
            "ffill", "bfill", "interpolate",
        )
        ensure_choice(self.strategy, name="strategy", choices=choices,
                      function="MissingOptions.validate")
        ensure_choice(self.categorical_strategy, name="categorical_strategy",
                      choices=choices, function="MissingOptions.validate")
        for col, strat in self.column_strategies.items():
            ensure_choice(strat, name=f"column_strategies[{col!r}]",
                          choices=choices, function="MissingOptions.validate")
        if not (0.0 <= self.row_drop_threshold <= 1.0):
            raise ConfigError(
                f"row_drop_threshold must be in [0.0, 1.0], got "
                f"{self.row_drop_threshold!r}",
                operation="MissingOptions.validate",
            )
        if not (0.0 <= self.col_drop_threshold <= 1.0):
            raise ConfigError(
                f"col_drop_threshold must be in [0.0, 1.0], got "
                f"{self.col_drop_threshold!r}",
                operation="MissingOptions.validate",
            )


@dataclass
class ColumnReport:
    """Per-column missingness snapshot."""

    column: str
    dtype: str
    total: int
    missing: int               # NaN cells (after sentinel standardization if enabled)
    missing_pct: float         # 0.0 .. 100.0
    sentinels_found: dict[str, int]  # disguised nulls hit, pre-standardization

    @property
    def has_missing(self) -> bool:
        return self.missing > 0


@dataclass
class MissingProfile:
    """Whole-DataFrame missingness profile."""

    columns: list[ColumnReport]
    rows_total: int
    cells_total: int
    cells_missing: int
    rows_with_any_missing: int
    rows_complete: int

    @property
    def cells_missing_pct(self) -> float:
        return (self.cells_missing / self.cells_total * 100.0) if self.cells_total else 0.0

    def to_dataframe(self) -> pd.DataFrame:
        """Long-form table suitable for the GUI / CLI."""
        rows = []
        for r in self.columns:
            top = max(r.sentinels_found.items(), key=lambda kv: kv[1], default=("", 0))
            rows.append({
                "column": r.column,
                "dtype": r.dtype,
                "missing": r.missing,
                "missing_pct": round(r.missing_pct, 2),
                "top_sentinel": top[0],
                "top_sentinel_count": top[1],
                "sentinel_total": sum(r.sentinels_found.values()),
            })
        return pd.DataFrame(rows)


@dataclass
class MissingResult:
    """Output of ``handle_missing``."""

    handled_df: pd.DataFrame
    profile_before: MissingProfile
    profile_after: MissingProfile
    changes: pd.DataFrame  # cols: row, column, old, new, action
    rows_dropped: int
    columns_dropped: list[str]
    cells_filled: int
    sentinels_standardized: int
    columns_processed: list[str]
    strategy_per_column: dict[str, Strategy]


# ---------------------------------------------------------------------------
# Profiling
# ---------------------------------------------------------------------------

def _select_columns(df: pd.DataFrame, options: MissingOptions) -> list[str]:
    """Pick the columns to operate on (mirrors text_clean._select_columns).

    Default: every column. Missing-value handling is meaningful for any
    dtype, unlike text cleaning which only touches strings.
    """
    if options.columns is not None:
        unknown = [c for c in options.columns if c not in df.columns]
        if unknown:
            raise InputValidationError(
                f"Columns not found in input: {unknown}",
                operation="handle_missing",
                suggestion=f"Available: {list(df.columns)}",
            )
        chosen: Iterable[str] = options.columns
    else:
        chosen = list(df.columns)
    skip = set(options.skip_columns)
    return [c for c in chosen if c not in skip]


def _standardize_sentinels(
    df: pd.DataFrame,
    columns: list[str],
    sentinels: Iterable[str],
) -> tuple[pd.DataFrame, list[dict[str, Any]], int]:
    """Replace sentinel strings with NaN in the selected columns.

    Returns ``(new_df, change_records, total_replacements)``. ``change_records``
    is appended to the audit table so the user can see exactly which cells
    were converted from "N/A" / "-" / etc. to a real null.

    Lazy-copy: rather than copying *df* up front, build per-column new
    value lists in a temporary dict and only materialise the output
    frame once we know at least one column actually changed. On clean
    files (no sentinels) we return *df* itself — a full-frame copy
    avoided, which on a 1 GB input saves a 1 GB allocation.
    """
    needles = {s.casefold(): s for s in sentinels}
    records: list[dict[str, Any]] = []
    total = 0
    # Per-column replacements only — keyed by column name.
    changed_cols: dict[str, list[Any]] = {}

    for col in columns:
        series = df[col]
        if not (pdtypes.is_object_dtype(series) or pdtypes.is_string_dtype(series)):
            continue
        new_values: list[Any] = []
        changed = False
        for row_idx, value in enumerate(series.tolist()):
            if value is None or (isinstance(value, float) and pd.isna(value)):
                new_values.append(value)
                continue
            if not isinstance(value, str):
                new_values.append(value)
                continue
            if _WHITESPACE_ONLY_RE.match(value):
                records.append({
                    "row": row_idx,
                    "column": col,
                    "old": value,
                    "new": np.nan,
                    "action": "standardize:whitespace",
                })
                new_values.append(np.nan)
                changed = True
                total += 1
                continue
            key = value.strip().casefold()
            if key in needles:
                records.append({
                    "row": row_idx,
                    "column": col,
                    "old": value,
                    "new": np.nan,
                    "action": f"standardize:{needles[key]}",
                })
                new_values.append(np.nan)
                changed = True
                total += 1
            else:
                new_values.append(value)
        if changed:
            changed_cols[col] = new_values

    if not changed_cols:
        # No sentinels found anywhere — return the input untouched and
        # let the caller decide whether downstream steps need a copy.
        return df, records, 0

    out = df.copy()
    for col, vals in changed_cols.items():
        out[col] = vals
    return out, records, total


def profile_missing(
    df: pd.DataFrame,
    options: Optional[MissingOptions] = None,
) -> MissingProfile:
    """Compute a per-column missingness profile.

    Sentinels are *not* mutated in *df*; this is a read-only inspection.
    The profile reports both raw NaN counts and which sentinel strings
    were hit so the GUI / CLI can show "12 disguised nulls (8 'N/A',
    4 '-')" alongside "47 real NaN".
    """
    ensure_dataframe(df, function="profile_missing")
    options = options or MissingOptions()
    columns = _select_columns(df, options)
    sentinels = options.sentinels if options.standardize_sentinels else []

    reports: list[ColumnReport] = []
    for col in columns:
        series = df[col]
        sentinels_hit = detect_sentinels(series, sentinels) if sentinels else {}
        # Effective missing = real-NaN count + sentinel hits (since they'd
        # become NaN once standardize_sentinels runs). This makes the
        # "before" profile match what the user sees post-standardization.
        nan_count = int(series.isna().sum())
        sentinel_count = sum(sentinels_hit.values())
        total = len(series)
        missing = nan_count + sentinel_count
        reports.append(ColumnReport(
            column=str(col),
            dtype=str(series.dtype),
            total=total,
            missing=missing,
            missing_pct=(missing / total * 100.0) if total else 0.0,
            sentinels_found=sentinels_hit,
        ))

    # For row-level stats use NaN ∪ sentinels in the selected columns.
    if columns and len(df):
        if sentinels:
            mask = pd.DataFrame(index=df.index)
            needles = {s.casefold() for s in sentinels}
            for col in columns:
                series = df[col]
                if pdtypes.is_object_dtype(series) or pdtypes.is_string_dtype(series):
                    sentinel_mask = series.apply(
                        lambda v: isinstance(v, str)
                        and (
                            bool(_WHITESPACE_ONLY_RE.match(v))
                            or v.strip().casefold() in needles
                        )
                    )
                    mask[col] = series.isna() | sentinel_mask
                else:
                    mask[col] = series.isna()
        else:
            mask = df[columns].isna()
        rows_with_any = int(mask.any(axis=1).sum())
        rows_complete = int((~mask.any(axis=1)).sum())
        cells_missing = int(mask.values.sum())
        cells_total = int(mask.size)
    else:
        rows_with_any = 0
        rows_complete = len(df)
        cells_missing = 0
        cells_total = len(df) * len(columns)

    return MissingProfile(
        columns=reports,
        rows_total=len(df),
        cells_total=cells_total,
        cells_missing=cells_missing,
        rows_with_any_missing=rows_with_any,
        rows_complete=rows_complete,
    )


# ---------------------------------------------------------------------------
# Imputation
# ---------------------------------------------------------------------------

def _resolve_strategy(
    col: str,
    series: pd.Series,
    options: MissingOptions,
) -> Strategy:
    """Effective strategy for *col*: per-column override → global → fallback.

    If the column is non-numeric and the selected strategy is numeric-only,
    fall back to ``options.categorical_strategy`` so the run doesn't crash
    halfway through. The fallback is logged so the audit trail records
    why a different strategy fired.
    """
    strat: Strategy = options.column_strategies.get(col, options.strategy)
    if strat in _NUMERIC_STRATEGIES and not pdtypes.is_numeric_dtype(series):
        logger.debug(
            "Column {!r}: strategy {!r} requires numeric dtype "
            "(got {}); falling back to {!r}",
            col, strat, series.dtype, options.categorical_strategy,
        )
        return options.categorical_strategy
    return strat


def _fill_value_for(
    col: str,
    series: pd.Series,
    strategy: Strategy,
    options: MissingOptions,
) -> Any:
    """Compute the scalar fill for *series* under *strategy*.

    Returns a sentinel ``object()`` when the strategy doesn't yield a
    single scalar (ffill/bfill/interpolate handle the fill themselves).
    """
    if strategy == "mean":
        return series.mean()
    if strategy == "median":
        return series.median()
    if strategy == "mode":
        modes = series.mode(dropna=True)
        return modes.iloc[0] if len(modes) else None
    if strategy == "constant":
        if col in options.column_fill_values:
            return options.column_fill_values[col]
        return options.fill_value
    return _NO_SCALAR


_NO_SCALAR = object()


def _apply_fill(
    df: pd.DataFrame,
    col: str,
    strategy: Strategy,
    options: MissingOptions,
    records: list[dict[str, Any]],
) -> int:
    """Apply *strategy* to a single column. Returns cells filled."""
    series = df[col]
    missing_mask = series.isna()
    if not missing_mask.any():
        return 0

    if strategy == "ffill":
        filled = series.ffill()
    elif strategy == "bfill":
        filled = series.bfill()
    elif strategy == "interpolate":
        # Interpolation is only defined for numeric series — guard so an
        # accidentally-routed object column produces no output rather
        # than a confusing TypeError.
        if not pdtypes.is_numeric_dtype(series):
            return 0
        filled = series.interpolate(method="linear", limit_direction="both")
    else:
        # Skip mean/median computation entirely on all-NaN numeric columns
        # so we don't trip numpy's "Mean of empty slice" RuntimeWarning.
        if (
            strategy in {"mean", "median"}
            and pdtypes.is_numeric_dtype(series)
            and series.dropna().empty
        ):
            return 0
        scalar = _fill_value_for(col, series, strategy, options)
        if scalar is _NO_SCALAR:
            return 0
        if scalar is None or (isinstance(scalar, float) and pd.isna(scalar)):
            # Nothing to fill with — e.g., all-NaN column under "mean".
            logger.debug(
                "Column {!r}: strategy {!r} produced no fill value (all-NaN?)",
                col, strategy,
            )
            return 0
        # Opt into pandas 2.x's future no-silent-downcast behaviour to
        # avoid the FutureWarning fired when fillna would auto-downcast
        # an object column. We then call infer_objects ourselves to
        # preserve the dtype the user would have ended up with.
        with pd.option_context("future.no_silent_downcasting", True):
            filled = series.fillna(scalar)
        if pdtypes.is_object_dtype(series):
            filled = filled.infer_objects(copy=False)

    cells = 0
    for row_idx in np.flatnonzero(missing_mask.values):
        old = series.iloc[row_idx]
        new = filled.iloc[row_idx]
        if pd.isna(new):
            # ffill/bfill at a leading/trailing NaN run can leave NaN in
            # place. Don't audit a no-op fill.
            continue
        records.append({
            "row": int(row_idx),
            "column": col,
            "old": old,
            "new": new,
            "action": f"fill:{strategy}",
        })
        cells += 1
    df[col] = filled
    return cells


def _apply_drops(
    df: pd.DataFrame,
    columns: list[str],
    strategy: Strategy,
    options: MissingOptions,
    records: list[dict[str, Any]],
) -> tuple[pd.DataFrame, int, list[str]]:
    """Drop rows / columns according to *strategy*.

    Returns ``(new_df, rows_dropped, columns_dropped)``.
    """
    out = df
    rows_dropped = 0
    cols_dropped: list[str] = []

    # Drop semantics (consistent across rows and columns): a row/column
    # is dropped when its missing fraction is *strictly greater* than the
    # threshold. The default threshold of 1.0 therefore means "never
    # drop" (no fraction can exceed 100%); 0.0 means "drop on any
    # missing"; intermediate values trigger when the missing share rises
    # above the chosen ceiling.
    if strategy in {"drop_col", "drop_both"} and columns:
        pct = out[columns].isna().mean()
        to_drop = [c for c, frac in pct.items() if frac > options.col_drop_threshold]
        if to_drop:
            for c in to_drop:
                records.append({
                    "row": -1,
                    "column": c,
                    "old": f"{int(out[c].isna().sum())} missing / {len(out)}",
                    "new": "",
                    "action": "drop_column",
                })
            out = out.drop(columns=to_drop)
            cols_dropped = to_drop
            columns = [c for c in columns if c not in to_drop]

    if strategy in {"drop_row", "drop_both"} and columns:
        sel = out[columns]
        frac = sel.isna().mean(axis=1)
        drop_mask = frac > options.row_drop_threshold
        rows_dropped = int(drop_mask.sum())
        if rows_dropped:
            for row_idx in np.flatnonzero(drop_mask.values):
                miss_cols = [c for c in columns if pd.isna(sel.iloc[row_idx][c])]
                records.append({
                    "row": int(row_idx),
                    "column": ",".join(miss_cols),
                    "old": "",
                    "new": "",
                    "action": "drop_row",
                })
            out = out.loc[~drop_mask].reset_index(drop=True)

    return out, rows_dropped, cols_dropped


def handle_missing(
    df: pd.DataFrame,
    options: Optional[MissingOptions] = None,
) -> MissingResult:
    """Detect and handle missing values in *df*.

    Pipeline placement (recommended, not enforced)
    ----------------------------------------------
    Run *after* the text cleaner (so NBSP-padded / zero-width-only
    cells are correctly detected as missing) and the format
    standardizer (so numeric imputation has numeric dtypes). Run
    *before* the deduplicator (so dedup doesn't merge a row with a
    missing email into a row that has one). See
    ``src.core.pipeline.SOFT_DEPENDENCIES``.

    Pipeline:
      1. Standardize disguised-null sentinels to ``NaN`` (audit-logged).
      2. Apply column drops (if strategy includes ``drop_col``).
      3. Apply row drops (if strategy includes ``drop_row``).
      4. Apply per-column fills (mean/median/mode/constant/ffill/bfill/
         interpolate). Per-column overrides win over the global strategy.

    The input DataFrame is not mutated.
    """
    ensure_dataframe(df, function="handle_missing")
    options = options or MissingOptions()
    options.validate()

    profile_before = profile_missing(df, options)
    columns = _select_columns(df, options)

    logger.debug(
        "handle_missing: rows={}, cols={}, strategy={}, scope_cols={}",
        len(df), len(df.columns), options.strategy, len(columns),
    )

    records: list[dict[str, Any]] = []
    sentinels_replaced = 0

    # ------------------------------------------------------------------
    # 1. Sentinel standardization
    # ------------------------------------------------------------------
    # ``out`` is the working frame. We track whether we own it (i.e.,
    # it's safe to mutate) so we can defer the copy until the first
    # actual mutation. ``_standardize_sentinels`` is lazy-copy: it
    # returns the input itself when no sentinels were found.
    if options.standardize_sentinels and options.sentinels and columns:
        out, sentinel_records, sentinels_replaced = _standardize_sentinels(
            df, columns, options.sentinels,
        )
        records.extend(sentinel_records)
        out_is_owned = sentinels_replaced > 0
    else:
        out = df
        out_is_owned = False

    def _ensure_owned() -> None:
        """Copy *df* on first mutation so we never write into the caller's frame."""
        nonlocal out, out_is_owned
        if not out_is_owned:
            out = out.copy()
            out_is_owned = True

    # ------------------------------------------------------------------
    # 2 + 3. Drops (column-first, then row)
    # ------------------------------------------------------------------
    rows_dropped = 0
    columns_dropped: list[str] = []
    global_strategy = options.strategy
    if global_strategy in _DROP_STRATEGIES:
        _ensure_owned()
        out, rows_dropped, columns_dropped = _apply_drops(
            out, columns, global_strategy, options, records,
        )
        # Update column scope after potential drops.
        columns = [c for c in columns if c not in columns_dropped]

    # ------------------------------------------------------------------
    # 4. Fills (per-column)
    # ------------------------------------------------------------------
    cells_filled = 0
    strategy_per_column: dict[str, Strategy] = {}
    for col in columns:
        strat = _resolve_strategy(col, out[col], options)
        strategy_per_column[col] = strat
        if strat in _FILL_STRATEGIES:
            # Defer the copy until we know a fill will actually write.
            # ``_apply_fill`` early-returns 0 when the column has no
            # missing cells; we only need an owned frame when it would
            # otherwise mutate. Easier: ensure ownership before any
            # column that *could* fill, which is still cheaper than the
            # upfront copy when ``options.standardize_sentinels`` found
            # nothing AND the strategy is a no-op fill.
            _ensure_owned()
            cells_filled += _apply_fill(out, col, strat, options, records)

    # No final blanket copy: when nothing mutated, the result *is* the
    # input. Callers must not assume identity-distinctness — this
    # mirrors pandas' own "views vs. copies" model. The buyer-facing
    # callers (GUI / CLI) only read the result, so this contract change
    # is benign and saves a 1 GB allocation on clean files.
    if not out_is_owned:
        logger.debug(
            "handle_missing: no mutations applied; returning input frame "
            "without copy (rows={}, cols={}).",
            len(out), len(out.columns),
        )

    # ------------------------------------------------------------------
    # Build audit + after-profile
    # ------------------------------------------------------------------
    changes_df = pd.DataFrame(
        records, columns=["row", "column", "old", "new", "action"],
    )
    profile_after = profile_missing(out, options)

    return MissingResult(
        handled_df=out,
        profile_before=profile_before,
        profile_after=profile_after,
        changes=changes_df,
        rows_dropped=rows_dropped,
        columns_dropped=columns_dropped,
        cells_filled=cells_filled,
        sentinels_standardized=sentinels_replaced,
        columns_processed=columns,
        strategy_per_column=strategy_per_column,
    )
