Kevin Carter (via Claude)
52801a5d34
GET /api/exam/corpus — read-only view over the seeded eb_specifications + eb_exams catalogue: board → subject → spec → papers grouped by paper_code+session, each showing which of QP/MS/ER exist AND are stored, with per-spec + global rollups (specs, papers, sessions, QP/MS/ER counts). Surfaces the collected exam bank so the app can display its state — e.g. AQA GCSE Physics 8463: 21 QP / 12 MS / 17 ER. Validated against the live catalogue (60 specs, 1178 docs, 535 papers) + a unit test (grouping, QP/MS/ER presence, stored-vs-catalogued). As-user (catalogue is public reference data). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
99 lines
4.0 KiB
Python
99 lines
4.0 KiB
Python
"""Exam-bank corpus coverage (/api/exam/corpus) — the state of the collected exam bank.
|
|
|
|
Read-only view over the seeded exam-board catalogue (eb_specifications + eb_exams): board → subject →
|
|
specification → papers, with per-session QP/MS/ER coverage and rollup counts. Shows what the app has
|
|
COLLECTED (question papers, mark schemes, examiner reports) so a teacher/admin can see the bank's state
|
|
and where science coverage is complete vs thin. Catalogue data is public reference — read as-the-user.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from fastapi import APIRouter, Depends
|
|
|
|
from modules.logger_tool import initialise_logger
|
|
from routers.exam.dependencies import ExamContext, get_exam_context
|
|
|
|
logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True)
|
|
|
|
router = APIRouter()
|
|
|
|
DOC_TYPES = ("QP", "MS", "ER")
|
|
|
|
|
|
def _rows(result: Any) -> List[Dict[str, Any]]:
|
|
data = getattr(result, "data", None)
|
|
if not data:
|
|
return []
|
|
return data if isinstance(data, list) else [data]
|
|
|
|
|
|
def _award_level(spec: Dict[str, Any]) -> str:
|
|
"""Best-effort GCSE / AS / A-level from the spec code (AQA GCSE = 8xxx, A-level/AS = 7xxx)."""
|
|
code = re.sub(r"\D", "", spec.get("award_code") or spec.get("spec_code") or "")
|
|
if code.startswith("8"):
|
|
return "GCSE"
|
|
if code.startswith("7"):
|
|
return "A-level"
|
|
return spec.get("award_level") or "Other"
|
|
|
|
|
|
@router.get("/corpus")
|
|
async def corpus_coverage(ctx: ExamContext = Depends(get_exam_context)) -> Dict[str, Any]:
|
|
specs = _rows(
|
|
ctx.supabase.table("eb_specifications")
|
|
.select("spec_code, exam_board_code, subject_code, award_code, first_teach").execute()
|
|
)
|
|
exams = _rows(
|
|
ctx.supabase.table("eb_exams")
|
|
.select("exam_code, spec_code, paper_code, tier, session, type_code, storage_loc").execute()
|
|
)
|
|
|
|
# group exam docs → per spec → per paper (paper_code + session) → which doc types are present
|
|
by_spec: Dict[str, Dict[str, Dict[str, Any]]] = {}
|
|
for e in exams:
|
|
sc = e.get("spec_code")
|
|
if not sc:
|
|
continue
|
|
key = f"{e.get('paper_code') or '?'}|{e.get('session') or '?'}"
|
|
paper = by_spec.setdefault(sc, {}).setdefault(key, {
|
|
"paper_code": e.get("paper_code"), "session": e.get("session"),
|
|
"tier": e.get("tier"), "docs": {}, "exam_codes": {},
|
|
})
|
|
dt = (e.get("type_code") or "").upper()
|
|
if dt in DOC_TYPES:
|
|
paper["docs"][dt] = bool(e.get("storage_loc"))
|
|
paper["exam_codes"][dt] = e.get("exam_code")
|
|
|
|
totals = {"specs": 0, "papers": 0, "sessions": 0, **{d: 0 for d in DOC_TYPES}}
|
|
boards: Dict[str, Dict[str, Any]] = {}
|
|
for s in specs:
|
|
sc = s["spec_code"]
|
|
papers_map = by_spec.get(sc, {})
|
|
if not papers_map:
|
|
continue
|
|
totals["specs"] += 1
|
|
board = s.get("exam_board_code") or "?"
|
|
level = _award_level(s)
|
|
papers = sorted(papers_map.values(), key=lambda p: (str(p["session"]), str(p["paper_code"])))
|
|
counts = {d: sum(1 for p in papers if p["docs"].get(d)) for d in DOC_TYPES}
|
|
for d in DOC_TYPES:
|
|
totals[d] += counts[d]
|
|
totals["papers"] += len(papers)
|
|
totals["sessions"] += len({p["session"] for p in papers})
|
|
spec_entry = {
|
|
"spec_code": sc, "subject": (s.get("subject_code") or "").title(), "level": level,
|
|
"board": board, "first_teach": s.get("first_teach"),
|
|
"n_papers": len(papers), "counts": counts, "papers": papers,
|
|
}
|
|
boards.setdefault(board, {"board": board, "specs": []})["specs"].append(spec_entry)
|
|
|
|
board_list = []
|
|
for board in sorted(boards):
|
|
specs_sorted = sorted(boards[board]["specs"], key=lambda x: (x["level"], x["subject"], x["spec_code"]))
|
|
board_list.append({"board": board, "n_specs": len(specs_sorted), "specs": specs_sorted})
|
|
|
|
return {"totals": totals, "boards": board_list}
|