Spaces:
Sleeping
Sleeping
File size: 10,834 Bytes
c1b1f3a 6869b3c c1b1f3a 6869b3c c1b1f3a 6869b3c c1b1f3a 41d8121 c1b1f3a 6869b3c c1b1f3a 41d8121 6869b3c 41d8121 c1b1f3a 41d8121 6869b3c 41d8121 6869b3c 41d8121 6869b3c 41d8121 c1b1f3a 6869b3c 41d8121 c1b1f3a 6869b3c c1b1f3a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 | import os
import time
import csv
import json
from datetime import datetime
from typing import Dict, Any, Optional
import pickle
from pathlib import Path
from typing import List
from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
import uvicorn
app = FastAPI(
title="Document Intelligence & Extraction API",
description="Two-stage pipeline: Document Classification & Field Extraction with Confidence Routing",
version="1.0.0"
)
# Global pipeline instance
PIPELINE = None
PIPELINE_ERROR = None
LOG_CSV = "service_requests.csv"
PROJECT_ROOT = Path(__file__).resolve().parents[2]
REPORTS_DIR = PROJECT_ROOT / "reports"
FRONTEND_DIR = PROJECT_ROOT.parent / "frontend" / "dist"
SAMPLE_DOCUMENTS = {
"invoice-ffcy0002": {
"label": "Commercial Invoice",
"description": "Invoice with supplier, buyer, line items, and total amount.",
"path": PROJECT_ROOT / "data/raw/invoice/ffcy0002.png",
"doc_class": "invoice",
},
"invoice-fflb0108": {
"label": "Purchase Invoice",
"description": "Invoice document for layout and field extraction testing.",
"path": PROJECT_ROOT / "data/raw/invoice/fflb0108.png",
"doc_class": "invoice",
},
"receipt-cord-000": {
"label": "Retail Receipt",
"description": "Receipt with merchant details, date, and total balance.",
"path": PROJECT_ROOT / "data/raw/receipt/cord_test-000_0.png",
"doc_class": "receipt",
},
"receipt-cord-001": {
"label": "Receipt Variant",
"description": "Receipt variation for OCR and classification comparison.",
"path": PROJECT_ROOT / "data/raw/receipt/cord_test-000_1.png",
"doc_class": "receipt",
},
"application-funsd-1160": {
"label": "Corporate Application Form",
"description": "Structured application form with applicant fields and signature block.",
"path": PROJECT_ROOT / "data/raw/application_form/funsd_0000971160.png",
"doc_class": "application_form",
},
"application-funsd-9556": {
"label": "Official Evaluation Form",
"description": "Form with headers, check boxes, and applicant fields.",
"path": PROJECT_ROOT / "data/raw/application_form/funsd_0000989556.png",
"doc_class": "application_form",
},
}
def log_api_request(doc_id: str, doc_class: str, confidence: float, needs_review: bool, latency_ms: float):
file_exists = os.path.exists(LOG_CSV)
with open(LOG_CSV, "a", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
if not file_exists:
writer.writerow(["timestamp", "doc_id", "doc_class", "confidence", "needs_review", "latency_ms"])
writer.writerow([datetime.utcnow().isoformat(), doc_id, doc_class, confidence, needs_review, latency_ms])
def load_champion_pipeline():
global PIPELINE, PIPELINE_ERROR
if PIPELINE is not None:
return PIPELINE
started_at = time.time()
try:
from ..pipeline.end_to_end import DocumentIntelligencePipeline
from ..pipeline.confidence_router import ConfidenceRouter
from ..extraction.cohere_extractor import CohereFieldExtractor
# Check for champion models on disk or initialize default
models_dir = REPORTS_DIR / "classification_artifacts"
# Logistic regression is the stable low-latency inference artifact for the
# web app. XGBoost remains available in analytics but can block first-use
# loading on small CPU Space instances.
cls_path = models_dir / "logistic_regression.pkl"
if not cls_path.exists():
cls_path = models_dir / "xgboost.pkl"
if cls_path.exists():
print(f"[API] Loading classifier: {cls_path.name}", flush=True)
with open(cls_path, "rb") as f:
classifier = pickle.load(f)
else:
from ..classification.train_classical import DocumentClassificationPipeline
classifier = DocumentClassificationPipeline("logistic_regression")
extractors = {
"invoice": CohereFieldExtractor("invoice"),
"receipt": CohereFieldExtractor("receipt"),
"application_form": CohereFieldExtractor("application_form")
}
PIPELINE = DocumentIntelligencePipeline(
classifier=classifier,
extractors_by_class=extractors,
router=ConfidenceRouter(t1_class_confidence=0.75, t2_field_confidence=0.70)
)
PIPELINE_ERROR = None
print(f"[API] Pipeline initialized in {round(time.time() - started_at, 2)}s", flush=True)
return PIPELINE
except Exception as exc:
PIPELINE_ERROR = f"{type(exc).__name__}: {exc}"
print(f"[API] Pipeline initialization failed: {PIPELINE_ERROR}", flush=True)
raise
@app.get("/health")
def health_check():
return {"status": "ok", "service": "document_intelligence_api", "timestamp": datetime.utcnow().isoformat()}
@app.get("/api/status")
def pipeline_status():
return {
"status": "ready" if PIPELINE is not None else "not_initialized",
"pipeline_initialized": PIPELINE is not None,
"pipeline_error": PIPELINE_ERROR,
}
@app.get("/api/samples")
def list_samples():
return [
{"id": key, "label": value["label"], "description": value["description"], "doc_class": value["doc_class"]}
for key, value in SAMPLE_DOCUMENTS.items()
if value["path"].exists()
]
@app.get("/api/samples/{sample_id}/preview")
def sample_preview(sample_id: str):
sample = SAMPLE_DOCUMENTS.get(sample_id)
if not sample or not sample["path"].exists():
raise HTTPException(status_code=404, detail="Sample document not found")
return FileResponse(sample["path"])
def process_path(path: str, doc_id: str):
try:
pipeline = load_champion_pipeline()
except Exception as exc:
raise HTTPException(status_code=503, detail=f"Pipeline initialization failed: {exc}") from exc
if pipeline is None:
raise HTTPException(status_code=503, detail="Pipeline not initialized")
res = pipeline.process_document(path, doc_id=doc_id)
log_api_request(
doc_id=doc_id,
doc_class=res["doc_class"],
confidence=res["class_confidence"],
needs_review=res["needs_review"],
latency_ms=res["processing_time_ms"]
)
return res
@app.post("/api/process")
@app.post("/process")
async def process_document_endpoint(file: UploadFile = File(...)):
temp_path = f"temp_{int(time.time()*1000)}_{file.filename}"
try:
content = await file.read()
if len(content) > 10 * 1024 * 1024:
raise HTTPException(status_code=413, detail="File size exceeds maximum allowed upload limit of 10 MB")
with open(temp_path, "wb") as f_out:
f_out.write(content)
return JSONResponse(content=process_path(temp_path, file.filename))
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
finally:
if os.path.exists(temp_path):
try:
os.remove(temp_path)
except Exception:
pass
@app.post("/api/process-sample/{sample_id}")
def process_sample(sample_id: str):
sample = SAMPLE_DOCUMENTS.get(sample_id)
if not sample or not sample["path"].exists():
raise HTTPException(status_code=404, detail="Sample document not found")
try:
return JSONResponse(content=process_path(str(sample["path"]), sample_id))
except HTTPException:
raise
except Exception as exc:
raise HTTPException(status_code=500, detail=str(exc)) from exc
@app.post("/api/process-batch")
async def process_batch(files: List[UploadFile] = File(...)):
if len(files) > 20:
raise HTTPException(status_code=400, detail="Batch size cannot exceed 20 files")
results = []
temp_paths = []
try:
for file in files:
content = await file.read()
if len(content) > 10 * 1024 * 1024:
results.append({"file_name": file.filename, "error": "File exceeds 10 MB"})
continue
temp_path = PROJECT_ROOT / "data/processed/uploads" / f"batch_{int(time.time()*1000)}_{file.filename}"
temp_path.parent.mkdir(parents=True, exist_ok=True)
temp_path.write_bytes(content)
temp_paths.append(temp_path)
try:
result = process_path(str(temp_path), file.filename)
results.append({
"file_name": file.filename,
"doc_class": result["doc_class"],
"confidence": result["class_confidence"],
"needs_review": result["needs_review"],
"latency_ms": result["processing_time_ms"],
"words": result["ocr_token_count"],
})
except Exception as exc:
results.append({"file_name": file.filename, "error": str(exc)})
return results
finally:
for path in temp_paths:
path.unlink(missing_ok=True)
@app.get("/api/analytics")
def analytics(model: str = "xgboost"):
artifact_dir = REPORTS_DIR / "classification_artifacts"
matrix = artifact_dir / f"normalized_confusion_matrix_{model}.png"
if not matrix.exists():
matrix = artifact_dir / f"confusion_matrix_{model}.png"
metrics_path = artifact_dir / f"classification_report_{model}.json"
metrics = {}
if metrics_path.exists():
with metrics_path.open("r", encoding="utf-8") as handle:
metrics = json.load(handle)
return {"model": model, "metrics": metrics, "matrix_url": f"/api/analytics/{model}/matrix" if matrix.exists() else None}
@app.get("/api/analytics/{model}/matrix")
def analytics_matrix(model: str):
artifact_dir = REPORTS_DIR / "classification_artifacts"
matrix = artifact_dir / f"normalized_confusion_matrix_{model}.png"
if not matrix.exists():
matrix = artifact_dir / f"confusion_matrix_{model}.png"
if not matrix.exists():
raise HTTPException(status_code=404, detail="Matrix artifact not found")
return FileResponse(matrix)
if FRONTEND_DIR.exists():
app.mount("/assets", StaticFiles(directory=FRONTEND_DIR / "assets"), name="assets")
@app.get("/{full_path:path}")
def frontend(full_path: str):
requested = FRONTEND_DIR / full_path
if full_path and requested.is_file():
return FileResponse(requested)
return FileResponse(FRONTEND_DIR / "index.html")
if __name__ == "__main__":
uvicorn.run("src.api.main:app", host="127.0.0.1", port=8000, reload=True)
|