src/preprocessing.py

"""
preprocessing.py
================
Text cleaning and prompt-building functions, extracted from the notebooks.
Every model needs a slightly different input format:

  - TF-IDF     : one flat string per question (3 strategies tried)
  - LSTM       : prompt + all 5 options concatenated, word-tokenized
  - DeBERTa    : question expanded into 5 (prompt, single option) pairs
  - RoBERTa    : 5 stacked (prompt, option) pairs per question (multiple-choice)
"""

from typing import List
import pandas as pd

from .config import OPTION_COLS, ANSWER_MAP


# ---------------------------------------------------------------------------
# Shared cleaning (identical in all 4 notebooks)
# ---------------------------------------------------------------------------
def clean_text(text: str) -> str:
    """Lowercase + strip whitespace. Applied to every text column at load time."""
    return str(text).lower().strip()


def clean_dataframe(df: pd.DataFrame, cols: List[str] = None) -> pd.DataFrame:
    """Apply clean_text() to prompt + option columns of a dataframe (returns a copy)."""
    cols = cols or (["prompt"] + OPTION_COLS)
    df = df.copy()
    for col in cols:
        if col in df.columns:
            df[col] = df[col].astype(str).str.lower().str.strip()
    return df


def stratified_split(train_df: pd.DataFrame, val_size: float = 0.2, seed: int = 42):
    """
    Manual per-class stratified split, identical across all 4 training notebooks.
    Keeps the per-answer-letter distribution the same in train and val.
    """
    import numpy as np

    np.random.seed(seed)
    train_idx, val_idx = [], []
    for ans in "ABCDE":
        idx = train_df[train_df["answer"] == ans].index.tolist()
        np.random.shuffle(idx)
        cut = int(len(idx) * (1.0 - val_size))
        train_idx += idx[:cut]
        val_idx += idx[cut:]

    tr = train_df.loc[train_idx].reset_index(drop=True)
    va = train_df.loc[val_idx].reset_index(drop=True)
    return tr, va


# ---------------------------------------------------------------------------
# TF-IDF text-builder strategies (02_baseline.ipynb)
# ---------------------------------------------------------------------------
def build_v1_simple(df: pd.DataFrame):
    """prompt + all options concatenated (naive baseline)"""
    return (df["prompt"] + " " + df["A"] + " " + df["B"] + " " +
            df["C"] + " " + df["D"] + " " + df["E"]).values


def build_v2_repeated(df: pd.DataFrame):
    """prompt repeated before EACH option: 'Q optA Q optB Q optC Q optD Q optE'"""
    q = df["prompt"]
    return (q + " " + df["A"] + " " + q + " " + df["B"] + " " +
            q + " " + df["C"] + " " + q + " " + df["D"] + " " + q + " " + df["E"]).values


def build_v3_labeled(df: pd.DataFrame):
    """explicit 'option a: ...' labels + trigrams — best-performing strategy"""
    return (df["prompt"] +
            " option a: " + df["A"] +
            " option b: " + df["B"] +
            " option c: " + df["C"] +
            " option d: " + df["D"] +
            " option e: " + df["E"]).values


TEXT_BUILDERS = {
    "v1_simple": build_v1_simple,
    "v2_repeated": build_v2_repeated,
    "v3_labeled": build_v3_labeled,
}


def build_tfidf_text(df: pd.DataFrame, strategy: str = "v3_labeled"):
    """Dispatch to the chosen TF-IDF text-builder strategy."""
    return TEXT_BUILDERS[strategy](df)


def build_tfidf_text_single(prompt: str, options: List[str], strategy: str = "v3_labeled") -> str:
    """Same as build_tfidf_text but for a single question (used at inference time)."""
    a, b, c, d, e = options
    prompt = clean_text(prompt)
    a, b, c, d, e = [clean_text(x) for x in (a, b, c, d, e)]

    if strategy == "v1_simple":
        return f"{prompt} {a} {b} {c} {d} {e}"
    if strategy == "v2_repeated":
        return (f"{prompt} {a} {prompt} {b} {prompt} {c} "
                f"{prompt} {d} {prompt} {e}")
    if strategy == "v3_labeled":
        return (f"{prompt} option a: {a} option b: {b} option c: {c} "
                f"option d: {d} option e: {e}")
    raise ValueError(f"Unknown strategy: {strategy}")


# ---------------------------------------------------------------------------
# LSTM: prompt + all options concatenated (03_lstm.ipynb)
# ---------------------------------------------------------------------------
def build_lstm_text(prompt: str, options: List[str]) -> str:
    """Combined text used by the LSTM tokenizer: prompt + A + B + C + D + E."""
    prompt = clean_text(prompt)
    a, b, c, d, e = [clean_text(x) for x in options]
    return f"{prompt} {a} {b} {c} {d} {e}"


# ---------------------------------------------------------------------------
# DeBERTa: expand each question into 5 (prompt, option) pairs (04_DeBERTa.ipynb)
# ---------------------------------------------------------------------------
def expand_pairs(df: pd.DataFrame, is_test: bool = False) -> pd.DataFrame:
    """
    Convert each question row into 5 rows: one per option.
    text  = 'prompt option_text'
    label = 1 if this option is correct else 0 (only when not is_test)
    """
    rows = []
    for _, r in df.iterrows():
        correct = r.get("answer", None)
        for opt in OPTION_COLS:
            rows.append({
                "id": r["id"],
                "option": opt,
                "text": f"{r['prompt']} {r[opt]}",
                "label": int(opt == correct) if not is_test else -1,
            })
    return pd.DataFrame(rows)


def build_deberta_pairs_single(prompt: str, options: List[str]) -> List[str]:
    """Same expansion, but for a single question at inference time.
    Returns a list of 5 strings, one per option, in A-E order."""
    prompt = clean_text(prompt)
    return [f"{prompt} {clean_text(opt)}" for opt in options]


# ---------------------------------------------------------------------------
# RoBERTa: 5 stacked (prompt, option) pairs for AutoModelForMultipleChoice
# (05_RoBERTa.ipynb) — tokenizer handles the pairing, this just prepares lists
# ---------------------------------------------------------------------------
def build_roberta_inputs_single(prompt: str, options: List[str]):
    """
    Returns (prompts_x5, options_x5) ready to feed straight into the
    tokenizer as tokenizer([prompt]*5, options, ...).
    """
    prompt = clean_text(prompt)
    options = [clean_text(opt) for opt in options]
    return [prompt] * 5, options