retrieval/parser.py

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


RETRIEVAL_SCHEMA_VERSION = "question-answer-v1"


REQUIRED_COLUMNS = {
    "id",
    "question",
    "answer",
    "source_id",
    "category_id",
    "difficulty",
    "flash",
    "is_deleted",
}


@dataclass(frozen=True)
class QuestionRecord:
    question_id: int
    question: str
    answer: str = ""
    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 = ""
    is_deleted: bool = False

    @property
    def retrieval_text(self) -> str:
        """Text used by dense retrieval, BM25, and reranking."""
        return f"Question: {self.question}\nAnswer: {self.answer}"

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


@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 _optional_positive_int(
    value: object,
    field: str,
    row_number: int,
) -> Optional[int]:
    text = _text(value)
    if not text:
        return None
    return _required_positive_int(text, field, row_number)


def _boolean(value: object, field: str, row_number: int) -> bool:
    text = _text(value).lower()
    if text in {"1", "true", "t", "yes", "y"}:
        return True
    if text in {"0", "false", "f", "no", "n", ""}:
        return False
    raise QuizCsvValidationError(
        f"Row {row_number}: {field} is not a boolean."
    )


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
    deleted_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."
                    )
                answer = _text(raw.get("answer"))
                if not answer:
                    raise QuizCsvValidationError(
                        f"Row {row_number}: answer is blank."
                    )
                if question_id in seen_ids:
                    raise QuizCsvValidationError(
                        f"Row {row_number}: duplicate id {question_id}."
                    )

                is_deleted = _boolean(
                    raw.get("is_deleted"),
                    "is_deleted",
                    row_number,
                )
                seen_ids.add(question_id)
                if is_deleted:
                    deleted_count += 1
                    continue

                difficulty = _optional_positive_int(
                    raw.get("difficulty"),
                    "difficulty",
                    row_number,
                )
                if difficulty is not None and difficulty not in {1, 2, 3}:
                    raise QuizCsvValidationError(
                        f"Row {row_number}: difficulty must be 1, 2, or 3."
                    )

                records.append(
                    QuestionRecord(
                        question_id=question_id,
                        question=question,
                        answer=answer,
                        source_id=_optional_positive_int(
                            raw.get("source_id"),
                            "source_id",
                            row_number,
                        ),
                        category_id=_optional_positive_int(
                            raw.get("category_id"),
                            "category_id",
                            row_number,
                        ),
                        difficulty=difficulty,
                        flash=_boolean(
                            raw.get("flash"),
                            "flash",
                            row_number,
                        ),
                        is_deleted=False,
                    )
                )
            except QuizCsvValidationError:
                invalid_count += 1
                raise

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

    return ParsedQuestions(
        records=records,
        source_sha256=file_sha256(path),
        invalid_count=invalid_count,
        deleted_count=deleted_count,
    )