"""Deduplication engine: matching, survivor selection, and merge.

Core algorithm:
1. Normalise columns → shadow ``_norm_*`` columns (computed once).
2. Pairwise comparison within each strategy → candidate pairs.
3. Union-find for transitive closure (A~B, B~C ⇒ one group).
4. Multi-strategy OR: feed all pairs from all strategies into the same union-find.
5. Survivor selection per group + optional field merge.
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Optional

import pandas as pd
from loguru import logger
from rapidfuzz import fuzz as rf_fuzz
from rapidfuzz import distance as rf_distance

from .normalizers import NormalizerType, get_normalizer


# ---------------------------------------------------------------------------
# Enums & data structures
# ---------------------------------------------------------------------------

class Algorithm(str, Enum):
    EXACT = "exact"
    LEVENSHTEIN = "levenshtein"
    JARO_WINKLER = "jaro_winkler"
    TOKEN_SET_RATIO = "token_set_ratio"


class SurvivorRule(str, Enum):
    KEEP_FIRST = "first"
    KEEP_LAST = "last"
    KEEP_MOST_COMPLETE = "most_complete"
    KEEP_MOST_RECENT = "most_recent"


@dataclass
class ColumnMatchStrategy:
    """How to match on a single column."""
    column: str
    algorithm: Algorithm = Algorithm.EXACT
    threshold: float = 100.0  # 0-100 scale
    normalizer: Optional[NormalizerType] = None

    def __post_init__(self) -> None:
        if not isinstance(self.threshold, (int, float)):
            raise TypeError(
                f"threshold must be a number, got {type(self.threshold).__name__}"
            )
        if not 0 <= self.threshold <= 100:
            raise ValueError(
                f"threshold must be in [0, 100]; got {self.threshold}. "
                "Match scores are on a 0–100 scale, so values outside this "
                "range either always match or never match."
            )


@dataclass
class MatchStrategy:
    """A set of column strategies combined with AND.

    Multiple ``MatchStrategy`` instances are combined with OR at the top level.
    """
    column_strategies: list[ColumnMatchStrategy]


@dataclass
class MatchResult:
    """One group of duplicate rows.

    ``row_indices`` and ``survivor_index`` are positional indexes into
    the *input* DataFrame (0-based, matching ``df.iloc[]``), not the
    output ``deduplicated_df`` (whose index is reset to 0..N-1). To map
    back to the original frame, use ``df.iloc[row_indices]``.
    """
    group_id: int
    row_indices: list[int]
    confidence: float            # min confidence across pairs in the group
    matched_on: list[str]        # column names that contributed to the match
    survivor_index: int          # index of the row to keep


@dataclass
class DeduplicationResult:
    """Full result of a deduplication run.

    ``deduplicated_df`` and ``removed_df`` both have their indexes reset
    to a fresh 0..N-1 range. ``match_groups[*].row_indices`` keeps the
    original positional indexes of the *input* frame so callers can
    cross-reference back to it (e.g., for an audit log).
    """
    original_row_count: int
    deduplicated_df: pd.DataFrame
    removed_df: pd.DataFrame
    match_groups: list[MatchResult]
    log_entries: list[str] = field(default_factory=list)
    is_preview: bool = True


# ---------------------------------------------------------------------------
# Union-Find
# ---------------------------------------------------------------------------

class _UnionFind:
    """Disjoint-set / union-find for transitive closure of match pairs."""

    def __init__(self, n: int):
        self._parent = list(range(n))
        self._rank = [0] * n

    def find(self, x: int) -> int:
        while self._parent[x] != x:
            self._parent[x] = self._parent[self._parent[x]]  # path halving
            x = self._parent[x]
        return x

    def union(self, a: int, b: int) -> None:
        ra, rb = self.find(a), self.find(b)
        if ra == rb:
            return
        if self._rank[ra] < self._rank[rb]:
            ra, rb = rb, ra
        self._parent[rb] = ra
        if self._rank[ra] == self._rank[rb]:
            self._rank[ra] += 1

    def groups(self) -> dict[int, list[int]]:
        """Return {root: [members]} for all non-singleton groups."""
        from collections import defaultdict
        g: dict[int, list[int]] = defaultdict(list)
        for i in range(len(self._parent)):
            g[self.find(i)].append(i)
        return {root: members for root, members in g.items() if len(members) > 1}


# ---------------------------------------------------------------------------
# Similarity computation
# ---------------------------------------------------------------------------

def _compute_similarity(val_a: str, val_b: str, algorithm: Algorithm) -> float:
    """Return similarity score on a 0-100 scale."""
    if algorithm == Algorithm.EXACT:
        return 100.0 if val_a == val_b else 0.0
    if algorithm == Algorithm.LEVENSHTEIN:
        return rf_fuzz.ratio(val_a, val_b)
    if algorithm == Algorithm.JARO_WINKLER:
        # rapidfuzz jaro_winkler_similarity returns 0-100
        return rf_distance.JaroWinkler.similarity(val_a, val_b) * 100
    if algorithm == Algorithm.TOKEN_SET_RATIO:
        return rf_fuzz.token_set_ratio(val_a, val_b)
    raise ValueError(f"Unknown algorithm: {algorithm}")


# ---------------------------------------------------------------------------
# Pair comparison
# ---------------------------------------------------------------------------

def _is_missing(value: Any) -> bool:
    """True for None, NaN, or pd.NA — used everywhere we need to treat
    a cell as "absent" rather than as the string ``"None"`` or ``"nan"``.

    Centralizing the check avoids the silent-merge bug where
    ``str(None)`` and ``str(nan)`` would both compare equal to themselves
    and produce a false 100% match in the deduplicator.
    """
    if value is None:
        return True
    if value is pd.NA:
        return True
    if isinstance(value, float) and pd.isna(value):
        return True
    return False


def _compare_pair(
    row_a: pd.Series,
    row_b: pd.Series,
    strategy: MatchStrategy,
    norm_prefix: str = "_norm_",
) -> tuple[bool, float, list[str]]:
    """Compare two rows using a single MatchStrategy (AND of column strategies).

    Returns ``(is_match, confidence, matched_columns)``.
    """
    min_score = 100.0
    matched_cols: list[str] = []

    for cs in strategy.column_strategies:
        col = f"{norm_prefix}{cs.column}" if cs.normalizer else cs.column
        raw_a = row_a.get(col, "")
        raw_b = row_b.get(col, "")

        va = "" if _is_missing(raw_a) else str(raw_a)
        vb = "" if _is_missing(raw_b) else str(raw_b)

        # Skip if both empty
        if not va and not vb:
            continue
        # If one empty and one not — no match for this column
        if not va or not vb:
            return False, 0.0, []

        score = _compute_similarity(va, vb, cs.algorithm)
        if score < cs.threshold:
            return False, 0.0, []
        min_score = min(min_score, score)
        matched_cols.append(cs.column)

    if not matched_cols:
        return False, 0.0, []

    return True, min_score, matched_cols


# ---------------------------------------------------------------------------
# Match-group finding
# ---------------------------------------------------------------------------

def _strategy_is_exact_only(strategy: MatchStrategy) -> bool:
    """True when every column in *strategy* matches via exact equality.

    Such strategies can be processed in **O(n)** via a single
    ``groupby`` on the composite key — rows with identical normalised
    values land in the same partition and need no pair-compare. The
    semantics are identical to running ``_compare_pair`` over every
    pair: exact at threshold 100 returns match iff both sides equal.
    """
    return all(
        cs.algorithm == Algorithm.EXACT and cs.threshold >= 100.0
        for cs in strategy.column_strategies
    )


def _strategy_columns(strategy: MatchStrategy, norm_prefix: str = "_norm_") -> list[str]:
    """Resolved column names (normalised shadow if present, else raw)."""
    return [
        f"{norm_prefix}{cs.column}" if cs.normalizer else cs.column
        for cs in strategy.column_strategies
    ]


def _match_exact_via_groupby(
    df: pd.DataFrame,
    strategy: MatchStrategy,
    uf: _UnionFind,
    pair_info: dict[tuple[int, int], tuple[float, list[str]]],
    norm_prefix: str = "_norm_",
) -> int:
    """O(n) exact-match fast path. Returns the number of pairs unioned.

    Mirrors :func:`_compare_pair` semantics for the EXACT-only case:
    rows where every key column is non-empty and equal across rows land
    in the same group. Rows with any empty value in the key columns are
    excluded — matching the "one empty + one non-empty = no match" rule
    in :func:`_compare_pair`.

    For each match group the function unions all members into ``uf``
    and records a single representative entry in ``pair_info`` (between
    the first two members) carrying confidence 100.0 and the strategy's
    column names. The group walk in :func:`_find_match_groups` finds
    that entry when computing per-group confidence.
    """
    cols = _strategy_columns(strategy, norm_prefix)
    matched_col_names = [cs.column for cs in strategy.column_strategies]

    # Build a string-typed view of just the key columns so groupby is
    # comparing apples to apples regardless of source dtype. Treat NaN
    # / None as empty to match the original ``_is_missing`` semantics.
    key_df = pd.DataFrame(
        {c: df[c].map(lambda v: "" if _is_missing(v) else str(v)) for c in cols},
        index=df.index,
    )
    # Eligible rows: every key column non-empty (after strip). Empty
    # rows can't match anyone under EXACT semantics, so excluding them
    # is equivalent to running them through ``_compare_pair`` and
    # getting False.
    nonempty = key_df.apply(lambda s: s.str.strip().astype(bool)).all(axis=1)
    eligible = key_df[nonempty]
    if eligible.empty:
        return 0

    pairs_unioned = 0
    # ``groupby.indices`` gives ``{key_tuple: ndarray_of_positions}``
    # in one pass — pandas does the bucket walk in C.
    grouped = eligible.groupby(list(eligible.columns), sort=False, dropna=False)
    for _, idx_array in grouped.indices.items():
        if len(idx_array) < 2:
            continue
        members = [int(i) for i in idx_array]
        base = members[0]
        # Record a single pair_info entry for the group; the rest are
        # transitively captured via union-find.
        pair_info[(min(base, members[1]), max(base, members[1]))] = (
            100.0,
            matched_col_names,
        )
        for m in members[1:]:
            uf.union(base, m)
            pairs_unioned += 1

    return pairs_unioned


def _build_block_keys(
    df: pd.DataFrame,
    blocking_columns: list[str],
    prefix_len: int,
    norm_prefix: str = "_norm_",
) -> pd.Series:
    """Compute a per-row block key from the first *prefix_len* chars of
    each blocking column. Rows with the same key go in the same block.

    Empty cells contribute ``""`` so a row with one empty blocking
    column still gets a deterministic key — those rows simply land in
    a block with other rows that share the same prefix on the other
    column(s).
    """
    parts: list[pd.Series] = []
    for col in blocking_columns:
        # Prefer the normalised shadow when present — it's lowercased
        # and dot/punct-stripped, which is exactly the canonical form
        # we want for a block key (so "John" and "JOHN" share a block).
        shadow = f"{norm_prefix}{col}"
        source = df[shadow] if shadow in df.columns else df.get(col)
        if source is None:
            raise ValueError(
                f"blocking_columns refers to unknown column {col!r}; "
                f"available: {list(df.columns)}"
            )
        s = source.map(lambda v: "" if _is_missing(v) else str(v))
        parts.append(s.str.strip().str.lower().str[:prefix_len])
    if not parts:
        return pd.Series([""] * len(df), index=df.index)
    key = parts[0]
    for p in parts[1:]:
        key = key + "\x1f" + p   # ASCII unit separator
    return key


def _find_match_groups(
    df: pd.DataFrame,
    strategies: list[MatchStrategy],
    *,
    progress_callback: Optional[Callable[[int, int], None]] = None,
    blocking_columns: Optional[list[str]] = None,
    blocking_prefix_len: int = 1,
) -> tuple[list[MatchResult], dict[tuple[int, int], tuple[float, list[str]]]]:
    """Pairwise comparison + union-find for transitive closure.

    Returns ``(match_groups, pair_info)`` where *pair_info* maps
    ``(i, j)`` → ``(confidence, matched_columns)`` for logging.

    Performance shape:

    - **Exact-only strategies** (every column EXACT @ 100) are peeled
      off and processed via O(n) groupby. The fallback "exact on all
      columns" path and standalone strong-key strategies (e.g.,
      ``email`` exact) both qualify and account for the vast majority
      of real-world dedup workloads.
    - **Fuzzy strategies** still do pairwise compare. When
      ``blocking_columns`` is supplied, only pairs sharing the same
      *prefix_len*-char prefix on those columns are compared — turning
      O(n²) into roughly O(n × avg_block_size). Default off so match
      semantics don't change unless the caller opts in.
    """
    n = len(df)
    uf = _UnionFind(n)
    pair_info: dict[tuple[int, int], tuple[float, list[str]]] = {}

    # Split strategies into exact-only and fuzzy. Order preserved
    # within each bucket so the (first-strategy-wins) ordering for
    # pair_info ties matches the prior behaviour as closely as
    # possible. Exact-only run first because they're cheap and
    # populate pair_info with the most informative (100%) score.
    exact_strategies = [s for s in strategies if _strategy_is_exact_only(s)]
    fuzzy_strategies = [s for s in strategies if not _strategy_is_exact_only(s)]

    for strategy in exact_strategies:
        _match_exact_via_groupby(df, strategy, uf, pair_info)

    if fuzzy_strategies:
        # Pair-compare path. Build the row-index iterator: full Cartesian
        # by default, or per-block when the caller supplied
        # ``blocking_columns``. Blocking is a recall/precision trade —
        # see the docstring.
        if blocking_columns:
            block_keys = _build_block_keys(
                df, blocking_columns, blocking_prefix_len,
            )
            blocks: dict[str, list[int]] = {}
            for pos, key in enumerate(block_keys.tolist()):
                blocks.setdefault(key, []).append(pos)
            block_iters = list(blocks.values())
        else:
            block_iters = [list(range(n))]

        total_pairs = sum(
            len(block) * (len(block) - 1) // 2 for block in block_iters
        )
        checked = 0

        for block in block_iters:
            block_len = len(block)
            for ii in range(block_len):
                i = block[ii]
                row_i = df.iloc[i]
                for jj in range(ii + 1, block_len):
                    j = block[jj]
                    # Preserve i<j ordering for pair_info keys regardless
                    # of how the block iterator sorts.
                    lo, hi = (i, j) if i < j else (j, i)
                    row_j = df.iloc[j]
                    for strategy in fuzzy_strategies:
                        is_match, confidence, cols = _compare_pair(
                            row_i, row_j, strategy
                        )
                        if is_match:
                            uf.union(i, j)
                            key = (lo, hi)
                            # Keep the highest-confidence match. Exact-
                            # path entries (100.0) always win, which
                            # matches the prior tiebreak.
                            if key not in pair_info or confidence > pair_info[key][0]:
                                pair_info[key] = (confidence, cols)
                            break  # OR logic: one strategy match is enough

                    checked += 1
                    if progress_callback and checked % 1000 == 0:
                        progress_callback(checked, total_pairs)

        if progress_callback:
            progress_callback(total_pairs, total_pairs)
    else:
        if progress_callback:
            progress_callback(0, 0)

    # Build MatchResult objects (survivor not yet selected)
    raw_groups = uf.groups()
    match_groups: list[MatchResult] = []
    for gid, (root, members) in enumerate(sorted(raw_groups.items())):
        # Confidence = min across all directly-recorded pairs in the
        # group. Transitive members (A→B and B→C imply A→C) may not have
        # a direct pair_info entry; we only count the recorded ones, so
        # the score reflects observed evidence rather than the optimistic
        # 100.0 default that masks weak links.
        observed_confidences: list[float] = []
        group_cols: set[str] = set()
        for idx_a, m in enumerate(members):
            for idx_b in range(idx_a + 1, len(members)):
                key = (min(m, members[idx_b]), max(m, members[idx_b]))
                if key in pair_info:
                    conf, cols = pair_info[key]
                    observed_confidences.append(conf)
                    group_cols.update(cols)

        if observed_confidences:
            group_confidence = min(observed_confidences)
        else:
            # Edge case: a group with no recorded pair info (shouldn't
            # happen for groups built from union-find on pair_info, but
            # be defensive). Fall back to 100.0 only for trivial groups.
            group_confidence = 100.0

        match_groups.append(MatchResult(
            group_id=gid,
            row_indices=members,
            confidence=round(group_confidence, 2),
            matched_on=sorted(group_cols),
            survivor_index=members[0],  # placeholder
        ))

    return match_groups, pair_info


# ---------------------------------------------------------------------------
# Survivor selection
# ---------------------------------------------------------------------------

def _select_survivor(
    group: MatchResult,
    df: pd.DataFrame,
    rule: SurvivorRule,
    date_column: Optional[str] = None,
) -> int:
    """Choose the survivor row index within a match group."""
    indices = group.row_indices

    if rule == SurvivorRule.KEEP_FIRST:
        return indices[0]

    if rule == SurvivorRule.KEEP_LAST:
        return indices[-1]

    if rule == SurvivorRule.KEEP_MOST_COMPLETE:
        # Fewest empty/blank cells wins
        best_idx = indices[0]
        best_empty = _count_empty(df.iloc[indices[0]])
        for idx in indices[1:]:
            empty = _count_empty(df.iloc[idx])
            if empty < best_empty:
                best_empty = empty
                best_idx = idx
        return best_idx

    if rule == SurvivorRule.KEEP_MOST_RECENT:
        if not date_column or date_column not in df.columns:
            from .errors import InputValidationError
            raise InputValidationError(
                "KEEP_MOST_RECENT requires date_column to be a column in df",
                operation="_select_survivor",
                column=date_column,
                suggestion=(
                    f"Got {date_column!r}; available columns: {list(df.columns)}"
                ),
            )
        best_idx = indices[0]
        best_date = _parse_date(df.iloc[indices[0]].get(date_column, ""))
        for idx in indices[1:]:
            d = _parse_date(df.iloc[idx].get(date_column, ""))
            if d is not None and (best_date is None or d > best_date):
                best_date = d
                best_idx = idx
        return best_idx

    return indices[0]


def _count_empty(row: pd.Series) -> int:
    """Count empty/blank cells in a row, ignoring internal shadow columns."""
    count = 0
    for col, val in row.items():
        if isinstance(col, str) and col.startswith("_norm_"):
            continue
        if pd.isna(val) or str(val).strip() == "":
            count += 1
    return count


def _parse_date(value: Any) -> Optional[pd.Timestamp]:
    """Best-effort date parse for survivor selection.

    Returns None for empty / unparseable values; logs at debug so a
    survivor-selection oddity ("the wrong row got kept") can be traced
    by enabling debug logs without changing code.
    """
    if value is None or (isinstance(value, float) and pd.isna(value)):
        return None
    try:
        return pd.to_datetime(value)
    except (TypeError, ValueError, pd.errors.OutOfBoundsDatetime) as e:
        logger.debug(
            "_parse_date: could not parse {!r} ({}): {}",
            value, type(e).__name__, e,
        )
        return None


# ---------------------------------------------------------------------------
# Merge mode
# ---------------------------------------------------------------------------

def _merge_group(df: pd.DataFrame, survivor_idx: int, loser_indices: list[int]) -> pd.Series:
    """Fill missing fields in survivor from losers (ordered by position)."""
    survivor = df.iloc[survivor_idx].copy()
    for col in survivor.index:
        if isinstance(col, str) and col.startswith("_norm_"):
            continue
        val = survivor[col]
        if pd.isna(val) or str(val).strip() == "":
            for loser_idx in loser_indices:
                candidate = df.iloc[loser_idx][col]
                if not pd.isna(candidate) and str(candidate).strip() != "":
                    survivor[col] = candidate
                    break
    return survivor


# ---------------------------------------------------------------------------
# Auto-detect strategies
# ---------------------------------------------------------------------------

# (pattern, normalizer, algorithm, threshold, is_strong_key)
# Strong keys (email, phone) can be standalone strategies.
# Weak keys (name, address) must be combined with a strong key via AND.
_COLUMN_TYPE_PATTERNS: list[tuple[re.Pattern, NormalizerType, Algorithm, float, bool]] = [
    (re.compile(r"e[-_]?mail", re.I), NormalizerType.EMAIL, Algorithm.EXACT, 100.0, True),
    (re.compile(r"phone|telephone|mobile|cell", re.I), NormalizerType.PHONE, Algorithm.EXACT, 100.0, True),
    (re.compile(r"^(name|full_name|customer_name|first_name|last_name|contact_name|respondent_name)$", re.I),
     NormalizerType.NAME, Algorithm.JARO_WINKLER, 85.0, False),
    (re.compile(r"address|street|addr", re.I), NormalizerType.ADDRESS, Algorithm.TOKEN_SET_RATIO, 80.0, False),
]


def build_default_strategies(df: pd.DataFrame) -> list[MatchStrategy]:
    """Auto-detect column types and build match strategies.

    Strategy logic:
    - Strong keys (email, phone): each gets its own standalone OR strategy.
    - Weak keys (name, address): combined with each strong key via AND to
      form additional strategies.  Weak keys never stand alone (too many
      false positives — "John" ≈ "Jon" at 93 % Jaro-Winkler).
    - If only weak keys are found (no strong keys), they're promoted to
      standalone strategies as a fallback.
    - If no columns match, exact match on all columns (drop_duplicates
      equivalent).
    """
    strong_cols: list[ColumnMatchStrategy] = []
    weak_cols: list[ColumnMatchStrategy] = []

    for col in df.columns:
        if col.startswith("_norm_"):
            continue
        for pattern, norm_type, algo, threshold, is_strong in _COLUMN_TYPE_PATTERNS:
            if pattern.search(col):
                cs = ColumnMatchStrategy(
                    column=col, algorithm=algo,
                    threshold=threshold, normalizer=norm_type,
                )
                if is_strong:
                    strong_cols.append(cs)
                else:
                    weak_cols.append(cs)
                break

    strategies: list[MatchStrategy] = []

    if strong_cols:
        # Each strong key is a standalone strategy (OR)
        for sc in strong_cols:
            strategies.append(MatchStrategy(column_strategies=[sc]))

        # Each weak key is paired with each strong key (AND) for extra recall
        for wc in weak_cols:
            for sc in strong_cols:
                strategies.append(MatchStrategy(column_strategies=[wc, sc]))
    elif weak_cols:
        # No strong keys — promote weak to standalone (best effort)
        for wc in weak_cols:
            strategies.append(MatchStrategy(column_strategies=[wc]))

    if strategies:
        return strategies

    # Fallback: exact match on all columns (equivalent to drop_duplicates)
    logger.info("No column patterns matched; using exact match on all columns")
    all_cols = [
        ColumnMatchStrategy(column=c, algorithm=Algorithm.EXACT, threshold=100.0)
        for c in df.columns
    ]
    return [MatchStrategy(column_strategies=all_cols)]


# ---------------------------------------------------------------------------
# Normalisation pass
# ---------------------------------------------------------------------------

def _apply_normalizations(df: pd.DataFrame, strategies: list[MatchStrategy]) -> pd.DataFrame:
    """Add ``_norm_*`` shadow columns for every column that has a normalizer.

    Normalizers are wrapped in a per-column ``lru_cache`` so repeat values
    (the common case in dedup workloads — the same phone, email, or
    address appears many times) skip re-parsing. ``phonenumbers.parse`` is
    the expensive call in this path; on a 1M-row file with 500k unique
    phones the cache cuts normalization time roughly in half.

    The cache lives only for the lifetime of this call (each invocation
    builds a fresh wrapper), so concurrent calls on different DataFrames
    don't share state and per-process memory doesn't grow unbounded.
    """
    from functools import lru_cache

    df = df.copy()
    seen: set[str] = set()
    for strategy in strategies:
        for cs in strategy.column_strategies:
            if cs.normalizer and cs.column not in seen and cs.column in df.columns:
                seen.add(cs.column)
                norm_fn = get_normalizer(cs.normalizer)

                @lru_cache(maxsize=None)
                def _cached(s: str, _fn=norm_fn) -> str:
                    return _fn(s)

                col_values = df[cs.column]
                norm_col = f"_norm_{cs.column}"
                # Pre-coerce to strings once via Series.map so the cache
                # key is always a ``str`` (matches what the unwrapped
                # apply did via ``fn(str(v))``).
                df[norm_col] = col_values.map(
                    lambda v, c=_cached: c(str(v))
                    if pd.notna(v) and str(v).strip()
                    else ""
                )
    return df


# ---------------------------------------------------------------------------
# Main entry point
# ---------------------------------------------------------------------------

def deduplicate(
    df: pd.DataFrame,
    *,
    strategies: Optional[list[MatchStrategy]] = None,
    survivor_rule: SurvivorRule = SurvivorRule.KEEP_FIRST,
    date_column: Optional[str] = None,
    merge: bool = False,
    preview: bool = True,
    review_callback: Optional[Callable] = None,
    progress_callback: Optional[Callable[[int, int], None]] = None,
    blocking_columns: Optional[list[str]] = None,
    blocking_prefix_len: int = 1,
) -> DeduplicationResult:
    """Run the full deduplication pipeline.

    Pipeline placement (recommended, not enforced)
    ----------------------------------------------
    Run *last* among the cleaning tools. Fuzzy matching is more
    accurate when:
      * text has been hygiened (NBSP padding doesn't make
        ``"Alice "`` look different from ``"Alice"``);
      * formats have been canonicalized (``+14155551234`` matches
        across rows where the source had ``(415) 555-1234`` and
        ``415.555.1234``);
      * missing values have been standardized (NaN matching is
        brittle; sentinel-laundered cells produce false matches).
    See ``src.core.pipeline.SOFT_DEPENDENCIES``.

    Parameters
    ----------
    df : input DataFrame
    strategies : matching strategies (auto-detected if None)
    survivor_rule : which row to keep per group
    date_column : used with ``KEEP_MOST_RECENT``
    merge : fill missing fields in survivor from losers
    preview : if True, result is informational only (no writes)
    review_callback : ``(group: MatchResult, df: DataFrame) -> bool|None``
        Called for each match group. Return True to accept, False to reject,
        None to skip (keep both rows). Used for interactive review.
    progress_callback : ``(current: int, total: int) -> None``
        Called periodically during pairwise comparison.
    blocking_columns : list of column names to partition fuzzy
        comparison by. When set, pairs only get compared if they
        share the same first *blocking_prefix_len* characters on
        every blocking column. Trades recall for speed; default
        ``None`` preserves the original full O(n²) semantics.
        Exact-only strategies always use the O(n) groupby path and
        are unaffected by this option.
    blocking_prefix_len : prefix length used to derive the block
        key. Smaller = bigger blocks = more recall, less speed.
        Default 1 (group by first character).

    Returns a ``DeduplicationResult``.
    """
    from .errors import ensure_dataframe, InputValidationError
    ensure_dataframe(df, function="deduplicate")
    if survivor_rule == SurvivorRule.KEEP_MOST_RECENT and not date_column:
        raise InputValidationError(
            "survivor_rule=KEEP_MOST_RECENT requires date_column",
            operation="deduplicate",
            suggestion=(
                "Pass date_column='created_at' (or whichever column holds "
                "the timestamp). Without it, 'most recent' has no reference."
            ),
        )
    if (
        survivor_rule == SurvivorRule.KEEP_MOST_RECENT
        and date_column
        and date_column not in df.columns
    ):
        raise InputValidationError(
            f"date_column={date_column!r} not found in input",
            operation="deduplicate",
            column=date_column,
            suggestion=f"Available columns: {list(df.columns)}",
        )

    log_entries: list[str] = []
    original_count = len(df)

    if strategies is None:
        strategies = build_default_strategies(df)
        log_entries.append(f"Auto-detected {len(strategies)} match strategies")

    if not strategies:
        # Empty input or no detectable strategies — return the input
        # unchanged with an empty match result, structurally consistent.
        logger.info("deduplicate: no strategies, returning input unchanged")
        return DeduplicationResult(
            original_row_count=original_count,
            deduplicated_df=df.reset_index(drop=True),
            removed_df=df.iloc[0:0].copy().reset_index(drop=True),
            match_groups=[],
            log_entries=log_entries + ["No strategies → no-op"],
            is_preview=preview,
        )

    # Validate every strategy references real columns — silent skip
    # would let a typo (``e_mail`` instead of ``email``) produce a
    # confidently-empty result.
    referenced = {cs.column for s in strategies for cs in s.column_strategies}
    missing = sorted(c for c in referenced if c not in df.columns)
    if missing:
        raise InputValidationError(
            f"Strategy references columns not present in the input: {missing}",
            operation="deduplicate",
            suggestion=(
                f"Available columns: {list(df.columns)}. "
                "Check for typos (e.g., 'e_mail' vs 'email') or for "
                "header rows that didn't get parsed."
            ),
        )

    # Log strategies
    for i, s in enumerate(strategies):
        cols_desc = ", ".join(
            f"{cs.column}({cs.algorithm.value}@{cs.threshold})"
            for cs in s.column_strategies
        )
        log_entries.append(f"Strategy {i}: {cols_desc}")
        logger.info("Strategy {}: {}", i, cols_desc)

    # Normalise
    df_work = _apply_normalizations(df, strategies)

    # Find matches
    match_groups, pair_info = _find_match_groups(
        df_work,
        strategies,
        progress_callback=progress_callback,
        blocking_columns=blocking_columns,
        blocking_prefix_len=blocking_prefix_len,
    )
    log_entries.append(f"Found {len(match_groups)} duplicate groups")
    logger.info("Found {} duplicate groups from {} rows", len(match_groups), original_count)

    # Interactive review
    if review_callback and match_groups:
        reviewed_groups: list[MatchResult] = []
        for group in match_groups:
            decision = review_callback(group, df_work)
            if decision is True:
                reviewed_groups.append(group)
                log_entries.append(f"Group {group.group_id}: accepted by reviewer")
            elif decision is False:
                log_entries.append(f"Group {group.group_id}: rejected by reviewer")
            else:
                log_entries.append(f"Group {group.group_id}: skipped by reviewer")
        match_groups = reviewed_groups

    # Survivor selection
    for group in match_groups:
        group.survivor_index = _select_survivor(group, df_work, survivor_rule, date_column)
        log_entries.append(
            f"Group {group.group_id}: survivor=row {group.survivor_index} "
            f"(rule={survivor_rule.value}, confidence={group.confidence}%)"
        )

    # Build result dataframes
    remove_indices: set[int] = set()
    merged_rows: dict[int, pd.Series] = {}

    for group in match_groups:
        survivor_idx = group.survivor_index
        losers = [i for i in group.row_indices if i != survivor_idx]
        remove_indices.update(losers)

        if merge and losers:
            merged = _merge_group(df_work, survivor_idx, losers)
            merged_rows[survivor_idx] = merged
            # Log merged fields
            original = df_work.iloc[survivor_idx]
            for col in original.index:
                if isinstance(col, str) and col.startswith("_norm_"):
                    continue
                orig_val = str(original[col]).strip()
                new_val = str(merged[col]).strip()
                if orig_val != new_val and not orig_val:
                    log_entries.append(
                        f"Group {group.group_id}: merged '{col}' "
                        f"into survivor from losers: '{new_val}'"
                    )

    # Build output DataFrames
    keep_indices = [i for i in range(len(df_work)) if i not in remove_indices]

    if merged_rows:
        rows = []
        for i in keep_indices:
            if i in merged_rows:
                rows.append(merged_rows[i])
            else:
                rows.append(df_work.iloc[i])
        deduplicated_df = pd.DataFrame(rows)
    else:
        deduplicated_df = df_work.iloc[keep_indices].copy()

    if remove_indices:
        removed_df = df_work.iloc[sorted(remove_indices)].copy()
    else:
        # Empty result: preserve column schema so downstream code can
        # rely on ``removed_df.columns == deduplicated_df.columns``.
        removed_df = df_work.iloc[0:0].copy()

    # Drop shadow columns from output
    norm_cols = [c for c in deduplicated_df.columns if str(c).startswith("_norm_")]
    deduplicated_df = deduplicated_df.drop(columns=norm_cols, errors="ignore")
    removed_df = removed_df.drop(columns=norm_cols, errors="ignore")

    # Reset index
    deduplicated_df = deduplicated_df.reset_index(drop=True)
    removed_df = removed_df.reset_index(drop=True)

    removed_count = original_count - len(deduplicated_df)
    log_entries.append(f"Result: {original_count} → {len(deduplicated_df)} rows ({removed_count} removed)")

    return DeduplicationResult(
        original_row_count=original_count,
        deduplicated_df=deduplicated_df,
        removed_df=removed_df,
        match_groups=match_groups,
        log_entries=log_entries,
        is_preview=preview,
    )
