custom_quiz/parser.py

import csv
import hashlib
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, List, Optional


# The retrieval index intentionally contains only the authoritative database ID
# and exact question text. All mutable eligibility/filter metadata is checked by
# the Supabase Edge Function against the live quiz_questions row.
REQUIRED_COLUMNS = {"id", "question"}


@dataclass(frozen=True)
class QuestionRecord:
    question_id: int
    question: str
    # Retained for backwards-compatible index deserialization. New two-column
    # CSV builds leave these unset because Supabase owns live filtering.
    source_id: Optional[int] = None
    category_id: Optional[int] = None
    chapter_id: Optional[int] = None
    difficulty: Optional[int] = None
    flash: bool = False
    event_seerah: bool = False
    updated_at: str = ""

    def metadata(self) -> Dict[str, object]:
        return {"question_id": self.question_id}


@dataclass(frozen=True)
class ParsedQuestions:
    records: List[QuestionRecord]
    source_sha256: str
    invalid_count: int
    deleted_count: int


class QuizCsvValidationError(ValueError):
    pass


def _text(value: object) -> str:
    text = str(value or "").strip()
    return "" if text.lower() == "null" else text


def _required_positive_int(value: object, field: str, row_number: int) -> int:
    text = _text(value)
    if not text:
        raise QuizCsvValidationError(f"Row {row_number}: {field} is required.")
    try:
        numeric = float(text)
    except ValueError as exc:
        raise QuizCsvValidationError(
            f"Row {row_number}: {field} is not an integer."
        ) from exc
    if not numeric.is_integer():
        raise QuizCsvValidationError(f"Row {row_number}: {field} is not an integer.")
    parsed = int(numeric)
    if parsed <= 0:
        raise QuizCsvValidationError(f"Row {row_number}: {field} must be positive.")
    return parsed


def file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def parse_questions_csv(path: Path) -> ParsedQuestions:
    if not path.is_file() or path.stat().st_size <= 0:
        raise QuizCsvValidationError("Quiz CSV is missing or empty.")

    records: List[QuestionRecord] = []
    seen_ids = set()
    invalid_count = 0

    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        reader = csv.DictReader(handle)
        fieldnames = {str(name or "").strip() for name in (reader.fieldnames or [])}
        missing = sorted(REQUIRED_COLUMNS - fieldnames)
        if missing:
            raise QuizCsvValidationError(
                "Quiz CSV is missing required columns: " + ", ".join(missing)
            )

        for row_number, raw in enumerate(reader, start=2):
            try:
                question_id = _required_positive_int(raw.get("id"), "id", row_number)
                question = _text(raw.get("question"))
                if not question:
                    raise QuizCsvValidationError(f"Row {row_number}: question is blank.")
                if question_id in seen_ids:
                    raise QuizCsvValidationError(
                        f"Row {row_number}: duplicate id {question_id}."
                    )
                seen_ids.add(question_id)
                records.append(
                    QuestionRecord(question_id=question_id, question=question)
                )
            except QuizCsvValidationError:
                invalid_count += 1
                raise

    if not records:
        raise QuizCsvValidationError("Quiz CSV contains no questions.")

    return ParsedQuestions(
        records=records,
        source_sha256=file_sha256(path),
        invalid_count=invalid_count,
        # Deleted rows cannot be represented in a two-column CSV. The export
        # excludes them, and the Edge Function rechecks is_deleted live.
        deleted_count=0,
    )