scripts/export_quiz_questions.py

#!/usr/bin/env python3
"""Export question-answer retrieval rows from Supabase.

The database stores ``answer`` as the correct option number. This exporter
resolves that number to the corresponding option text and writes a compact RAG
CSV where ``answer`` contains the actual answer. The API embeds and searches a
combined ``Question: ...\nAnswer: ...`` document while retaining the remaining
columns as filter metadata. The Edge Function still performs final live
validation against the database.
"""

from __future__ import annotations

import csv
import os
import sys
import tempfile
from pathlib import Path
from typing import Dict, Iterable, List

import requests

SOURCE_FIELDS = (
    "id",
    "question",
    "opt_1",
    "opt_2",
    "opt_3",
    "opt_4",
    "answer",
    "source_id",
    "category_id",
    "difficulty",
    "flash",
    "is_deleted",
)
OUTPUT_FIELDS = (
    "id",
    "question",
    "answer",
    "source_id",
    "category_id",
    "difficulty",
    "flash",
    "is_deleted",
)
PAGE_SIZE = 1000


def require_env(name: str) -> str:
    value = os.getenv(name, "").strip()
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value


def fetch_rows(base_url: str, service_key: str) -> Iterable[Dict[str, object]]:
    endpoint = f"{base_url.rstrip('/')}/rest/v1/quiz_questions"
    offset = 0
    headers = {
        "apikey": service_key,
        "Authorization": f"Bearer {service_key}",
        "Accept": "application/json",
        "Range-Unit": "items",
    }

    while True:
        response = requests.get(
            endpoint,
            headers={
                **headers,
                "Range": f"{offset}-{offset + PAGE_SIZE - 1}",
            },
            params={
                "select": ",".join(SOURCE_FIELDS),
                "is_deleted": "eq.false",
                "order": "id.asc",
            },
            timeout=60,
        )
        response.raise_for_status()
        rows = response.json()
        if not isinstance(rows, list):
            raise RuntimeError("Supabase returned a non-list response.")
        for row in rows:
            if isinstance(row, dict):
                yield row
        if len(rows) < PAGE_SIZE:
            break
        offset += PAGE_SIZE


def _positive_int(value: object, field: str, row_number: int) -> int:
    try:
        parsed = int(str(value).strip())
    except (TypeError, ValueError) as exc:
        raise RuntimeError(
            f"Row {row_number}: invalid {field} {value!r}."
        ) from exc
    if parsed <= 0:
        raise RuntimeError(f"Row {row_number}: {field} must be positive.")
    return parsed


def prepare_rows(rows: Iterable[Dict[str, object]]) -> List[Dict[str, object]]:
    prepared: List[Dict[str, object]] = []
    seen = set()

    for row_number, row in enumerate(rows, start=2):
        question_id = _positive_int(row.get("id"), "id", row_number)
        if question_id in seen:
            raise RuntimeError(f"Row {row_number}: duplicate id {question_id}.")

        question = str(row.get("question") or "").strip()
        if not question:
            raise RuntimeError(f"Row {row_number}: blank question.")

        answer_index = _positive_int(row.get("answer"), "answer", row_number)
        if answer_index not in {1, 2, 3, 4}:
            raise RuntimeError(
                f"Row {row_number}: answer must be an option number from 1 to 4."
            )
        answer = str(row.get(f"opt_{answer_index}") or "").strip()
        if not answer:
            raise RuntimeError(
                f"Row {row_number}: correct option opt_{answer_index} is blank."
            )
        if row.get("is_deleted") is True:
            raise RuntimeError(
                f"Row {row_number}: deleted question {question_id} was exported."
            )

        prepared.append(
            {
                "id": question_id,
                "question": question,
                "answer": answer,
                "source_id": row.get("source_id"),
                "category_id": row.get("category_id"),
                "difficulty": row.get("difficulty"),
                "flash": row.get("flash"),
                "is_deleted": False,
            }
        )
        seen.add(question_id)

    if not prepared:
        raise RuntimeError("The export contains no active quiz questions.")
    return prepared


def write_atomic(rows: List[Dict[str, object]], output: Path) -> None:
    output.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile(
        "w",
        encoding="utf-8",
        newline="",
        dir=output.parent,
        prefix=f".{output.name}.",
        suffix=".tmp",
        delete=False,
    ) as handle:
        temp_path = Path(handle.name)
        writer = csv.DictWriter(
            handle,
            fieldnames=OUTPUT_FIELDS,
            extrasaction="ignore",
        )
        writer.writeheader()
        writer.writerows(rows)
    os.replace(temp_path, output)


def main() -> int:
    try:
        base_url = require_env("QUIZ_EXPORT_SUPABASE_URL")
        service_key = require_env("QUIZ_EXPORT_SERVICE_ROLE_KEY")
        output = Path(
            os.getenv(
                "QUIZ_EXPORT_OUTPUT",
                "sources/custom-topic-quiz-generator/quiz_questions.csv",
            )
        ).resolve()
        rows = prepare_rows(fetch_rows(base_url, service_key))
        write_atomic(rows, output)
        print(f"Exported {len(rows)} active question-answer rows to {output}")
        return 0
    except Exception as error:
        print(f"Export failed: {error}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())