S6-1: /api/exam analyse endpoint — structure → ghost-region suggestions (async)

The canvas "Auto-detect" path (spec R3.1/R3.3/R3.5/R5.4). Async because extraction on
an image-only paper takes ~15-60s (Card-12 grounding):

  POST /api/exam/templates/{id}/analyse -> {job_id}   (BackgroundTasks)
  GET  /api/exam/jobs/{id}              -> {status, counts, suggestions}

suggestions = questions + response_areas + boundaries, each source:'ai',confirmed:false
with confidence + page-fraction bounds + deterministic uuids — rendered as ghosts the
teacher confirms/dismisses; never overwrites manual shapes; persistence stays PUT /templates/{id}.

Access is as-the-user (E1/E2): the template is read under RLS (unseen -> 404, IDOR-safe)
and a job is only visible to the user who started it.

The heavy pipeline stays OUT of the API behind modules/services/exam_extract.py — an HTTP
client (aiohttp) to the exam-structure extraction service (the docling-exam-spike pipeline,
GPU-attached, deterministic), mirroring how the app calls Docling Serve/Ollama. Configured
via EXAM_EXTRACT_URL; unset -> the job fails cleanly as 'extraction unavailable'.

Follow-on (S6-1b): stand up that extraction service (POST /api/extract: source -> suggestions)
wrapping scripts/structure.py + scripts/analyse.py.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
CC Worker 2026-07-02 19:45:26 +00:00
parent e269e67f27
commit c0458f5528
4 changed files with 178 additions and 0 deletions

View File

@ -181,3 +181,7 @@ GOOGLE_CLIENT_SECRETS_FILE=
# Node Filesystem (for document storage)
# =============================================================================
NODE_FILESYSTEM_PATH=/tmp/cc-nodes
# Exam-structure extraction service (docling-exam-spike pipeline) for /api/exam analyse (S6-1)
EXAM_EXTRACT_URL=
EXAM_EXTRACT_TIMEOUT=180

View File

@ -0,0 +1,71 @@
"""Exam structure-EXTRACTION client — the seam between the /api/exam analyse endpoint and the
deterministic exam-structure pipeline (Card 12 = GO; verdict in docling-exam-spike).
Why a service, not an in-process port: the pipeline is heavy (Docling + OCR + a question/mark/geometry
parser) and already exists, GPU-attached, in the spike. Following the same pattern the app uses for
Docling Serve and Ollama (§2 infra map), the API calls it over HTTP rather than vendoring thousands of
lines. This module is the ONLY place the API knows about that service.
Contract (the extraction service returns the app's ghost-region 'analyse' suggestions directly — the
mapping lives in the pipeline, `scripts/analyse.py` in the spike; see results/ANALYSE-BRIDGE.md):
POST {EXAM_EXTRACT_URL}/api/extract {"source": {...}, "exam_code": "AQA-PHYS-8463-1H-22-JUN"}
-> {"status": "complete",
"coordinate_space": "page_fraction",
"suggestions": {"questions": [...], "response_areas": [...], "boundaries": [...]}}
Every suggestion is `source:'ai', confirmed:false` with a `confidence` float and page-fraction bounds
(the canvas scales by the PDF render size, R3.2); uuids are deterministic so re-analysing is stable
(§2 join keys). The API never trusts these as final the teacher confirms/dismisses (R3.1/R3.3).
"""
from __future__ import annotations
import os
from typing import Any, Dict, Optional
import aiohttp
from modules.logger_tool import initialise_logger
logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True)
# The exam-structure extraction service (the docling-exam-spike pipeline deployed as a service).
EXAM_EXTRACT_URL = os.getenv("EXAM_EXTRACT_URL", "").rstrip("/")
EXAM_EXTRACT_TIMEOUT = float(os.getenv("EXAM_EXTRACT_TIMEOUT", "180"))
class ExtractionUnavailable(RuntimeError):
"""The extraction service is not configured/reachable — surfaced to the job as a clean failure."""
async def analyse_source(source: Dict[str, Any], exam_code: Optional[str] = None) -> Dict[str, Any]:
"""Run the extraction pipeline on a template's source paper → ghost-region suggestions.
`source` identifies the paper for the service to fetch/render: a storage-resolvable pointer
(e.g. {"storage_url": "..."} or {"file_id": "..."}) and/or a catalogue {"exam_code": "..."}.
Resolution + PDF rendering happen service-side; the API only forwards the pointer + exam_code.
"""
if not EXAM_EXTRACT_URL:
raise ExtractionUnavailable(
"EXAM_EXTRACT_URL is not set — the exam-structure extraction service is not wired. "
"Deploy the docling-exam-spike pipeline as the extraction service and set EXAM_EXTRACT_URL."
)
payload = {"source": source, "exam_code": exam_code}
try:
timeout = aiohttp.ClientTimeout(total=EXAM_EXTRACT_TIMEOUT)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.post(f"{EXAM_EXTRACT_URL}/api/extract", json=payload) as resp:
resp.raise_for_status()
data = await resp.json()
except aiohttp.ClientError as exc:
logger.error(f"exam extraction service call failed: {exc}")
raise ExtractionUnavailable(f"extraction service error: {exc}") from exc
sugg = (data or {}).get("suggestions") or {}
# normalise: the three suggestion arrays are always present so the canvas can render partial results
return {
"coordinate_space": data.get("coordinate_space", "page_fraction"),
"questions": sugg.get("questions", []),
"response_areas": sugg.get("response_areas", []),
"boundaries": sugg.get("boundaries", []),
"meta": data.get("meta"),
}

View File

@ -8,9 +8,11 @@ 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.analyse import router as analyse_router
router = APIRouter()
router.include_router(templates_router)
router.include_router(batches_router)
router.include_router(analyse_router)
__all__ = ["router"]

101
routers/exam/analyse.py Normal file
View File

@ -0,0 +1,101 @@
"""AI auto-map: analyse a template's source paper → ghost-region suggestions (spec R3.1/R3.3/R3.5/R5.4).
The teacher's "Auto-detect remaining" button hits this. Because extraction on an image-only paper takes
~15-60s (Card-12 grounding), it runs ASYNC: POST returns a job_id, the canvas polls GET /jobs/{id}, and on
completion gets `suggestions` (questions + response_areas + boundaries) to render as GHOSTS
(`source:'ai', confirmed:false`, opacity by confidence). The teacher confirms/dismisses; nothing here is
authoritative and it never overwrites manual shapes (R3.1/R3.3). Persisting confirmed shapes is the
existing PUT /templates/{id} path.
Access is as-the-user (E1/E2): the template is read under RLS (a template the caller can't see reads back
absent 404, IDOR-safe), and a job is only visible to the user who started it. The heavy pipeline lives
behind `modules.services.exam_extract` (the docling-exam-spike deployed as the extraction service).
"""
from __future__ import annotations
import os
import time
import uuid
from typing import Any, Dict
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
from modules.logger_tool import initialise_logger
from modules.services.exam_extract import ExtractionUnavailable, analyse_source
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()
# In-process job store. R5.4 allows BackgroundTasks + a status row for Sprint-4; a durable `exam_jobs`
# table (survives restarts / multiple workers) is the follow-on. Keyed by job_id; scoped to its user.
JOBS: Dict[str, Dict[str, Any]] = {}
_JOB_TTL = 3600 # forget finished jobs after an hour so the dict doesn't grow unbounded
def _gc() -> None:
now = time.time()
for jid in [j for j, v in JOBS.items() if now - v.get("created", now) > _JOB_TTL]:
JOBS.pop(jid, None)
def _template_or_404(ctx: ExamContext, template_id: str) -> Dict[str, Any]:
"""Read the template AS THE USER — RLS makes a non-owned/other-institute template read as absent."""
res = (
ctx.supabase.table("exam_templates")
.select("id, exam_code, source_file_id, institute_id")
.eq("id", template_id)
.execute()
)
rows = getattr(res, "data", None) or []
if not rows:
raise HTTPException(status_code=404, detail="Template not found")
return rows[0] if isinstance(rows, list) else rows
async def _run_job(job_id: str, source: Dict[str, Any], exam_code: str | None) -> None:
job = JOBS.get(job_id)
if not job:
return
job["status"] = "running"
try:
sugg = await analyse_source(source, exam_code)
job.update(status="complete", progress=100, suggestions=sugg,
counts={k: len(sugg.get(k, [])) for k in ("questions", "response_areas", "boundaries")})
except ExtractionUnavailable as exc:
logger.warning(f"analyse job {job_id}: extraction unavailable: {exc}")
job.update(status="error", error=str(exc))
except Exception as exc: # noqa: BLE001 - the job must record any failure, not crash the worker
logger.error(f"analyse job {job_id} failed: {exc}")
job.update(status="error", error=f"{type(exc).__name__}: {exc}")
@router.post("/templates/{template_id}/analyse")
async def start_analyse(
template_id: str,
background: BackgroundTasks,
ctx: ExamContext = Depends(get_exam_context),
) -> Dict[str, Any]:
"""Kick off extraction for this template's source paper → {job_id}. Poll GET /jobs/{job_id}."""
_gc()
tmpl = _template_or_404(ctx, template_id)
job_id = str(uuid.uuid4())
JOBS[job_id] = {"id": job_id, "template_id": template_id, "user_id": ctx.user_id,
"status": "queued", "progress": 0, "created": time.time()}
# the extraction service resolves + renders the source itself; the API only forwards the pointer
source = {"file_id": tmpl.get("source_file_id"), "exam_code": tmpl.get("exam_code")}
background.add_task(_run_job, job_id, source, tmpl.get("exam_code"))
return {"job_id": job_id, "status": "queued"}
@router.get("/jobs/{job_id}")
async def get_job(job_id: str, ctx: ExamContext = Depends(get_exam_context)) -> Dict[str, Any]:
"""Poll an analyse job. On `complete`, `suggestions` = ghost regions for the canvas to render."""
job = JOBS.get(job_id)
if not job or job.get("user_id") != ctx.user_id: # a job is only visible to the user who started it
raise HTTPException(status_code=404, detail="Job not found")
out = {k: job.get(k) for k in ("id", "template_id", "status", "progress", "error", "counts")}
if job.get("status") == "complete":
out["suggestions"] = job.get("suggestions")
return out