content_pipeline.py

"""
Recall — Module A: Content Pipeline.  OWNER: Frank

document -> clean text -> chunks -> list[Card]

Runs in STUB mode out of the box (returns a hardcoded demo deck). Replace the
TODO bodies with real logic; the public signatures must NOT change — app.py and
learning_engine depend on them.
"""
from __future__ import annotations

import os
import re

import llm
from schema import Card, new_card, validate_card

# STUB is owned by llm (single source of truth) and read dynamically as
# `llm.STUB` so every module agrees and runtime/reload changes are honored.


# ---- Public interface ------------------------------------------------------

class ExtractionError(Exception):
    """Raised for bad input (no file, corrupt/empty/image-only PDF) with a
    user-facing message. app.py catches this and shows it — never a crash."""


def extract_text(file) -> str:
    """
    file: a path (str) or a Gradio file object with `.name`. Handles PDF + .txt.
    Returns plain text. Paste-text path in app.py bypasses this.

    Raises ExtractionError (with a friendly message) for bad inputs: no file,
    a missing path, a corrupt/password-protected PDF, an image-only/scanned PDF
    with no selectable text, or an empty file.
    """
    if llm.STUB:
        return ("Photosynthesis is the process by which plants convert light "
                "energy into chemical energy. It occurs in the chloroplasts. "
                "The Calvin cycle fixes carbon dioxide into glucose.")

    if file is None:
        raise ExtractionError("No file provided — upload a PDF/.txt or paste notes.")

    path = getattr(file, "name", file)
    if not os.path.isfile(path):
        raise ExtractionError("Couldn't find that file. Try uploading it again.")

    if str(path).lower().endswith(".pdf"):
        from pypdf import PdfReader
        try:
            reader = PdfReader(path)
            text = "\n\n".join((page.extract_text() or "") for page in reader.pages)
        except Exception:
            raise ExtractionError(
                "Couldn't read that PDF — it may be corrupted or password-protected."
            )
        if not _normalize(text):
            raise ExtractionError(
                "This PDF has no selectable text (looks scanned/image-only). "
                "Try a text-based PDF, or paste the notes instead."
            )
    else:
        with open(path, "r", encoding="utf-8", errors="ignore") as f:
            text = f.read()
        if not _normalize(text):
            raise ExtractionError("That file is empty — nothing to study from.")

    return _normalize(text)


def generate_deck(text: str, n: int = 12) -> list[Card]:
    """
    Turn source text into ~n good question cards. Quality over quantity.
    Always returns valid Cards (bad model output is skipped, never crashes).
    """
    if llm.STUB:
        return [
            new_card(
                "What does photosynthesis convert light energy into?",
                "Chemical energy (stored as glucose).",
                topic="Photosynthesis",
                source_chunk=text[:120],
                difficulty=1,
            ),
            new_card(
                "Where in the plant cell does photosynthesis occur?",
                "In the chloroplasts.",
                topic="Cell Biology",
                source_chunk=text[:120],
                difficulty=1,
            ),
            new_card(
                "What does the Calvin cycle fix carbon dioxide into?",
                "Glucose.",
                topic="Photosynthesis",
                source_chunk=text[:120],
                difficulty=2,
            ),
        ]

    cards: list[Card] = []
    for chunk in chunk_text(text):
        cards.extend(_cards_from_chunk(chunk))
        if len(cards) >= n:
            break
    return _dedupe(cards)[:n]


# ---- Internals (Frank implements) ------------------------------------------

def _normalize(text: str) -> str:
    return " ".join(text.split())


def chunk_text(text: str, size: int = 2500) -> list[str]:
    """
    Public chunker (used by generate_deck and by app.py's debug panel).

    Character-based splitter targeting ~500–800 tokens per chunk (1 token ≈ 4 chars,
    so size=2500 ≈ 625 tokens). Breaks at sentence boundaries where possible.
    """
    # Split into sentences on . ! ? followed by whitespace or end-of-string.
    sentences = re.split(r'(?<=[.!?])\s+', text.strip())
    chunks: list[str] = []
    current: list[str] = []
    current_len = 0

    for sentence in sentences:
        slen = len(sentence)
        if current and current_len + 1 + slen > size:
            chunks.append(" ".join(current))
            current = [sentence]
            current_len = slen
        else:
            current.append(sentence)
            current_len += (1 + slen) if current_len else slen

    if current:
        chunks.append(" ".join(current))

    return chunks or [text]


def _cards_from_chunk(chunk: str) -> list[Card]:
    """Ask the model for JSON cards, parse defensively, validate each."""
    messages = [
        {"role": "system", "content": (
            "You are a quiz generator. Given a study passage, produce 3 to 5 quiz questions "
            "that test different aspects of the material at varying difficulty levels.\n\n"
            "GROUNDING RULES (critical):\n"
            "- Every question AND its answer must be answerable using ONLY the passage below.\n"
            "- Do NOT introduce facts, names, numbers, or examples that are not in the passage.\n"
            "- If the passage is too short or thin for a question, write fewer questions "
            "rather than inventing content.\n"
            "- Quote or paraphrase the passage's own wording in the reference answer.\n\n"
            "Return ONLY a JSON array. Each element must have exactly these keys:\n"
            "  question   — the question text (clear, specific, one sentence)\n"
            "  answer     — a concise reference answer (1-3 sentences), grounded in the passage\n"
            "  topic      — a short tag naming the concept (e.g. 'Cell Biology')\n"
            "  difficulty — integer: 1 (direct recall), 2 (application/explanation), "
            "3 (synthesis/comparison)\n\n"
            "Mix all three difficulty levels across the questions. "
            "No prose, no markdown fences, no explanation outside the JSON array."
        )},
        {"role": "user", "content": f"Generate quiz questions from this passage:\n\n{chunk}"},
    ]
    # Strict-JSON with one repair pass: chat_json feeds a malformed reply back to
    # the model demanding clean JSON before giving up. A chunk that still won't
    # parse is skipped (returns []), never crashing the deck.
    data = llm.chat_json(messages, max_tokens=600, retries=1)
    if not isinstance(data, list):
        return []
    out: list[Card] = []
    for item in data:
        if not isinstance(item, dict):
            continue
        try:
            difficulty = max(1, min(3, int(item.get("difficulty") or 1)))
        except (ValueError, TypeError):
            difficulty = 1
        card = new_card(
            question=str(item.get("question", "")).strip(),
            answer=str(item.get("answer", "")).strip(),
            topic=str(item.get("topic", "General")).strip() or "General",
            source_chunk=chunk,
            difficulty=difficulty,
        )
        if validate_card(card):
            out.append(card)
    return out


MIN_QUESTION_CHARS = 10   # shorter than this isn't a real question
MIN_ANSWER_CHARS = 2      # an answer needs at least a token of substance


def _norm_key(text: str) -> str:
    """Normalize for near-identical matching: lowercase, strip punctuation,
    collapse whitespace. 'What is X?' and 'what is x' collapse to one key."""
    return re.sub(r"[^a-z0-9 ]", "", text.lower()).strip()


def _dedupe(cards: list[Card]) -> list[Card]:
    """Drop near-identical questions and low-quality cards.

    Near-identical = same question once punctuation/case/whitespace is
    normalized away. Low-quality = a question/answer too short to be useful.
    ~12 great cards beats 40 mediocre ones.
    """
    seen: set[str] = set()
    out: list[Card] = []
    for c in cards:
        question = c["question"].strip()
        answer = c["answer"].strip()
        if len(question) < MIN_QUESTION_CHARS or len(answer) < MIN_ANSWER_CHARS:
            continue  # drop low-quality
        key = _norm_key(question)
        if not key or key in seen:
            continue  # drop empty/near-duplicate
        seen.add(key)
        out.append(c)
    return out


# STRETCH(Frank): difficulty dial — backs Arturo's UI toggle (NAH-32).
def regenerate(card: Card, direction: str) -> Card:
    """Rewrite a card harder or easier on the SAME concept, grounded in the same
    source_chunk. direction: 'harder' | 'easier'. Always returns a valid Card —
    on any failure it falls back to the original so the UI never breaks.
    """
    step = 1 if direction == "harder" else -1
    new_difficulty = max(1, min(3, card["difficulty"] + step))

    if llm.STUB:
        prefix = "[harder] " if direction == "harder" else "[easier] "
        return new_card(
            prefix + card["question"],
            card["answer"],
            topic=card["topic"],
            source_chunk=card["source_chunk"],
            difficulty=new_difficulty,
            parent_id=card.get("parent_id"),
        )

    level = ("more challenging (synthesis/comparison/application)" if direction == "harder"
             else "simpler (direct recall)")
    messages = [
        {"role": "system", "content": (
            "You rewrite a single quiz question to a different difficulty while "
            "keeping the SAME underlying concept. Stay grounded in the passage — "
            "do not introduce facts that are not in it. "
            "Return ONLY a JSON object with keys: question, answer, topic."
        )},
        {"role": "user", "content": (
            f"Passage:\n{card['source_chunk']}\n\n"
            f"Current question: {card['question']}\n"
            f"Current answer: {card['answer']}\n"
            f"Topic: {card['topic']}\n\n"
            f"Rewrite it to be {level}, same concept."
        )},
    ]
    data = llm.chat_json(messages, max_tokens=300, retries=1)
    if isinstance(data, dict):
        out = new_card(
            str(data.get("question", "")).strip(),
            str(data.get("answer", "")).strip(),
            topic=str(data.get("topic", card["topic"])).strip() or card["topic"],
            source_chunk=card["source_chunk"],
            difficulty=new_difficulty,
            parent_id=card.get("parent_id"),
        )
        if validate_card(out):
            return out
    return card  # safe fallback — never break the study loop