api/server.py

"""
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 os
import threading
import time
import uuid
import urllib.request
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import Annotated, Any, Dict, List, Optional
from urllib.parse import urlparse

import httpx
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 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"

MAX_UPLOAD_BYTES = 100 * 1024 * 1024

MAX_WORKERS = min(32, (os.cpu_count() or 1) + 4)
_thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS)

_converter = DocumentConverter()

logger.info("Thread pool initialised with %d workers", MAX_WORKERS)


# ---------------------------------------------------------------------------
# Self-ping
# ---------------------------------------------------------------------------

PING_URL = os.environ.get("PING_URL", "https://validops-us-data-extract.hf.space/health")
PING_INTERVAL_SECONDS = 30 * 60


def _ping_once() -> None:
    ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
    try:
        with urllib.request.urlopen(PING_URL, timeout=10) as resp:
            logger.info("self_ping | status=%d | url=%s | ts=%s", resp.status, PING_URL, ts)
    except Exception as exc:
        logger.warning("self_ping | failed | url=%s | error=%s | ts=%s", PING_URL, exc, ts)


def _ping_loop() -> None:
    logger.info("self_ping | scheduler started | interval_minutes=30 | url=%s", PING_URL)
    while True:
        time.sleep(PING_INTERVAL_SECONDS)
        _ping_once()


def _start_ping_scheduler() -> None:
    thread = threading.Thread(target=_ping_loop, name="self-ping", daemon=True)
    thread.start()


# ---------------------------------------------------------------------------
# Lifespan
# ---------------------------------------------------------------------------

@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info(
        "MarkItDown & PDF Conversion API starting | version=2.1.0 | host=0.0.0.0:7860",
    )
    _start_ping_scheduler()
    yield
    logger.info("MarkItDown & PDF Conversion API shutting down")


# ---------------------------------------------------------------------------
# Application
# ---------------------------------------------------------------------------

app = FastAPI(
    title="MarkItDown & PDF Conversion API",
    description=(
        "Unified API for document-to-Markdown conversion (Microsoft MarkItDown + RapidOCR) "
        "and PDF-to-image conversion with async job support."
    ),
    version="2.1.0",
    docs_url="/docs",
    redoc_url="/redoc",
    lifespan=lifespan,
)

# -- Middleware --
app.add_middleware(RateLimitMiddleware)
app.add_middleware(GZipMiddleware, minimum_size=1000)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.middleware("http")
async def request_context_middleware(request: Request, call_next):
    request_id = str(uuid.uuid4())
    start = time.perf_counter()

    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..."}


# ---------------------------------------------------------------------------
# MarkItDown router — all routes under /api/v1/markitdown
# ---------------------------------------------------------------------------

markitdown_router = APIRouter(dependencies=[Depends(require_api_key)])

# -- Models --
class ConversionMetadata(BaseModel):
    source: str
    char_count: int
    word_count: int
    line_count: int
    file_size_bytes: int
    mime_type: str
    content_hash: str
    token_estimate: int


class ConversionResponse(BaseModel):
    success: bool
    time_ms: float
    content: str
    return_json: bool = False
    json_content: Optional[Any] = None
    metadata: Optional[ConversionMetadata] = None
    error_message: Optional[str] = None


class UrlRequest(BaseModel):
    url: str
    return_json: bool = False
    mappings: Optional[Dict[str, Dict[str, Any]]] = None
    model_config = {"populate_by_name": True}

    @field_validator("url")
    @classmethod
    def validate_scheme(cls, v: str) -> str:
        if not v.startswith(("http://", "https://")):
            raise ValueError("Only http/https URLs are supported.")
        return v


class BatchUrlRequest(BaseModel):
    urls: List[str]

    @field_validator("urls")
    @classmethod
    def validate_urls(cls, v: List[str]) -> List[str]:
        for url in v:
            if not url.startswith(("http://", "https://")):
                raise ValueError(f"Invalid URL scheme: {url}")
        if len(v) > 20:
            raise ValueError("Maximum 20 URLs per batch request.")
        return v


class BatchFileResult(BaseModel):
    filename: str
    success: bool
    time_ms: float
    content: Optional[str] = None
    error: Optional[str] = None
    metadata: Optional[ConversionMetadata] = None


class BatchResponse(BaseModel):
    total: int
    succeeded: int
    failed: int
    total_time_ms: float
    results: List[BatchFileResult]


# -- Helpers --
def _build_metadata(result: ConversionResult) -> ConversionMetadata:
    return ConversionMetadata(
        source=result.source,
        char_count=result.char_count,
        word_count=result.word_count,
        line_count=result.line_count,
        file_size_bytes=result.file_size_bytes,
        mime_type=result.mime_type,
        content_hash=result.content_hash,
        token_estimate=result.token_estimate,
    )


async def _build_response(
    result: ConversionResult,
    *,
    return_json: bool = False,
    filename: Optional[str] = None,
    raw_data: Optional[bytes] = None,
    mappings: Optional[Dict[str, Dict[str, Any]]] = None,
) -> ConversionResponse:
    json_content: Optional[Any] = None
    error_message: Optional[str] = None

    if return_json and filename:
        loop = asyncio.get_running_loop()
        json_result = await loop.run_in_executor(
            _thread_pool, extract, filename, result.markdown, mappings, raw_data
        )
        if "error" in json_result:
            error_message = json_result["error"]
        else:
            json_content = json_result

    return ConversionResponse(
        success=True,
        time_ms=round(result.duration_ms, 3),
        content=result.markdown,
        return_json=return_json,
        json_content=json_content,
        metadata=_build_metadata(result),
        error_message=error_message,
    )


def _raise_for_error(outcome: ConversionError) -> None:
    status_map = {
        "FileNotFoundError": status.HTTP_404_NOT_FOUND,
        "ValueError":        status.HTTP_422_UNPROCESSABLE_ENTITY,
        "PermissionError":   status.HTTP_403_FORBIDDEN,
    }
    code = status_map.get(outcome.error_type, status.HTTP_500_INTERNAL_SERVER_ERROR)
    raise HTTPException(
        status_code=code,
        detail={
            "success":    False,
            "error_type": outcome.error_type,
            "message":    outcome.message,
            "time_ms":    round(outcome.duration_ms, 3),
        },
    )


def _batch_result_from_error(name: str, err: ConversionError) -> BatchFileResult:
    return BatchFileResult(
        filename=name,
        success=False,
        time_ms=round(err.duration_ms, 3),
        error=err.message,
    )


def _batch_result_from_ok(result: ConversionResult) -> BatchFileResult:
    return BatchFileResult(
        filename=result.source,
        success=True,
        time_ms=round(result.duration_ms, 3),
        content=result.markdown,
        metadata=_build_metadata(result),
    )


# -- Routes --
@markitdown_router.get("/formats", tags=["MarkItDown"], summary="Supported file formats by category")
async def list_formats():
    by_category = {
        "documents": [e for e in SUPPORTED_EXTENSIONS if e in {".pdf", ".docx", ".doc", ".epub"}],
        "office":    [e for e in SUPPORTED_EXTENSIONS if e in {".pptx", ".ppt", ".xlsx", ".xls"}],
        "data":      [e for e in SUPPORTED_EXTENSIONS if e in {".csv", ".json", ".xml"}],
        "web":       [e for e in SUPPORTED_EXTENSIONS if e in {".htm", ".html"}],
        "text":      [e for e in SUPPORTED_EXTENSIONS if e in {".txt", ".md", ".rst"}],
        "images":    [e for e in SUPPORTED_EXTENSIONS if e in {".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp", ".tiff"}],
        "audio":     [e for e in SUPPORTED_EXTENSIONS if e in {".mp3", ".wav", ".ogg", ".flac"}],
        "archives":  [e for e in SUPPORTED_EXTENSIONS if e in {".zip"}],
    }
    return {
        "success":        True,
        "total_count":    len(SUPPORTED_EXTENSIONS),
        "all_extensions": sorted(SUPPORTED_EXTENSIONS),
        "by_category":    {k: sorted(v) for k, v in by_category.items()},
    }


@markitdown_router.get("/spacy-labels", tags=["MarkItDown"], summary="Available spaCy NER labels")
async def list_spacy_labels():
    from extraction.spacy_extractor import VALID_SPACY_LABELS

    return {
        "success":      True,
        "spacy_labels": VALID_SPACY_LABELS,
        "source_types": {
            "entity":     "Extract using spaCy NER labels (ORG, PERSON, DATE, etc.)",
            "regex":      "Extract using custom regular expressions",
            "token_attr": "Extract using token attributes (text, pos_, tag_, etc.)",
        },
        "example_mappings": {
            "company": {"source_type": "entity", "label": "ORG"},
            "person":  {"source_type": "entity", "label": "PERSON"},
            "date":    {"source_type": "entity", "label": "DATE"},
            "money":   {"source_type": "entity", "label": "MONEY"},
            "email":   {"source_type": "regex",  "pattern": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"},
            "phone":   {"source_type": "regex",  "pattern": r"\b\d{3}-\d{3}-\d{4}\b"},
        },
    }


@markitdown_router.post(
    "/convert/file",
    response_model=ConversionResponse,
    tags=["MarkItDown"],
    summary="Convert an uploaded file to Markdown",
)
async def convert_file(
    file: Annotated[UploadFile, File(description="File to convert")],
    plain_text: bool = Form(False),
    return_json: bool = Form(False),
    mappings: Optional[str] = Form(
        None,
        description='JSON string defining spaCy extraction rules. Example: {"company": {"source_type": "entity", "label": "ORG"}}',
    ),
):
    if file is None:
        raise HTTPException(status_code=400, detail={"success": False, "message": "No file provided."})

    parsed_mappings: Optional[Dict[str, Any]] = None
    if mappings:
        import json as _json
        try:
            parsed_mappings = _json.loads(mappings)
        except _json.JSONDecodeError:
            raise HTTPException(status_code=400, detail={"success": False, "message": "Invalid JSON in mappings parameter."})

    logger.info("convert_file | filename=%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."})

    loop = asyncio.get_running_loop()
    outcome = await loop.run_in_executor(_thread_pool, _converter.convert_stream, raw, file.filename or "upload")

    if isinstance(outcome, ConversionError):
        logger.error("convert_file | conversion failed | filename=%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)

    if plain_text:
        return PlainTextResponse(outcome.markdown)

    return await _build_response(outcome, return_json=return_json, filename=file.filename, raw_data=raw, mappings=parsed_mappings)


@markitdown_router.post(
    "/convert/url",
    response_model=ConversionResponse,
    tags=["MarkItDown"],
    summary="Convert a public URL to Markdown",
)
async def convert_url(body: UrlRequest):
    logger.info("convert_url | url=%s", body.url)
    parsed = urlparse(body.url)
    filename = Path(parsed.path).name or "url_content"
    loop = asyncio.get_running_loop()

    if body.return_json:
        try:
            async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client:
                resp = await client.get(body.url)
                resp.raise_for_status()
        except httpx.HTTPError as exc:
            logger.error("convert_url | fetch failed | url=%s | error=%s", body.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."})

        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)

    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)


@markitdown_router.post(
    "/batch/files",
    response_model=BatchResponse,
    tags=["MarkItDown"],
    summary="Convert multiple files (up to 10)",
)
async def batch_files(
    files: Annotated[List[UploadFile], File(description="Files to convert — maximum 10")],
):
    if not files:
        raise HTTPException(status_code=400, detail={"success": False, "message": "No files provided."})
    if len(files) > 10:
        raise HTTPException(status_code=400, detail={"success": False, "message": "Maximum 10 files per batch."})

    batch_start = time.perf_counter()
    logger.info("batch_files | count=%d", len(files))

    async def _process_file(f: UploadFile) -> BatchFileResult:
        if f is None:
            return BatchFileResult(filename="unknown", success=False, time_ms=0, error="File object is None.")
        raw = await f.read()
        if len(raw) > MAX_UPLOAD_BYTES:
            return BatchFileResult(filename=f.filename or "unknown", success=False, time_ms=0, error="File exceeds 100 MB limit.")
        loop = asyncio.get_running_loop()
        outcome = await loop.run_in_executor(_thread_pool, _converter.convert_stream, raw, f.filename or "upload")
        return _batch_result_from_error(f.filename or "unknown", outcome) if isinstance(outcome, ConversionError) else _batch_result_from_ok(outcome)

    results = await asyncio.gather(*[_process_file(f) for f in files])
    total_ms = round((time.perf_counter() - batch_start) * 1000, 3)
    succeeded = sum(1 for r in results if r.success)
    logger.info("batch_files | done | succeeded=%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)


@markitdown_router.post(
    "/batch/urls",
    response_model=BatchResponse,
    tags=["MarkItDown"],
    summary="Convert multiple URLs (up to 20)",
)
async def batch_urls(body: BatchUrlRequest):
    batch_start = time.perf_counter()
    logger.info("batch_urls | count=%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)

    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)


# ---------------------------------------------------------------------------
# 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:
    import uvicorn
    uvicorn.run("api.server:app", host=host, port=port, reload=reload)