#!/usr/bin/env python3
"""Fast docling server using DocumentConverter singleton.

A lightweight FastAPI server optimized for hybrid PDF processing:
1. Using a single DocumentConverter instance (no per-request initialization)
2. Returns only JSON (DoclingDocument format) - markdown/HTML generated by Java

Usage:
    opendataloader-pdf-hybrid [--port PORT] [--host HOST] [--ocr-lang LANG]
                              [--force-ocr | --no-ocr] [--ocr-engine ENGINE] [--psm N]
                              [--device DEVICE]
                              [--enrich-formula] [--enrich-picture-description]
                              [--max-file-size MB]

    # Default: http://localhost:5002
    opendataloader-pdf-hybrid

    # Custom port
    opendataloader-pdf-hybrid --port 5003

    # Disable OCR when the PDF already has reliable embedded text (#387)
    opendataloader-pdf-hybrid --no-ocr

    # Use Tesseract for a language EasyOCR doesn't support (#439)
    opendataloader-pdf-hybrid --ocr-engine tesseract --ocr-lang mal

    # Chinese + English OCR with force full-page OCR
    opendataloader-pdf-hybrid --ocr-lang "ch_sim,en" --force-ocr

    # Korean OCR
    opendataloader-pdf-hybrid --ocr-lang "ko"

    # Explicitly use Apple Silicon GPU (MPS)
    opendataloader-pdf-hybrid --device mps

    # Force CPU-only processing
    opendataloader-pdf-hybrid --device cpu

    # With formula enrichment (LaTeX extraction)
    opendataloader-pdf-hybrid --enrich-formula

    # With picture description (alt text generation)
    opendataloader-pdf-hybrid --enrich-picture-description

    # Combined: OCR + enrichments
    opendataloader-pdf-hybrid --ocr-lang "en" --enrich-formula --enrich-picture-description

API Endpoints:
    GET  /health              - Health check
    POST /v1/convert/file     - Convert PDF to JSON

The /v1/convert/file endpoint parameters:
    - files: PDF file (multipart/form-data)
    - page_ranges: Page range to process (optional)

Requirements:
    Install with hybrid extra: pip install opendataloader-pdf[hybrid]
"""

import argparse
import asyncio
import logging
import os
import re
import sys
import tempfile
import threading
import time
import traceback

# Enable docling per-step pipeline profiling (layout, ocr, table_structure, etc.)
# Must be set before docling settings singleton is instantiated.
os.environ.setdefault("DOCLING_DEBUG_PROFILE_PIPELINE_TIMINGS", "true")
from contextlib import asynccontextmanager
from typing import Any, Optional

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)

# Configuration
DEFAULT_HOST = "0.0.0.0"
DEFAULT_PORT = 5002
MAX_FILE_SIZE = 0  # No file size limit by default (0 = unlimited)
UPLOAD_CHUNK_SIZE = 1024 * 1024  # 1MB chunks for streaming upload

# OCR engine kinds we filter out from `--ocr-engine` choices.
# `kserve_v2_ocr` requires a separate KServe v2 / Triton inference server, which
# is unsuitable for the local hybrid mode this server is designed for.
# Single source of truth shared by `create_converter`, `main()` argparse, and tests
# to avoid drift between production and test code.
_OCR_ENGINE_DENYLIST = frozenset({"kserve_v2_ocr"})


def _non_negative_int(value: str) -> int:
    """Argparse type validator that rejects negative integers."""
    parsed = int(value)
    if parsed < 0:
        raise argparse.ArgumentTypeError("--max-file-size must be >= 0")
    return parsed


# Global converter instance (initialized on startup with CLI options)
converter = None

# Serialize converter.convert() calls. The converter singleton was designed for
# sequential use; this lock keeps that guarantee while allowing the event loop
# to stay responsive via asyncio.to_thread().
_convert_lock = threading.Lock()

# Regex matching lone surrogates (U+D800..U+DFFF) and null characters
_INVALID_UNICODE_RE = re.compile(r"[\ud800-\udfff\x00]")


def _extract_failed_pages_from_errors(errors: list[str]) -> list[int]:
    """Extract failed page numbers from error messages.

    Docling error messages follow the pattern "Page N: <error>" (e.g.,
    "Page 26: std::bad_alloc") or, when no error description is available,
    a bare "Page N" with no colon. Even when docling includes failed pages
    in the pages dict as empty entries, the error messages reliably
    indicate which pages actually failed.

    Args:
        errors: List of error message strings.

    Returns:
        Sorted list of 1-indexed page numbers that failed.
    """
    failed = set()
    page_pattern = re.compile(r"^Page\s+(\d+)(?::|$)")
    for msg in errors:
        m = page_pattern.match(msg)
        if m:
            failed.add(int(m.group(1)))
    return sorted(failed)


def extract_timings(result: Any) -> dict[str, Any]:
    """Extract per-step pipeline timings from a Docling ConversionResult.

    Requires DOCLING_DEBUG_PROFILE_PIPELINE_TIMINGS=true (set at module level).
    Returns a dict keyed by step name (e.g. "layout", "ocr", "table_structure")
    with total_s, avg_s, and count for each step.
    """
    timings_out: dict[str, Any] = {}
    raw_timings = getattr(result, "timings", None)
    if not raw_timings:
        return timings_out
    for name, item in raw_timings.items():
        try:
            timings_out[name] = {
                "total_s": round(item.total(), 4),
                "avg_s": round(item.avg(), 4) if item.count > 0 else 0.0,
                "count": item.count,
            }
        except Exception:
            # Gracefully skip if ProfilingItem API changes
            pass
    return timings_out


def build_conversion_response(
    status_value: str,
    json_content: dict,
    processing_time: float,
    errors: list[str],
    requested_pages: tuple[int, int] | None,
    total_pages: int | None = None,
    timings: dict[str, Any] | None = None,
) -> dict:
    """Build a structured conversion response with status and failed page info.

    When Docling encounters errors (e.g., std::bad_alloc in PDF preprocessing),
    it may still include failed pages as empty entries in the pages dict.
    This function combines two strategies to detect failed pages:
    1. Parse page numbers from error messages ("Page N: <error>")
    2. Detect pages missing from the output pages dict (gap detection)
    Both results are merged (union) since each catches a different failure mode.

    Args:
        status_value: Docling ConversionStatus value as string (e.g., "success", "partial_success").
        json_content: The exported document dict from Docling.
        processing_time: Time taken for conversion in seconds.
        errors: List of error message strings from Docling.
        requested_pages: Tuple of (start, end) 1-indexed page range, or None for all pages.
        total_pages: Total page count of the input document (from Docling InputDocument).
                     Used to detect boundary page failures when requested_pages is None.

    Returns:
        Response dict with status, document, errors, failed_pages, and processing_time.
    """
    failed_pages: list[int] = []

    if status_value == "partial_success":
        # Strategy 1: Extract failed pages from error messages (reliable —
        # docling may include failed pages as empty entries in the pages dict,
        # making gap detection ineffective)
        error_failed = set(_extract_failed_pages_from_errors(errors))

        # Strategy 2: Detect pages missing from the pages dict (catches
        # failures that don't produce "Page N:" error messages)
        gap_failed: set[int] = set()
        pages_dict = json_content.get("pages", {})
        present_pages = set()
        for k in pages_dict.keys():
            try:
                present_pages.add(int(k))
            except (ValueError, TypeError):
                logger.warning("Unexpected non-integer page key in Docling output: %r", k)

        if requested_pages:
            expected_pages = set(range(requested_pages[0], requested_pages[1] + 1))
        elif total_pages is not None:
            expected_pages = set(range(1, total_pages + 1))
        elif present_pages:
            logger.warning(
                "No page range or total_pages available; boundary page failures cannot be detected"
            )
            expected_pages = set(range(min(present_pages), max(present_pages) + 1))
        else:
            expected_pages = set()

        gap_failed = expected_pages - present_pages

        # Union: each strategy catches a different failure mode
        failed_pages = sorted(error_failed | gap_failed)

    response: dict[str, Any] = {
        "status": status_value,
        "document": {
            "json_content": json_content,
        },
        "processing_time": processing_time,
        "errors": errors,
        "failed_pages": failed_pages,
    }

    if timings:
        response["timings"] = timings

    return response


def sanitize_unicode(data: Any) -> Any:
    """Recursively replace lone surrogates and null characters with U+FFFD.

    Docling OCR can produce lone surrogates (U+D800-U+DFFF) and null characters
    from PDFs with malformed font encodings. These pass through json.dumps(ensure_ascii=False)
    but fail on .encode('utf-8') in Starlette's JSONResponse.render(), causing
    UnicodeEncodeError and a 500 response.

    This mirrors the Java-side TextProcessor.replaceUndefinedCharacters().

    Args:
        data: Arbitrary data structure (dict, list, str, or primitive) from
              Docling's export_to_dict() output.

    Returns:
        The same structure with problematic characters replaced by U+FFFD.
    """
    if isinstance(data, str):
        return _INVALID_UNICODE_RE.sub("\ufffd", data)
    if isinstance(data, dict):
        return {k: sanitize_unicode(v) for k, v in data.items()}
    if isinstance(data, list):
        return [sanitize_unicode(item) for item in data]
    return data


def _get_loop_setting() -> str:
    """Return the uvicorn event loop setting appropriate for the current platform.

    uvloop is not supported on Windows, so we force 'asyncio' there.
    On other platforms, 'auto' lets uvicorn use uvloop if available.
    """
    if sys.platform == "win32":
        return "asyncio"
    return "auto"


def _check_dependencies():
    """Check if hybrid dependencies are installed."""
    missing = []
    try:
        import uvicorn  # noqa: F401
    except ImportError:
        missing.append("uvicorn")
    try:
        import fastapi  # noqa: F401
    except ImportError:
        missing.append("fastapi")
    try:
        import docling  # noqa: F401
    except ImportError:
        missing.append("docling")

    if missing:
        raise ImportError(
            f"Missing dependencies: {', '.join(missing)}. "
            "Install with: pip install opendataloader-pdf[hybrid]"
        )


def _check_ocr_engine_available(engine_kind: str) -> tuple[bool, str]:
    """Probe runtime prerequisites for the selected OCR engine.

    Called at startup so a missing binary or Python package surfaces before the
    first request rather than mid-conversion. Returns (ok, error_message).

    `easyocr` ships with the `[hybrid]` extra and `auto` defers engine choice to
    docling at conversion time, so both are treated as always available. Other
    engines pull in extra system or Python dependencies that are not declared
    in pyproject and must be installed by the user.
    """
    import importlib.util
    import shutil

    if engine_kind in ("auto", "easyocr"):
        return True, ""

    if engine_kind == "tesseract":
        # docling's TesseractCliOcrOptions shells out to the `tesseract` binary.
        if shutil.which("tesseract") is None:
            return False, (
                "OCR engine 'tesseract' selected but the `tesseract` binary was not "
                "found on PATH. Install Tesseract for your platform "
                "(e.g., `brew install tesseract`, `apt install tesseract-ocr`, "
                "`choco install tesseract`) and ensure the language data files for "
                "your --ocr-lang are installed (e.g., `tesseract-ocr-mal` for Malayalam)."
            )
        return True, ""

    if engine_kind == "tesserocr":
        if importlib.util.find_spec("tesserocr") is None:
            return False, (
                "OCR engine 'tesserocr' selected but the `tesserocr` Python package "
                "is not installed. Install it with `pip install tesserocr` "
                "(libtesseract must also be available at runtime; on Windows install "
                "a wheel that matches your installed Tesseract major version)."
            )
        return True, ""

    if engine_kind == "rapidocr":
        # rapidocr's default backend is ONNX Runtime; docling 2.91 imports
        # `from rapidocr import EngineType, RapidOCR` and routes through it.
        if importlib.util.find_spec("rapidocr") is None:
            return False, (
                "OCR engine 'rapidocr' selected but the `rapidocr` Python package "
                "is not installed. Install it with `pip install rapidocr onnxruntime`."
            )
        if importlib.util.find_spec("onnxruntime") is None:
            return False, (
                "OCR engine 'rapidocr' selected but `onnxruntime` is not installed. "
                "rapidocr's default backend requires it. Install with "
                "`pip install onnxruntime` (or, for GPU, `pip install onnxruntime-gpu`)."
            )
        return True, ""

    if engine_kind == "ocrmac":
        if sys.platform != "darwin":
            return False, (
                "OCR engine 'ocrmac' selected but this is not macOS "
                f"(platform={sys.platform!r}). The ocrmac engine uses Apple's Vision "
                "framework and only runs on macOS."
            )
        if importlib.util.find_spec("ocrmac") is None:
            return False, (
                "OCR engine 'ocrmac' selected but the `ocrmac` Python package "
                "is not installed. Install it with `pip install ocrmac`."
            )
        return True, ""

    # Unknown engine kind — fail closed. argparse `choices` filters the CLI
    # surface, so this branch is reachable via direct programmatic calls or
    # when docling registers a new engine kind we haven't added a probe for.
    # Falling through to True would silently green-light unverified engines
    # and defer the failure to the first conversion request.
    return False, (
        f"OCR engine {engine_kind!r} is not recognized by the availability "
        "probe. Add a probe branch for this engine kind in "
        "`_check_ocr_engine_available()`."
    )


def create_converter(
    force_full_page_ocr: bool = False,
    disable_ocr: bool = False,
    ocr_engine: str = "easyocr",
    psm: int | None = None,
    ocr_lang: list[str] | None = None,
    enrich_formula: bool = False,
    enrich_picture_description: bool = False,
    picture_description_prompt: str | None = None,
    device: str = "auto",
):
    """Create a DocumentConverter with the specified options.

    Args:
        force_full_page_ocr: If True, force OCR on all pages regardless of text content.
                            If False (default), OCR only where needed.
        disable_ocr: If True, disable OCR entirely (do_ocr=False). Useful when input PDFs
                    already have reliable embedded text — prevents duplicate text extraction
                    from images (charts, diagrams, screenshots). Mutually exclusive with
                    force_full_page_ocr at the CLI level.
        ocr_engine: OCR engine kind to use. Engine availability is delegated to docling's
                    factory (`get_ocr_factory`). Each engine has its own license, language
                    coverage, and accuracy characteristics; this project does not validate
                    engine accuracy. Default: "easyocr" (preserves prior behavior).
        psm: Tesseract Page Segmentation Mode. Only applied when ocr_engine is
             "tesseract" or "tesserocr". Ignored otherwise. Range and semantics
             are owned by Tesseract / docling; see `tesseract --help-extra`.
        ocr_lang: List of OCR language codes. The code system depends on the chosen engine
                  (EasyOCR uses 'ko,en', Tesseract uses 'kor,eng', RapidOCR uses
                  'english,chinese', ocrmac uses 'en-US'). If None, the engine's default
                  languages are used.
        enrich_formula: If True, enable formula enrichment (LaTeX extraction).
        enrich_picture_description: If True, enable picture description (alt text generation).
        picture_description_prompt: Custom prompt forwarded to the VLM. If None or blank/whitespace-only, docling's default prompt is used.
        device: Accelerator device for model inference. Options: "auto", "cpu", "cuda", "mps", "xpu".
                "auto" lets Docling select the best available device. Default: "auto".
    """
    from docling.datamodel.accelerator_options import AcceleratorOptions
    from docling.datamodel.base_models import InputFormat
    from docling.datamodel.pipeline_options import (
        AcceleratorOptions,
        PdfPipelineOptions,
        PictureDescriptionVlmOptions,
        TableFormerMode,
        TableStructureOptions,
        TesseractCliOcrOptions,
        TesseractOcrOptions,
    )
    from docling.document_converter import DocumentConverter, PdfFormatOption
    from docling.models.factories import get_ocr_factory

    # Delegate engine selection to docling's factory. We block external plugins for
    # security/reproducibility; the module-level _OCR_ENGINE_DENYLIST filters
    # engines unsuitable for hybrid local mode (e.g., remote inference servers).
    ocr_factory = get_ocr_factory(allow_external_plugins=False)
    if ocr_engine in _OCR_ENGINE_DENYLIST:
        # Programmatic callers (importing this module) bypass argparse `choices`,
        # so enforce the denylist here too. Without this, the module-level claim
        # that `_OCR_ENGINE_DENYLIST` is shared across CLI and create_converter
        # would only be true at the CLI layer.
        available = sorted(set(ocr_factory.registered_kind) - _OCR_ENGINE_DENYLIST)
        raise ValueError(
            f"OCR engine '{ocr_engine}' is not supported in hybrid local mode "
            f"(filtered by _OCR_ENGINE_DENYLIST). Available engines: {available}"
        )
    try:
        ocr_options = ocr_factory.create_options(
            kind=ocr_engine,
            force_full_page_ocr=force_full_page_ocr,
        )
    except RuntimeError as e:
        # Library-friendly error type so programmatic callers can catch and retry
        # with a different engine. main() relies on argparse `choices` to gate
        # invalid CLI input, so this branch is reached only via direct calls.
        available = sorted(set(ocr_factory.registered_kind) - _OCR_ENGINE_DENYLIST)
        raise ValueError(
            f"Unknown ocr_engine '{ocr_engine}': {e}\nAvailable engines: {available}"
        ) from e

    if ocr_lang:
        ocr_options.lang = ocr_lang

    # Tesseract-only: Page Segmentation Mode
    if psm is not None and isinstance(
        ocr_options, (TesseractOcrOptions, TesseractCliOcrOptions)
    ):
        ocr_options.psm = psm

    # Configure picture description options with custom prompt.
    # When picture_description_prompt is None or blank, omit the field so
    # docling's built-in default prompt is used. A blank string would otherwise
    # silently produce empty-prompt output — same class of silent-flag bug
    # as PDFDLOSP-20 reported.
    picture_description_options = None
    if enrich_picture_description:
        vlm_kwargs: dict[str, Any] = {
            "repo_id": "HuggingFaceTB/SmolVLM-256M-Instruct",
        }
        if picture_description_prompt and picture_description_prompt.strip():
            vlm_kwargs["prompt"] = picture_description_prompt
        picture_description_options = PictureDescriptionVlmOptions(**vlm_kwargs)

    pipeline_kwargs = {
        "do_ocr": not disable_ocr,
        "do_table_structure": True,
        "ocr_options": ocr_options,
        "table_structure_options": TableStructureOptions(mode=TableFormerMode.ACCURATE),
        "do_formula_enrichment": enrich_formula,
        "do_picture_description": enrich_picture_description,
        "generate_picture_images": enrich_picture_description,
        "accelerator_options": AcceleratorOptions(device=device),
    }
    if picture_description_options is not None:
        pipeline_kwargs["picture_description_options"] = picture_description_options

    if device != "auto":
        pipeline_kwargs["accelerator_options"] = AcceleratorOptions(device=device)

    pipeline_options = PdfPipelineOptions(**pipeline_kwargs)

    return DocumentConverter(
        format_options={
            InputFormat.PDF: PdfFormatOption(pipeline_options=pipeline_options)
        }
    )


def create_app(
    force_ocr: bool = False,
    disable_ocr: bool = False,
    ocr_engine: str = "easyocr",
    psm: int | None = None,
    ocr_lang: list[str] | None = None,
    enrich_formula: bool = False,
    enrich_picture_description: bool = False,
    picture_description_prompt: str | None = None,
    max_file_size: int = MAX_FILE_SIZE,
    device: str = "auto",
):
    """Create and configure the FastAPI application.

    Args:
        force_ocr: If True, force full-page OCR on all pages.
        disable_ocr: If True, disable OCR entirely. Mutually exclusive with force_ocr at CLI.
        ocr_engine: OCR engine kind (delegated to docling's get_ocr_factory). Default: "easyocr".
        psm: Tesseract Page Segmentation Mode. Only applied for Tesseract engines.
        ocr_lang: List of OCR language codes (engine-specific format).
        enrich_formula: If True, enable formula enrichment (LaTeX extraction).
        enrich_picture_description: If True, enable picture description (alt text generation).
        picture_description_prompt: Custom prompt forwarded to the VLM. If None or blank/whitespace-only, docling's default prompt is used.
        max_file_size: Maximum file size in bytes. 0 means no limit (default).
        device: Accelerator device for model inference ("auto", "cpu", "cuda", "mps", "xpu").
    """
    from fastapi import FastAPI, File, Form, UploadFile
    from fastapi.responses import JSONResponse

    # Profile converters: initialized lazily on first /v1/profile/file request
    profile_converters: dict[str, Any] = {}

    @asynccontextmanager
    async def lifespan(_app: FastAPI):
        """Lifespan context manager for startup and shutdown events."""
        global converter
        lang_str = ",".join(ocr_lang) if ocr_lang else "default"
        enrichments = []
        if enrich_formula:
            enrichments.append("formula")
        if enrich_picture_description:
            enrichments.append("picture-description")
        enrichment_str = ",".join(enrichments) if enrichments else "none"
        logger.info(
            f"Initializing DocumentConverter "
            f"(do_ocr={not disable_ocr}, ocr_engine={ocr_engine}, force_ocr={force_ocr}, "
            f"lang={lang_str}, enrichments={enrichment_str}, device={device})..."
        )
        start = time.perf_counter()

        converter = create_converter(
            force_full_page_ocr=force_ocr,
            disable_ocr=disable_ocr,
            ocr_engine=ocr_engine,
            psm=psm,
            ocr_lang=ocr_lang,
            enrich_formula=enrich_formula,
            enrich_picture_description=enrich_picture_description,
            picture_description_prompt=picture_description_prompt,
            device=device,
        )

        elapsed = time.perf_counter() - start
        logger.info(f"DocumentConverter initialized in {elapsed:.2f}s")
        yield
        # Cleanup on shutdown (if needed)

    app = FastAPI(
        title="Docling Fast Server",
        description="Fast PDF conversion using docling SDK with singleton pattern",
        version="1.0.0",
        lifespan=lifespan,
    )

    @app.get("/health")
    def health():
        """Health check endpoint."""
        return {"status": "ok"}

    @app.post("/v1/convert/file")
    async def convert_file(
        files: UploadFile = File(...),
        page_ranges: Optional[str] = Form(default=None),
    ):
        """Convert PDF file to JSON (DoclingDocument format).

        Only JSON output is provided - markdown and HTML are generated by
        Java processors for consistent reading order application.

        Args:
            files: The PDF file to convert
            page_ranges: Page range string "start-end" (e.g., "1-5") (optional)

        Returns:
            JSON response with document content.
        """
        global converter

        if converter is None:
            return JSONResponse(
                {"status": "failure", "errors": ["Server not initialized"]},
                status_code=503,
            )

        # Parse page_ranges string to tuple
        page_range_tuple = None
        if page_ranges:
            try:
                parts = page_ranges.split("-")
                if len(parts) == 2:
                    page_range_tuple = (int(parts[0]), int(parts[1]))
            except ValueError:
                pass

        # Stream upload to temp file and enforce size incrementally
        tmp_path = None
        total_size = 0
        with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
            tmp_path = tmp.name
            while True:
                chunk = await files.read(UPLOAD_CHUNK_SIZE)
                if not chunk:
                    break
                total_size += len(chunk)
                if max_file_size > 0 and total_size > max_file_size:
                    tmp.close()
                    os.unlink(tmp_path)
                    return JSONResponse(
                        {
                            "status": "failure",
                            "errors": [f"File size exceeds maximum allowed ({max_file_size // (1024*1024)}MB)"],
                        },
                        status_code=413,
                    )
                tmp.write(chunk)

        try:
            def _do_convert():
                with _convert_lock:
                    t0 = time.perf_counter()
                    if page_range_tuple:
                        res = converter.convert(tmp_path, page_range=page_range_tuple)
                    else:
                        res = converter.convert(tmp_path)
                    return res, time.perf_counter() - t0

            result, processing_time = await asyncio.to_thread(_do_convert)

            # Export to JSON (DoclingDocument format)
            json_content = result.document.export_to_dict()

            # Sanitize lone surrogates and null chars from OCR output to prevent
            # UnicodeEncodeError in Starlette's JSONResponse.render()
            json_content = sanitize_unicode(json_content)

            # Extract status and errors from Docling ConversionResult
            from docling.datamodel.base_models import ConversionStatus

            status_value = result.status.value if hasattr(result.status, "value") else str(result.status)
            errors = [getattr(e, "error_message", str(e)) for e in result.errors] if result.errors else []

            # Get total page count for accurate failed-page detection
            input_page_count = getattr(result.input, "page_count", None) if result.input else None

            if result.status == ConversionStatus.PARTIAL_SUCCESS:
                logger.warning(
                    "Docling returned partial_success: %d error(s), failed_pages will be reported",
                    len(errors),
                )

            # Extract per-step pipeline timings (layout, ocr, table_structure, etc.)
            step_timings = extract_timings(result)

            response = build_conversion_response(
                status_value=status_value,
                json_content=json_content,
                processing_time=processing_time,
                errors=errors,
                requested_pages=page_range_tuple,
                total_pages=input_page_count,
                timings=step_timings,
            )

            return JSONResponse(response)

        except Exception as e:
            logger.error(f"PDF conversion failed: {e}\n{traceback.format_exc()}")
            return JSONResponse(
                {
                    "status": "failure",
                    "errors": ["PDF conversion failed. Check server logs for details."],
                },
                status_code=500,
            )
        finally:
            if tmp_path and os.path.exists(tmp_path):
                os.unlink(tmp_path)

    def _ensure_profile_converters():
        """Lazily initialize profile converters on first use."""
        if profile_converters:
            return
        profiles = {
            "base": dict(enrich_formula=False, enrich_picture_description=False),
            "picture": dict(enrich_formula=False, enrich_picture_description=True),
            "formula": dict(enrich_formula=True, enrich_picture_description=False),
        }
        for name, opts in profiles.items():
            logger.info(f"Initializing profile converter: {name} ({opts})")
            t0 = time.perf_counter()
            profile_converters[name] = create_converter(
                force_full_page_ocr=force_ocr,
                disable_ocr=disable_ocr,
                ocr_engine=ocr_engine,
                psm=psm,
                ocr_lang=ocr_lang,
                picture_description_prompt=picture_description_prompt,
                device=device,
                **opts,
            )
            logger.info(f"  {name} initialized in {time.perf_counter() - t0:.2f}s")

    @app.post("/v1/profile/file")
    async def profile_file(
        files: UploadFile = File(...),
    ):
        """Run the same PDF through base / +picture / +picture+formula converters.

        Returns per-profile timings for cost comparison.
        """
        _ensure_profile_converters()

        # Stream upload to temp file
        tmp_path = None
        with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
            tmp_path = tmp.name
            while True:
                chunk = await files.read(UPLOAD_CHUNK_SIZE)
                if not chunk:
                    break
                tmp.write(chunk)

        try:
            results = {}
            for profile_name, conv in profile_converters.items():
                def _run(c=conv):
                    with _convert_lock:
                        t0 = time.perf_counter()
                        res = c.convert(tmp_path)
                        return res, time.perf_counter() - t0

                result, wall_time = await asyncio.to_thread(_run)
                timings = extract_timings(result)

                # Count pictures and formulas
                json_content = result.document.export_to_dict()
                pictures = json_content.get("pictures", [])
                pics_total = len(pictures)
                pics_described = sum(
                    1 for p in pictures
                    if p.get("captions") or p.get("annotations")
                )
                texts = json_content.get("texts", [])
                formulas_total = sum(1 for t in texts if t.get("label") == "formula")
                tables_total = len(json_content.get("tables", []))

                results[profile_name] = {
                    "wall_time_s": round(wall_time, 3),
                    "timings": timings,
                    "pictures": {
                        "total": pics_total,
                        "with_description": pics_described,
                    },
                    "formulas": formulas_total,
                    "tables": tables_total,
                }

            return JSONResponse({"status": "success", "profiles": results})

        except Exception as e:
            logger.error(f"Profile failed: {e}\n{traceback.format_exc()}")
            return JSONResponse(
                {"status": "failure", "errors": ["Internal server error"]},
                status_code=500,
            )
        finally:
            if tmp_path and os.path.exists(tmp_path):
                os.unlink(tmp_path)

    return app


def main():
    """Run the server."""
    _check_dependencies()
    import uvicorn

    parser = argparse.ArgumentParser(description="Docling Fast Server for opendataloader-pdf")
    parser.add_argument(
        "--host",
        default=DEFAULT_HOST,
        help=f"Host to bind to (default: {DEFAULT_HOST})",
    )
    parser.add_argument(
        "--port",
        type=int,
        default=DEFAULT_PORT,
        help=f"Port to bind to (default: {DEFAULT_PORT})",
    )
    parser.add_argument(
        "--log-level",
        default="info",
        choices=["debug", "info", "warning", "error"],
        help="Log level (default: info)",
    )
    # OCR mode: --force-ocr and --no-ocr are mutually exclusive (opposite intents)
    ocr_mode_group = parser.add_mutually_exclusive_group()
    ocr_mode_group.add_argument(
        "--force-ocr",
        action="store_true",
        help="Force full-page OCR on all pages, even pages with embedded text. "
             "Use for scanned PDFs where embedded text is unreliable. "
             "Mutually exclusive with --no-ocr.",
    )
    ocr_mode_group.add_argument(
        "--no-ocr",
        action="store_true",
        help="Disable OCR entirely. Use when input PDFs already have reliable embedded text — "
             "prevents duplicate text extraction from images (charts, diagrams, screenshots). "
             "Mutually exclusive with --force-ocr.",
    )

    # Engine selection — delegated to docling's factory.
    # Choices are computed at parse time; the module-level _OCR_ENGINE_DENYLIST
    # filters engines unsuitable for hybrid local mode (single source of truth
    # shared with create_converter and tests).
    try:
        from docling.models.factories import get_ocr_factory as _get_ocr_factory
        _ocr_engine_choices = sorted(
            set(_get_ocr_factory(allow_external_plugins=False).registered_kind)
            - _OCR_ENGINE_DENYLIST
        )
    except ImportError:
        _ocr_engine_choices = ["easyocr"]
    parser.add_argument(
        "--ocr-engine",
        type=str,
        default="easyocr",
        choices=_ocr_engine_choices,
        help=f"OCR engine. Available: {', '.join(_ocr_engine_choices)}. "
             "Use 'auto' for engine auto-selection per page (delegates the choice to docling). "
             "Each engine has its own license, language coverage, and accuracy characteristics; "
             "this server does not validate engine accuracy. "
             "Default: easyocr (preserves prior behavior).",
    )
    parser.add_argument(
        "--psm",
        type=int,
        default=None,
        help="Tesseract Page Segmentation Mode. Applied only when --ocr-engine is "
             "'tesseract' or 'tesserocr'; ignored for other engines. See "
             "`tesseract --help-extra` for valid values.",
    )
    parser.add_argument(
        "--ocr-lang",
        type=str,
        default=None,
        help="OCR languages (comma-separated). Code system depends on --ocr-engine: "
             "EasyOCR uses ISO 639-1 ('ko,en'), Tesseract uses ISO 639-2 ('kor,eng'), "
             "RapidOCR uses 'english,chinese', ocrmac uses BCP-47 ('en-US'). "
             "If omitted, the engine's default languages are used.",
    )
    parser.add_argument(
        "--enrich-formula",
        action="store_true",
        default=False,
        help="Enable formula enrichment model (LaTeX extraction)",
    )
    parser.add_argument(
        "--no-enrich-formula",
        action="store_false",
        dest="enrich_formula",
    )
    parser.add_argument(
        "--enrich-picture-description",
        action="store_true",
        default=False,
        help="Enable picture description model (alt text generation using SmolVLM)",
    )
    parser.add_argument(
        "--no-enrich-picture-description",
        action="store_false",
        dest="enrich_picture_description",
    )
    parser.add_argument(
        "--picture-description-prompt",
        type=str,
        default=None,
        help="Custom prompt for picture description. If unset or blank/whitespace-only, uses docling's default prompt.",
    )
    parser.add_argument(
        "--max-file-size",
        type=_non_negative_int,
        default=MAX_FILE_SIZE,
        help="Maximum upload file size in MB. 0 means no limit (default: 0).",
    )
    parser.add_argument(
        "--device",
        type=str,
        default="auto",
        choices=["auto", "cpu", "cuda", "mps", "xpu"],
        help="Accelerator device for model inference: auto (default), cpu, cuda, mps (Apple Silicon), xpu (Intel GPU).",
    )
    args = parser.parse_args()

    # Parse ocr_lang
    ocr_lang = None
    if args.ocr_lang:
        ocr_lang = [lang.strip() for lang in args.ocr_lang.split(",") if lang.strip()]

    # Warn when --no-ocr makes other OCR flags inert. We let it through (rather
    # than failing) because the combination is unambiguous — OCR is off — but
    # silently dropping user-supplied flags is unfriendly.
    if args.no_ocr:
        # argparse cannot distinguish explicit `--ocr-engine easyocr` from the
        # implicit default; peek at argv so an explicitly-typed default value
        # is still treated as a user-supplied (inert) flag and reported.
        argv = sys.argv[1:]
        ocr_engine_explicit = any(
            t == "--ocr-engine" or t.startswith("--ocr-engine=") for t in argv
        )
        ignored = []
        if ocr_engine_explicit:
            ignored.append(f"--ocr-engine {args.ocr_engine}")
        if ocr_lang:
            ignored.append(f"--ocr-lang {args.ocr_lang}")
        if args.psm is not None:
            ignored.append(f"--psm {args.psm}")
        if ignored:
            logger.warning(
                "OCR is disabled (--no-ocr); the following flag(s) will have no "
                "effect: %s",
                ", ".join(ignored),
            )

    # Probe engine availability at startup (only when OCR is on). A missing
    # `tesseract` binary or Python package surfaces here as a clear, actionable
    # error rather than as a deferred runtime exception during the first request.
    if not args.no_ocr:
        ok, err = _check_ocr_engine_available(args.ocr_engine)
        if not ok:
            logger.error(err)
            sys.exit(2)

    # Build enrichment log message
    enrichments = []
    if args.enrich_formula:
        enrichments.append("formula")
    if args.enrich_picture_description:
        enrichments.append("picture-description")

    # Log accelerator detection
    try:
        import torch
        if torch.cuda.is_available():
            gpu_name = torch.cuda.get_device_name(0)
            cuda_version = torch.version.cuda
            logger.info(f"Accelerator: CUDA — {gpu_name} (CUDA {cuda_version})")
        elif hasattr(torch.backends, "mps") and torch.backends.mps.is_available():
            logger.info("Accelerator: MPS (Apple Silicon)")
        elif hasattr(torch, "xpu") and torch.xpu.is_available():
            logger.info("Accelerator: XPU (Intel GPU)")
        else:
            logger.info("Accelerator: CPU (no GPU detected)")
    except ImportError:
        logger.info("Accelerator: CPU (PyTorch not installed)")
    if args.device != "auto":
        logger.info(f"Device override: --device {args.device}")

    # Convert MB to bytes (0 stays 0 = unlimited)
    max_file_size_bytes = args.max_file_size * 1024 * 1024 if args.max_file_size > 0 else 0

    logger.info(f"Starting Docling Fast Server on http://{args.host}:{args.port}")
    psm_str = f", psm={args.psm}" if args.psm is not None else ""
    logger.info(
        f"OCR settings: do_ocr={not args.no_ocr}, ocr_engine={args.ocr_engine}, "
        f"force_ocr={args.force_ocr}, lang={ocr_lang or 'default'}{psm_str}"
    )
    if max_file_size_bytes > 0:
        logger.info(f"Max file size: {args.max_file_size}MB")
    else:
        logger.info("Max file size: unlimited")
    if enrichments:
        logger.info(f"Enrichments enabled: {', '.join(enrichments)}")

    app = create_app(
        force_ocr=args.force_ocr,
        disable_ocr=args.no_ocr,
        ocr_engine=args.ocr_engine,
        psm=args.psm,
        ocr_lang=ocr_lang,
        enrich_formula=args.enrich_formula,
        enrich_picture_description=args.enrich_picture_description,
        picture_description_prompt=args.picture_description_prompt,
        max_file_size=max_file_size_bytes,
        device=args.device,
    )
    uvicorn.run(
        app,
        host=args.host,
        port=args.port,
        log_level=args.log_level,
        loop=_get_loop_setting(),
    )


if __name__ == "__main__":
    main()