prompt_card/preprocess.py

"""Preprocess: per-message English filter, min-20 gate, most-recent-500 cap, focus helper.

The English filter is deterministic (langdetect seeded). Very short / low-signal strings that
langdetect cannot classify fall back to an ASCII-letter heuristic (assume English) so that terse
but valid English prompts ("fix this", "why?") are not dropped.
"""
from __future__ import annotations

import re
from dataclasses import dataclass, field

from langdetect import detect, DetectorFactory, LangDetectException

from .schema import Conversation, Turn, ROLE_USER

DetectorFactory.seed = 0

MIN_USER_MSGS = 20
CAP = 500

_LATIN = re.compile(r"[A-Za-z]")
_NON_ASCII = re.compile(r"[^\x00-\x7f]")


class InsufficientData(Exception):
    """Raised when there are fewer than MIN_USER_MSGS English user messages to score."""


@dataclass
class Prepared:
    conversations: list[Conversation]  # English-only turns
    user_prompts: list[str] = field(default_factory=list)  # flat, most-recent CAP


def is_english(text: str) -> bool:
    t = (text or "").strip()
    if not t:
        return False
    # Heavy non-ASCII content (CJK, etc.) is clearly not English.
    if _NON_ASCII.search(t) and len(_NON_ASCII.findall(t)) >= max(3, len(t) * 0.2):
        return False
    try:
        return detect(t) == "en"
    except LangDetectException:
        # Too short for langdetect; keep if it contains Latin letters.
        return bool(_LATIN.search(t))


def _filter_english(conv: Conversation) -> Conversation:
    turns = [t for t in conv.turns if is_english(t.text)]
    return Conversation(provider=conv.provider, created_at=conv.created_at, turns=turns)


def multi_message_sessions(conversations: list[Conversation]) -> list[list[str]]:
    """User-prompt lists per conversation, keeping only sessions with >=2 user prompts
    (focus needs at least two messages to measure coherence)."""
    sessions = []
    for c in conversations:
        prompts = [t.text for t in c.turns if t.role == ROLE_USER]
        if len(prompts) >= 2:
            sessions.append(prompts)
    return sessions


def preprocess(
    conversations: list[Conversation],
    *,
    cap: int = CAP,
    min_user_msgs: int = MIN_USER_MSGS,
) -> Prepared:
    filtered = [_filter_english(c) for c in conversations]
    filtered = [c for c in filtered if c.turns]

    user_prompts = [t.text for c in filtered for t in c.turns if t.role == ROLE_USER]
    if len(user_prompts) < min_user_msgs:
        raise InsufficientData(
            f"need >= {min_user_msgs} English user messages, found {len(user_prompts)}"
        )

    # most-recent cap (assumes chronological order within/across conversations)
    user_prompts = user_prompts[-cap:]
    return Prepared(conversations=filtered, user_prompts=user_prompts)