""" MarkItDown & PDF Conversion API — FastAPI server. All routes are under `/api/v1` except root-level `/` and `/ping`. MarkItDown conversion (under /api/v1/markitdown): POST /api/v1/markitdown/convert/file POST /api/v1/markitdown/convert/url POST /api/v1/markitdown/batch/files POST /api/v1/markitdown/batch/urls GET /api/v1/markitdown/formats GET /api/v1/markitdown/spacy-labels PDF-to-image conversion (under /api/v1): POST /api/v1/convert POST /api/v1/convert/url POST /api/v1/convert/async POST /api/v1/convert/async/url GET /api/v1/jobs/{id} GET /api/v1/files/{id}/{fn} GET /api/v1/health GET /api/v1/ready System: GET / Root info GET /ping Liveness check & Docker HEALTHCHECK target GET /docs Swagger UI GET /redoc ReDoc UI GET /metrics Prometheus metrics Nested-JSON key extractor (under /api/v1/keys): POST /api/v1/keys/extract """ from __future__ import annotations import asyncio import concurrent.futures import json import os import time import uuid from contextlib import asynccontextmanager from pathlib import Path from typing import Annotated, Any, Dict, List, Optional from urllib.parse import urlparse import httpx import structlog from fastapi import APIRouter, Depends, FastAPI, File, Form, HTTPException, Request, UploadFile, status from fastapi.responses import JSONResponse, PlainTextResponse from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.gzip import GZipMiddleware from prometheus_client import make_asgi_app from pydantic import BaseModel, Field, field_validator from app.api.routes import router as pdf_router from app.core.auth import require_api_key from app.core.config import settings from app.core.exceptions import AppException from app.core.logging import configure_logging from app.core.rate_limit import RateLimitMiddleware from app.services.ping import start_self_ping from app.utils.cleanup import cleanup_loop from core import ConversionError, ConversionResult, DocumentConverter, SUPPORTED_EXTENSIONS from extraction.generic_json_extractor import extract from extraction.keys_extractor import Extractor as KeysExtractor from logger import get_logger configure_logging() logger = get_logger(__name__) APP_NAME = "reconciliation-file-processing-service" APP_VERSION = "2.2.0" MAX_UPLOAD_BYTES = int(os.getenv("MAX_UPLOAD_BYTES", str(100 * 1024 * 1024))) MAX_WORKERS = min(32, (os.cpu_count() or 1) + 4) _thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) _converter = DocumentConverter() logger.info("thread_pool_initialised", workers=MAX_WORKERS) # --------------------------------------------------------------------------- # Lifespan # --------------------------------------------------------------------------- @asynccontextmanager async def lifespan(app: FastAPI): logger.info("api_starting", version=APP_VERSION, host="0.0.0.0:7860") ping_task = await start_self_ping() cleanup_task = asyncio.create_task(cleanup_loop()) yield if ping_task: ping_task.cancel() cleanup_task.cancel() logger.info("api_shutting_down") # --------------------------------------------------------------------------- # Application # --------------------------------------------------------------------------- app = FastAPI( title="MarkItDown & PDF Conversion API", description=( "Unified API for document-to-Markdown conversion (Microsoft MarkItDown + RapidOCR) " "and PDF-to-image conversion with async job support." ), version=APP_VERSION, docs_url="/docs", redoc_url="/redoc", lifespan=lifespan, ) # -- Middleware (order matters: outermost first) -- app.add_middleware(RateLimitMiddleware) app.add_middleware(GZipMiddleware, minimum_size=1000) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) @app.middleware("http") async def request_context_middleware(request: Request, call_next): request_id = str(uuid.uuid4()) start = time.perf_counter() structlog.contextvars.clear_contextvars() structlog.contextvars.bind_contextvars( request_id=request_id, method=request.method, path=request.url.path, ) response = await call_next(request) elapsed = round((time.perf_counter() - start) * 1000, 2) logger.info("request_completed", status=response.status_code, duration_ms=elapsed) response.headers["X-Request-ID"] = request_id response.headers["X-Response-Time-Ms"] = str(elapsed) return response # -- Exception handlers -- def _request_id(request: Request) -> str: return getattr(request.state, "request_id", "") @app.exception_handler(AppException) async def app_exception_handler(request: Request, exc: AppException): logger.warning("app_exception", detail=exc.detail, status_code=exc.status_code) return JSONResponse( status_code=exc.status_code, content={"error": exc.detail, "request_id": _request_id(request)}, ) @app.exception_handler(Exception) async def generic_exception_handler(request: Request, exc: Exception): logger.exception("unhandled_exception", exc_info=exc) return JSONResponse( status_code=500, content={"error": "Internal server error", "request_id": _request_id(request)}, ) # -- Root-level routes (only / , /ping) -- @app.get("/", tags=["System"], summary="Root", include_in_schema=False) async def root(): return {"service": APP_NAME, "version": APP_VERSION, "status": "running"} @app.get("/ping", tags=["System"], summary="Ping", include_in_schema=False) async def ping(): return {"message": f"{APP_NAME} is running..."} # --------------------------------------------------------------------------- # MarkItDown router — all routes under /api/v1/markitdown # --------------------------------------------------------------------------- markitdown_router = APIRouter(dependencies=[Depends(require_api_key)]) # -- Models -- class ConversionMetadata(BaseModel): source: str char_count: int word_count: int line_count: int file_size_bytes: int mime_type: str content_hash: str token_estimate: int class ConversionResponse(BaseModel): success: bool time_ms: float content: str return_json: bool = False json_content: Optional[Any] = None metadata: Optional[ConversionMetadata] = None error_message: Optional[str] = None class UrlRequest(BaseModel): url: str return_json: bool = False mappings: Optional[Dict[str, Dict[str, Any]]] = None model_config = {"populate_by_name": True} @field_validator("url") @classmethod def validate_scheme(cls, v: str) -> str: if not v.startswith(("http://", "https://")): raise ValueError("Only http/https URLs are supported.") return v class BatchUrlRequest(BaseModel): urls: List[str] @field_validator("urls") @classmethod def validate_urls(cls, v: List[str]) -> List[str]: for url in v: if not url.startswith(("http://", "https://")): raise ValueError(f"Invalid URL scheme: {url}") if len(v) > 20: raise ValueError("Maximum 20 URLs per batch request.") return v class BatchFileResult(BaseModel): filename: str success: bool time_ms: float content: Optional[str] = None error: Optional[str] = None metadata: Optional[ConversionMetadata] = None class BatchResponse(BaseModel): total: int succeeded: int failed: int total_time_ms: float results: List[BatchFileResult] # -- Helpers -- def _build_metadata(result: ConversionResult) -> ConversionMetadata: return ConversionMetadata( source=result.source, char_count=result.char_count, word_count=result.word_count, line_count=result.line_count, file_size_bytes=result.file_size_bytes, mime_type=result.mime_type, content_hash=result.content_hash, token_estimate=result.token_estimate, ) async def _build_response( result: ConversionResult, *, return_json: bool = False, filename: Optional[str] = None, raw_data: Optional[bytes] = None, mappings: Optional[Dict[str, Dict[str, Any]]] = None, ) -> ConversionResponse: json_content: Optional[Any] = None error_message: Optional[str] = None if return_json and filename: loop = asyncio.get_running_loop() json_result = await loop.run_in_executor( _thread_pool, extract, filename, result.markdown, mappings, raw_data ) if "error" in json_result: error_message = json_result["error"] else: json_content = json_result return ConversionResponse( success=True, time_ms=round(result.duration_ms, 3), content=result.markdown, return_json=return_json, json_content=json_content, metadata=_build_metadata(result), error_message=error_message, ) def _raise_for_error(outcome: ConversionError) -> None: status_map = { "FileNotFoundError": status.HTTP_404_NOT_FOUND, "ValueError": status.HTTP_422_UNPROCESSABLE_ENTITY, "PermissionError": status.HTTP_403_FORBIDDEN, } code = status_map.get(outcome.error_type, status.HTTP_500_INTERNAL_SERVER_ERROR) raise HTTPException( status_code=code, detail={ "success": False, "error_type": outcome.error_type, "message": outcome.message, "time_ms": round(outcome.duration_ms, 3), }, ) def _batch_result_from_error(name: str, err: ConversionError) -> BatchFileResult: return BatchFileResult( filename=name, success=False, time_ms=round(err.duration_ms, 3), error=err.message, ) def _batch_result_from_ok(result: ConversionResult) -> BatchFileResult: return BatchFileResult( filename=result.source, success=True, time_ms=round(result.duration_ms, 3), content=result.markdown, metadata=_build_metadata(result), ) # -- Routes -- @markitdown_router.get("/formats", tags=["MarkItDown"], summary="Supported file formats by category") async def list_formats(): by_category = { "documents": [e for e in SUPPORTED_EXTENSIONS if e in {".pdf", ".docx", ".doc", ".epub"}], "office": [e for e in SUPPORTED_EXTENSIONS if e in {".pptx", ".ppt", ".xlsx", ".xls"}], "data": [e for e in SUPPORTED_EXTENSIONS if e in {".csv", ".json", ".xml"}], "web": [e for e in SUPPORTED_EXTENSIONS if e in {".htm", ".html"}], "text": [e for e in SUPPORTED_EXTENSIONS if e in {".txt", ".md", ".rst"}], "images": [e for e in SUPPORTED_EXTENSIONS if e in {".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp", ".tiff"}], "audio": [e for e in SUPPORTED_EXTENSIONS if e in {".mp3", ".wav", ".ogg", ".flac"}], "archives": [e for e in SUPPORTED_EXTENSIONS if e in {".zip"}], } return { "success": True, "total_count": len(SUPPORTED_EXTENSIONS), "all_extensions": sorted(SUPPORTED_EXTENSIONS), "by_category": {k: sorted(v) for k, v in by_category.items()}, } @markitdown_router.get("/spacy-labels", tags=["MarkItDown"], summary="Available spaCy NER labels") async def list_spacy_labels(): from extraction.spacy_extractor import VALID_SPACY_LABELS return { "success": True, "spacy_labels": VALID_SPACY_LABELS, "source_types": { "entity": "Extract using spaCy NER labels (ORG, PERSON, DATE, etc.)", "regex": "Extract using custom regular expressions", "token_attr": "Extract using token attributes (text, pos_, tag_, etc.)", }, "example_mappings": { "company": {"source_type": "entity", "label": "ORG"}, "person": {"source_type": "entity", "label": "PERSON"}, "date": {"source_type": "entity", "label": "DATE"}, "money": {"source_type": "entity", "label": "MONEY"}, "email": {"source_type": "regex", "pattern": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"}, "phone": {"source_type": "regex", "pattern": r"\b\d{3}-\d{3}-\d{4}\b"}, }, } @markitdown_router.post( "/convert/file", response_model=ConversionResponse, tags=["MarkItDown"], summary="Convert an uploaded file to Markdown", ) async def convert_file( file: Annotated[UploadFile, File(description="File to convert")], plain_text: bool = Form(False), return_json: bool = Form(False), mappings: Optional[str] = Form( None, description='JSON string defining spaCy extraction rules. Example: {"company": {"source_type": "entity", "label": "ORG"}}', ), ): if file is None: raise HTTPException(status_code=400, detail={"success": False, "message": "No file provided."}) parsed_mappings: Optional[Dict[str, Any]] = None if mappings: try: parsed_mappings = json.loads(mappings) except json.JSONDecodeError: raise HTTPException(status_code=400, detail={"success": False, "message": "Invalid JSON in mappings parameter."}) logger.info("convert_file", filename=file.filename) raw = await file.read() if len(raw) > MAX_UPLOAD_BYTES: logger.warning("convert_file | file too large", filename=file.filename, size=len(raw)) raise HTTPException(status_code=413, detail={"success": False, "message": "File exceeds 100 MB limit."}) loop = asyncio.get_running_loop() outcome = await loop.run_in_executor(_thread_pool, _converter.convert_stream, raw, file.filename or "upload") if isinstance(outcome, ConversionError): logger.error("convert_file | conversion failed", filename=file.filename, error=outcome.message) _raise_for_error(outcome) logger.info("convert_file | success", filename=file.filename, chars=outcome.char_count, time_ms=round(outcome.duration_ms, 1)) if plain_text: return PlainTextResponse(outcome.markdown) return await _build_response(outcome, return_json=return_json, filename=file.filename, raw_data=raw, mappings=parsed_mappings) @markitdown_router.post( "/convert/url", response_model=ConversionResponse, tags=["MarkItDown"], summary="Convert a public URL to Markdown", ) async def convert_url(body: UrlRequest): logger.info("convert_url", url=body.url) parsed = urlparse(body.url) filename = Path(parsed.path).name or "url_content" loop = asyncio.get_running_loop() if body.return_json: try: async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client: resp = await client.get(body.url) resp.raise_for_status() except httpx.HTTPError as exc: logger.error("convert_url | fetch failed", url=body.url, error=str(exc)) raise HTTPException(status_code=400, detail={"success": False, "message": f"Failed to fetch URL: {exc}"}) raw_data = resp.content if len(raw_data) > MAX_UPLOAD_BYTES: raise HTTPException(status_code=413, detail={"success": False, "message": "File exceeds 100 MB limit."}) outcome = await loop.run_in_executor(_thread_pool, _converter.convert_stream, raw_data, filename) if isinstance(outcome, ConversionError): logger.error("convert_url | conversion failed", url=body.url, error=outcome.message) _raise_for_error(outcome) logger.info("convert_url | success", url=body.url, chars=outcome.char_count, time_ms=round(outcome.duration_ms, 1)) return await _build_response(outcome, return_json=body.return_json, filename=filename, raw_data=raw_data, mappings=body.mappings) outcome = await loop.run_in_executor(_thread_pool, _converter.convert_url, body.url) if isinstance(outcome, ConversionError): logger.error("convert_url | conversion failed", url=body.url, error=outcome.message) _raise_for_error(outcome) logger.info("convert_url | success", url=body.url, chars=outcome.char_count, time_ms=round(outcome.duration_ms, 1)) return await _build_response(outcome, return_json=body.return_json, filename=filename, mappings=body.mappings) @markitdown_router.post( "/batch/files", response_model=BatchResponse, tags=["MarkItDown"], summary="Convert multiple files (up to 10)", ) async def batch_files( files: Annotated[List[UploadFile], File(description="Files to convert — maximum 10")], ): if not files: raise HTTPException(status_code=400, detail={"success": False, "message": "No files provided."}) if len(files) > 10: raise HTTPException(status_code=400, detail={"success": False, "message": "Maximum 10 files per batch."}) batch_start = time.perf_counter() logger.info("batch_files", count=len(files)) async def _process_file(f: UploadFile) -> BatchFileResult: if f is None: return BatchFileResult(filename="unknown", success=False, time_ms=0, error="File object is None.") raw = await f.read() if len(raw) > MAX_UPLOAD_BYTES: return BatchFileResult(filename=f.filename or "unknown", success=False, time_ms=0, error="File exceeds 100 MB limit.") loop = asyncio.get_running_loop() outcome = await loop.run_in_executor(_thread_pool, _converter.convert_stream, raw, f.filename or "upload") return _batch_result_from_error(f.filename or "unknown", outcome) if isinstance(outcome, ConversionError) else _batch_result_from_ok(outcome) results = await asyncio.gather(*[_process_file(f) for f in files]) total_ms = round((time.perf_counter() - batch_start) * 1000, 3) succeeded = sum(1 for r in results if r.success) logger.info("batch_files | done", succeeded=succeeded, failed=len(results) - succeeded, total_ms=round(total_ms, 1)) return BatchResponse(total=len(results), succeeded=succeeded, failed=len(results) - succeeded, total_time_ms=total_ms, results=results) @markitdown_router.post( "/batch/urls", response_model=BatchResponse, tags=["MarkItDown"], summary="Convert multiple URLs (up to 20)", ) async def batch_urls(body: BatchUrlRequest): batch_start = time.perf_counter() logger.info("batch_urls", count=len(body.urls)) async def _process_url(url: str) -> BatchFileResult: loop = asyncio.get_running_loop() outcome = await loop.run_in_executor(_thread_pool, _converter.convert_url, url) return _batch_result_from_error(url, outcome) if isinstance(outcome, ConversionError) else _batch_result_from_ok(outcome) results = await asyncio.gather(*[_process_url(url) for url in body.urls]) total_ms = round((time.perf_counter() - batch_start) * 1000, 3) succeeded = sum(1 for r in results if r.success) logger.info("batch_urls | done", succeeded=succeeded, failed=len(results) - succeeded, total_ms=round(total_ms, 1)) return BatchResponse(total=len(results), succeeded=succeeded, failed=len(results) - succeeded, total_time_ms=total_ms, results=results) # --------------------------------------------------------------------------- # Keys extractor router — nested-JSON key/value lookup # --------------------------------------------------------------------------- keys_router = APIRouter(dependencies=[Depends(require_api_key)]) class KeysExtractRequest(BaseModel): data: Any = Field(..., description="JSON object or array to search") key_names: List[str] = Field(..., min_length=1, description="Key names to look up at any depth") result_limit: Optional[int] = Field( None, ge=1, description="Maximum values to return per key. Omit (or pass null) to return the full, uncapped result.", ) @field_validator("key_names") @classmethod def _validate_key_names(cls, v: List[str]) -> List[str]: for kn in v: if not isinstance(kn, str) or not kn: raise ValueError("each key_name must be a non-empty string") return v class KeysExtractResponse(BaseModel): success: bool time_ms: float data: Dict[str, List[Any]] error_message: Optional[str] = None @keys_router.post( "/extract", response_model=KeysExtractResponse, tags=["Keys Extractor"], summary="Extract all values for given key names from a nested JSON object", ) async def extract_keys(body: KeysExtractRequest): """Recursively walk any JSON object/array and return every value attached to the supplied key names, regardless of how deeply nested they are.""" start = time.perf_counter() logger.info( "keys_extract | start", key_count=len(body.key_names), limit=body.result_limit, ) if not isinstance(body.data, (dict, list)): return KeysExtractResponse( success=False, time_ms=round((time.perf_counter() - start) * 1000, 3), data={}, error_message="`data` must be a JSON object or array", ) try: loop = asyncio.get_running_loop() results = await loop.run_in_executor( _thread_pool, KeysExtractor(body.data, body.key_names, body.result_limit).extract, ) except (TypeError, ValueError) as exc: logger.warning("keys_extract | invalid input", error=str(exc)) return KeysExtractResponse( success=False, time_ms=round((time.perf_counter() - start) * 1000, 3), data={}, error_message=str(exc), ) except Exception as exc: logger.exception("keys_extract | failed") return KeysExtractResponse( success=False, time_ms=round((time.perf_counter() - start) * 1000, 3), data={}, error_message=f"extraction failed: {exc}", ) elapsed = round((time.perf_counter() - start) * 1000, 3) total_values = sum(len(v) for v in results.values()) logger.info( "keys_extract | done", keys=len(body.key_names), total_values=total_values, time_ms=elapsed, ) return KeysExtractResponse( success=True, time_ms=elapsed, data=results, error_message=None, ) # --------------------------------------------------------------------------- # Include all routers # --------------------------------------------------------------------------- app.include_router(markitdown_router, prefix="/api/v1/markitdown") app.include_router(pdf_router, prefix="/api/v1") app.include_router(keys_router, prefix="/api/v1/keys") # -- Mount Prometheus metrics -- app.mount("/metrics", make_asgi_app()) # --------------------------------------------------------------------------- # Server runner # --------------------------------------------------------------------------- def run_server(host: str = "0.0.0.0", port: int = 7860, reload: bool = False) -> None: import uvicorn uvicorn.run("api.server:app", host=host, port=port, reload=reload)