"""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)