"""CSV-normalization gate.

A file enters the tool pages only after passing the gate. The gate has
two paths:

1. **Auto-fix** — apply every algorithm flagged ``confidence="high"``.
2. **Review** — show the user a preview of medium/low-confidence findings
   and accept an explicit per-finding decision before applying.

The gate produces a :class:`NormalizationResult` containing the cleaned
DataFrame, the bytes representation, and a structured audit log of every
fix that ran. Tool pages are guarded by :func:`is_normalized` against
the result and the original list of findings.
"""

from __future__ import annotations

import io
from dataclasses import dataclass, field
from pathlib import Path
from typing import Literal, Optional

import pandas as pd

from .analyze import Finding, analyze
from .fixes import get_fix


DecisionAction = Literal["auto", "skip", "modified"]


@dataclass
class Decision:
    """One user-recorded choice for a finding.

    Attributes
    ----------
    finding_id
        The :class:`Finding` id this decision applies to.
    action
        ``"auto"`` to run the registered fix as-is, ``"skip"`` to leave
        it alone (the gate logs it as waived), ``"modified"`` to run the
        fix with a custom payload (e.g. user-edited null sentinel list).
    payload
        Optional kwargs forwarded to the fix function. Required for
        ``"modified"``; ignored for ``"skip"``.
    """

    finding_id: str
    action: DecisionAction
    payload: Optional[dict] = None


@dataclass
class FixApplied:
    """One fix that ran during a gate pass."""

    finding_id: str
    fix_action: str
    cells_changed: int
    decision: DecisionAction


@dataclass
class NormalizationResult:
    """Output of a gate pass.

    Attributes
    ----------
    cleaned_df
        DataFrame after every applied fix. The downstream tool pages
        consume this directly.
    cleaned_bytes
        UTF-8 encoded CSV of *cleaned_df* — the canonical artifact for
        round-tripping into another tool that re-parses.
    applied
        Audit log of fixes that ran.
    skipped_findings
        Findings the user explicitly waived (decision = ``"skip"``).
    pending_findings
        Findings still requiring a user decision before the gate is
        considered passed. Empty on a successful gate pass.
    blocking_findings
        Severity=error findings that have no decision and no auto-fix.
        Non-empty means the gate is blocked and the file cannot enter
        tool pages.
    """

    cleaned_df: pd.DataFrame
    cleaned_bytes: bytes
    applied: list[FixApplied] = field(default_factory=list)
    skipped_findings: list[Finding] = field(default_factory=list)
    pending_findings: list[Finding] = field(default_factory=list)
    blocking_findings: list[Finding] = field(default_factory=list)

    @property
    def passed(self) -> bool:
        return not self.pending_findings and not self.blocking_findings


def _df_to_bytes(df: pd.DataFrame) -> bytes:
    buf = io.StringIO()
    df.to_csv(buf, index=False, lineterminator="\n")
    return buf.getvalue().encode("utf-8")


def _is_actionable(f: Finding) -> bool:
    """Does this finding still need attention from the gate?

    Pre-applied fixes (BOM strip, etc. — already done during read) are
    not actionable. Findings without a registered fix_action are not
    actionable here either; severity=error ones become blockers.
    """
    if f.pre_applied:
        return False
    if not f.fix_action:
        return False
    return get_fix(f.fix_action) is not None


def auto_fix(
    df: pd.DataFrame, findings: list[Finding],
) -> NormalizationResult:
    """Apply every fix flagged ``confidence="high"``.

    Returns a :class:`NormalizationResult`. Medium / low / unknown
    confidence findings are surfaced as ``pending_findings`` and the
    result is *not* considered passed until the user decides on them.
    """
    decisions: list[Decision] = [
        Decision(finding_id=f.id, action="auto")
        for f in findings
        if _is_actionable(f) and f.confidence == "high"
    ]
    return apply_decisions(df, findings, decisions)


def apply_decisions(
    df: pd.DataFrame, findings: list[Finding], decisions: list[Decision],
) -> NormalizationResult:
    """Apply *decisions* to *df* in finding order.

    Findings with no matching decision are categorized:

    * ``severity=error`` -> ``blocking_findings``
    * Otherwise -> ``pending_findings`` (user still owes us a decision)

    Pre-applied findings are recorded once in the audit log with
    ``cells_changed=0`` so callers can render "what was already done."
    """
    decision_by_id = {d.finding_id: d for d in decisions}

    # One working copy is taken up front; each fix mutates it in place.
    # Without this, N fixes against a 6 GB DataFrame allocate 6 GB N
    # times — peak RSS in the 1.25 GB stress test would otherwise climb
    # past 30 GB. Callers that need the original frame already passed it
    # by reference; we copy here exactly once.
    out = df.copy()
    applied: list[FixApplied] = []
    skipped: list[Finding] = []
    pending: list[Finding] = []
    blocking: list[Finding] = []

    for f in findings:
        if f.pre_applied:
            applied.append(FixApplied(
                finding_id=f.id,
                fix_action=f.fix_action,
                cells_changed=0,
                decision="auto",
            ))
            continue

        decision = decision_by_id.get(f.id)
        if decision is None:
            if f.severity == "error":
                blocking.append(f)
            elif _is_actionable(f):
                pending.append(f)
            # else: informational with no fix; ignore.
            continue

        if decision.action == "skip":
            skipped.append(f)
            continue

        fix_fn = get_fix(f.fix_action)
        if fix_fn is None:
            # Decision references a fix we don't have; treat as pending.
            pending.append(f)
            continue

        payload = dict(decision.payload or {})
        # Per-column fixes (lowercase_email) can carry the column from
        # the finding when the user didn't override it.
        if f.column and "column" not in payload:
            payload["column"] = f.column
        # Tell fixes that opt in (the vectorized helpers in fixes.py)
        # to mutate ``out`` in place so we don't allocate per-fix.
        payload.setdefault("inplace", True)

        # Fixes that don't accept the inplace kwarg simply drop it.
        try:
            out, changed = fix_fn(out, payload)
        except TypeError:
            payload.pop("inplace", None)
            out, changed = fix_fn(out, payload)
        applied.append(FixApplied(
            finding_id=f.id,
            fix_action=f.fix_action,
            cells_changed=changed,
            decision=decision.action,
        ))

    return NormalizationResult(
        cleaned_df=out,
        cleaned_bytes=_df_to_bytes(out),
        applied=applied,
        skipped_findings=skipped,
        pending_findings=pending,
        blocking_findings=blocking,
    )


def is_normalized(
    findings: list[Finding], result: Optional[NormalizationResult],
) -> bool:
    """True iff *result* satisfies the gate against *findings*.

    The gate passes when:

    * A result exists, and
    * It has no blocking findings, and
    * It has no pending (undecided) actionable findings.

    Re-run analysis on the cleaned bytes to confirm the high-confidence
    detectors no longer fire — that's the contract the tool pages rely
    on. Callers who want the cheap check can pass ``result.passed``
    directly; this function is the strict version.
    """
    if result is None:
        return False
    if not result.passed:
        return False
    # Re-analyze the cleaned bytes; high-confidence detectors must be silent.
    rerun = analyze(result.cleaned_df)
    for f in rerun:
        if f.confidence == "high" and _is_actionable(f):
            return False
    return True


def gate_summary(result: NormalizationResult) -> dict:
    """One-line-per-key summary suitable for logging or the CLI."""
    return {
        "passed": result.passed,
        "fixes_applied": len(result.applied),
        "cells_changed": sum(a.cells_changed for a in result.applied),
        "skipped": [f.id for f in result.skipped_findings],
        "pending": [f.id for f in result.pending_findings],
        "blocking": [f.id for f in result.blocking_findings],
    }
