pipeline/fr_docs/api.py

"""FR-Docs classification service.

    uvicorn fr_docs.api:app --port 8080

POST /classify  (multipart file upload) ->
    {
      "filename": "...", "format": "pdf", "is_scanned": false,
      "prediction": {"category_id": "invoice", "confidence": 0.93, ...},
      "routing": {"route": "structured", ...}
    }

Set FR_DOCS_CHECKPOINT=/path/to/checkpoint to switch from the v0 embedding
engine to the fine-tuned ModernBERT model. Nothing else changes.
"""

import os
import tempfile
from pathlib import Path

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse

from .classifier import get_classifier
from .extract import extract
from .routing import route
from .sensitive import SensitiveScanner, redact, summarize
from .taxonomy import load_taxonomy
from .ui import PAGE

app = FastAPI(title="FR-Docs", version="0.1.0")
_classifier = None
_scanner = None


def scanner():
    global _scanner
    if _scanner is None:
        _scanner = SensitiveScanner()
    return _scanner


@app.get("/", response_class=HTMLResponse)
def home():
    return PAGE


def classifier():
    global _classifier
    if _classifier is None:
        _classifier = get_classifier(os.environ.get("FR_DOCS_CHECKPOINT"))
    return _classifier


@app.get("/health")
def health():
    return {"status": "ok", "engine": type(classifier()).__name__}


@app.get("/taxonomy")
def taxonomy():
    return [
        {"id": c.id, "label": c.label, "group": c.group_label}
        for c in load_taxonomy()
    ]


@app.post("/classify")
async def classify(file: UploadFile = File(...)):
    suffix = Path(file.filename or "upload").suffix
    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
        tmp.write(await file.read())
        tmp_path = tmp.name
    try:
        extraction = extract(tmp_path)
        prediction = classifier().predict(extraction.text)
        try:
            findings = scanner().scan(extraction.text)
            scan_warnings = scanner().warnings
        except Exception as exc:  # belt and braces: classification always succeeds
            findings, scan_warnings = [], [f"sensitive scan failed: {exc}"]
        preview_src = extraction.text[:2000]
        preview_findings = [f for f in findings if f.end <= 2000]
        return {
            "filename": file.filename,
            "format": extraction.format,
            "is_scanned": extraction.is_scanned,
            "char_count": extraction.char_count,
            "warnings": extraction.warnings,
            "prediction": prediction.to_dict(),
            "routing": route(prediction, extraction.is_scanned),
            "sensitive": {
                "counts": summarize(findings),
                "total": len(findings),
                "findings": [f.to_dict() for f in findings[:200]],
                "redacted_preview": redact(preview_src, preview_findings),
                "gliner_active": scanner().gliner_active,
                "warnings": scan_warnings,
            },
        }
    finally:
        os.unlink(tmp_path)