feat: updated

.dockerignore

@@ -0,0 +1,27 @@
+__pycache__
+*.pyc
+*.pyo
+*.pyd
+.git
+.gitignore
+.env
+.venv
+venv
+*.log
+logs
+.vscode
+.idea
+.claude
+__pycache__
+.pytest_cache
+.ruff_cache
+.mypy_cache
+*.md
+!README.md
+docs
+tmp
+temp
+.DS_Store
+Thumbs.db
+*.swp
+*.swo

Dockerfile

@@ -1,54 +1,55 @@
 # ─────────────────────────────────────────────────────────────
-# MarkItDown API — Production Dockerfile
+# Reconciliation File Processing Service — Production Dockerfile
 # Port: 7860
-# Uses Microsoft MarkItDown and RapidOCR for document processing.
-# JSON extraction: pandas for CSV/XLS/XLSX, spaCy NER for all other formats.
 # ─────────────────────────────────────────────────────────────
 
 FROM python:3.12-slim
 
-LABEL maintainer="MarkItDown API"
-LABEL description="Document-to-Markdown API with Microsoft MarkItDown and RapidOCR"
+LABEL maintainer="Reconciliation API"
+LABEL description="Document-to-Markdown & PDF-to-image API"
 LABEL version="2.2.0"
 
-# ── System dependencies ───────────────────────────────────────
+# ── System dependencies ──────────────────────────────────────
 RUN apt-get update && apt-get install -y --no-install-recommends \
     curl \
     ffmpeg \
     libmagic1 \
+    poppler-utils \
     && rm -rf /var/lib/apt/lists/*
 
-# ── Non-root user ─────────────────────────────────────────────
+# ── Non-root user ────────────────────────────────────────────
 RUN groupadd --gid 1000 appuser && \
     useradd --uid 1000 --gid appuser --shell /bin/bash --create-home appuser
 
 WORKDIR /app
 
-# ── Python base dependencies ──────────────────────────────────
+# ── Python dependencies ──────────────────────────────────────
 COPY requirements.txt .
 RUN pip install --no-cache-dir --upgrade pip && \
     pip install --no-cache-dir -r requirements.txt && \
     python -m spacy download en_core_web_sm
 
-# ── Application code ──────────────────────────────────────────
+# ── Application code ─────────────────────────────────────────
 COPY --chown=appuser:appuser . .
 
-# ── Ensure startup script exists and is executable ────────────
+# ── Ensure startup script exists and is executable ──────────
 RUN test -f /app/start.sh || (echo "ERROR: start.sh not found" && exit 1) && \
     chmod +x /app/start.sh
 
-# ── Create necessary directories with proper permissions ───────
-RUN mkdir -p /app/logs && \
-    chown -R appuser:appuser /app/logs
+# ── Create runtime directories ──────────────────────────────
+RUN mkdir -p /app/logs /tmp/pdf2img_outputs /tmp/pdf2img_temp && \
+    chown -R appuser:appuser /app/logs /tmp/pdf2img_outputs /tmp/pdf2img_temp
 
 USER appuser
 
 ENV PYTHONPATH=/app
 ENV PYTHONUNBUFFERED=1
+ENV OUTPUT_DIR=/tmp/pdf2img_outputs
+ENV TEMP_DIR=/tmp/pdf2img_temp
 
 EXPOSE 7860
 
 HEALTHCHECK --interval=30s --timeout=10s --start-period=30s --retries=3 \
-    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:7860/health')" || exit 1
+    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:7860/ping')" || exit 1
 
-CMD ["/bin/bash", "/app/start.sh"]
+CMD ["/bin/bash", "/app/start.sh"]

api/server.py

@@ -1,30 +1,43 @@
 """
-MarkItDown API — FastAPI server.
-
-This module defines the FastAPI application, all request/response models,
-route handlers, and the application lifespan. There is no browser-facing UI;
-the application is a pure REST API intended for programmatic consumption.
-
-Routes
-------
-    POST /convert/file      Convert an uploaded file to Markdown.
-    POST /convert/url       Convert a public URL to Markdown.
-    POST /batch/files       Convert up to 10 files in a single request.
-    POST /batch/urls        Convert up to 20 URLs in a single request.
-    GET  /health            Liveness check returning uptime and version.
-    GET  /info              Server metadata (version, platform, limits).
-    GET  /formats           Supported file extensions grouped by category.
-    GET  /spacy-labels      Available spaCy NER labels for field extraction.
+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
+  GET  /api/v1/ping
+
+System:
+  GET  /              Root info
+  GET  /ping          Liveness check & Docker HEALTHCHECK target
+  GET  /docs          Swagger UI
+  GET  /redoc         ReDoc UI
+  GET  /metrics       Prometheus metrics
 """
 
 from __future__ import annotations
 
 import asyncio
 import concurrent.futures
-import datetime
 import os
 import threading
 import time
+import uuid
 import urllib.request
 from contextlib import asynccontextmanager
 from datetime import datetime, timezone
@@ -33,27 +46,31 @@ from typing import Annotated, Any, Dict, List, Optional
 from urllib.parse import urlparse
 
 import httpx
-import uvicorn
-from fastapi import FastAPI, File, Form, HTTPException, UploadFile, status
-from fastapi.responses import PlainTextResponse
+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 core import ConversionError, ConversionResult, DocumentConverter, SUPPORTED_EXTENSIONS
 from extraction.generic_json_extractor import extract
-from extraction.label_mapper import validate_mappings
 from logger import get_logger
-
+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
+
+configure_logging()
 logger = get_logger(__name__)
 
 _START_TIME = time.time()
 APP_NAME = "reconciliation-file-processing-service"
 
-# Maximum accepted upload size (100 MB).
 MAX_UPLOAD_BYTES = 100 * 1024 * 1024
 
-# Thread pool for CPU-bound conversion work running alongside the async event loop.
 MAX_WORKERS = min(32, (os.cpu_count() or 1) + 4)
 _thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS)
 
@@ -67,11 +84,10 @@ logger.info("Thread pool initialised with %d workers", MAX_WORKERS)
 # ---------------------------------------------------------------------------
 
 PING_URL = os.environ.get("PING_URL", "https://validops-us-data-extract.hf.space/health")
-PING_INTERVAL_SECONDS = 30 * 60  # 30 minutes
+PING_INTERVAL_SECONDS = 30 * 60
 
 
 def _ping_once() -> None:
-    """Send a single HTTP GET to PING_URL and log the outcome."""
     ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
     try:
         with urllib.request.urlopen(PING_URL, timeout=10) as resp:
@@ -81,7 +97,6 @@ def _ping_once() -> None:
 
 
 def _ping_loop() -> None:
-    """Background loop: sleep PING_INTERVAL_SECONDS, ping, repeat."""
     logger.info("self_ping | scheduler started | interval_minutes=30 | url=%s", PING_URL)
     while True:
         time.sleep(PING_INTERVAL_SECONDS)
@@ -89,7 +104,6 @@ def _ping_loop() -> None:
 
 
 def _start_ping_scheduler() -> None:
-    """Start the self-ping daemon thread. Called once from lifespan startup."""
     thread = threading.Thread(target=_ping_loop, name="self-ping", daemon=True)
     thread.start()
 
@@ -100,14 +114,12 @@ def _start_ping_scheduler() -> None:
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    """Application lifespan handler — runs startup and shutdown logic."""
     logger.info(
-        "MarkItDown API starting | version=2.1.0 | host=0.0.0.0:7860 | started_at=%s",
-        datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC"),
+        "MarkItDown & PDF Conversion API starting | version=2.1.0 | host=0.0.0.0:7860",
     )
     _start_ping_scheduler()
     yield
-    logger.info("MarkItDown API shutting down")
+    logger.info("MarkItDown & PDF Conversion API shutting down")
 
 
 # ---------------------------------------------------------------------------
@@ -115,23 +127,19 @@ async def lifespan(app: FastAPI):
 # ---------------------------------------------------------------------------
 
 app = FastAPI(
-    title="MarkItDown API",
+    title="MarkItDown & PDF Conversion API",
     description=(
-        "Document-to-Markdown conversion API powered by Microsoft MarkItDown "
-        "and RapidOCR. Accepts file uploads and public URLs; returns structured "
-        "Markdown with optional JSON field extraction."
+        "Unified API for document-to-Markdown conversion (Microsoft MarkItDown + RapidOCR) "
+        "and PDF-to-image conversion with async job support."
     ),
     version="2.1.0",
     docs_url="/docs",
     redoc_url="/redoc",
-    openapi_tags=[
-        {"name": "Convert", "description": "Single-file or single-URL conversion"},
-        {"name": "Batch",   "description": "Bulk conversion — up to 10 files or 20 URLs"},
-        {"name": "System",  "description": "Health, server info, and supported formats"},
-    ],
     lifespan=lifespan,
 )
 
+# -- Middleware --
+app.add_middleware(RateLimitMiddleware)
 app.add_middleware(GZipMiddleware, minimum_size=1000)
 app.add_middleware(
     CORSMiddleware,
@@ -141,13 +149,66 @@ app.add_middleware(
 )
 
 
+@app.middleware("http")
+async def request_context_middleware(request: Request, call_next):
+    request_id = str(uuid.uuid4())
+    start = time.perf_counter()
+
+    import structlog as _structlog
+    _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=%d | duration_ms=%.2f", response.status_code, elapsed)
+    response.headers["X-Request-ID"] = request_id
+    response.headers["X-Response-Time-Ms"] = str(elapsed)
+    return response
+
+
+# -- Exception handlers --
+@app.exception_handler(AppException)
+async def app_exception_handler(request: Request, exc: AppException):
+    logger.warning("app_exception | detail=%s | status_code=%d", exc.detail, exc.status_code)
+    return JSONResponse(
+        status_code=exc.status_code,
+        content={"error": exc.detail, "request_id": getattr(request.state, "request_id", "")},
+    )
+
+
+@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": getattr(request.state, "request_id", "")},
+    )
+
+
+# -- Root-level routes (only / , /ping , /health) --
+@app.get("/", tags=["System"], summary="Root", include_in_schema=False)
+async def root():
+    return {"service": APP_NAME, "version": "2.1.0", "status": "running"}
+
+
+@app.get("/ping", tags=["System"], summary="Ping", include_in_schema=False)
+async def ping():
+    return {"message": f"{APP_NAME} is running..."}
+
+
 # ---------------------------------------------------------------------------
-# Request / Response models
+# MarkItDown router — all routes under /api/v1/markitdown
 # ---------------------------------------------------------------------------
 
-class ConversionMetadata(BaseModel):
-    """File-level statistics attached to every successful conversion."""
+markitdown_router = APIRouter(dependencies=[Depends(require_api_key)])
 
+# -- Models --
+class ConversionMetadata(BaseModel):
     source: str
     char_count: int
     word_count: int
@@ -159,8 +220,6 @@ class ConversionMetadata(BaseModel):
 
 
 class ConversionResponse(BaseModel):
-    """Standard response envelope for single-item conversion endpoints."""
-
     success: bool
     time_ms: float
     content: str
@@ -171,12 +230,9 @@ class ConversionResponse(BaseModel):
 
 
 class UrlRequest(BaseModel):
-    """Request body for /convert/url."""
-
     url: str
     return_json: bool = False
     mappings: Optional[Dict[str, Dict[str, Any]]] = None
-
     model_config = {"populate_by_name": True}
 
     @field_validator("url")
@@ -188,8 +244,6 @@ class UrlRequest(BaseModel):
 
 
 class BatchUrlRequest(BaseModel):
-    """Request body for /batch/urls."""
-
     urls: List[str]
 
     @field_validator("urls")
@@ -204,8 +258,6 @@ class BatchUrlRequest(BaseModel):
 
 
 class BatchFileResult(BaseModel):
-    """Per-item result within a batch response."""
-
     filename: str
     success: bool
     time_ms: float
@@ -215,8 +267,6 @@ class BatchFileResult(BaseModel):
 
 
 class BatchResponse(BaseModel):
-    """Aggregate response for batch endpoints."""
-
     total: int
     succeeded: int
     failed: int
@@ -224,12 +274,8 @@ class BatchResponse(BaseModel):
     results: List[BatchFileResult]
 
 
-# ---------------------------------------------------------------------------
-# Internal helpers
-# ---------------------------------------------------------------------------
-
+# -- Helpers --
 def _build_metadata(result: ConversionResult) -> ConversionMetadata:
-    """Map a ConversionResult to its API metadata representation."""
     return ConversionMetadata(
         source=result.source,
         char_count=result.char_count,
@@ -250,7 +296,6 @@ async def _build_response(
     raw_data: Optional[bytes] = None,
     mappings: Optional[Dict[str, Dict[str, Any]]] = None,
 ) -> ConversionResponse:
-    """Construct a ConversionResponse, optionally running JSON extraction."""
     json_content: Optional[Any] = None
     error_message: Optional[str] = None
 
@@ -276,7 +321,6 @@ async def _build_response(
 
 
 def _raise_for_error(outcome: ConversionError) -> None:
-    """Translate a ConversionError into an appropriate HTTPException."""
     status_map = {
         "FileNotFoundError": status.HTTP_404_NOT_FOUND,
         "ValueError":        status.HTTP_422_UNPROCESSABLE_ENTITY,
@@ -313,28 +357,14 @@ def _batch_result_from_ok(result: ConversionResult) -> BatchFileResult:
     )
 
 
-# ---------------------------------------------------------------------------
-# System endpoints
-# ---------------------------------------------------------------------------
-
-@app.get("/", tags=["System"], summary="Root", include_in_schema=False)
-async def root():
-    return {"service": APP_NAME, "version": "2.1.0", "status": "running"}
-
-
-@app.get("/ping", tags=["System"], summary="Ping", include_in_schema=False)
-async def ping():
-    return {"message": f"{APP_NAME} is running..."}
-
-
-@app.get("/formats", tags=["System"], summary="Supported file formats by category")
+# -- Routes --
+@markitdown_router.get("/formats", tags=["MarkItDown"], summary="Supported file formats by category")
 async def list_formats():
-    """Return all supported file extensions, grouped by document category."""
     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 {".html", ".htm"}],
+        "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"}],
@@ -348,9 +378,8 @@ async def list_formats():
     }
 
 
-@app.get("/spacy-labels", tags=["System"], summary="Available spaCy NER labels for field extraction")
+@markitdown_router.get("/spacy-labels", tags=["MarkItDown"], summary="Available spaCy NER labels")
 async def list_spacy_labels():
-    """Return spaCy Named Entity Recognition labels available for structured extraction mappings."""
     from extraction.spacy_extractor import VALID_SPACY_LABELS
 
     return {
@@ -372,14 +401,10 @@ async def list_spacy_labels():
     }
 
 
-# ---------------------------------------------------------------------------
-# Convert endpoints
-# ---------------------------------------------------------------------------
-
-@app.post(
+@markitdown_router.post(
     "/convert/file",
     response_model=ConversionResponse,
-    tags=["Convert"],
+    tags=["MarkItDown"],
     summary="Convert an uploaded file to Markdown",
 )
 async def convert_file(
@@ -388,26 +413,11 @@ async def convert_file(
     return_json: bool = Form(False),
     mappings: Optional[str] = Form(
         None,
-        description=(
-            "JSON string defining spaCy field extraction rules. "
-            "Example: {\"company\": {\"source_type\": \"entity\", \"label\": \"ORG\"}}"
-        ),
+        description='JSON string defining spaCy extraction rules. Example: {"company": {"source_type": "entity", "label": "ORG"}}',
     ),
 ):
-    """Convert a single uploaded file to Markdown.
-
-    Set ``return_json=true`` to also receive structured JSON extraction:
-    - CSV / XLS / XLSX files: automatic tabular extraction.
-    - All other files: provide ``mappings`` with spaCy extraction rules.
-
-    On success, ``json_content`` contains the extracted data.
-    On extraction failure, ``json_content`` is null and ``error_message`` is populated.
-    """
     if file is None:
-        raise HTTPException(
-            status_code=400,
-            detail={"success": False, "message": "No file provided."},
-        )
+        raise HTTPException(status_code=400, detail={"success": False, "message": "No file provided."})
 
     parsed_mappings: Optional[Dict[str, Any]] = None
     if mappings:
@@ -415,152 +425,84 @@ async def convert_file(
         try:
             parsed_mappings = _json.loads(mappings)
         except _json.JSONDecodeError:
-            raise HTTPException(
-                status_code=400,
-                detail={"success": False, "message": "Invalid JSON in mappings parameter."},
-            )
+            raise HTTPException(status_code=400, detail={"success": False, "message": "Invalid JSON in mappings parameter."})
 
     logger.info("convert_file | filename=%s", file.filename)
-
     raw = await file.read()
     if len(raw) > MAX_UPLOAD_BYTES:
         logger.warning("convert_file | file too large | filename=%s | size=%d", file.filename, len(raw))
-        raise HTTPException(
-            status_code=413,
-            detail={"success": False, "message": "File exceeds 100 MB limit."},
-        )
+        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"
-    )
+    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=%s | error=%s", file.filename, outcome.message)
         _raise_for_error(outcome)
 
-    logger.info(
-        "convert_file | success | filename=%s | chars=%d | time_ms=%.1f",
-        file.filename,
-        outcome.char_count,
-        outcome.duration_ms,
-    )
+    logger.info("convert_file | success | filename=%s | chars=%d | time_ms=%.1f", file.filename, outcome.char_count, outcome.duration_ms)
 
     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,
-    )
+    return await _build_response(outcome, return_json=return_json, filename=file.filename, raw_data=raw, mappings=parsed_mappings)
 
 
-@app.post(
+@markitdown_router.post(
     "/convert/url",
     response_model=ConversionResponse,
-    tags=["Convert"],
+    tags=["MarkItDown"],
     summary="Convert a public URL to Markdown",
 )
 async def convert_url(body: UrlRequest):
-    """Convert a public HTTP/HTTPS URL to Markdown.
-
-    When ``return_json=true``, the URL content is fetched as raw bytes first
-    to enable binary-aware extraction (e.g. Excel files served over HTTP).
-    """
     logger.info("convert_url | url=%s", body.url)
     parsed = urlparse(body.url)
     filename = Path(parsed.path).name or "url_content"
-
     loop = asyncio.get_running_loop()
 
     if body.return_json:
-        # Fetch raw bytes so binary formats (XLSX, etc.) can be properly parsed.
         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=%s | error=%s", body.url, exc)
-            raise HTTPException(
-                status_code=400,
-                detail={"success": False, "message": f"Failed to fetch URL: {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."},
-            )
+            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
-        )
+        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=%s | error=%s", body.url, outcome.message)
             _raise_for_error(outcome)
 
-        logger.info(
-            "convert_url | success | url=%s | chars=%d | time_ms=%.1f",
-            body.url, outcome.char_count, outcome.duration_ms,
-        )
-        return await _build_response(
-            outcome,
-            return_json=body.return_json,
-            filename=filename,
-            raw_data=raw_data,
-            mappings=body.mappings,
-        )
+        logger.info("convert_url | success | url=%s | chars=%d | time_ms=%.1f", body.url, outcome.char_count, outcome.duration_ms)
+        return await _build_response(outcome, return_json=body.return_json, filename=filename, raw_data=raw_data, mappings=body.mappings)
 
-    # Standard conversion without binary fetch.
     outcome = await loop.run_in_executor(_thread_pool, _converter.convert_url, body.url)
     if isinstance(outcome, ConversionError):
         logger.error("convert_url | conversion failed | url=%s | error=%s", body.url, outcome.message)
         _raise_for_error(outcome)
 
-    logger.info(
-        "convert_url | success | url=%s | chars=%d | time_ms=%.1f",
-        body.url, outcome.char_count, outcome.duration_ms,
-    )
-    return await _build_response(
-        outcome,
-        return_json=body.return_json,
-        filename=filename,
-        mappings=body.mappings,
-    )
-
+    logger.info("convert_url | success | url=%s | chars=%d | time_ms=%.1f", body.url, outcome.char_count, outcome.duration_ms)
+    return await _build_response(outcome, return_json=body.return_json, filename=filename, mappings=body.mappings)
 
-# ---------------------------------------------------------------------------
-# Batch endpoints
-# ---------------------------------------------------------------------------
 
-@app.post(
+@markitdown_router.post(
     "/batch/files",
     response_model=BatchResponse,
-    tags=["Batch"],
+    tags=["MarkItDown"],
     summary="Convert multiple files (up to 10)",
 )
 async def batch_files(
     files: Annotated[List[UploadFile], File(description="Files to convert — maximum 10")],
 ):
-    """Convert up to 10 uploaded files in a single request.
-
-    Files are processed concurrently. Per-item results include success/error
-    details, timing, and content metadata.
-    """
     if not files:
-        raise HTTPException(
-            status_code=400,
-            detail={"success": False, "message": "No files provided."},
-        )
+        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."},
-        )
+        raise HTTPException(status_code=400, detail={"success": False, "message": "Maximum 10 files per batch."})
 
     batch_start = time.perf_counter()
     logger.info("batch_files | count=%d", len(files))
@@ -568,91 +510,59 @@ async def batch_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.",
-            )
-
+            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)
-        )
+        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=%d | failed=%d | total_ms=%.1f", succeeded, len(results) - succeeded, total_ms)
 
-    return BatchResponse(
-        total=len(results),
-        succeeded=succeeded,
-        failed=len(results) - succeeded,
-        total_time_ms=total_ms,
-        results=results,
-    )
+    return BatchResponse(total=len(results), succeeded=succeeded, failed=len(results) - succeeded, total_time_ms=total_ms, results=results)
 
 
-@app.post(
+@markitdown_router.post(
     "/batch/urls",
     response_model=BatchResponse,
-    tags=["Batch"],
+    tags=["MarkItDown"],
     summary="Convert multiple URLs (up to 20)",
 )
 async def batch_urls(body: BatchUrlRequest):
-    """Convert up to 20 public URLs in a single request.
-
-    URLs are processed concurrently. Per-item results include success/error
-    details, timing, and content metadata.
-    """
     batch_start = time.perf_counter()
     logger.info("batch_urls | count=%d", 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)
-        )
+        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=%d | failed=%d | total_ms=%.1f", succeeded, len(results) - succeeded, total_ms)
 
-    return BatchResponse(
-        total=len(results),
-        succeeded=succeeded,
-        failed=len(results) - succeeded,
-        total_time_ms=total_ms,
-        results=results,
-    )
+    return BatchResponse(total=len(results), succeeded=succeeded, failed=len(results) - succeeded, total_time_ms=total_ms, results=results)
 
 
 # ---------------------------------------------------------------------------
-# Server runner (used when invoking this module directly)
+# Include all routers
+# ---------------------------------------------------------------------------
+
+app.include_router(markitdown_router, prefix="/api/v1/markitdown")
+app.include_router(pdf_router, prefix="/api/v1")
+
+# -- 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:
-    """Start the uvicorn server programmatically."""
     import uvicorn
-
-    uvicorn.run(
-        "api.server:app",
-        host=host,
-        port=port,
-        reload=reload,
-    )
+    uvicorn.run("api.server:app", host=host, port=port, reload=reload)

app/api/routes.py

@@ -0,0 +1,256 @@
+from __future__ import annotations
+
+import uuid
+from pathlib import Path
+from typing import Literal, Optional
+
+import structlog
+from fastapi import APIRouter, BackgroundTasks, Depends, File, Query, UploadFile
+from fastapi.responses import FileResponse, JSONResponse
+
+from app.core.auth import require_api_key
+from app.core.config import settings
+from app.core.exceptions import InvalidParameterError, JobNotFoundError
+from app.models.schemas import (
+    AsyncJobResponse,
+    ConversionParams,
+    ConversionResult,
+    HealthResponse,
+    ImageFormat,
+    JobStatus,
+    JobStatusResponse,
+)
+from app.services.conversion import conversion_service
+from app.services.file_service import upload_file_to_service
+from app.services.upload_orchestrator import convert_and_upload
+from app.utils.validators import fetch_pdf_from_url, read_and_validate_pdf
+
+logger = structlog.get_logger(__name__)
+router = APIRouter()
+
+
+_jobs: dict[str, JobStatusResponse] = {}
+
+_MEDIA_TYPES: dict[str, str] = {
+    "png": "image/png",
+    "jpg": "image/jpeg",
+    "jpeg": "image/jpeg",
+}
+
+
+def build_conversion_params(
+    format: ImageFormat = Query(ImageFormat.PNG, description="Output image format"),
+    dpi: int = Query(settings.DEFAULT_DPI, ge=72, le=settings.MAX_DPI, description="Render DPI (72–600)"),
+    quality: int = Query(settings.DEFAULT_JPEG_QUALITY, ge=1, le=100, description="JPEG/WEBP quality (1–100)"),
+    pages: Optional[str] = Query(None, description="Page spec: '1', '1-3', '1,3,5-7'. Empty = all pages."),
+    grayscale: bool = Query(False, description="Convert output to grayscale"),
+    transparent_bg: bool = Query(False, description="Preserve transparency (PNG only)"),
+    split_page: bool = Query(False, description="True: each page separate image. False: stitch into one tall image."),
+) -> ConversionParams:
+    try:
+        return ConversionParams(
+            format=format,
+            dpi=dpi,
+            quality=quality,
+            pages=pages,
+            grayscale=grayscale,
+            transparent_bg=transparent_bg,
+            split_page=split_page,
+        )
+    except Exception as exc:
+        raise InvalidParameterError(str(exc)) from exc
+
+
+def _upload_mode_param(
+    upload_mode: Optional[Literal["per_page", "merged"]] = Query(
+        None,
+        description=(
+            "Upload behaviour after conversion. "
+            "'per_page' uploads each page image immediately as it is rendered (lowest latency). "
+            "'merged' waits for all pages to render, then uploads them. "
+            "Omit to use the server default (PDF_UPLOAD_MODE env var)."
+        ),
+    ),
+) -> Optional[str]:
+    return upload_mode
+
+
+@router.post(
+    "/convert",
+    response_model=ConversionResult,
+    summary="Convert PDF to images — file upload (sync)",
+    tags=["PDF Conversion"],
+)
+async def convert_sync_upload(
+    file: UploadFile = File(...),
+    params: ConversionParams = Depends(build_conversion_params),
+    upload_mode: Optional[str] = Depends(_upload_mode_param),
+    _: str = Depends(require_api_key),
+):
+    """
+    Upload a PDF and convert all (or selected) pages to JPEG or PNG images synchronously.
+
+    - Each page is rendered in **parallel** (split into individual page blobs when split_page=true).
+    - Converted images are uploaded to the File Upload Service automatically.
+    - Returns file URLs and metadata for every converted page.
+    - Supported output formats: JPEG, PNG.
+    """
+    pdf_bytes = await read_and_validate_pdf(file)
+    job_id = str(uuid.uuid4())
+    return await convert_and_upload(pdf_bytes, params, job_id, upload_mode)
+
+
+@router.post(
+    "/convert/url",
+    response_model=ConversionResult,
+    summary="Convert PDF to images — URL (sync)",
+    tags=["PDF Conversion"],
+)
+async def convert_sync_url(
+    pdf_url: str = Query(..., description="Publicly accessible HTTPS URL to a PDF"),
+    params: ConversionParams = Depends(build_conversion_params),
+    upload_mode: Optional[str] = Depends(_upload_mode_param),
+    _: str = Depends(require_api_key),
+):
+    """Fetch a PDF from a public URL and convert it. Private/loopback IPs are blocked."""
+    pdf_bytes = await fetch_pdf_from_url(pdf_url)
+    job_id = str(uuid.uuid4())
+    return await convert_and_upload(pdf_bytes, params, job_id, upload_mode)
+
+
+@router.post(
+    "/convert/async",
+    response_model=AsyncJobResponse,
+    status_code=202,
+    summary="Submit async conversion job — file upload",
+    tags=["PDF Conversion"],
+)
+async def convert_async_upload(
+    background_tasks: BackgroundTasks,
+    file: UploadFile = File(...),
+    params: ConversionParams = Depends(build_conversion_params),
+    upload_mode: Optional[str] = Depends(_upload_mode_param),
+    _: str = Depends(require_api_key),
+):
+    """Submit a PDF for async conversion. Returns 202 immediately; poll /jobs/{job_id}."""
+    pdf_bytes = await read_and_validate_pdf(file)
+    job_id = str(uuid.uuid4())
+    _jobs[job_id] = JobStatusResponse(job_id=job_id, status=JobStatus.PENDING, progress=0.0)
+    background_tasks.add_task(_run_conversion_job, job_id, pdf_bytes, params, upload_mode)
+    return AsyncJobResponse(
+        job_id=job_id,
+        status=JobStatus.PENDING,
+        message="Job accepted. Poll the status URL for progress.",
+        status_url=f"/api/v1/jobs/{job_id}",
+    )
+
+
+@router.post(
+    "/convert/async/url",
+    response_model=AsyncJobResponse,
+    status_code=202,
+    summary="Submit async conversion job — URL",
+    tags=["PDF Conversion"],
+)
+async def convert_async_url(
+    background_tasks: BackgroundTasks,
+    pdf_url: str = Query(..., description="Publicly accessible HTTPS URL to a PDF"),
+    params: ConversionParams = Depends(build_conversion_params),
+    upload_mode: Optional[str] = Depends(_upload_mode_param),
+    _: str = Depends(require_api_key),
+):
+    """Submit a URL-based PDF conversion job asynchronously."""
+    pdf_bytes = await fetch_pdf_from_url(pdf_url)
+    job_id = str(uuid.uuid4())
+    _jobs[job_id] = JobStatusResponse(job_id=job_id, status=JobStatus.PENDING, progress=0.0)
+    background_tasks.add_task(_run_conversion_job, job_id, pdf_bytes, params, upload_mode)
+    return AsyncJobResponse(
+        job_id=job_id,
+        status=JobStatus.PENDING,
+        message="Job accepted. Poll the status URL for progress.",
+        status_url=f"/api/v1/jobs/{job_id}",
+    )
+
+
+@router.get(
+    "/jobs/{job_id}",
+    response_model=JobStatusResponse,
+    summary="Get async job status",
+    tags=["Jobs"],
+)
+async def get_job_status(
+    job_id: str,
+    _: str = Depends(require_api_key),
+):
+    job = _jobs.get(job_id)
+    if not job:
+        raise JobNotFoundError(job_id)
+    return job
+
+
+@router.get(
+    "/files/{job_id}/{filename}",
+    summary="Download a converted image",
+    tags=["Files"],
+)
+async def download_file(
+    job_id: str,
+    filename: str,
+    _: str = Depends(require_api_key),
+):
+    output_root = Path(settings.OUTPUT_DIR).resolve()
+    file_path = (output_root / job_id / filename).resolve()
+
+    try:
+        file_path.relative_to(output_root)
+    except ValueError:
+        raise JobNotFoundError(f"{job_id}/{filename}")
+
+    if not file_path.exists() or not file_path.is_file():
+        raise JobNotFoundError(f"{job_id}/{filename}")
+
+    ext = file_path.suffix.lstrip(".").lower()
+    media_type = _MEDIA_TYPES.get(ext, "application/octet-stream")
+    return FileResponse(path=str(file_path), media_type=media_type, filename=filename)
+
+
+@router.get("/health", response_model=HealthResponse, summary="Liveness probe", tags=["Observability"])
+async def health():
+    return HealthResponse(status="ok", version=settings.VERSION, environment=settings.ENV)
+
+
+@router.get("/ready", response_model=HealthResponse, summary="Readiness probe", tags=["Observability"])
+async def ready():
+    try:
+        probe = Path(settings.OUTPUT_DIR) / ".probe"
+        probe.touch()
+        probe.unlink()
+    except OSError as exc:
+        return JSONResponse(status_code=503, content={"status": "unavailable", "detail": str(exc)})
+    return HealthResponse(status="ready", version=settings.VERSION, environment=settings.ENV)
+
+
+async def _run_conversion_job(
+    job_id: str,
+    pdf_bytes: bytes,
+    params: ConversionParams,
+    upload_mode: Optional[str],
+) -> None:
+    _jobs[job_id] = JobStatusResponse(job_id=job_id, status=JobStatus.PROCESSING, progress=10.0)
+    try:
+        result = await convert_and_upload(pdf_bytes, params, job_id, upload_mode)
+        _jobs[job_id] = JobStatusResponse(
+            job_id=job_id,
+            status=JobStatus.COMPLETED,
+            progress=100.0,
+            result=result,
+        )
+        logger.info("async_job_complete", job_id=job_id)
+    except Exception as exc:
+        logger.exception("async_job_failed", job_id=job_id)
+        _jobs[job_id] = JobStatusResponse(
+            job_id=job_id,
+            status=JobStatus.FAILED,
+            progress=0.0,
+            error=str(exc),
+        )

app/banner.py

@@ -0,0 +1,21 @@
+import pyfiglet
+from rich.console import Console
+from rich.rule import Rule
+from rich.text import Text
+
+from app.core.config import get_settings
+
+_console = Console()
+
+
+def print_banner() -> None:
+    settings = get_settings()
+
+    art = pyfiglet.figlet_format("ValidOps", font="slant")
+
+    _console.print(Text(art, style="bold cyan"))
+    _console.print(f"  [bold white]{'Service:':<14}[/bold white]  [cyan]{settings.APP_NAME}[/cyan]")
+    _console.print(f"  [bold white]{'Version:':<14}[/bold white]  [cyan]v{settings.VERSION}[/cyan]")
+    _console.print(f"  [bold white]{'Environment:':<14}[/bold white]  [cyan]{settings.ENV}[/cyan]")
+    _console.print(Rule(style="dim cyan"))
+    _console.print()

app/core/auth.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+import secrets
+
+from fastapi import Security
+from fastapi.security import APIKeyHeader
+
+from app.core.config import settings
+from app.core.exceptions import AuthenticationError
+
+_api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
+
+
+async def require_api_key(api_key: str | None = Security(_api_key_header)) -> str:
+    if not settings.API_KEY:
+        return api_key or ""
+    if not api_key:
+        raise AuthenticationError("X-API-Key header is required")
+    if not secrets.compare_digest(api_key, settings.API_KEY):
+        raise AuthenticationError("Invalid API key")
+    return api_key

app/core/config.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+from functools import lru_cache
+from typing import List, Literal
+
+from pydantic import Field, field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        case_sensitive=False,
+        extra="ignore",
+    )
+
+    APP_NAME: str = "reconciliation-report-service"
+    VERSION: str = "1.0.0"
+    ENV: Literal["development", "staging", "production"] = "production"
+    DEBUG: bool = False
+
+    HOST: str = "0.0.0.0"
+    PORT: int = 7860
+    WORKERS: int = 4
+    ALLOWED_ORIGINS: List[str] = ["*"]
+
+    API_KEY: str = Field(default="", description="Secret key required in X-API-Key header")
+
+    SELF_PING_ENABLED: bool = False
+    SELF_PING_URL: str = "https://validops-east-2-instance-2.hf.space/ping"
+    SELF_PING_INTERVAL_SECONDS: int = 300
+    HF_TOKEN: str = Field(
+        default="",
+        description="Hugging Face Bearer token used exclusively for self-ping",
+    )
+
+    MAX_FILE_SIZE_MB: int = 50
+    MAX_PAGES: int = 100
+
+    DEFAULT_DPI: int = 200
+    MAX_DPI: int = 600
+    DEFAULT_FORMAT: str = "JPEG"
+    DEFAULT_JPEG_QUALITY: int = 95
+    SUPPORTED_FORMATS: List[str] = ["PNG", "JPEG", "WEBP", "TIFF", "BMP"]
+
+    PAGE_RENDER_WORKERS: int = 8
+
+    PDF_UPLOAD_MODE: Literal["per_page", "merged"] = "per_page"
+
+    UPLOAD_CONCURRENCY: int = 4
+    FILE_SERVICE_TIMEOUT: float = 600.0
+    FILE_SERVICE_CONNECT_TIMEOUT: float = 100.0
+
+    OUTPUT_DIR: str = "/tmp/pdf2img_outputs"
+    TEMP_DIR: str = "/tmp/pdf2img_temp"
+    OUTPUT_TTL_SECONDS: int = 3600
+
+    LOG_LEVEL: str = "INFO"
+    LOG_FORMAT: Literal["json", "console"] = "console"
+
+    RATE_LIMIT: int = 100
+    RATE_LIMIT_WINDOW: int = 60
+
+    REDIS_URL: str = Field(default="", description="Redis connection URL for rate limiting")
+
+    
+    FILE_SERVICE_URL: str = Field(default="", description="External file upload service endpoint")
+    FILE_SERVICE_API_KEY: str = Field(default="", description="API key for external file upload service")
+    FILE_SERVICE_BEARER_TOKEN: str = Field(default="", description="Bearer token for external file upload service")
+
+    @property
+    def MAX_FILE_SIZE_BYTES(self) -> int:
+        return self.MAX_FILE_SIZE_MB * 1024 * 1024
+
+    @property
+    def rate_limit_redis_configured(self) -> bool:
+        return bool(self.WORKERS > 1 and self.REDIS_URL)
+
+    @field_validator("DEFAULT_FORMAT")
+    @classmethod
+    def normalise_format(cls, v: str) -> str:
+        return v.upper()
+
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()
+
+
+settings = get_settings()

app/core/exceptions.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from fastapi import status
+
+
+class AppException(Exception):
+    def __init__(self, detail: str, status_code: int = status.HTTP_500_INTERNAL_SERVER_ERROR):
+        self.detail = detail
+        self.status_code = status_code
+        super().__init__(detail)
+
+
+class AuthenticationError(AppException):
+    def __init__(self, detail: str = "Unauthorized"):
+        super().__init__(detail, status.HTTP_401_UNAUTHORIZED)
+
+
+class InvalidFileTypeError(AppException):
+    def __init__(self, detail: str = "Unsupported file type"):
+        super().__init__(detail, status.HTTP_415_UNSUPPORTED_MEDIA_TYPE)
+
+
+class FileTooLargeError(AppException):
+    def __init__(self, max_mb: int):
+        super().__init__(
+            f"File exceeds the maximum allowed size of {max_mb} MB",
+            status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
+        )
+
+
+class TooManyPagesError(AppException):
+    def __init__(self, max_pages: int):
+        super().__init__(
+            f"PDF exceeds the maximum allowed page count of {max_pages}",
+            status.HTTP_422_UNPROCESSABLE_ENTITY,
+        )
+
+
+class ConversionError(AppException):
+    def __init__(self, detail: str = "PDF conversion failed"):
+        super().__init__(detail, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+
+class JobNotFoundError(AppException):
+    def __init__(self, job_id: str):
+        super().__init__(f"Job '{job_id}' not found", status.HTTP_404_NOT_FOUND)
+
+
+class InvalidParameterError(AppException):
+    def __init__(self, detail: str):
+        super().__init__(detail, status.HTTP_422_UNPROCESSABLE_ENTITY)
+
+
+class FileServiceError(AppException):
+    def __init__(self, detail: str = "File upload service error"):
+        super().__init__(detail, status.HTTP_502_BAD_GATEWAY)
+
+
+class CSVGenerationError(AppException):
+    def __init__(self, detail: str = "CSV generation failed"):
+        super().__init__(detail, status.HTTP_500_INTERNAL_SERVER_ERROR)

app/core/logging.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+import logging
+import sys
+
+import structlog
+
+from app.core.config import settings
+
+
+def configure_logging() -> None:
+    pre_chain = [
+        structlog.contextvars.merge_contextvars,
+        structlog.stdlib.add_log_level,
+        structlog.processors.TimeStamper(fmt="iso"),
+    ]
+
+    is_prod = settings.ENV == "production"
+
+    if is_prod:
+        renderer = structlog.processors.JSONRenderer()
+    else:
+        renderer = structlog.dev.ConsoleRenderer(colors=True)
+        pre_chain.insert(0, structlog.stdlib.add_logger_name)
+
+    structlog.configure(
+        processors=[
+            structlog.stdlib.filter_by_level,
+            *pre_chain,
+            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
+        ],
+        logger_factory=structlog.stdlib.LoggerFactory(),
+        wrapper_class=structlog.stdlib.BoundLogger,
+        cache_logger_on_first_use=True,
+    )
+
+    formatter = structlog.stdlib.ProcessorFormatter(
+        foreign_pre_chain=pre_chain,
+        processors=[
+            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
+            renderer,
+        ],
+    )
+
+    handler = logging.StreamHandler(sys.stdout)
+    handler.setFormatter(formatter)
+
+    root_logger = logging.getLogger()
+    root_logger.handlers = [handler]
+    root_logger.setLevel(settings.LOG_LEVEL.upper())
+
+    for lib in (
+        "uvicorn.access",
+        "uvicorn.error",
+        "multipart",
+        "httpx",
+        "httpcore",
+        "redis",
+        "PIL",
+        "asyncio",
+    ):
+        logging.getLogger(lib).setLevel(logging.WARNING)

app/core/rate_limit.py

@@ -0,0 +1,218 @@
+from __future__ import annotations
+
+import ipaddress
+import json
+import structlog
+import time
+from typing import Optional, Tuple
+from uuid import uuid4
+
+import redis.asyncio as aioredis
+from redis.asyncio.retry import Retry
+from redis.backoff import ExponentialBackoff
+from redis.exceptions import (
+    BusyLoadingError,
+    ConnectionError as RedisConnectionError,
+    NoScriptError,
+    RedisError,
+    TimeoutError as RedisTimeoutError,
+)
+from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
+from starlette.requests import Request
+from starlette.responses import Response
+
+from app.core.config import settings
+
+logger = structlog.get_logger(__name__)
+
+_EXEMPT_PATHS: frozenset[str] = frozenset({
+    "/api/v1/health",
+    "/api/v1/ready",
+    "/api/v1/ping",
+    "/docs",
+    "/redoc",
+    "/openapi.json",
+    "/metrics",
+})
+
+_TRUSTED_PROXY_NETWORKS: tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...] = (
+    ipaddress.ip_network("10.0.0.0/8"),
+    ipaddress.ip_network("172.16.0.0/12"),
+    ipaddress.ip_network("192.168.0.0/16"),
+    ipaddress.ip_network("127.0.0.0/8"),
+    ipaddress.ip_network("fc00::/7"),
+    ipaddress.ip_network("::1/128"),
+)
+
+_RATE_LIMIT_SCRIPT = """
+local key        = KEYS[1]
+local now        = tonumber(ARGV[1])
+local window     = tonumber(ARGV[2])
+local limit      = tonumber(ARGV[3])
+local unique_id  = ARGV[4]
+local cutoff     = now - window
+
+redis.call('zremrangebyscore', key, '-inf', cutoff)
+local count = redis.call('zcard', key)
+
+if count < limit then
+    redis.call('zadd', key, now, unique_id)
+    redis.call('expire', key, window + 1)
+    return {0, limit - count - 1}
+end
+
+redis.call('expire', key, window + 1)
+return {1, 0}
+"""
+
+
+def _is_trusted_proxy(host: str) -> bool:
+    try:
+        addr = ipaddress.ip_address(host)
+        return any(addr in net for net in _TRUSTED_PROXY_NETWORKS)
+    except ValueError:
+        return False
+
+
+def _extract_client_ip(request: Request) -> str:
+    direct_host = request.client.host if request.client else None
+
+    if direct_host and _is_trusted_proxy(direct_host):
+        forwarded_for = request.headers.get("x-forwarded-for", "")
+        if forwarded_for:
+            for candidate in reversed([ip.strip() for ip in forwarded_for.split(",")]):
+                if candidate and not _is_trusted_proxy(candidate):
+                    return candidate
+
+        real_ip = request.headers.get("x-real-ip", "").strip()
+        if real_ip and not _is_trusted_proxy(real_ip):
+            return real_ip
+
+    return direct_host or "unknown"
+
+
+class RateLimitMiddleware(BaseHTTPMiddleware):
+    def __init__(self, app, **kwargs) -> None:
+        super().__init__(app, **kwargs)
+        self._pool: Optional[aioredis.ConnectionPool] = None
+        self._client: Optional[aioredis.Redis] = None
+        self._script_sha: Optional[str] = None
+
+    def _build_pool(self) -> aioredis.ConnectionPool:
+        retry = Retry(
+            ExponentialBackoff(cap=0.5, base=0.1),
+            retries=2,
+            supported_errors=(BusyLoadingError, RedisConnectionError, RedisTimeoutError),
+        )
+        return aioredis.ConnectionPool.from_url(
+            settings.REDIS_URL,
+            encoding="utf-8",
+            decode_responses=True,
+            socket_connect_timeout=1,
+            socket_timeout=1,
+            socket_keepalive=True,
+            retry=retry,
+            retry_on_timeout=True,
+            max_connections=max(settings.WORKERS * 2, 10),
+            health_check_interval=30,
+        )
+
+    async def _get_client(self) -> aioredis.Redis:
+        if self._pool is None:
+            self._pool = self._build_pool()
+        if self._client is None:
+            self._client = aioredis.Redis(connection_pool=self._pool)
+        return self._client
+
+    async def _load_script(self, client: aioredis.Redis) -> str:
+        if self._script_sha is None:
+            self._script_sha = await client.script_load(_RATE_LIMIT_SCRIPT)
+        return self._script_sha
+
+    async def _check_rate_limit(self, client_ip: str) -> Tuple[bool, int]:
+        now_ms = int(time.time() * 1000)
+        window_ms = settings.RATE_LIMIT_WINDOW * 1000
+        key = f"rl:{client_ip}"
+        unique_id = f"{now_ms}:{uuid4().hex}"
+
+        client = await self._get_client()
+
+        try:
+            sha = await self._load_script(client)
+            result = await client.evalsha(
+                sha,
+                1,
+                key,
+                str(now_ms),
+                str(window_ms),
+                str(settings.RATE_LIMIT),
+                unique_id,
+            )
+        except NoScriptError:
+            self._script_sha = None
+            sha = await self._load_script(client)
+            result = await client.evalsha(
+                sha,
+                1,
+                key,
+                str(now_ms),
+                str(window_ms),
+                str(settings.RATE_LIMIT),
+                unique_id,
+            )
+
+        return bool(result[0]), int(result[1])
+
+    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
+        if request.url.path in _EXEMPT_PATHS:
+            return await call_next(request)
+
+        if not settings.rate_limit_redis_configured:
+            return await call_next(request)
+
+        client_ip = _extract_client_ip(request)
+
+        try:
+            is_limited, remaining = await self._check_rate_limit(client_ip)
+        except (RedisConnectionError, RedisTimeoutError) as exc:
+            logger.error("redis_unavailable", error=str(exc))
+            return self._service_unavailable_response()
+        except RedisError as exc:
+            logger.error("redis_error", error=str(exc))
+            return self._service_unavailable_response()
+
+        if is_limited:
+            request_id = getattr(request.state, "request_id", str(uuid4()))
+            return self._rate_limit_response(request_id)
+
+        response = await call_next(request)
+        response.headers["X-RateLimit-Limit"] = str(settings.RATE_LIMIT)
+        response.headers["X-RateLimit-Remaining"] = str(remaining)
+        response.headers["X-RateLimit-Window"] = str(settings.RATE_LIMIT_WINDOW)
+        return response
+
+    def _rate_limit_response(self, request_id: str) -> Response:
+        return Response(
+            content=json.dumps({
+                "error": "Too many requests. Please slow down.",
+                "request_id": request_id,
+            }),
+            status_code=429,
+            media_type="application/json",
+            headers={
+                "Retry-After": str(settings.RATE_LIMIT_WINDOW),
+                "X-RateLimit-Limit": str(settings.RATE_LIMIT),
+                "X-RateLimit-Remaining": "0",
+                "X-RateLimit-Window": str(settings.RATE_LIMIT_WINDOW),
+            },
+        )
+
+    def _service_unavailable_response(self) -> Response:
+        return Response(
+            content=json.dumps(
+                {"error": "Service temporarily unavailable. Please retry shortly."}
+            ),
+            status_code=503,
+            media_type="application/json",
+            headers={"Retry-After": "5"},
+        )

app/main.py

@@ -0,0 +1,95 @@
+from __future__ import annotations
+
+import time
+import uuid
+from contextlib import asynccontextmanager
+
+import structlog
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.middleware.gzip import GZipMiddleware
+from fastapi.responses import JSONResponse
+from prometheus_client import make_asgi_app
+
+from app.api.routes import router
+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
+configure_logging()
+logger = structlog.get_logger(__name__)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    logger.info("startup", service=settings.APP_NAME, version=settings.VERSION, env=settings.ENV)
+    yield
+    logger.info("shutdown", service=settings.APP_NAME)
+
+
+app = FastAPI(
+    title=settings.APP_NAME,
+    description="High-performance PDF to Image conversion and CSV report generation API",
+    version=settings.VERSION,
+    docs_url="/docs",
+    redoc_url="/redoc",
+    lifespan=lifespan,
+)
+
+app.add_middleware(RateLimitMiddleware)
+app.add_middleware(GZipMiddleware, minimum_size=1000)
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=settings.ALLOWED_ORIGINS,
+    allow_credentials=True,
+    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()
+    request.state.request_id = request_id
+
+    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_code=response.status_code, duration_ms=elapsed)
+    response.headers["X-Request-ID"] = request_id
+    response.headers["X-Response-Time-Ms"] = str(elapsed)
+    return response
+
+
+@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.state.request_id},
+    )
+
+
+@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.state.request_id},
+    )
+
+
+@app.get("/")
+async def root():
+    return {"message": f"{settings.APP_NAME} v{settings.VERSION}"}
+
+app.include_router(router, prefix="/api/v1")
+app.mount("/metrics", make_asgi_app())

app/models/schemas.py

@@ -0,0 +1,162 @@
+from __future__ import annotations
+
+import re
+from enum import Enum
+from typing import Any, Dict, List, Literal, Optional
+
+from pydantic import BaseModel, Field, field_validator, model_validator
+
+from app.core.config import settings
+
+_PAGE_SPEC_RE = re.compile(r"^\s*(\d+(-\d+)?)(\s*,\s*(\d+(-\d+)?))*\s*$")
+
+
+class ImageFormat(str, Enum):
+    PNG = "PNG"
+    JPEG = "JPEG"
+
+
+class JobStatus(str, Enum):
+    PENDING = "pending"
+    PROCESSING = "processing"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+
+class ConversionParams(BaseModel):
+    format: ImageFormat = Field(ImageFormat.PNG)
+    dpi: int = Field(settings.DEFAULT_DPI, ge=72, le=settings.MAX_DPI)
+    quality: int = Field(85, ge=1, le=100)
+    pages: Optional[str] = Field(None)
+    grayscale: bool = Field(False)
+    transparent_bg: bool = Field(False)
+    split_page: Optional[bool] = Field(
+        False,
+        description=(
+            "Controls output granularity. "
+            "True: each page → separate image URL. "
+            "False (default): all pages stitched into one tall image → single URL."
+        ),
+    )
+
+    @field_validator("format", mode="before")
+    @classmethod
+    def normalise_format(cls, v: str) -> str:
+        if isinstance(v, str):
+            upper = v.upper()
+            if upper == "JPG":
+                return "JPEG"
+            return upper
+        return v
+
+    @field_validator("pages")
+    @classmethod
+    def validate_page_spec_syntax(cls, v: Optional[str]) -> Optional[str]:
+        if v is None or v.strip() == "":
+            return None
+        if not _PAGE_SPEC_RE.match(v):
+            raise ValueError(
+                f"Invalid page spec '{v}'. Use comma-separated numbers or ranges, "
+                "e.g. '1', '1-3', '1,3,5-7'."
+            )
+        for token in v.split(","):
+            token = token.strip()
+            if "-" in token:
+                parts = token.split("-", 1)
+                start, end = int(parts[0]), int(parts[1])
+                if start < 1:
+                    raise ValueError(f"Page numbers are 1-indexed; got '{token}'")
+                if start > end:
+                    raise ValueError(f"Invalid range '{token}': start must be <= end")
+            else:
+                if int(token) < 1:
+                    raise ValueError(f"Page numbers are 1-indexed; got '{token}'")
+        return v
+
+    @model_validator(mode="after")
+    def validate_transparent_bg(self) -> ConversionParams:
+        if self.transparent_bg and self.format != ImageFormat.PNG:
+            raise ValueError("transparent_bg is only supported for PNG output")
+        return self
+
+    def parse_pages(self, total_pages: int) -> List[int]:
+        if not self.pages:
+            return list(range(total_pages))
+        indices: set[int] = set()
+        for token in self.pages.split(","):
+            token = token.strip()
+            if "-" in token:
+                parts = token.split("-", 1)
+                start, end = int(parts[0]) - 1, int(parts[1]) - 1
+                indices.update(range(start, end + 1))
+            else:
+                indices.add(int(token) - 1)
+        valid = sorted(i for i in indices if 0 <= i < total_pages)
+        if not valid:
+            raise ValueError(
+                f"Page spec '{self.pages}' contains no pages within "
+                f"the document's {total_pages} page(s)"
+            )
+        return valid
+
+
+class PageResult(BaseModel):
+    page_number: int
+    download_url: str
+    width: int
+    height: int
+    size_bytes: int
+    format: str
+    file_id: Optional[str] = None
+    file_url: Optional[str] = None
+    upload_error: Optional[str] = None
+
+
+class UploadSummary(BaseModel):
+    upload_mode: str
+    total_uploaded: int
+    failed_uploads: int
+    files: List["UploadedFileInfo"]
+
+
+class ConversionResult(BaseModel):
+    job_id: str
+    status: JobStatus
+    total_pages: int
+    converted_pages: int
+    pages: List[PageResult]
+    duration_ms: float
+    upload_summary: Optional[UploadSummary] = None
+
+
+class AsyncJobResponse(BaseModel):
+    job_id: str
+    status: JobStatus
+    message: str
+    status_url: str
+
+
+class JobStatusResponse(BaseModel):
+    job_id: str
+    status: JobStatus
+    progress: float = Field(..., ge=0, le=100)
+    result: Optional[ConversionResult] = None
+    error: Optional[str] = None
+
+
+class HealthResponse(BaseModel):
+    status: str
+    version: str
+    environment: str
+
+
+class ErrorResponse(BaseModel):
+    error: str
+    request_id: Optional[str] = None
+
+
+class UploadedFileInfo(BaseModel):
+    file_id: str
+    file_url: str
+    filename: str
+    size_bytes: int

app/services/conversion.py

@@ -0,0 +1,341 @@
+from __future__ import annotations
+
+import asyncio
+import io
+import time
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+import structlog
+from PIL import Image
+
+from app.core.config import settings
+from app.core.exceptions import ConversionError, InvalidParameterError, TooManyPagesError
+from app.models.schemas import ConversionParams, PageResult
+
+logger = structlog.get_logger(__name__)
+
+_SUPPORTED_FORMATS = {"JPEG", "PNG"}
+_MAX_MEMORY_ESTIMATE_MB = 512
+
+_render_pool = ThreadPoolExecutor(
+    max_workers=settings.PAGE_RENDER_WORKERS,
+    thread_name_prefix="page_renderer",
+)
+
+
+class ConversionService:
+    def __init__(self) -> None:
+        self._output_dir = Path(settings.OUTPUT_DIR)
+        self._output_dir.mkdir(parents=True, exist_ok=True)
+
+    async def convert(
+        self,
+        pdf_bytes: bytes,
+        params: ConversionParams,
+        job_id: Optional[str] = None,
+    ) -> Tuple[List[PageResult], float]:
+        """
+        Full pipeline:
+          1. Validate PDF + determine page indices
+          2. Render pages (always individually for parallelism)
+          3a. split_page=True / omitted  → return one PageResult per page
+          3b. split_page=False           → stitch all pages into one tall image,
+                                           return a single PageResult
+        """
+        job_id = job_id or str(uuid.uuid4())
+        job_dir = self._output_dir / job_id
+        job_dir.mkdir(parents=True, exist_ok=True)
+
+        start = time.perf_counter()
+        loop = asyncio.get_event_loop()
+
+        try:
+            page_indices, total_pages = await loop.run_in_executor(
+                _render_pool, _validate_and_get_indices, pdf_bytes, params
+            )
+        except (ConversionError, InvalidParameterError, TooManyPagesError):
+            raise
+        except Exception as exc:
+            logger.exception("pdf_validation_failed", job_id=job_id)
+            raise ConversionError(str(exc)) from exc
+
+        logger.info(
+            "conversion_started",
+            job_id=job_id,
+            total_pages=total_pages,
+            rendering=len(page_indices),
+            dpi=params.dpi,
+            format=params.format,
+            split_page=params.split_page,
+        )
+
+        try:
+            page_blobs: dict[int, bytes] = await loop.run_in_executor(
+                _render_pool, _split_pdf_pages, pdf_bytes, page_indices
+            )
+        except Exception as exc:
+            logger.exception("pdf_split_failed", job_id=job_id)
+            raise ConversionError(f"Failed to split PDF into pages: {exc}") from exc
+
+        page_results = await self._render_parallel(
+            page_blobs, params, job_dir, job_id, loop
+        )
+
+        if params.split_page is False:
+            try:
+                stitched = await loop.run_in_executor(
+                    _render_pool, _stitch_pages, page_results, params, job_dir
+                )
+                page_results = [stitched]
+                logger.info("pages_stitched", job_id=job_id, total_pages=len(page_blobs))
+            except Exception as exc:
+                logger.exception("stitch_failed", job_id=job_id)
+                raise ConversionError(f"Failed to stitch pages into single image: {exc}") from exc
+
+        duration_ms = round((time.perf_counter() - start) * 1000, 2)
+        logger.info("conversion_complete", job_id=job_id, pages=len(page_results), duration_ms=duration_ms)
+        return page_results, duration_ms
+
+    async def _render_parallel(
+        self,
+        page_blobs: dict[int, bytes],
+        params: ConversionParams,
+        job_dir: Path,
+        job_id: str,
+        loop: asyncio.AbstractEventLoop,
+    ) -> List[PageResult]:
+        ordered_indices = sorted(page_blobs.keys())
+        render_coros = [
+            loop.run_in_executor(
+                _render_pool,
+                _render_single_page,
+                page_idx,
+                page_blobs[page_idx],
+                params,
+                job_dir,
+            )
+            for page_idx in ordered_indices
+        ]
+
+        results_by_idx: dict[int, PageResult] = {}
+        errors: list[str] = []
+
+        raw_results = await asyncio.gather(*render_coros, return_exceptions=True)
+
+        for page_idx, outcome in zip(ordered_indices, raw_results):
+            if isinstance(outcome, Exception):
+                logger.error("page_render_failed", job_id=job_id, page=page_idx + 1, error=str(outcome))
+                errors.append(f"Page {page_idx + 1}: {outcome}")
+            else:
+                results_by_idx[page_idx] = outcome
+                logger.debug("page_rendered", job_id=job_id, page=page_idx + 1, size_bytes=outcome.size_bytes)
+
+        if not results_by_idx:
+            raise ConversionError(f"All pages failed to render. Errors: {'; '.join(errors)}")
+
+        if errors:
+            logger.warning("some_pages_failed", job_id=job_id, failed=len(errors), succeeded=len(results_by_idx))
+
+        return [results_by_idx[i] for i in sorted(results_by_idx)]
+
+
+def _validate_and_get_indices(
+    pdf_bytes: bytes,
+    params: ConversionParams,
+) -> Tuple[List[int], int]:
+    from pypdf import PdfReader
+    from pypdf.errors import PdfReadError
+
+    fmt = params.format.value if hasattr(params.format, "value") else str(params.format)
+    if fmt not in _SUPPORTED_FORMATS:
+        raise InvalidParameterError(
+            f"Unsupported format '{fmt}'. Only JPEG (JPG) and PNG are supported."
+        )
+
+    try:
+        reader = PdfReader(io.BytesIO(pdf_bytes))
+        if reader.is_encrypted:
+            from pypdf import PasswordType
+            result = reader.decrypt("")
+            if result == PasswordType.NOT_DECRYPTED:
+                raise ConversionError("PDF is password-protected. Please provide an unlocked PDF.")
+        total_pages = len(reader.pages)
+    except ConversionError:
+        raise
+    except PdfReadError as exc:
+        raise ConversionError(f"Malformed PDF: {exc}") from exc
+
+    if total_pages == 0:
+        raise ConversionError("PDF contains no pages")
+    if total_pages > settings.MAX_PAGES:
+        raise TooManyPagesError(settings.MAX_PAGES)
+
+    page_indices = params.parse_pages(total_pages)
+    _guard_memory(len(page_indices), params.dpi)
+
+    return page_indices, total_pages
+
+
+def _split_pdf_pages(pdf_bytes: bytes, page_indices: List[int]) -> dict[int, bytes]:
+    """Extract each requested page into its own single-page PDF blob for parallel rendering."""
+    from pypdf import PdfReader, PdfWriter
+
+    reader = PdfReader(io.BytesIO(pdf_bytes))
+    blobs: dict[int, bytes] = {}
+
+    for page_idx in page_indices:
+        writer = PdfWriter()
+        writer.add_page(reader.pages[page_idx])
+        buf = io.BytesIO()
+        writer.write(buf)
+        blobs[page_idx] = buf.getvalue()
+
+    logger.debug("pdf_split_complete", pages=len(blobs))
+    return blobs
+
+
+def _render_single_page(
+    page_idx: int,
+    page_pdf_bytes: bytes,
+    params: ConversionParams,
+    job_dir: Path,
+) -> PageResult:
+    """Render one single-page PDF blob to an image file on disk."""
+    from pdf2image import convert_from_bytes
+    from pdf2image.exceptions import PDFInfoNotInstalledError, PDFPageCountError, PDFSyntaxError
+
+    fmt = params.format.value if hasattr(params.format, "value") else params.format
+
+    try:
+        pil_images = convert_from_bytes(
+            page_pdf_bytes,
+            dpi=params.dpi,
+            fmt="ppm",
+            thread_count=1,
+            transparent=params.transparent_bg and fmt == "PNG",
+            use_pdftocairo=True,
+        )
+    except (PDFInfoNotInstalledError, PDFPageCountError, PDFSyntaxError) as exc:
+        raise ConversionError(f"Page {page_idx + 1} render failed: {exc}") from exc
+
+    if not pil_images:
+        raise ConversionError(f"Page {page_idx + 1}: renderer returned no image")
+
+    img = pil_images[0]
+
+    if fmt == "JPEG" and img.mode in ("RGBA", "LA", "P"):
+        if img.mode == "P":
+            img = img.convert("RGBA")
+        bg = Image.new("RGB", img.size, (255, 255, 255))
+        mask = img.split()[-1] if img.mode in ("RGBA", "LA") else None
+        bg.paste(img, mask=mask)
+        img = bg
+    elif fmt == "JPEG" and img.mode != "RGB":
+        img = img.convert("RGB")
+
+    if params.grayscale:
+        img = img.convert("L")
+
+    ext = "jpg" if fmt == "JPEG" else "png"
+    filename = f"page_{page_idx + 1:04d}.{ext}"
+    out_path = job_dir / filename
+
+    save_kwargs: dict = {}
+    if fmt == "JPEG":
+        save_kwargs["quality"] = params.quality
+        save_kwargs["optimize"] = True
+    if fmt == "PNG":
+        save_kwargs["optimize"] = True
+
+    try:
+        img.save(str(out_path), format=fmt, **save_kwargs)
+    except OSError as exc:
+        raise ConversionError(f"Failed to write {filename}: {exc}") from exc
+
+    stat = out_path.stat()
+    return PageResult(
+        page_number=page_idx + 1,
+        download_url=f"/api/v1/files/{job_dir.name}/{filename}",
+        width=img.width,
+        height=img.height,
+        size_bytes=stat.st_size,
+        format=fmt,
+    )
+
+
+def _stitch_pages(
+    page_results: List[PageResult],
+    params: ConversionParams,
+    job_dir: Path,
+) -> PageResult:
+    """
+    Stitch all rendered page images vertically into a single tall image.
+    Called only when split_page=False.
+    Pages are placed top-to-bottom in document order with no gaps.
+    """
+    fmt = params.format.value if hasattr(params.format, "value") else params.format
+    ext = "jpg" if fmt == "JPEG" else "png"
+
+    images: list[Image.Image] = []
+    for pr in sorted(page_results, key=lambda p: p.page_number):
+        path = job_dir / Path(pr.download_url).name
+        img = Image.open(str(path))
+        if fmt == "JPEG" and img.mode != "RGB":
+            img = img.convert("RGB")
+        elif fmt == "PNG" and img.mode not in ("RGB", "RGBA", "L"):
+            img = img.convert("RGB")
+        images.append(img)
+
+    if not images:
+        raise ConversionError("No rendered page images found to stitch")
+
+    total_width  = max(im.width  for im in images)
+    total_height = sum(im.height for im in images)
+
+    mode = images[0].mode
+    stitched = Image.new(mode, (total_width, total_height), color=(255, 255, 255) if mode == "RGB" else 255)
+
+    y_offset = 0
+    for img in images:
+        stitched.paste(img, (0, y_offset))
+        y_offset += img.height
+
+    out_filename = f"stitched.{ext}"
+    out_path = job_dir / out_filename
+
+    save_kwargs: dict = {}
+    if fmt == "JPEG":
+        save_kwargs["quality"] = params.quality
+        save_kwargs["optimize"] = True
+    if fmt == "PNG":
+        save_kwargs["optimize"] = True
+
+    stitched.save(str(out_path), format=fmt, **save_kwargs)
+
+    stat = out_path.stat()
+    return PageResult(
+        page_number=1,
+        download_url=f"/api/v1/files/{job_dir.name}/{out_filename}",
+        width=stitched.width,
+        height=stitched.height,
+        size_bytes=stat.st_size,
+        format=fmt,
+    )
+
+
+def _guard_memory(page_count: int, dpi: int) -> None:
+    pixels_per_page = (dpi * 8.5) * (dpi * 11)
+    bytes_per_page = pixels_per_page * 3
+    estimated_mb = (bytes_per_page * page_count) / (1024 * 1024)
+    if estimated_mb > _MAX_MEMORY_ESTIMATE_MB:
+        raise InvalidParameterError(
+            f"Requested conversion would require approximately {estimated_mb:.0f} MB of memory "
+            f"({page_count} pages at {dpi} DPI). "
+            f"Reduce DPI or use the 'pages' parameter to select fewer pages."
+        )
+
+
+conversion_service = ConversionService()

app/services/file_service.py

@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+import httpx
+import structlog
+
+from app.core.config import settings
+from app.core.exceptions import FileServiceError
+from app.models.schemas import UploadedFileInfo
+
+logger = structlog.get_logger(__name__)
+
+
+async def upload_file_to_service(
+    file_bytes: bytes,
+    filename: str,
+    content_type: str = "application/octet-stream",
+) -> UploadedFileInfo:
+    timeout = httpx.Timeout(settings.FILE_SERVICE_TIMEOUT, connect=settings.FILE_SERVICE_CONNECT_TIMEOUT)
+    async with httpx.AsyncClient(timeout=timeout) as client:
+        try:
+            response = await client.post(
+                settings.FILE_SERVICE_URL,
+                headers={
+                    "X-API-Key": settings.FILE_SERVICE_API_KEY,
+                    "Authorization": f"Bearer {settings.FILE_SERVICE_BEARER_TOKEN}"
+                },
+                files={"files": (filename, file_bytes, content_type)},
+            )
+            response.raise_for_status()
+        except httpx.HTTPStatusError as exc:
+            logger.error(
+                "file_service_http_error",
+                status_code=exc.response.status_code,
+                body=exc.response.text,
+            )
+            raise FileServiceError(
+                f"File service returned HTTP {exc.response.status_code}"
+            ) from exc
+        except httpx.RequestError as exc:
+            logger.error("file_service_request_error", error=str(exc))
+            raise FileServiceError(f"File service unreachable: {exc}") from exc
+
+    payload = response.json()
+    uploaded = _extract_uploaded_file(payload, filename, len(file_bytes))
+    logger.info("file_uploaded", file_id=uploaded.file_id, filename=filename)
+    return uploaded
+
+
+def _extract_uploaded_file(
+    payload: object,
+    filename: str,
+    size_bytes: int,
+) -> UploadedFileInfo:
+    if isinstance(payload, list):
+        if not payload:
+            raise FileServiceError("File service returned an empty list")
+        first = payload[0]
+        if not isinstance(first, dict):
+            raise FileServiceError(f"Unexpected item type in file service list: {type(first)}")
+        entry = first.get("data") or first
+        return _entry_to_info(entry, filename, size_bytes)
+
+    if isinstance(payload, dict):
+        nested = payload.get("files") or payload.get("data")
+        if isinstance(nested, list) and nested:
+            entry = nested[0].get("data") or nested[0]
+            return _entry_to_info(entry, filename, size_bytes)
+        return _entry_to_info(payload, filename, size_bytes)
+
+    raise FileServiceError(f"Unrecognised file service response type: {type(payload)}")
+
+
+def _entry_to_info(entry: dict, filename: str, size_bytes: int) -> UploadedFileInfo:
+    file_id = (
+        entry.get("fileid")
+        or entry.get("file_id")
+        or entry.get("id")
+        or entry.get("_id")
+        or ""
+    )
+    file_url = (
+        entry.get("file_url")
+        or entry.get("url")
+        or entry.get("path")
+        or ""
+    )
+    resolved_filename = (
+        entry.get("filename")
+        or entry.get("name")
+        or filename
+    )
+    resolved_size = int(entry.get("size") or entry.get("size_bytes") or size_bytes)
+
+    return UploadedFileInfo(
+        file_id=str(file_id),
+        file_url=str(file_url),
+        filename=str(resolved_filename),
+        size_bytes=resolved_size,
+    )

app/services/ping.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+import asyncio
+
+import httpx
+import structlog
+
+from app.core.config import settings
+
+logger = structlog.get_logger(__name__)
+
+
+async def start_self_ping() -> None:
+    if not settings.SELF_PING_ENABLED or not settings.SELF_PING_URL:
+        logger.info("self_ping_disabled")
+        return
+    asyncio.create_task(_ping_loop())
+
+
+async def _ping_loop() -> None:
+    logger.info(
+        "self_ping_started",
+        url=settings.SELF_PING_URL,
+        interval=settings.SELF_PING_INTERVAL_SECONDS,
+    )
+    async with httpx.AsyncClient(timeout=10.0) as client:
+        while True:
+            await asyncio.sleep(settings.SELF_PING_INTERVAL_SECONDS)
+            await _execute_ping(client)
+
+
+async def _execute_ping(client: httpx.AsyncClient) -> None:
+    headers: dict[str, str] = {}
+    if settings.HF_TOKEN:
+        headers["Authorization"] = f"Bearer {settings.HF_TOKEN}"
+
+    try:
+        response = await client.get(settings.SELF_PING_URL, headers=headers)
+        logger.info("self_ping_success", status_code=response.status_code)
+    except httpx.RequestError as exc:
+        logger.warning("self_ping_failed", error=str(exc))

app/services/upload_orchestrator.py

@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+"""
+Upload orchestrator for PDF conversion results.
+
+Workflow:
+  1. Convert PDF pages to JPEG or PNG images.
+  2. Upload each image to the File Upload Service concurrently.
+  3. Return clean ConversionResult with uploaded file URLs.
+
+Only JPEG and PNG output formats are supported.
+TIFF / merged-file mode has been removed.
+"""
+
+import asyncio
+import time
+from pathlib import Path
+from typing import List, Optional
+
+import structlog
+
+from app.core.config import settings
+from app.models.schemas import (
+    ConversionParams,
+    ConversionResult,
+    JobStatus,
+    PageResult,
+    UploadSummary,
+    UploadedFileInfo,
+)
+from app.services.conversion import conversion_service
+from app.services.file_service import upload_file_to_service
+
+logger = structlog.get_logger(__name__)
+
+
+async def convert_and_upload(
+    pdf_bytes: bytes,
+    params: ConversionParams,
+    job_id: str,
+    upload_mode: Optional[str] = None,
+) -> ConversionResult:
+    """
+    Full pipeline:
+      1. Render PDF pages to JPEG or PNG images.
+      2. Upload every image to the File Upload Service.
+      3. Return ConversionResult with file URLs per page.
+    """
+    start = time.perf_counter()
+
+    pages, _conv_ms = await conversion_service.convert(pdf_bytes, params, job_id)
+    pages, upload_summary = await _upload_pages(pages, params, job_id)
+
+    duration_ms = round((time.perf_counter() - start) * 1000, 2)
+
+    return ConversionResult(
+        job_id=job_id,
+        status=JobStatus.COMPLETED,
+        total_pages=len(pages),
+        converted_pages=len(pages),
+        pages=pages,
+        duration_ms=duration_ms,
+        upload_summary=upload_summary,
+    )
+
+
+async def _upload_pages(
+    pages: List[PageResult],
+    params: ConversionParams,
+    job_id: str,
+) -> tuple[List[PageResult], UploadSummary]:
+    semaphore = asyncio.Semaphore(settings.UPLOAD_CONCURRENCY)
+    fmt = params.format.value if hasattr(params.format, "value") else str(params.format)
+    ext = "jpg" if fmt == "JPEG" else "png"
+    content_type = "image/jpeg" if fmt == "JPEG" else "image/png"
+
+    async def _upload_one(pr: PageResult) -> PageResult:
+        async with semaphore:
+            file_path = _resolve_local_path(pr.download_url)
+            if not file_path.exists():
+                pr.upload_error = f"Local file not found: {file_path}"
+                logger.error("upload_file_missing", page=pr.page_number, path=str(file_path))
+                return pr
+            try:
+                file_bytes = file_path.read_bytes()
+                if file_path.stem == "stitched":
+                    filename = f"{job_id}_stitched.{ext}"
+                else:
+                    filename = f"{job_id}_page_{pr.page_number:04d}.{ext}"
+                info = await upload_file_to_service(file_bytes, filename, content_type)
+                pr.file_id = info.file_id
+                pr.file_url = info.file_url
+                logger.debug("page_uploaded", job_id=job_id, page=pr.page_number, file_id=info.file_id)
+            except Exception as exc:
+                pr.upload_error = str(exc)
+                logger.error("page_upload_failed", job_id=job_id, page=pr.page_number, error=str(exc))
+            return pr
+
+    updated_pages = list(await asyncio.gather(*[_upload_one(pr) for pr in pages]))
+
+    succeeded = [p for p in updated_pages if p.file_id]
+    failed    = [p for p in updated_pages if p.upload_error]
+
+    summary = UploadSummary(
+        upload_mode="per_page",
+        total_uploaded=len(succeeded),
+        failed_uploads=len(failed),
+        files=[
+            UploadedFileInfo(
+                file_id=p.file_id or "",
+                file_url=p.file_url or "",
+                filename=(
+                    f"{job_id}_stitched.{ext}"
+                    if Path(_resolve_local_path(p.download_url)).stem == "stitched"
+                    else f"{job_id}_page_{p.page_number:04d}.{ext}"
+                ),
+                size_bytes=p.size_bytes,
+            )
+            for p in succeeded
+        ],
+    )
+
+    logger.info("upload_complete", job_id=job_id, uploaded=len(succeeded), failed=len(failed))
+    return updated_pages, summary
+
+
+def _resolve_local_path(download_url: str) -> Path:
+    parts = download_url.lstrip("/").split("/")
+    if len(parts) >= 5:
+        return Path(settings.OUTPUT_DIR) / parts[3] / parts[4]
+    raise ValueError(f"Cannot resolve local path from URL: {download_url}")

app/utils/cleanup.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+import asyncio
+import shutil
+import time
+from pathlib import Path
+
+import structlog
+
+from app.core.config import settings
+
+logger = structlog.get_logger(__name__)
+
+
+async def cleanup_loop() -> None:
+    output_dir = Path(settings.OUTPUT_DIR)
+    while True:
+        await asyncio.sleep(600)
+        await _sweep(output_dir)
+
+
+async def _sweep(output_dir: Path) -> None:
+    if not output_dir.exists():
+        return
+
+    now = time.time()
+    removed = 0
+
+    for job_dir in output_dir.iterdir():
+        if not job_dir.is_dir():
+            continue
+        if now - job_dir.stat().st_mtime > settings.OUTPUT_TTL_SECONDS:
+            try:
+                shutil.rmtree(job_dir)
+                removed += 1
+            except Exception:
+                logger.exception("cleanup_error", path=str(job_dir))
+
+    if removed:
+        logger.info("cleanup_complete", removed_jobs=removed)

app/utils/validators.py

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+import io
+import ipaddress
+import socket
+from urllib.parse import urlparse
+
+import httpx
+import structlog
+from fastapi import UploadFile
+
+from app.core.config import settings
+from app.core.exceptions import FileTooLargeError, InvalidFileTypeError, InvalidParameterError
+
+logger = structlog.get_logger(__name__)
+
+_PDF_MAGIC = b"%PDF"
+_MAX_URL_REDIRECTS = 5
+_ALLOWED_SCHEMES = {"http", "https"}
+_PRIVATE_RANGES = [
+    ipaddress.ip_network("10.0.0.0/8"),
+    ipaddress.ip_network("172.16.0.0/12"),
+    ipaddress.ip_network("192.168.0.0/16"),
+    ipaddress.ip_network("127.0.0.0/8"),
+    ipaddress.ip_network("169.254.0.0/16"),
+    ipaddress.ip_network("::1/128"),
+    ipaddress.ip_network("fc00::/7"),
+]
+
+
+async def read_and_validate_pdf(upload: UploadFile) -> bytes:
+    buffer = io.BytesIO()
+    total = 0
+    chunk_size = 64 * 1024
+
+    while chunk := await upload.read(chunk_size):
+        total += len(chunk)
+        if total > settings.MAX_FILE_SIZE_BYTES:
+            raise FileTooLargeError(settings.MAX_FILE_SIZE_MB)
+        buffer.write(chunk)
+
+    if total == 0:
+        raise InvalidFileTypeError("Uploaded file is empty")
+
+    pdf_bytes = buffer.getvalue()
+    _assert_pdf_magic(pdf_bytes)
+    logger.debug("pdf_upload_validated", size_bytes=total)
+    return pdf_bytes
+
+
+async def fetch_pdf_from_url(url: str) -> bytes:
+    _validate_url_scheme(url)
+    _validate_url_host(url)
+
+    async with httpx.AsyncClient(
+        follow_redirects=True,
+        max_redirects=_MAX_URL_REDIRECTS,
+        timeout=httpx.Timeout(30.0, connect=10.0),
+    ) as client:
+        try:
+            response = await client.get(url)
+            response.raise_for_status()
+        except httpx.HTTPStatusError as exc:
+            raise InvalidFileTypeError(
+                f"Failed to fetch PDF from URL: HTTP {exc.response.status_code}"
+            ) from exc
+        except httpx.RequestError as exc:
+            raise InvalidFileTypeError(f"Failed to fetch PDF from URL: {exc}") from exc
+
+        _validate_url_host(str(response.url))
+
+        content_length = response.headers.get("content-length")
+        if content_length and int(content_length) > settings.MAX_FILE_SIZE_BYTES:
+            raise FileTooLargeError(settings.MAX_FILE_SIZE_MB)
+
+        pdf_bytes = b""
+        async for chunk in response.aiter_bytes(chunk_size=64 * 1024):
+            pdf_bytes += chunk
+            if len(pdf_bytes) > settings.MAX_FILE_SIZE_BYTES:
+                raise FileTooLargeError(settings.MAX_FILE_SIZE_MB)
+
+    if len(pdf_bytes) == 0:
+        raise InvalidFileTypeError("URL returned an empty response")
+
+    _assert_pdf_magic(pdf_bytes)
+    logger.debug("pdf_url_validated", url=url, size_bytes=len(pdf_bytes))
+    return pdf_bytes
+
+
+def _assert_pdf_magic(data: bytes) -> None:
+    if not data.startswith(_PDF_MAGIC):
+        raise InvalidFileTypeError(
+            "File does not appear to be a valid PDF (missing PDF magic bytes)"
+        )
+
+
+def _validate_url_scheme(url: str) -> None:
+    parsed = urlparse(url)
+    if parsed.scheme.lower() not in _ALLOWED_SCHEMES:
+        raise InvalidParameterError(
+            f"URL scheme '{parsed.scheme}' is not allowed. Only http and https are permitted."
+        )
+
+
+def _validate_url_host(url: str) -> None:
+    parsed = urlparse(url)
+    hostname = parsed.hostname
+    if not hostname:
+        raise InvalidParameterError("URL has no valid hostname")
+
+    try:
+        resolved_ip = socket.gethostbyname(hostname)
+        addr = ipaddress.ip_address(resolved_ip)
+    except (socket.gaierror, ValueError):
+        raise InvalidParameterError(f"Could not resolve hostname: {hostname}")
+
+    for private_range in _PRIVATE_RANGES:
+        if addr in private_range:
+            raise InvalidParameterError(
+                "Requests to private or loopback IP addresses are not permitted (SSRF protection)"
+            )

requirements.txt

@@ -23,3 +23,17 @@ spacy>=3.7.0
 # CLI banner
 pyfiglet>=1.0.2
 rich>=13.0.0
+
+# PDF rendering and conversion
+pdf2image>=1.17.0
+pypdf>=5.1.0
+
+# Structured logging and metrics
+structlog>=24.4.0
+prometheus-client>=0.21.1
+
+# Pydantic settings
+pydantic-settings>=2.7.1
+
+# Rate limiting (optional — requires Redis)
+redis[hiredis]>=6.4.0