scripts/export_true_false_questions.py

#!/usr/bin/env python3
"""Export true/false retrieval rows from Supabase.

Writes a compact RAG CSV for the custom-topic-quiz-generator backend. The API
embeds and searches ``Question: ...\nAnswer: True/False`` while preserving source,
category, and difficulty metadata for filtering. The Edge Function/database still
perform final live validation before serving questions to the app.
"""

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",
    "answer",
    "correct_answer",
    "source_id",
    "category_id",
    "difficulty",
    "is_deleted",
)
OUTPUT_FIELDS = (
    "id",
    "question",
    "answer",
    "source_id",
    "category_id",
    "difficulty",
    "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/true_false_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 _answer_text(row: Dict[str, object], row_number: int) -> str:
    raw = str(row.get("answer") or row.get("correct_answer") or "").strip().lower()
    if raw in {"true", "t", "1", "yes", "y"}:
        return "True"
    if raw in {"false", "f", "0", "2", "no", "n"}:
        return "False"
    raise RuntimeError(
        f"Row {row_number}: answer/correct_answer must resolve to true or false."
    )


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.")
        if row.get("is_deleted") is True:
            raise RuntimeError(
                f"Row {row_number}: deleted true/false question {question_id} was exported."
            )

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

    if not prepared:
        raise RuntimeError("The export contains no active true/false 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(
                "TRUE_FALSE_EXPORT_OUTPUT",
                "sources/custom-topic-quiz-generator/true_false_questions.csv",
            )
        ).resolve()
        rows = prepare_rows(fetch_rows(base_url, service_key))
        write_atomic(rows, output)
        print(f"Exported {len(rows)} active true/false 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())