#!/usr/bin/env python3 """HTTP API that serves the pipeline (preprocess -> ocr -> structure) to the Android app. Runs the same in-process functions the numbered host scripts drive via `docker run` per stage, but skips the per-stage container start-up since a live server has no need for that inspectability - the host scripts remain the way to inspect each stage independently during pipeline development. Jobs are processed asynchronously so the app can show real per-stage progress: POST /jobs returns immediately with a job id; GET /jobs/{id} reports which of preprocessing/ocr/extracting is currently running, and the final result once done. Job state lives in memory only - this is a personal, LAN-only, single-instance server with no need for persistence here (the Android app is the system of record for confirmed receipts). """ import os import shutil import tempfile import threading import time import uuid import requests from fastapi import FastAPI, HTTPException, UploadFile from fastapi.responses import JSONResponse from adaptive import preprocess_and_ocr_adaptive from extract import structure OLLAMA_BASE_URL = os.environ.get("OLLAMA_BASE_URL", "http://127.0.0.1:11434") OLLAMA_MODEL = os.environ.get("OLLAMA_MODEL", "qwen2.5:3b") app = FastAPI(title="fuelish") _jobs: dict[str, dict] = {} _jobs_lock = threading.Lock() def _set_job(job_id: str, **fields) -> None: with _jobs_lock: _jobs[job_id].update(fields) def _ensure_model_pulled() -> None: """Best-effort: wait for Ollama to come up (compose starts `api` and `ollama` together, with no ordering guarantee beyond container start) and pull OLLAMA_MODEL if it isn't already present, so a bare `docker compose up` is enough without a manual `ollama pull` step. Runs in a background thread - jobs submitted before this finishes will simply fail with a clear error from Ollama.""" tags = None for _ in range(60): try: tags = requests.get(f"{OLLAMA_BASE_URL}/api/tags", timeout=5).json() break except requests.RequestException: time.sleep(2) if tags is None: return have = {m["name"] for m in tags.get("models", [])} if OLLAMA_MODEL in have or any(m.startswith(OLLAMA_MODEL + ":") for m in have): return try: requests.post( f"{OLLAMA_BASE_URL}/api/pull", json={"name": OLLAMA_MODEL}, timeout=1800, ) except requests.RequestException: pass @app.on_event("startup") def _on_startup() -> None: threading.Thread(target=_ensure_model_pulled, daemon=True).start() @app.get("/health") def health() -> dict: return {"status": "ok"} def _process_job(job_id: str, image_path: str, scratch_dir: str) -> None: try: _set_job(job_id, status="preprocessing") adaptive_result = preprocess_and_ocr_adaptive(image_path, scratch_dir) _set_job(job_id, status="extracting") result = structure(adaptive_result["ocr_text"]) _set_job(job_id, status="done", result=result) except Exception as exc: # noqa: BLE001 - reported to the client, not swallowed _set_job(job_id, status="error", error=str(exc)) finally: shutil.rmtree(scratch_dir, ignore_errors=True) @app.post("/jobs", status_code=201) async def create_job(image: UploadFile) -> JSONResponse: job_id = uuid.uuid4().hex scratch_dir = tempfile.mkdtemp(prefix=f"fuelish-{job_id}-") image_path = os.path.join(scratch_dir, "upload" + os.path.splitext(image.filename or "")[1]) with open(image_path, "wb") as f: f.write(await image.read()) with _jobs_lock: _jobs[job_id] = {"status": "queued", "result": None, "error": None, "_created": time.time()} thread = threading.Thread(target=_process_job, args=(job_id, image_path, scratch_dir), daemon=True) thread.start() return JSONResponse({"job_id": job_id}, status_code=201) @app.get("/jobs/{job_id}") def get_job(job_id: str) -> dict: with _jobs_lock: job = _jobs.get(job_id) if job is None: raise HTTPException(status_code=404, detail="job not found") return {"status": job["status"], "result": job["result"], "error": job["error"]}