import os import io import re import json import zipfile import asyncio import subprocess import tempfile import threading from datetime import datetime, timezone from pathlib import Path import openpyxl from openpyxl.styles import Font, PatternFill, Alignment from huggingface_hub import HfApi, hf_hub_download from fastapi import FastAPI, File, UploadFile, Form, HTTPException, Security from fastapi.responses import JSONResponse from fastapi.security.api_key import APIKeyHeader import httpx from sarvamai import SarvamAI app = FastAPI( title="Document Extraction API", description="Extract structured data from documents using Sarvam Vision", version="3.0.0" ) # --------------------------------------------------------------------------- # Supported document types # --------------------------------------------------------------------------- SUPPORTED_TYPES = { "income_certificate": "Income certificate issued by a government authority", "caste_certificate": "Caste certificate issued by a government authority", "domicile_certificate": "Domicile / residence certificate issued by a government authority", "ssc_certificate": "SSC (Secondary School Certificate) / Class 10 marksheet or passing certificate issued by an education board", "hsc_certificate": "HSC (Higher Secondary Certificate) / Class 12 marksheet or passing certificate issued by an education board", } SARVAM_API_KEY = os.environ.get("SARVAM_API_KEY", "") API_SECRET_KEY = os.environ.get("API_SECRET_KEY", "") HF_TOKEN = os.environ.get("HF_TOKEN", "") # --------------------------------------------------------------------------- # Logging config — edit these two constants to match your HF dataset repo # --------------------------------------------------------------------------- LOG_REPO_ID = "PrathameshRaut/VisionModelLogs" LOG_FILENAME = "extraction_logs.xlsx" LOG_HEADERS = [ "timestamp_utc", "filename", "file_size_bytes", "file_type", "document_type", "parameters_requested", "model_raw_thinking", "extracted_json", "status", "error_detail", ] # A threading lock so concurrent requests don't corrupt the xlsx _log_lock = threading.Lock() # --------------------------------------------------------------------------- # Auth — X-API-Key header guard # --------------------------------------------------------------------------- api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False) async def verify_api_key(key: str = Security(api_key_header)): if not API_SECRET_KEY: raise HTTPException(status_code=500, detail="API_SECRET_KEY is not configured on the server") if not key or key != API_SECRET_KEY: raise HTTPException( status_code=401, detail="Invalid or missing API key. Send it as request header: X-API-Key: " ) # --------------------------------------------------------------------------- # Excel log helpers # --------------------------------------------------------------------------- def _style_header_row(ws): header_fill = PatternFill("solid", start_color="1F4E79") header_font = Font(bold=True, color="FFFFFF", name="Arial", size=10) for cell in ws[1]: cell.fill = header_fill cell.font = header_font cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True) def _download_or_create_workbook() -> openpyxl.Workbook: """Download the existing log workbook from HF, or create a fresh one.""" if not HF_TOKEN: raise RuntimeError("HF_TOKEN secret is not set — cannot write logs") try: local_path = hf_hub_download( repo_id=LOG_REPO_ID, filename=LOG_FILENAME, repo_type="dataset", token=HF_TOKEN, force_download=True, ) wb = openpyxl.load_workbook(local_path) except Exception: # File doesn't exist yet — start fresh wb = openpyxl.Workbook() ws = wb.active ws.title = "Logs" ws.append(LOG_HEADERS) _style_header_row(ws) col_widths = { "A": 22, "B": 30, "C": 16, "D": 14, "E": 22, "F": 35, "G": 60, "H": 60, "I": 12, "J": 40, } for col, width in col_widths.items(): ws.column_dimensions[col].width = width ws.row_dimensions[1].height = 28 return wb def _upload_workbook(wb: openpyxl.Workbook): """Save workbook to a buffer and push to HF dataset repo.""" buf = io.BytesIO() wb.save(buf) buf.seek(0) api = HfApi(token=HF_TOKEN) api.upload_file( path_or_fileobj=buf, path_in_repo=LOG_FILENAME, repo_id=LOG_REPO_ID, repo_type="dataset", commit_message=f"Add log entry {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S')} UTC", ) def append_log( filename: str, file_size: int, file_type: str, document_type: str, parameters: list[str], raw_thinking: str, extracted_json: dict | None, status: str, error_detail: str = "", ): """Thread-safe: download → append row → upload.""" if not HF_TOKEN: return row = [ datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S"), filename, file_size, file_type, document_type, ", ".join(parameters), (raw_thinking or "")[:5000], json.dumps(extracted_json, ensure_ascii=False) if extracted_json else "", status, error_detail[:1000], ] with _log_lock: try: wb = _download_or_create_workbook() ws = wb.active ws.append(row) data_font = Font(name="Arial", size=10) wrap_align = Alignment(vertical="top", wrap_text=True) row_idx = ws.max_row for cell in ws[row_idx]: cell.font = data_font cell.alignment = wrap_align if row_idx % 2 == 0: fill = PatternFill("solid", start_color="DCE6F1") for cell in ws[row_idx]: cell.fill = fill _upload_workbook(wb) except Exception as e: print(f"[LOG WARNING] Failed to write log entry: {e}") # --------------------------------------------------------------------------- # File conversion — DOCX/PPTX → PDF via LibreOffice; images/PDFs pass through # --------------------------------------------------------------------------- NATIVE_EXTENSIONS = {".pdf", ".png", ".jpg", ".jpeg"} NATIVE_MIME = {"application/pdf", "image/png", "image/jpeg", "image/jpg"} def prepare_file(file_bytes: bytes, content_type: str, filename: str) -> tuple[bytes, str]: suffix = Path(filename).suffix.lower() if suffix in NATIVE_EXTENSIONS or content_type in NATIVE_MIME: return file_bytes, filename with tempfile.TemporaryDirectory() as tmpdir: src = os.path.join(tmpdir, "input" + suffix) with open(src, "wb") as f: f.write(file_bytes) result = subprocess.run( ["libreoffice", "--headless", "--convert-to", "pdf", "--outdir", tmpdir, src], capture_output=True, timeout=180 ) if result.returncode != 0: raise HTTPException( status_code=500, detail=f"File conversion to PDF failed: {result.stderr.decode()}" ) pdfs = [f for f in os.listdir(tmpdir) if f.endswith(".pdf")] if not pdfs: raise HTTPException(status_code=500, detail="PDF conversion produced no output") out_path = os.path.join(tmpdir, pdfs[0]) with open(out_path, "rb") as f: return f.read(), Path(filename).stem + ".pdf" # --------------------------------------------------------------------------- # Text cleanup — strip base64-encoded image blobs from OCR output # --------------------------------------------------------------------------- def clean_raw_text(text: str) -> str: # Pattern 1 — markdown image embed text = re.sub( r'!\[[^\]]*\]\(data:[a-zA-Z]+/[a-zA-Z+\-]+;base64,[A-Za-z0-9+/=\s]+\)', '', text, flags=re.DOTALL, ) # Pattern 2 — bare data URI text = re.sub( r'data:[a-zA-Z]+/[a-zA-Z+\-]+;base64,[A-Za-z0-9+/=\s]{100,}', '', text, flags=re.DOTALL, ) return text.strip() # --------------------------------------------------------------------------- # Step 1: Extract raw text via Sarvam Document Intelligence SDK # --------------------------------------------------------------------------- def extract_text_with_sdk(file_bytes: bytes, filename: str) -> str: if not SARVAM_API_KEY: raise HTTPException(status_code=500, detail="SARVAM_API_KEY is not configured on the server") # FIX: increased timeout so the polling loop doesn't give up too early client = SarvamAI( api_subscription_key=SARVAM_API_KEY, httpx_client=httpx.Client(timeout=180), ) suffix = Path(filename).suffix.lower() with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp: tmp.write(file_bytes) tmp_path = tmp.name zip_path = tmp_path + "_output.zip" try: job = client.document_intelligence.create_job(language="en-IN", output_format="md") job.upload_file(tmp_path) job.start() status = job.wait_until_complete() if status.job_state not in ("Completed", "PartiallyCompleted"): raise HTTPException( status_code=500, detail=f"Sarvam document processing ended with state: {status.job_state}" ) job.download_output(zip_path) with zipfile.ZipFile(zip_path, "r") as z: md_files = sorted(n for n in z.namelist() if n.endswith(".md")) if not md_files: raise HTTPException(status_code=500, detail="No markdown output found in Sarvam result") full_text = "\n\n".join( z.read(name).decode("utf-8", errors="replace") for name in md_files ) return full_text finally: for path in (tmp_path, zip_path): try: os.unlink(path) except FileNotFoundError: pass # --------------------------------------------------------------------------- # Step 2: Extract structured JSON via SarvamAI SDK chat completions # --------------------------------------------------------------------------- def strip_think_tags(text: str) -> str: text = re.sub(r".*?", "", text, flags=re.DOTALL) text = re.sub(r".*$", "", text, flags=re.DOTALL) return text.strip() async def extract_json_via_chat(doc_type: str, raw_text: str, parameters: list[str]) -> dict: if not SARVAM_API_KEY: raise HTTPException(status_code=500, detail="SARVAM_API_KEY is not configured on the server") description = SUPPORTED_TYPES[doc_type] param_list = "\n".join(f" - {p}" for p in parameters) # Truncate to avoid blowing context window raw_text = raw_text[:6000] if len(raw_text) > 6000 else raw_text prompt = f"""You are a document data extraction assistant. Below is the full text extracted from a {description}. --- DOCUMENT TEXT START --- {raw_text} --- DOCUMENT TEXT END --- Extract the following fields from the document text above: {param_list} Rules: 1. Return ONLY a valid JSON object. No explanations, no markdown, no code fences. 2. Use exactly the field names listed above as JSON keys. 3. If a field is not found or not legible, set its value to null. 4. Do not invent or infer values not explicitly present in the document. 5. Respond everything in english language only. Also convert date into proper DD/MM/YYYY format. Respond with the JSON object only.""" def _call_sdk() -> str: # FIX: increased timeout — sarvam-105b reasoning can take >60s client = SarvamAI( api_subscription_key=SARVAM_API_KEY, httpx_client=httpx.Client(timeout=180), ) # FIX: retry up to 3 times — 105b occasionally returns null content for attempt in range(3): response = client.chat.completions( model="sarvam-105b", messages=[{"role": "user", "content": prompt}], reasoning_effort=None, temperature=0, top_p=1, max_tokens=4096, ) # FIX: guard against null content before calling .strip() content = response.choices[0].message.content if content: return content.strip() print( f"[WARN] Attempt {attempt + 1}/3: Sarvam returned null content " f"(finish_reason={response.choices[0].finish_reason!r}), retrying..." ) raise ValueError("Sarvam returned null content after 3 attempts") try: raw = await asyncio.to_thread(_call_sdk) except ValueError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc except Exception as exc: raise HTTPException(status_code=500, detail=f"Sarvam chat error: {exc}") from exc # Remove blocks raw = strip_think_tags(raw) # Strip markdown code fences if present if raw.startswith("```"): raw = raw.split("\n", 1)[-1] if raw.endswith("```"): raw = raw[: raw.rfind("```")] raw = raw.strip() try: return json.loads(raw) except json.JSONDecodeError: return {"raw_response": raw, "parse_error": "Model did not return valid JSON"} # --------------------------------------------------------------------------- # Routes # --------------------------------------------------------------------------- @app.get("/") def root(): return { "message": "Document Extraction API powered by Sarvam Vision", "supported_document_types": list(SUPPORTED_TYPES.keys()), "docs": "/docs" } @app.get("/types") def get_supported_types(): return {"supported_types": {k: {"description": v} for k, v in SUPPORTED_TYPES.items()}} @app.post("/extract", dependencies=[Security(verify_api_key)]) async def extract_document( file: UploadFile = File(..., description="Document file (image, PDF, DOCX, PPTX, etc.)"), document_type: str = Form(..., description="One of: income_certificate, caste_certificate, domicile_certificate"), parameters: str = Form(..., description="Comma-separated field names to extract. E.g: name,income,date_of_issue") ): """ Extract structured JSON data from an uploaded document. Requires header: **X-API-Key: your-secret-key** """ if document_type not in SUPPORTED_TYPES: raise HTTPException( status_code=400, detail={ "error": f"Unsupported document_type: '{document_type}'", "supported_types": list(SUPPORTED_TYPES.keys()) } ) param_list = [p.strip() for p in parameters.split(",") if p.strip()] if not param_list: raise HTTPException( status_code=400, detail="'parameters' must contain at least one field name (e.g. 'name,income,date_of_issue')" ) file_bytes = await file.read() if not file_bytes: raise HTTPException(status_code=400, detail="Uploaded file is empty") content_type = file.content_type or "" filename = file.filename or "document" file_size = len(file_bytes) raw_text = "" extracted = {} try: processed_bytes, processed_name = prepare_file(file_bytes, content_type, filename) except HTTPException as exc: asyncio.get_event_loop().run_in_executor( None, append_log, filename, file_size, content_type, document_type, param_list, "", None, "error", str(exc.detail) ) raise try: raw_text = await asyncio.to_thread(extract_text_with_sdk, processed_bytes, processed_name) except HTTPException as exc: asyncio.get_event_loop().run_in_executor( None, append_log, filename, file_size, content_type, document_type, param_list, "", None, "error", str(exc.detail) ) raise # Strip base64 image blobs before sending to chat model raw_text = clean_raw_text(raw_text) try: extracted = await extract_json_via_chat(document_type, raw_text, param_list) except HTTPException as exc: asyncio.get_event_loop().run_in_executor( None, append_log, filename, file_size, content_type, document_type, param_list, raw_text, None, "error", str(exc.detail) ) raise # Fire-and-forget logging (don't block the response) asyncio.get_event_loop().run_in_executor( None, append_log, filename, file_size, content_type, document_type, param_list, raw_text, extracted, "success", "" ) return JSONResponse(content={ "document_type": document_type, "filename": filename, "parameters_requested": param_list, "extracted_data": extracted }) @app.get("/health") def health(): return {"status": "ok"}