"""Preprocess phase orchestrator — the single source of truth for sub-stage order.

The phase is modeled as: load the pool, then fold a list of small frame→frame ``Stage``s over it. Exposing
the stages as data (``preprocess_stages``) means the orchestrator, the CLI, and the Streamlit explorer all
drive the SAME sequence — so each sub-stage can be run, previewed, and debugged independently. New stages
(enrich → materialize → expand) land by appending one line below.

The expensive part (parsing the 465MB JSONL) is cached inside ``load_candidates`` itself, not here — see its
docstring. validate/materialize/quality always run fresh on every call: they're cheap (~2-3s combined) and
``quality`` APPENDS to the ``text`` column, so caching+replaying them would double-append and corrupt it.
"""

from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass
from datetime import date
from pathlib import Path

import polars as pl

from common.checkpoint import checkpoint
from common.logging import get_logger
from config import Settings, load_settings

from .load import load_candidates
from .materialize import materialize
from .quality import quality
from .validate import validate

# Fixed reference "now" all recency/age features measure against. Recency features are relative deltas
# (``as_of - date``), so a fixed anchor shifts every candidate uniformly — it keeps the pass deterministic
# and avoids an O(n) scan for the data's max date. It sits at/after the dataset horizon so deltas stay ≥ 0.
DATASET_AS_OF: date = date(2026, 6, 6)


@dataclass(frozen=True)
class Stage:
    """One preprocess sub-stage: a label + a frame→frame transform (each logs its own ``step()``)."""

    name: str
    run: Callable[[pl.DataFrame], pl.DataFrame]


def preprocess_stages(as_of: date) -> list[Stage]:
    """The ordered frame→frame sub-stages applied AFTER the initial load. Order is the contract."""
    return [
        Stage("validate", lambda frame: validate(frame, as_of)),
        Stage("materialize", lambda frame: materialize(frame, as_of)),
        Stage("quality", quality),
        # enrich + expand_skills intentionally omitted (raw flat_skills; experience_band computed inline).
    ]


def run_preprocess(
    candidates: str | Path, limit: int | None = None, settings: Settings | None = None
) -> tuple[pl.DataFrame, date]:
    """Run the full preprocess phase: load → validate → … → returns (frame, reference date).

    Validation runs on raw input (before enrichment) so flags reflect what the candidate submitted.
    Each phase's output is written to a Parquet checkpoint when ``settings.debug.checkpoints`` is on.
    ``load_candidates`` transparently serves from its own Parquet cache for the real, unlimited run — see its
    docstring; this orchestrator doesn't need to know that.
    """
    active = settings or load_settings()
    as_of = DATASET_AS_OF
    frame = load_candidates(candidates, limit=limit)
    checkpoint(0, "load", frame, active)
    for index, stage in enumerate(preprocess_stages(as_of), start=1):
        frame = stage.run(frame)
        checkpoint(index, stage.name, frame, active)
    get_logger("preprocess").info(
        "preprocess.done", rows=frame.height, columns=frame.width, as_of=str(as_of)
    )
    return frame, as_of