From 52801a5d340c025c701e5057fa9370124702ef72 Mon Sep 17 00:00:00 2001 From: "Kevin Carter (via Claude)" Date: Thu, 2 Jul 2026 22:33:07 +0000 Subject: [PATCH] Exam bank: corpus coverage endpoint (state of the collected exam bank) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- routers/exam/__init__.py | 2 + routers/exam/corpus.py | 98 +++++++++++++++++++++++++++++++++++++++ tests/test_exam_corpus.py | 48 +++++++++++++++++++ 3 files changed, 148 insertions(+) create mode 100644 routers/exam/corpus.py create mode 100644 tests/test_exam_corpus.py diff --git a/routers/exam/__init__.py b/routers/exam/__init__.py index 3aa820d..449e62c 100644 --- a/routers/exam/__init__.py +++ b/routers/exam/__init__.py @@ -9,10 +9,12 @@ from fastapi import APIRouter from routers.exam.templates import router as templates_router from routers.exam.batches import router as batches_router from routers.exam.bank import router as bank_router +from routers.exam.corpus import router as corpus_router router = APIRouter() router.include_router(templates_router) router.include_router(batches_router) router.include_router(bank_router) +router.include_router(corpus_router) __all__ = ["router"] diff --git a/routers/exam/corpus.py b/routers/exam/corpus.py new file mode 100644 index 0000000..14a0180 --- /dev/null +++ b/routers/exam/corpus.py @@ -0,0 +1,98 @@ +"""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} diff --git a/tests/test_exam_corpus.py b/tests/test_exam_corpus.py new file mode 100644 index 0000000..409a5a0 --- /dev/null +++ b/tests/test_exam_corpus.py @@ -0,0 +1,48 @@ +"""Test the exam-bank corpus coverage grouping (/api/exam/corpus).""" +from fastapi import FastAPI +from fastapi.testclient import TestClient +from routers.exam.corpus import router +from routers.exam.dependencies import ExamContext, get_exam_context + +TEACHER = "00000000-0000-0000-0000-000000000001" +INST_A = "10000000-0000-0000-0000-000000000001" + + +class FakeResult: + def __init__(self, data): self.data = data + + +class FakeQuery: + def __init__(self, store, table): self.store, self.table = store, table + def select(self, *_a, **_k): return self + def execute(self): return FakeResult(list(self.store.get(self.table, []))) + + +class FakeSupabase: + def __init__(self, store): self.store = store + def table(self, name): return FakeQuery(self.store, name) + + +def _client(store): + app = FastAPI(); app.include_router(router, prefix="/api/exam") + app.dependency_overrides[get_exam_context] = lambda: ExamContext(TEACHER, "tok", FakeSupabase(store), [INST_A]) + return TestClient(app) + + +def test_corpus_groups_papers_by_session_with_qp_ms_er_coverage(): + store = { + "eb_specifications": [{"spec_code": "AQA-PHYS-8463", "exam_board_code": "AQA", "subject_code": "PHYSICS", "award_code": "8463", "first_teach": "2016"}], + "eb_exams": [ + {"exam_code": "a", "spec_code": "AQA-PHYS-8463", "paper_code": "8463/1H", "session": "2022-Jun", "type_code": "QP", "storage_loc": "cc.examboards/a.pdf"}, + {"exam_code": "b", "spec_code": "AQA-PHYS-8463", "paper_code": "8463/1H", "session": "2022-Jun", "type_code": "MS", "storage_loc": "cc.examboards/b.pdf"}, + {"exam_code": "c", "spec_code": "AQA-PHYS-8463", "paper_code": "8463/1H", "session": "2022-Jun", "type_code": "ER", "storage_loc": None}, + {"exam_code": "d", "spec_code": "AQA-PHYS-8463", "paper_code": "8463/2H", "session": "2023-Jun", "type_code": "QP", "storage_loc": "cc.examboards/d.pdf"}, + ], + } + body = _client(store).get("/api/exam/corpus").json() + assert body["totals"]["specs"] == 1 and body["totals"]["papers"] == 2 + spec = body["boards"][0]["specs"][0] + assert spec["level"] == "GCSE" and spec["subject"] == "Physics" + assert spec["counts"] == {"QP": 2, "MS": 1, "ER": 0} # ER present but not stored → not counted + p = next(p for p in spec["papers"] if p["paper_code"] == "8463/1H") + assert p["docs"] == {"QP": True, "MS": True, "ER": False}