diff --git a/.env.example b/.env.example index 8ee1234..1f2d4cf 100644 --- a/.env.example +++ b/.env.example @@ -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 diff --git a/modules/services/exam_extract.py b/modules/services/exam_extract.py new file mode 100644 index 0000000..8cd9f03 --- /dev/null +++ b/modules/services/exam_extract.py @@ -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"), + } diff --git a/routers/exam/__init__.py b/routers/exam/__init__.py index 35e6488..e5e5072 100644 --- a/routers/exam/__init__.py +++ b/routers/exam/__init__.py @@ -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"] diff --git a/routers/exam/analyse.py b/routers/exam/analyse.py new file mode 100644 index 0000000..c92837f --- /dev/null +++ b/routers/exam/analyse.py @@ -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