From 1051ed333213c4a0c70b7b649466595c9b6b9a82 Mon Sep 17 00:00:00 2001 From: Chaim Date: Wed, 3 Jun 2026 09:56:21 +0000 Subject: [PATCH] feat(legal-aid): async parse jobs to survive multi-minute vision latency MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude Vision via ai-gateway's claude-code OAuth provider runs ~140s per rendered page (a 3-page report measured at 417s), which blows past any safe synchronous HTTP timeout — the EspoCRM controller's 240s curl gave up with "0 bytes received" while ai-gateway kept working. parse-pdf-report now accepts ?mode=async: it returns {jobId, status:"pending"} immediately and runs the parse in the background, persisting the result/error to a file-based job store under /opt/data (survives multi-worker + restarts). Poll GET /parse-pdf-report/jobs/{jobId}. Default sync mode is unchanged so n8n and other server-side callers keep working. Refs prod ai-gateway log id cee4c17c (claudeMs=417513). Co-Authored-By: Claude Opus 4.8 (1M context) --- api/routes/legal_aid.py | 164 ++++++++++++++++++++++++++++++++++++---- 1 file changed, 148 insertions(+), 16 deletions(-) diff --git a/api/routes/legal_aid.py b/api/routes/legal_aid.py index bbb594c..dd0e632 100644 --- a/api/routes/legal_aid.py +++ b/api/routes/legal_aid.py @@ -17,12 +17,15 @@ result. Auth is X-Api-Key as for the other admin endpoints. from __future__ import annotations +import asyncio import base64 import io import json import logging import os import re +import time +import uuid from typing import Any from fastapi import APIRouter, File, HTTPException, Query, Request, UploadFile @@ -44,6 +47,67 @@ def _verify_admin(request: Request) -> None: _PDF_RENDER_DPI = 180 _PDF_MAX_PAGES = 8 # safety cap; real reports are 1-2 pages +# --- Async job store --------------------------------------------------- +# Claude Vision via ai-gateway's claude-code OAuth provider runs ~140s per +# rendered page (a 3-page report measured at 417s). That far exceeds the +# EspoCRM controller's synchronous curl timeout, so manual uploads run as a +# background job: POST ?mode=async returns a jobId immediately and the caller +# polls GET /parse-pdf-report/jobs/{jobId}. Jobs are persisted as files under +# /opt/data (a named volume) so they survive across uvicorn workers and short +# restarts. n8n / other server-side callers can still use the default sync mode. +_JOBS_DIR = os.environ.get("LEGAL_AID_JOBS_DIR", "/opt/data/legal-aid-jobs") +_JOB_TTL_SECONDS = 3600 # finished jobs are reaped after 1h +_running_jobs: set[asyncio.Task] = set() + + +def _jobs_dir() -> str: + os.makedirs(_JOBS_DIR, exist_ok=True) + return _JOBS_DIR + + +def _job_path(job_id: str) -> str: + # job_id is a server-generated uuid hex; sanitize defensively anyway so a + # crafted id from the status route can never escape the jobs directory. + safe = re.sub(r"[^a-f0-9]", "", job_id.lower()) + if not safe: + safe = "_invalid_" + return os.path.join(_jobs_dir(), f"{safe}.json") + + +def _write_job(job_id: str, data: dict[str, Any]) -> None: + path = _job_path(job_id) + tmp = f"{path}.tmp" + with open(tmp, "w", encoding="utf-8") as f: + json.dump(data, f, ensure_ascii=False) + os.replace(tmp, path) # atomic publish + + +def _read_job(job_id: str) -> dict[str, Any] | None: + path = _job_path(job_id) + if not os.path.exists(path): + return None + try: + with open(path, encoding="utf-8") as f: + return json.load(f) + except (OSError, ValueError): + return None + + +def _cleanup_old_jobs() -> None: + try: + now = time.time() + for name in os.listdir(_jobs_dir()): + if not name.endswith(".json"): + continue + p = os.path.join(_JOBS_DIR, name) + try: + if now - os.path.getmtime(p) > _JOB_TTL_SECONDS: + os.remove(p) + except OSError: + pass + except OSError: + pass + _EXTRACT_PROMPT = """\ אתה מחלץ נתונים מדוח תשלומים חודשי של הסיוע המשפטי (Legal Aid, עברית RTL). @@ -87,25 +151,15 @@ _EXTRACT_PROMPT = """\ """ -@router.post("/parse-pdf-report") -async def parse_pdf_report( - request: Request, - file: UploadFile = File(..., description="Legal Aid monthly payment report PDF"), - subject: str | None = None, - debug: bool = Query(default=False), +async def _parse_pdf_to_payload( + pdf_bytes: bytes, subject: str | None, debug: bool ) -> dict[str, Any]: - """Parse a Legal Aid monthly payment-report PDF via Claude Vision. + """Render the PDF pages and run Claude Vision to produce the ingest payload. - Auth: X-Api-Key. - Body: multipart with `file` (the PDF bytes). - Optional `subject` query: if supplied, used to extract `[סימוכין NNN]` → - sysref. Manual uploads typically omit this. - Optional `debug=true` query: also returns the raw model response and per- - page render info under `_debug` for diagnosing extraction issues. + Shared by both the synchronous route and the async job runner. Raises + HTTPException on any failure (the job runner converts those to a stored + error status). """ - _verify_admin(request) - - pdf_bytes = await file.read() if not pdf_bytes: raise HTTPException(status_code=400, detail="Empty file") if pdf_bytes[:4] != b"%PDF": @@ -167,6 +221,84 @@ async def parse_pdf_report( return out +async def _run_parse_job( + job_id: str, pdf_bytes: bytes, subject: str | None, debug: bool +) -> None: + """Background runner: parse the PDF and persist the result/error to disk.""" + try: + result = await _parse_pdf_to_payload(pdf_bytes, subject, debug) + _write_job( + job_id, + {"jobId": job_id, "status": "done", "result": result, "finishedAt": time.time()}, + ) + except HTTPException as e: + _write_job( + job_id, + {"jobId": job_id, "status": "error", "detail": str(e.detail), "finishedAt": time.time()}, + ) + except Exception as e: # noqa: BLE001 — anything unexpected becomes a stored error + logger.exception("Legal Aid async parse job %s failed", job_id) + _write_job( + job_id, + {"jobId": job_id, "status": "error", "detail": str(e), "finishedAt": time.time()}, + ) + + +@router.post("/parse-pdf-report") +async def parse_pdf_report( + request: Request, + file: UploadFile = File(..., description="Legal Aid monthly payment report PDF"), + subject: str | None = None, + mode: str = Query(default="sync", description="'sync' (inline) or 'async' (job + poll)"), + debug: bool = Query(default=False), +) -> dict[str, Any]: + """Parse a Legal Aid monthly payment-report PDF via Claude Vision. + + Auth: X-Api-Key. + Body: multipart with `file` (the PDF bytes). + Optional `subject` query: if supplied, used to extract `[סימוכין NNN]` → + sysref. Manual uploads typically omit this. + Optional `mode=async`: returns `{jobId, status:"pending"}` immediately and + runs the parse in the background; poll `/parse-pdf-report/jobs/{jobId}`. + Default `sync` runs inline (the vision call can take several minutes — + only safe for callers without a short request timeout, e.g. n8n). + Optional `debug=true`: also returns the raw model response + page count. + """ + _verify_admin(request) + + pdf_bytes = await file.read() + # Validate up front so an obviously-bad upload fails fast in BOTH modes + # (rather than surfacing as a background-job error a moment later). + if not pdf_bytes: + raise HTTPException(status_code=400, detail="Empty file") + if pdf_bytes[:4] != b"%PDF": + raise HTTPException(status_code=400, detail="File is not a PDF") + + if mode == "async": + _cleanup_old_jobs() + job_id = uuid.uuid4().hex + _write_job(job_id, {"jobId": job_id, "status": "pending", "createdAt": time.time()}) + task = asyncio.create_task(_run_parse_job(job_id, pdf_bytes, subject, debug)) + _running_jobs.add(task) # keep a strong ref so the task isn't GC'd mid-run + task.add_done_callback(_running_jobs.discard) + return {"jobId": job_id, "status": "pending"} + + return await _parse_pdf_to_payload(pdf_bytes, subject, debug) + + +@router.get("/parse-pdf-report/jobs/{job_id}") +async def get_parse_job(request: Request, job_id: str) -> dict[str, Any]: + """Poll an async parse job. Returns `{status: pending|done|error, ...}`. + + `done` carries `result` (the ingest payload); `error` carries `detail`. + """ + _verify_admin(request) + job = _read_job(job_id) + if job is None: + raise HTTPException(status_code=404, detail="Job not found (expired or invalid)") + return job + + # ---------------------------------------------------------------------- # AI Gateway / Claude Vision call # ----------------------------------------------------------------------