Merge exam-bank corpus coverage endpoint

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"]

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}

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}