"""Upload-time data quality analyzer.

Runs a fast, read-only scan over an uploaded file (or DataFrame) and
returns a list of :class:`Finding` objects. Each finding names the issue,
how many cells/rows are affected, and which downstream tool can address
it. The GUI consumes findings to badge tool nav items; the CLI prints
them as a table.

The analyzer is *purely advisory*: it never mutates data, never runs a
tool, and is safe to skip. Treat it as a guided onboarding step, not a
hard gate on the upload flow.
"""

from __future__ import annotations

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

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

from .io import RepairResult, repair_bytes, detect_encoding, detect_delimiter

Severity = Literal["info", "warn", "error"]
Confidence = Literal["high", "medium", "low"]


# Tool identifiers — match the 0N_<name> convention used by the script set.
# Listed here so detectors stay decoupled from the GUI's display layer.
TOOL_TEXT_CLEANER = "02_text_cleaner"
TOOL_MISSING_HANDLER = "04_missing_handler"
TOOL_DEDUPLICATOR = "01_deduplicator"
TOOL_FORMAT_STANDARDIZER = "03_format_standardizer"


# Stable fix-action ids. These name the algorithm that resolves a finding;
# the normalize layer dispatches on this id. Keep in sync with fixes.py.
FIX_TRIM_WHITESPACE = "trim_whitespace"
FIX_STRIP_NBSP = "strip_nbsp_unicode_whitespace"
FIX_STRIP_ZERO_WIDTH = "strip_zero_width"
FIX_FOLD_SMART_PUNCT = "fold_smart_punctuation"
FIX_CLEAN_HEADERS = "clean_headers"
FIX_NORMALIZE_LINE_ENDINGS = "normalize_line_endings"
FIX_STRIP_BOM = "strip_bom"
FIX_STRIP_NUL = "strip_nul"
FIX_FOLD_SMART_QUOTES_BYTE = "fold_smart_quotes_byte"
FIX_REPAIR_UNQUOTED_DELIM = "repair_unquoted_delimiters"
FIX_LOWERCASE_EMAIL = "lowercase_email_column"
FIX_REPLACE_NULL_SENTINELS = "replace_null_sentinels"
FIX_REPAIR_MOJIBAKE = "repair_mojibake"
FIX_NONE = ""  # informational — nothing to apply

# Replacement character (U+FFFD) inserted when a decoder gave up on a byte.
# Anything more than a tiny ratio of it in the loaded text is a strong
# signal that the encoding was wrong.
_REPLACEMENT_CHAR = "�"


@dataclass
class Finding:
    """One issue the analyzer surfaced.

    Attributes
    ----------
    id
        Stable identifier (``"smart_quotes_in_data"``); used for GUI lookup
        and downloadable JSON exports. Never localized.
    severity
        ``"info"`` (FYI), ``"warn"`` (likely needs cleanup),
        ``"error"`` (will block downstream work).
    confidence
        ``"high"`` — round-trip-safe algorithmic fix, eligible for auto-fix.
        ``"medium"`` — right call in the common case but has known
        false-positive shapes; user should preview before applying.
        ``"low"`` — heuristic; the wrong call corrupts data; opt-in only.
        Independent of severity: a ``warn`` finding can be high-confidence
        (NBSP strip) and an ``info`` finding can be low-confidence (mojibake).
    fix_action
        Stable id naming the algorithm that resolves this finding. Empty
        string for informational findings with no associated fix.
    tool
        Tool id that can address the finding, or empty string for purely
        informational findings.
    count
        Number of cells (or rows) affected.
    description
        Single-sentence human summary used for banners and tooltips.
    column
        Column name when scoped to one column; ``None`` for whole-frame /
        file-level findings.
    samples
        Up to a handful of ``(row, column, value)`` tuples for the GUI
        to render. Cap at five so the JSON export stays compact.
    """

    id: str
    severity: Severity
    tool: str
    count: int
    description: str
    column: Optional[str] = None
    samples: list[tuple[int, str, str]] = field(default_factory=list)
    confidence: Confidence = "high"
    fix_action: str = FIX_NONE
    # True when the fix already ran during the pre-parse repair pass
    # (e.g. BOM strip, byte-level smart-quote fold). The gate treats these
    # as already-resolved; the review page still surfaces them so the
    # user can see what was auto-applied during read.
    pre_applied: bool = False


# ---------------------------------------------------------------------------
# Per-cell character classes (kept independent of text_clean to avoid an
# import cycle and to keep the analyzer self-contained).
# ---------------------------------------------------------------------------

_SMART_QUOTE_CHARS = set("“”‘’„‟«»′″")
_DASH_ELLIPSIS_CHARS = set("–—―−…")
_NBSP_LIKE_CHARS = set("      　")
_ZERO_WIDTH_CHARS = set("​‌‍⁠‎‏﻿­")

_NULL_LIKE = {
    "n/a", "na", "nan", "null", "none", "#n/a", "#na", "-", "--",
    "tbd", "unknown", "n.a.", "(null)",
    # Pandas-specific: NA values stringified via str(pd.NA) → "<NA>".
    "<na>",
}

# Mojibake fingerprints: classic UTF-8-as-cp1252 corruptions.
_MOJIBAKE_PATTERNS = re.compile(
    r"Ã[©¨¢¤¶]"   # café -> cafÃ©, étage -> Ã©tage etc.
    r"|â€[™œžs˜“”]"            # don't -> donâ€™t
    r"|Â[ -¿]"
)

_LEADING_ZERO_ID_RE = re.compile(r"^0\d{2,}$")
_DIGITS_RE = re.compile(r"^\d+$")
_EMAIL_LIKE_COL = re.compile(r"e?[ -_]?mail|^email|address$", re.IGNORECASE)


def _has_any(text: str, chars: set[str]) -> bool:
    return any(c in chars for c in text)


def _samples(rows: Iterable[tuple[int, str, str]], limit: int = 5) -> list[tuple[int, str, str]]:
    out: list[tuple[int, str, str]] = []
    for item in rows:
        out.append(item)
        if len(out) >= limit:
            break
    return out


# Compiled character classes for the vectorized hot path. Pandas
# ``str.contains(pattern, regex=True)`` runs in C and beats per-cell
# Python loops by 50-100× on multi-million-row inputs.
_SMART_PUNCT_RE = "[" + re.escape("".join(_SMART_QUOTE_CHARS | _DASH_ELLIPSIS_CHARS)) + "]"
_NBSP_RE = "[" + re.escape("".join(_NBSP_LIKE_CHARS)) + "]"
_ZW_RE = "[" + re.escape("".join(_ZERO_WIDTH_CHARS)) + "]"
_NULL_LIKE_RE = (
    r"^\s*(?:" + "|".join(re.escape(s) for s in sorted(_NULL_LIKE, key=len, reverse=True)) + r")\s*$"
)
_WHITESPACE_PAD_RE = r"^\s|\s$|  "  # leading, trailing, or any double-space


def _str_columns(df: pd.DataFrame) -> list[str]:
    """Object-dtype columns — only ones that can hold Python strings."""
    return [
        c for c in df.columns
        if pdtypes.is_object_dtype(df[c]) or pdtypes.is_string_dtype(df[c])
    ]


def _vec_match_count(
    df: pd.DataFrame, pattern: str, *, sample_limit: int = 5,
) -> tuple[int, list[tuple[int, str, str]]]:
    """Count cells in object-dtype columns that match *pattern*; collect samples.

    Vectorized: uses ``Series.str.contains`` (C-implemented) per column
    plus a single ``.sum()`` for the count and ``.head(N)`` for samples.
    Returns ``(total_cells_matched, sample_tuples)``.
    """
    total = 0
    samples: list[tuple[int, str, str]] = []
    for col in _str_columns(df):
        ser = df[col]
        # ``.str.contains`` returns NaN for non-strings; ``na=False`` makes
        # it boolean. Only object-dtype columns are scanned, so this is
        # safe and avoids the per-cell isinstance check in the old loop.
        mask = ser.str.contains(pattern, regex=True, na=False)
        n = int(mask.sum())
        if not n:
            continue
        total += n
        if len(samples) < sample_limit:
            for idx in mask[mask].index[:sample_limit - len(samples)]:
                samples.append((int(idx), str(col), str(ser.iloc[idx])))
    return total, samples


# ---------------------------------------------------------------------------
# Detectors
# ---------------------------------------------------------------------------

def _detect_smart_punctuation(df: pd.DataFrame) -> list[Finding]:
    affected_cells, sample_rows = _vec_match_count(df, _SMART_PUNCT_RE)
    if not affected_cells:
        return []
    return [Finding(
        id="smart_punctuation_in_data",
        severity="warn",
        tool=TOOL_TEXT_CLEANER,
        count=affected_cells,
        description=(
            f"{affected_cells} cell(s) contain curly quotes, em/en dashes, "
            f"or ellipsis characters. These break string equality joins and "
            f"regex patterns."
        ),
        samples=sample_rows,
        confidence="high",
        fix_action=FIX_FOLD_SMART_PUNCT,
    )]


def _detect_invisible_chars(df: pd.DataFrame) -> list[Finding]:
    nbsp_cells, nbsp_samples = _vec_match_count(df, _NBSP_RE)
    zw_cells, zw_samples = _vec_match_count(df, _ZW_RE)
    findings: list[Finding] = []
    if nbsp_cells:
        findings.append(Finding(
            id="nbsp_or_unicode_whitespace",
            severity="warn",
            tool=TOOL_TEXT_CLEANER,
            count=nbsp_cells,
            description=(
                f"{nbsp_cells} cell(s) contain non-breaking or other Unicode "
                f"spaces. These look identical to a regular space but break "
                f"join keys."
            ),
            samples=nbsp_samples,
            confidence="high",
            fix_action=FIX_STRIP_NBSP,
        ))
    if zw_cells:
        findings.append(Finding(
            id="zero_width_or_invisible",
            severity="warn",
            tool=TOOL_TEXT_CLEANER,
            count=zw_cells,
            description=(
                f"{zw_cells} cell(s) contain zero-width or invisible "
                f"characters (ZWSP, ZWJ, soft hyphen, BOM, bidi marks)."
            ),
            samples=zw_samples,
            confidence="high",
            fix_action=FIX_STRIP_ZERO_WIDTH,
        ))
    # Headers carry the same risks; flag separately so the user sees that
    # df["Email"] vs df["Email​"] is the issue.
    bad_headers = [
        c for c in df.columns
        if isinstance(c, str) and (
            c != c.strip()
            or _has_any(c, _NBSP_LIKE_CHARS)
            or _has_any(c, _ZERO_WIDTH_CHARS)
            or _has_any(c, _SMART_QUOTE_CHARS)
        )
    ]
    if bad_headers:
        findings.append(Finding(
            id="dirty_column_headers",
            severity="warn",
            tool=TOOL_TEXT_CLEANER,
            count=len(bad_headers),
            description=(
                f"{len(bad_headers)} column header(s) contain whitespace, "
                f"smart quotes, or invisible characters. These break "
                f"df['col'] lookups."
            ),
            samples=[(0, h, h) for h in bad_headers[:5]],
            confidence="high",
            fix_action=FIX_CLEAN_HEADERS,
        ))
    return findings


def _detect_whitespace_padding(df: pd.DataFrame) -> list[Finding]:
    affected, samples = _vec_match_count(df, _WHITESPACE_PAD_RE)
    if not affected:
        return []
    return [Finding(
        id="whitespace_padding",
        severity="warn",
        tool=TOOL_TEXT_CLEANER,
        count=affected,
        description=(
            f"{affected} cell(s) have leading/trailing whitespace or "
            f"multi-space internal runs. Common cause of failed joins."
        ),
        samples=samples,
        confidence="high",
        fix_action=FIX_TRIM_WHITESPACE,
    )]


def _detect_null_like_sentinels(df: pd.DataFrame) -> list[Finding]:
    affected = 0
    samples: list[tuple[int, str, str]] = []
    cols_with_sentinels: set[str] = set()
    for col in _str_columns(df):
        ser = df[col]
        # Case-insensitive match: lowercase the series once, then test.
        lowered = ser.str.strip().str.lower()
        mask = lowered.isin(_NULL_LIKE)
        n = int(mask.sum())
        if not n:
            continue
        affected += n
        cols_with_sentinels.add(str(col))
        if len(samples) < 5:
            for idx in mask[mask].index[:5 - len(samples)]:
                samples.append((int(idx), str(col), str(ser.iloc[idx])))
    if not affected:
        return []
    return [Finding(
        id="null_like_sentinels",
        severity="info",
        tool=TOOL_MISSING_HANDLER,
        count=affected,
        description=(
            f"{affected} cell(s) across {len(cols_with_sentinels)} column(s) "
            f"look like disguised nulls (N/A, NaN, None, '-'). Decide what "
            f"counts as missing in the missing-value handler."
        ),
        samples=samples,
        confidence="medium",
        fix_action=FIX_REPLACE_NULL_SENTINELS,
    )]


def _detect_mojibake(df: pd.DataFrame) -> list[Finding]:
    affected, samples = _vec_match_count(df, _MOJIBAKE_PATTERNS.pattern)
    if not affected:
        return []
    return [Finding(
        id="suspected_mojibake",
        severity="info",
        tool=TOOL_TEXT_CLEANER,
        count=affected,
        description=(
            f"{affected} cell(s) match common UTF-8-as-cp1252 mojibake "
            f"patterns (Ã©, â€™, etc.). Auto-repair is opt-in (Tier 2)."
        ),
        samples=samples,
        confidence="low",
        fix_action=FIX_REPAIR_MOJIBAKE,
    )]


# Date-shaped patterns for the inconsistent-format detector.
_DATE_FORMAT_PATTERNS: dict[str, str] = {
    "iso":      r"^\d{4}-\d{1,2}-\d{1,2}$",
    "us_slash": r"^\d{1,2}/\d{1,2}/\d{2,4}$",
    "eu_dot":   r"^\d{1,2}\.\d{1,2}\.\d{2,4}$",
    "eu_slash": r"^\d{1,2}/\d{1,2}/\d{4}$",  # may overlap us_slash; resolved by us_slash first
}
_DATE_FORMAT_RE: dict[str, "re.Pattern"] = {
    name: re.compile(pat) for name, pat in _DATE_FORMAT_PATTERNS.items()
}


def _detect_inconsistent_date_format(df: pd.DataFrame) -> list[Finding]:
    """Flag columns whose date-shaped values use multiple incompatible formats.

    A column is "date-shaped" if more than half its non-empty values
    match one of the recognized date regexes. If two or more distinct
    formats are present (each format counted with one or more matches)
    AND the column is overall date-shaped, emit a finding routed to
    the format standardizer.

    Earlier versions required two matches per format, which missed a
    legitimate real-world case: a column with N different date formats
    where each appears once — that's still inconsistent and worth
    flagging. The 50% date-shaped overall threshold still prevents
    false positives on free-text columns that happen to contain a
    couple of date-like substrings.
    """
    findings: list[Finding] = []
    for col in df.columns:
        try:
            ser = df[col].dropna().astype(str)
        except (TypeError, ValueError) as e:
            logger.debug(
                "date-format detector: skipping {!r} ({}): {}",
                col, type(e).__name__, e,
            )
            continue
        nonempty = ser[ser.str.strip().astype(bool)]
        if len(nonempty) < 4:
            continue
        format_counts: dict[str, int] = {}
        for name, pat in _DATE_FORMAT_RE.items():
            count = int(nonempty.str.match(pat).sum())
            if count >= 1:
                format_counts[name] = count
        if len(format_counts) < 2:
            continue
        total_date_shaped = sum(format_counts.values())
        if total_date_shaped < len(nonempty) * 0.35:
            continue
        format_summary = ", ".join(
            f"{n}({c})" for n, c in sorted(
                format_counts.items(), key=lambda kv: -kv[1]
            )
        )
        samples_idx = nonempty.head(5)
        samples = [(int(i), str(col), str(v)) for i, v in samples_idx.items()]
        findings.append(Finding(
            id="inconsistent_date_format",
            severity="info",
            tool=TOOL_FORMAT_STANDARDIZER,
            count=int(total_date_shaped),
            description=(
                f"Column '{col}' contains dates in multiple formats: "
                f"{format_summary}. Run format standardizer to normalize."
            ),
            column=str(col),
            samples=samples,
            confidence="medium",
            fix_action=FIX_NONE,
        ))
    return findings


# Phone / currency / boolean format regexes used by the inconsistency
# detectors below. Each map is name → regex pattern. Recognized formats
# are deliberately overlapping at the column level (one cell matches
# only one); the inconsistency detector fires when a single column has
# values matching ≥2 distinct format names.
_PHONE_FORMAT_RE = {
    "plain_10":    re.compile(r"^\d{10}$"),
    "us_paren":    re.compile(r"^\(\d{3}\)\s*\d{3}[\s.\-]?\d{4}$"),
    "us_dash":     re.compile(r"^\d{3}-\d{3}-\d{4}$"),
    "us_dot":      re.compile(r"^\d{3}\.\d{3}\.\d{4}$"),
    "us_space":    re.compile(r"^\d{3}\s\d{3}\s\d{4}$"),
    "plus_country":re.compile(r"^\+\d{1,3}[\s.\-]\d.*\d$"),
    "intl_plus":   re.compile(r"^\+\d{2,15}$"),
    "country_prefix": re.compile(r"^1[\s.\-]\d{3}[\s.\-]\d{3}[\s.\-]\d{4}$"),
    "extension":   re.compile(r"^.*\b(ext|x)\.?\s*\d+$", re.IGNORECASE),
}

_CURRENCY_FORMAT_RE = {
    "us_with_symbol":  re.compile(r"^\$\s*\d{1,3}(,\d{3})*(\.\d{2})?$"),
    "us_no_symbol":    re.compile(r"^\d{1,3}(,\d{3})*\.\d{2}$"),
    "us_plain":        re.compile(r"^\$\d+\.\d+$"),  # $1234.56 — no thousands
    "us_with_suffix":  re.compile(r"^\d{1,3}(,\d{3})*\.\d{2}\s*(USD|EUR|GBP|CAD|AUD)$"),
    "us_with_prefix":  re.compile(r"^(USD|EUR|GBP|CAD|AUD)\s+\d{1,3}(,\d{3})*\.\d{2}$"),
    "trailing_symbol": re.compile(r"^\d.*[\$€£¥]$"),  # 1234.56$
    "eu_dot_comma":    re.compile(r"^[€£¥₹]?\s*\d{1,3}([.\s]\d{3})*,\d{2}$"),
    "eu_no_decimals":  re.compile(r"^[€£¥₹]\s*\d{1,3}(,\d{3})*$"),  # ¥1,234
    "in_lakh":         re.compile(r"^[₹]\s*\d{1,2}(,\d{2})+(,\d{3})(\.\d{1,2})?$"),  # ₹1,23,456.78
    "swiss_apostrophe":re.compile(r"^\d{1,3}('\d{3})+(\.\d{2})?$"),  # 1'234.56
    "parens_negative": re.compile(r"^\(\s*[\$€£¥]?\d.*\)$"),
    "negative_prefix": re.compile(r"^-[\$€£¥]?\d.*$"),
    "negative_after":  re.compile(r"^[\$€£¥]-\d.*$"),  # $-100.00
}

_BOOL_TRUE = {"yes", "y", "true", "t", "1"}
_BOOL_FALSE = {"no", "n", "false", "f", "0"}


def _detect_inconsistent_phone_format(df: pd.DataFrame) -> list[Finding]:
    """Same shape as the date detector — fire on a phone-shaped column
    when ≥2 distinct phone formats are present."""
    findings: list[Finding] = []
    for col in df.columns:
        try:
            ser = df[col].dropna().astype(str)
        except (TypeError, ValueError):
            continue
        nonempty = ser[ser.str.strip().astype(bool)]
        if len(nonempty) < 4:
            continue
        format_counts: dict[str, int] = {}
        for name, pat in _PHONE_FORMAT_RE.items():
            count = int(nonempty.str.match(pat).sum())
            if count >= 1:
                format_counts[name] = count
        if len(format_counts) < 2:
            continue
        total = sum(format_counts.values())
        if total < len(nonempty) * 0.35:
            continue
        summary = ", ".join(
            f"{n}({c})" for n, c in sorted(
                format_counts.items(), key=lambda kv: -kv[1]
            )
        )
        samples_idx = nonempty.head(5)
        samples = [(int(i), str(col), str(v)) for i, v in samples_idx.items()]
        findings.append(Finding(
            id="inconsistent_phone_format",
            severity="info",
            tool=TOOL_FORMAT_STANDARDIZER,
            count=int(total),
            description=(
                f"Column '{col}' contains phone numbers in multiple "
                f"formats: {summary}. Run format standardizer to normalize."
            ),
            column=str(col),
            samples=samples,
            confidence="medium",
            fix_action=FIX_NONE,
        ))
    return findings


def _detect_inconsistent_currency_format(df: pd.DataFrame) -> list[Finding]:
    """Fire when a currency-shaped column mixes formats."""
    findings: list[Finding] = []
    for col in df.columns:
        try:
            ser = df[col].dropna().astype(str)
        except (TypeError, ValueError):
            continue
        nonempty = ser[ser.str.strip().astype(bool)]
        if len(nonempty) < 4:
            continue
        format_counts: dict[str, int] = {}
        for name, pat in _CURRENCY_FORMAT_RE.items():
            count = int(nonempty.str.match(pat).sum())
            if count >= 1:
                format_counts[name] = count
        if len(format_counts) < 2:
            continue
        total = sum(format_counts.values())
        if total < len(nonempty) * 0.35:
            continue
        summary = ", ".join(
            f"{n}({c})" for n, c in sorted(
                format_counts.items(), key=lambda kv: -kv[1]
            )
        )
        samples_idx = nonempty.head(5)
        samples = [(int(i), str(col), str(v)) for i, v in samples_idx.items()]
        findings.append(Finding(
            id="inconsistent_currency_format",
            severity="info",
            tool=TOOL_FORMAT_STANDARDIZER,
            count=int(total),
            description=(
                f"Column '{col}' contains currency values in multiple "
                f"formats: {summary}. Run format standardizer to normalize."
            ),
            column=str(col),
            samples=samples,
            confidence="medium",
            fix_action=FIX_NONE,
        ))
    return findings


def _detect_inconsistent_boolean_format(df: pd.DataFrame) -> list[Finding]:
    """Fire when a boolean-valued column mixes representations
    (e.g. ``Yes`` / ``Y`` / ``true`` / ``1`` in the same column)."""
    findings: list[Finding] = []
    bool_tokens = _BOOL_TRUE | _BOOL_FALSE
    for col in df.columns:
        try:
            ser = df[col].dropna().astype(str)
        except (TypeError, ValueError):
            continue
        nonempty = ser[ser.str.strip().astype(bool)]
        if len(nonempty) < 4:
            continue
        lowered = nonempty.str.strip().str.lower()
        bool_mask = lowered.isin(bool_tokens)
        if bool_mask.sum() < len(nonempty) * 0.8:
            continue
        # Distinct underlying tokens — case-insensitive count of the
        # different surface forms used in the column.
        distinct = set(lowered[bool_mask].unique())
        if len(distinct) < 3:
            # 2 distinct tokens is the normal yes/no shape; only flag
            # when there are at least 3 distinct surface forms.
            continue
        summary = ", ".join(sorted(distinct))
        samples_idx = nonempty.head(5)
        samples = [(int(i), str(col), str(v)) for i, v in samples_idx.items()]
        findings.append(Finding(
            id="inconsistent_boolean_format",
            severity="info",
            tool=TOOL_FORMAT_STANDARDIZER,
            count=int(bool_mask.sum()),
            description=(
                f"Column '{col}' uses mixed boolean representations: "
                f"{summary}. Run format standardizer to normalize."
            ),
            column=str(col),
            samples=samples,
            confidence="medium",
            fix_action=FIX_NONE,
        ))
    return findings


def _detect_mixed_case_email(df: pd.DataFrame) -> list[Finding]:
    findings: list[Finding] = []
    for col in df.columns:
        if not isinstance(col, str) or not _EMAIL_LIKE_COL.search(col):
            continue
        # Drop NaN/None *before* astype(str), otherwise None becomes the
        # string "None" — which contains both upper "N" and lower "one"
        # and would trigger a false-positive mixed-case finding on a
        # column that has no real emails at all.
        ser = df[col].dropna().astype(str)
        nonempty = ser[ser.str.strip().astype(bool)]
        if nonempty.empty:
            continue
        has_upper = nonempty.str.contains(r"[A-Z]", regex=True, na=False).any()
        has_lower = nonempty.str.contains(r"[a-z]", regex=True, na=False).any()
        if has_upper and has_lower:
            head = nonempty.head(5)
            samples = [(int(i), col, str(v)) for i, v in head.items()]
            findings.append(Finding(
                id="mixed_case_email_column",
                severity="info",
                tool=TOOL_TEXT_CLEANER,
                count=int(len(nonempty)),
                description=(
                    f"Column '{col}' has mixed case across email values. "
                    f"Lowercasing emails before dedup avoids false negatives."
                ),
                column=col,
                samples=samples,
                confidence="medium",
                fix_action=FIX_LOWERCASE_EMAIL,
            ))
    return findings


def _detect_near_duplicates(df: pd.DataFrame) -> list[Finding]:
    """Detect duplicate rows that differ only in case or padding.

    Cheap pass: lowercase + strip every string column, then count
    ``df.duplicated()``. Catches the most common dedup signal (the same
    customer entered twice with subtle formatting differences) without
    paying the cost of fuzzy matching. Anything more sophisticated belongs
    in tool 01.

    Skips the full ``df.copy()`` that previously doubled peak memory on
    1 GB files — builds only the normalized string columns (the columns
    that change) and references the rest by view so pandas reuses the
    underlying buffer.
    """
    if len(df) < 2:
        return []
    columns = {}
    for col in df.columns:
        s = df[col]
        if pdtypes.is_object_dtype(s) or pdtypes.is_string_dtype(s):
            # Skip the redundant ``astype(str)`` when the column is
            # already a string dtype — saves a column-sized allocation
            # per textual column.
            base = s if pdtypes.is_string_dtype(s) else s.astype(str)
            columns[col] = base.str.strip().str.lower()
        else:
            columns[col] = s
    norm = pd.DataFrame(columns, copy=False)
    dup_mask = norm.duplicated(keep=False)
    n_dupes = int(dup_mask.sum())
    if n_dupes < 2:
        return []
    # ``n_groups`` is the count of unique duplicate signatures; each
    # group contains 2+ rows. ``n_extra`` is rows that would be removed
    # by dedup (total in groups minus one survivor per group) — that's
    # the number the user usually wants ("remove X to fix").
    n_groups = int(norm[dup_mask].drop_duplicates().shape[0])
    n_extra = n_dupes - n_groups
    samples: list[tuple[int, str, str]] = []
    for i in df[dup_mask].index[:5]:
        # Render the first textual column's value as a sample.
        col_name = next(
            (c for c in df.columns if isinstance(df[c].iloc[i], str)),
            df.columns[0],
        )
        samples.append((int(i), str(col_name), str(df[col_name].iloc[i])))
    return [Finding(
        id="near_duplicate_rows",
        severity="info",
        tool=TOOL_DEDUPLICATOR,
        count=n_extra,
        description=(
            f"{n_extra} extra copy(ies) across {n_groups} duplicate group(s) "
            f"({n_dupes} rows total) — duplicates after stripping whitespace "
            f"and lowercasing string columns. Run the deduplicator to merge "
            f"or remove."
        ),
        samples=samples,
        confidence="medium",
        fix_action=FIX_NONE,  # routed to dedup tool, not auto-fixed here
    )]


def _detect_leading_zero_ids(df: pd.DataFrame) -> list[Finding]:
    """Informational: a column where most values are zero-padded digit IDs.

    Worth surfacing because Excel re-opens often strip them — the user
    should know they're there before any Excel round-trip.
    """
    findings: list[Finding] = []
    for col in _str_columns(df):
        ser = df[col].astype(str)
        nonempty = ser[ser.str.strip().astype(bool)]
        n = int(len(nonempty))
        if n < 5:
            continue
        digit_count = int(nonempty.str.match(r"^\d+$").sum())
        leading_zero_mask = nonempty.str.match(r"^0\d{2,}$")
        leading_zero_count = int(leading_zero_mask.sum())
        # >80% are zero-padded digit IDs of the same length-ish.
        if digit_count >= 0.8 * n and leading_zero_count >= 0.5 * n:
            head = nonempty[leading_zero_mask].head(5)
            samples = [(int(i), str(col), str(v)) for i, v in head.items()]
            findings.append(Finding(
                id="leading_zero_ids",
                severity="info",
                tool="",
                count=leading_zero_count,
                description=(
                    f"Column '{col}' contains zero-padded numeric IDs "
                    f"({leading_zero_count}/{n}). Excel will strip "
                    f"the zeros on round-trip unless saved as text."
                ),
                column=str(col),
                samples=samples,
                confidence="low",
                fix_action=FIX_NONE,  # informational only
            ))
    return findings


def _count_row_terminators(raw: bytes) -> tuple[int, int, int]:
    """Count CRLF / LF / CR sequences that act as *row* terminators.

    Walks the bytes tracking quoted-region state so that line breaks
    inside multi-line quoted cells (e.g. an address column) are not
    counted. Without this, files that legitimately have CRLF at row
    boundaries plus LF inside quoted cells get false-positive
    ``mixed_line_endings`` findings.
    """
    n_crlf = n_lf = n_cr = 0
    in_quotes = False
    i = 0
    n = len(raw)
    while i < n:
        b = raw[i]
        if b == 0x22:  # ASCII double quote — toggles quoted region.
            # Doubled quote inside a quoted cell is an escape, not an exit.
            if in_quotes and i + 1 < n and raw[i + 1] == 0x22:
                i += 2
                continue
            in_quotes = not in_quotes
            i += 1
            continue
        if not in_quotes:
            if b == 0x0D:  # CR
                if i + 1 < n and raw[i + 1] == 0x0A:
                    n_crlf += 1
                    i += 2
                    continue
                n_cr += 1
            elif b == 0x0A:  # LF
                n_lf += 1
        i += 1
    return n_crlf, n_lf, n_cr


def _detect_lying_bom(raw: bytes) -> list[Finding]:
    """Flag files whose UTF-8 BOM disagrees with the body bytes.

    The "lying BOM" pattern is a file that starts with the UTF-8 BOM
    (``EF BB BF``) but whose body cannot be decoded as UTF-8 — typically
    a cp1252 export that someone hand-prepended a BOM to in an attempt to
    make Excel happy. The encoding detector recovers transparently
    (returns cp1252), but the user should still be told their file is
    misrepresenting itself so the next downstream tool doesn't get
    surprised.
    """
    if not raw[:3] == b"\xef\xbb\xbf":
        return []
    try:
        raw[3:].decode("utf-8")
        return []  # honest BOM — body is real UTF-8
    except UnicodeDecodeError:
        pass
    return [Finding(
        id="encoding_lying_bom",
        severity="warn",
        tool="",
        count=1,
        description=(
            "File starts with a UTF-8 BOM, but the body bytes are not "
            "valid UTF-8 — the BOM is misleading. The encoding detector "
            "recovered by falling back to a single-byte codepage; you "
            "may want to re-save the file with a matching encoding."
        ),
        confidence="high",
        fix_action=FIX_NONE,
    )]


def _detect_mixed_line_endings(raw: bytes) -> list[Finding]:
    """Flag files that mix CRLF, LF, and bare CR row terminators.

    Mixed endings are a classic disaster pattern after multi-source concat
    (Windows + macOS + Linux exports stitched together). Counts only the
    terminators that act as row separators, so embedded newlines inside
    quoted multi-line cells don't create false positives. Operates on raw
    bytes only — DataFrame-mode :func:`analyze` skips this detector.
    """
    if not raw:
        return []
    n_crlf, n_lf, n_cr = _count_row_terminators(raw)
    kinds_present = sum(1 for n in (n_crlf, n_lf, n_cr) if n > 0)
    if kinds_present <= 1:
        return []
    breakdown = []
    if n_crlf:
        breakdown.append(f"{n_crlf} CRLF")
    if n_lf:
        breakdown.append(f"{n_lf} LF")
    if n_cr:
        breakdown.append(f"{n_cr} CR")
    return [Finding(
        id="mixed_line_endings",
        severity="warn",
        tool=TOOL_TEXT_CLEANER,
        count=kinds_present,
        description=(
            f"File mixes {kinds_present} line-ending styles "
            f"({', '.join(breakdown)}). Naive splits on one style produce "
            f"ghost rows or merged lines. Run the text cleaner to normalize."
        ),
        confidence="high",
        fix_action=FIX_NORMALIZE_LINE_ENDINGS,
    )]


def _detect_encoding_uncertainty(df: pd.DataFrame) -> list[Finding]:
    """Flag DataFrames whose loaded text contains U+FFFD replacement chars.

    The replacement character is what Python's decoder substitutes for
    bytes it could not interpret under ``errors="replace"``. Any non-zero
    count is a strong signal that the encoding picked by the loader was
    wrong for at least part of the file — classic lying-BOM, mixed-encoding,
    or wrong-codepage symptom. The user has to pick: re-upload with an
    explicit encoding, or accept the loss.
    """
    affected_cells, sample_rows = _vec_match_count(df, re.escape(_REPLACEMENT_CHAR))
    bad_headers = [
        c for c in df.columns
        if isinstance(c, str) and _REPLACEMENT_CHAR in c
    ]
    if not affected_cells and not bad_headers:
        return []
    location = []
    if affected_cells:
        location.append(f"{affected_cells} cell(s)")
    if bad_headers:
        location.append(f"{len(bad_headers)} header(s)")
    return [Finding(
        id="encoding_uncertain",
        severity="error",
        tool="",
        count=affected_cells + len(bad_headers),
        description=(
            f"{' and '.join(location)} contain U+FFFD replacement characters, "
            f"which means the file's encoding could not be decoded cleanly. "
            f"Re-upload with an explicit encoding (e.g. cp1252, latin-1) "
            f"or fix the source. Continuing risks silent data loss."
        ),
        samples=sample_rows,
        confidence="low",
        fix_action=FIX_NONE,
    )]


def _findings_from_repair(repair: RepairResult) -> list[Finding]:
    """Synthesize findings from a :class:`RepairResult`.

    Each repair kind maps to a single info-severity finding so the GUI
    shows the user what the parser quietly fixed before they reached the
    tool pages.
    """
    if not repair.changed and not repair.unrepairable_lines:
        return []
    summary = repair.summary()
    findings: list[Finding] = []
    if "strip_bom" in summary:
        findings.append(Finding(
            id="csv_bom_stripped",
            severity="info",
            tool=TOOL_TEXT_CLEANER,
            count=1,
            description="UTF-8 BOM at file start was removed before parsing.",
            confidence="high",
            fix_action=FIX_STRIP_BOM,
            pre_applied=True,
        ))
    if "strip_nul" in summary:
        nul_action = next(a for a in repair.actions if a.kind == "strip_nul")
        findings.append(Finding(
            id="csv_nul_stripped",
            severity="warn",
            tool=TOOL_TEXT_CLEANER,
            count=1,
            description=(
                f"Embedded NUL bytes in the file were stripped before "
                f"parsing ({nul_action.detail})."
            ),
            confidence="high",
            fix_action=FIX_STRIP_NUL,
            pre_applied=True,
        ))
    if "fold_smart_quote" in summary:
        action = next(a for a in repair.actions if a.kind == "fold_smart_quote")
        findings.append(Finding(
            id="csv_smart_quotes_folded",
            severity="info",
            tool=TOOL_TEXT_CLEANER,
            count=1,
            description=(
                f"Smart double quotes were folded to ASCII before parsing "
                f"({action.detail})."
            ),
            confidence="high",
            fix_action=FIX_FOLD_SMART_QUOTES_BYTE,
            pre_applied=True,
        ))
    if "normalize_line_endings" in summary:
        action = next(a for a in repair.actions if a.kind == "normalize_line_endings")
        findings.append(Finding(
            id="csv_line_endings_normalized",
            severity="info",
            tool=TOOL_TEXT_CLEANER,
            count=1,
            description=(
                f"Line endings were normalized to LF before parsing "
                f"({action.detail})."
            ),
            confidence="high",
            fix_action=FIX_NORMALIZE_LINE_ENDINGS,
            pre_applied=True,
        ))
    if "transcode_to_utf8" in summary:
        action = next(a for a in repair.actions if a.kind == "transcode_to_utf8")
        findings.append(Finding(
            id="csv_transcoded_to_utf8",
            severity="info",
            tool="",
            count=1,
            description=(
                f"File was transcoded from a wide encoding to UTF-8 before "
                f"parsing ({action.detail})."
            ),
            confidence="high",
            fix_action=FIX_NONE,
            pre_applied=True,
        ))
    if "decode_replaced" in summary:
        action = next(a for a in repair.actions if a.kind == "decode_replaced")
        findings.append(Finding(
            id="encoding_decode_failed",
            severity="error",
            tool="",
            count=1,
            description=(
                f"Some bytes could not be decoded under the detected "
                f"encoding ({action.detail}). Replacement characters "
                f"(U+FFFD) were inserted; the file likely uses a different "
                f"encoding or mixes encodings. Re-upload with --encoding."
            ),
            confidence="low",
            fix_action=FIX_NONE,
        ))
    if "quote_unquoted_delim" in summary:
        n = summary["quote_unquoted_delim"]
        findings.append(Finding(
            id="csv_unquoted_delimiters_repaired",
            severity="warn",
            tool="",
            count=n,
            description=(
                f"{n} row(s) had a delimiter inside an unquoted field "
                f"(e.g. '$1,500.00') and were merged during pre-parse repair."
            ),
            confidence="medium",
            fix_action=FIX_REPAIR_UNQUOTED_DELIM,
            pre_applied=True,
        ))
    if repair.unrepairable_lines:
        n = len(repair.unrepairable_lines)
        findings.append(Finding(
            id="csv_unrepairable_rows",
            severity="error",
            tool="",
            count=n,
            description=(
                f"{n} row(s) had ambiguous structural problems and were "
                f"left as-is. Inspect lines: "
                f"{repair.unrepairable_lines[:10]}"
            ),
            confidence="low",
            fix_action=FIX_NONE,
        ))
    return findings


# ---------------------------------------------------------------------------
# Public entry point
# ---------------------------------------------------------------------------

def analyze(
    source: pd.DataFrame | str | Path,
    *,
    sample_rows: int = 1000,
    repair_result: Optional[RepairResult] = None,
    encoding_override: Optional[str] = None,
) -> list[Finding]:
    """Run all detectors against *source* and return a list of findings.

    Parameters
    ----------
    source
        Either a DataFrame already in memory or a path to a CSV/Excel file.
        Paths are read with the same encoding/delimiter detection as
        :func:`read_file`. Only the first *sample_rows* are scanned.
    sample_rows
        Cap on how many rows to scan. Defaults to 1000 — enough to detect
        every per-cell pollution pattern without paying for a multi-GB read.
    repair_result
        Optional :class:`RepairResult` from a prior pre-parse pass; used
        to synthesize ``csv_*`` findings so the user sees what the parser
        quietly fixed.
    encoding_override
        When set, skip charset detection and decode with this encoding
        instead. Used by the Review page to let the user correct
        misdetections (cp1250-vs-cp1252 ambiguity, KOI8-R surfacing as
        Shift_JIS, etc.). Only applies when *source* is a path.
    """
    raw_for_byte_scan: Optional[bytes] = None
    if isinstance(source, (str, Path)):
        df, internal_repair, raw_for_byte_scan = _load_for_analysis(
            Path(source), sample_rows=sample_rows,
            encoding_override=encoding_override,
        )
        # Caller-supplied repair_result wins over the internally produced one,
        # since the caller may have used non-default repair flags.
        if repair_result is None:
            repair_result = internal_repair
    else:
        df = source.head(sample_rows).copy() if len(source) > sample_rows else source.copy()

    findings: list[Finding] = []
    if raw_for_byte_scan is not None and not raw_for_byte_scan.strip():
        findings.append(Finding(
            id="empty_input",
            severity="error",
            tool="",
            count=0,
            description="Input file is empty (zero bytes or whitespace only).",
            confidence="low",
            fix_action=FIX_NONE,
        ))
        return findings
    if df.empty and df.columns.empty and raw_for_byte_scan is not None:
        # Non-empty bytes but the parser couldn't extract a header row.
        findings.append(Finding(
            id="empty_input",
            severity="error",
            tool="",
            count=0,
            description=(
                "Input file has no parseable rows or columns "
                "(only line endings, BOM, or whitespace)."
            ),
            confidence="low",
            fix_action=FIX_NONE,
        ))
    if repair_result is not None:
        findings.extend(_findings_from_repair(repair_result))
    if raw_for_byte_scan is not None:
        findings.extend(_detect_mixed_line_endings(raw_for_byte_scan))
        findings.extend(_detect_lying_bom(raw_for_byte_scan))
    findings.extend(_detect_encoding_uncertainty(df))
    findings.extend(_detect_smart_punctuation(df))
    findings.extend(_detect_invisible_chars(df))
    findings.extend(_detect_whitespace_padding(df))
    findings.extend(_detect_null_like_sentinels(df))
    findings.extend(_detect_mojibake(df))
    findings.extend(_detect_mixed_case_email(df))
    findings.extend(_detect_inconsistent_date_format(df))
    findings.extend(_detect_inconsistent_phone_format(df))
    findings.extend(_detect_inconsistent_currency_format(df))
    findings.extend(_detect_inconsistent_boolean_format(df))
    findings.extend(_detect_leading_zero_ids(df))
    findings.extend(_detect_near_duplicates(df))
    return findings


def _load_for_analysis(
    path: Path, *, sample_rows: int, encoding_override: Optional[str] = None,
    fold_quotes: bool = True,
) -> tuple[pd.DataFrame, Optional[RepairResult], Optional[bytes]]:
    """Read just enough of *path* to scan, with the same robust pre-parse
    repair the tool pages will use.

    Returns ``(df, repair_result, raw_bytes)``. The repair result and raw
    bytes are *None* for Excel files since the byte-level repair step
    (BOM/NUL/smart-quote folding) and line-ending scan are CSV-specific.
    An empty CSV returns an empty DataFrame plus the (empty) raw bytes;
    the caller synthesizes an ``empty_input`` finding from that.

    When *encoding_override* is set, it replaces the detected encoding
    entirely — the user has explicitly told us what the file is. The
    delimiter is still detected (it's separate from encoding choice).

    *fold_quotes* defaults to True so the byte-level smart-quote fold
    runs as part of the repair pass (correct for CSV parsing). Pass
    False when the caller needs a content-preserving decode for
    identity round-trip checks (encoding corpus tests, format-fidelity
    audits).
    """
    suffix = path.suffix.lower()
    if suffix in (".xlsx", ".xls"):
        df = pd.read_excel(
            path, dtype=str, keep_default_na=False, engine="openpyxl",
            nrows=sample_rows,
        )
        return df, None, None

    # Sample-mode budget: read just enough head bytes to satisfy
    # ``sample_rows`` plus a generous margin. On a 1 GB file with 180-byte
    # rows, sampling 1000 rows is ~180 KB — three orders of magnitude
    # smaller than slurping the whole file. Pathological wide rows could
    # exceed the budget; the loop below grows it once if pandas reports
    # we got fewer than ``sample_rows`` rows.
    file_size = path.stat().st_size
    if file_size == 0:
        return pd.DataFrame(), None, b""

    enc = encoding_override or detect_encoding(path)
    delim = detect_delimiter(path, enc)

    # 4 KB minimum (covers headers + a few rows of even very-wide data).
    # 256 bytes per row is a generous estimate; doubled budget gives slack.
    head_budget = max(4 * 1024, sample_rows * 256 * 2)
    head_budget = min(head_budget, file_size)
    head_was_full = (head_budget >= file_size)

    with path.open("rb") as fh:
        head = fh.read(head_budget)

    if not head.strip():
        return pd.DataFrame(), None, head

    repair = repair_bytes(head, encoding=enc, delimiter=delim, fold_quotes=fold_quotes)
    import io as _io
    try:
        df = pd.read_csv(
            _io.BytesIO(repair.repaired_bytes),
            encoding="utf-8", delimiter=delim,
            dtype=str, keep_default_na=False, on_bad_lines="warn",
            nrows=sample_rows,
        )
    except pd.errors.EmptyDataError:
        return pd.DataFrame(), repair, head

    # If the head budget was too tight, pandas may have returned fewer rows
    # than asked. Retry once with the full file. In practice this almost
    # never trips; the 2× row-size multiplier above handles 99% of inputs.
    if not head_was_full and len(df) < sample_rows:
        full_raw = path.read_bytes()
        full_repair = repair_bytes(
            full_raw, encoding=enc, delimiter=delim, fold_quotes=fold_quotes,
        )
        try:
            df = pd.read_csv(
                _io.BytesIO(full_repair.repaired_bytes),
                encoding="utf-8", delimiter=delim,
                dtype=str, keep_default_na=False, on_bad_lines="warn",
                nrows=sample_rows,
            )
        except pd.errors.EmptyDataError:
            return pd.DataFrame(), full_repair, full_raw
        return df, full_repair, full_raw

    return df, repair, head


def to_dict(finding: Finding) -> dict[str, Any]:
    """JSON-friendly representation; used by the CLI ``--json`` output."""
    return {
        "id": finding.id,
        "severity": finding.severity,
        "confidence": finding.confidence,
        "fix_action": finding.fix_action,
        "pre_applied": finding.pre_applied,
        "tool": finding.tool,
        "count": finding.count,
        "description": finding.description,
        "column": finding.column,
        "samples": [
            {"row": r, "column": c, "value": v}
            for r, c, v in finding.samples
        ],
    }


def findings_by_tool(findings: list[Finding]) -> dict[str, list[Finding]]:
    """Group findings by tool id; useful for the GUI sidebar badges."""
    out: dict[str, list[Finding]] = {}
    for f in findings:
        if not f.tool:
            continue
        out.setdefault(f.tool, []).append(f)
    return out
