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

System:
  GET  /              Root info
  GET  /ping          Liveness check & Docker HEALTHCHECK target
  GET  /docs          Swagger UI
  GET  /redoc         ReDoc UI
  GET  /metrics       Prometheus metrics

Nested-JSON key extractor (under /api/v1/keys):
  POST /api/v1/keys/extract
"""

from __future__ import annotations

import asyncio
import concurrent.futures
import json
import os
import time
import uuid
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Annotated, Any, Dict, List, Optional
from urllib.parse import urlparse

import httpx
import structlog
from fastapi import APIRouter, Depends, FastAPI, File, Form, HTTPException, Request, UploadFile, status
from fastapi.responses import JSONResponse, PlainTextResponse
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from prometheus_client import make_asgi_app
from pydantic import BaseModel, Field, field_validator

from app.api.routes import router as pdf_router
from app.core.auth import require_api_key
from app.core.config import settings
from app.core.exceptions import AppException
from app.core.logging import configure_logging
from app.core.rate_limit import RateLimitMiddleware
from app.services.ping import start_self_ping
from app.utils.cleanup import cleanup_loop
from core import ConversionError, ConversionResult, DocumentConverter, SUPPORTED_EXTENSIONS
from extraction.generic_json_extractor import extract
from extraction.keys_extractor import Extractor as KeysExtractor
from logger import get_logger

configure_logging()
logger = get_logger(__name__)

APP_NAME = "reconciliation-file-processing-service"
APP_VERSION = "2.2.0"

MAX_UPLOAD_BYTES = int(os.getenv("MAX_UPLOAD_BYTES", str(100 * 1024 * 1024)))
MAX_WORKERS = min(32, (os.cpu_count() or 1) + 4)

_thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS)
_converter = DocumentConverter()

logger.info("thread_pool_initialised", workers=MAX_WORKERS)



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

@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("api_starting", version=APP_VERSION, host="0.0.0.0:7860")
    ping_task = await start_self_ping()
    cleanup_task = asyncio.create_task(cleanup_loop())
    yield
    if ping_task:
        ping_task.cancel()
    cleanup_task.cancel()
    logger.info("api_shutting_down")


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

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

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


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

    structlog.contextvars.clear_contextvars()
    structlog.contextvars.bind_contextvars(
        request_id=request_id,
        method=request.method,
        path=request.url.path,
    )

    response = await call_next(request)
    elapsed = round((time.perf_counter() - start) * 1000, 2)

    logger.info("request_completed", status=response.status_code, duration_ms=elapsed)
    response.headers["X-Request-ID"] = request_id
    response.headers["X-Response-Time-Ms"] = str(elapsed)
    return response


# -- Exception handlers --
def _request_id(request: Request) -> str:
    return getattr(request.state, "request_id", "")


@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    logger.warning("app_exception", detail=exc.detail, status_code=exc.status_code)
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": exc.detail, "request_id": _request_id(request)},
    )


@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
    logger.exception("unhandled_exception", exc_info=exc)
    return JSONResponse(
        status_code=500,
        content={"error": "Internal server error", "request_id": _request_id(request)},
    )


# -- Root-level routes (only / , /ping) --
@app.get("/", tags=["System"], summary="Root", include_in_schema=False)
async def root():
    return {"service": APP_NAME, "version": APP_VERSION, "status": "running"}


@app.get("/ping", tags=["System"], summary="Ping", include_in_schema=False)
async def ping():
    return {"message": f"{APP_NAME} is running..."}



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

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

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


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


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

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


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

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


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


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


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


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

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

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


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


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


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


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


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

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


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

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

    logger.info("convert_file", filename=file.filename)
    raw = await file.read()
    if len(raw) > MAX_UPLOAD_BYTES:
        logger.warning("convert_file | file too large", filename=file.filename, size=len(raw))
        raise HTTPException(status_code=413, detail={"success": False, "message": "File exceeds 100 MB limit."})

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

    if isinstance(outcome, ConversionError):
        logger.error("convert_file | conversion failed", filename=file.filename, error=outcome.message)
        _raise_for_error(outcome)

    logger.info("convert_file | success", filename=file.filename, chars=outcome.char_count, time_ms=round(outcome.duration_ms, 1))

    if plain_text:
        return PlainTextResponse(outcome.markdown)

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


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

    if body.return_json:
        try:
            async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client:
                resp = await client.get(body.url)
                resp.raise_for_status()
        except httpx.HTTPError as exc:
            logger.error("convert_url | fetch failed", url=body.url, error=str(exc))
            raise HTTPException(status_code=400, detail={"success": False, "message": f"Failed to fetch URL: {exc}"})

        raw_data = resp.content
        if len(raw_data) > MAX_UPLOAD_BYTES:
            raise HTTPException(status_code=413, detail={"success": False, "message": "File exceeds 100 MB limit."})

        outcome = await loop.run_in_executor(_thread_pool, _converter.convert_stream, raw_data, filename)
        if isinstance(outcome, ConversionError):
            logger.error("convert_url | conversion failed", url=body.url, error=outcome.message)
            _raise_for_error(outcome)

        logger.info("convert_url | success", url=body.url, chars=outcome.char_count, time_ms=round(outcome.duration_ms, 1))
        return await _build_response(outcome, return_json=body.return_json, filename=filename, raw_data=raw_data, mappings=body.mappings)

    outcome = await loop.run_in_executor(_thread_pool, _converter.convert_url, body.url)
    if isinstance(outcome, ConversionError):
        logger.error("convert_url | conversion failed", url=body.url, error=outcome.message)
        _raise_for_error(outcome)

    logger.info("convert_url | success", url=body.url, chars=outcome.char_count, time_ms=round(outcome.duration_ms, 1))
    return await _build_response(outcome, return_json=body.return_json, filename=filename, mappings=body.mappings)


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

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

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

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

    return BatchResponse(total=len(results), succeeded=succeeded, failed=len(results) - succeeded, total_time_ms=total_ms, results=results)


@markitdown_router.post(
    "/batch/urls",
    response_model=BatchResponse,
    tags=["MarkItDown"],
    summary="Convert multiple URLs (up to 20)",
)
async def batch_urls(body: BatchUrlRequest):
    batch_start = time.perf_counter()
    logger.info("batch_urls", count=len(body.urls))

    async def _process_url(url: str) -> BatchFileResult:
        loop = asyncio.get_running_loop()
        outcome = await loop.run_in_executor(_thread_pool, _converter.convert_url, url)
        return _batch_result_from_error(url, outcome) if isinstance(outcome, ConversionError) else _batch_result_from_ok(outcome)

    results = await asyncio.gather(*[_process_url(url) for url in body.urls])
    total_ms = round((time.perf_counter() - batch_start) * 1000, 3)
    succeeded = sum(1 for r in results if r.success)
    logger.info("batch_urls | done", succeeded=succeeded, failed=len(results) - succeeded, total_ms=round(total_ms, 1))

    return BatchResponse(total=len(results), succeeded=succeeded, failed=len(results) - succeeded, total_time_ms=total_ms, results=results)


# ---------------------------------------------------------------------------
# Keys extractor router — nested-JSON key/value lookup
# ---------------------------------------------------------------------------

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


class KeysExtractRequest(BaseModel):
    data: Any = Field(..., description="JSON object or array to search")
    key_names: List[str] = Field(..., min_length=1, description="Key names to look up at any depth")
    result_limit: Optional[int] = Field(
        None,
        ge=1,
        description="Maximum values to return per key. Omit (or pass null) to return the full, uncapped result.",
    )

    @field_validator("key_names")
    @classmethod
    def _validate_key_names(cls, v: List[str]) -> List[str]:
        for kn in v:
            if not isinstance(kn, str) or not kn:
                raise ValueError("each key_name must be a non-empty string")
        return v


class KeysExtractResponse(BaseModel):
    success: bool
    time_ms: float
    data: Dict[str, List[Any]]
    error_message: Optional[str] = None


@keys_router.post(
    "/extract",
    response_model=KeysExtractResponse,
    tags=["Keys Extractor"],
    summary="Extract all values for given key names from a nested JSON object",
)
async def extract_keys(body: KeysExtractRequest):
    """Recursively walk any JSON object/array and return every value attached
    to the supplied key names, regardless of how deeply nested they are."""
    start = time.perf_counter()
    logger.info(
        "keys_extract | start",
        key_count=len(body.key_names),
        limit=body.result_limit,
    )

    if not isinstance(body.data, (dict, list)):
        return KeysExtractResponse(
            success=False,
            time_ms=round((time.perf_counter() - start) * 1000, 3),
            data={},
            error_message="`data` must be a JSON object or array",
        )

    try:
        loop = asyncio.get_running_loop()
        results = await loop.run_in_executor(
            _thread_pool,
            KeysExtractor(body.data, body.key_names, body.result_limit).extract,
        )
    except (TypeError, ValueError) as exc:
        logger.warning("keys_extract | invalid input", error=str(exc))
        return KeysExtractResponse(
            success=False,
            time_ms=round((time.perf_counter() - start) * 1000, 3),
            data={},
            error_message=str(exc),
        )
    except Exception as exc:
        logger.exception("keys_extract | failed")
        return KeysExtractResponse(
            success=False,
            time_ms=round((time.perf_counter() - start) * 1000, 3),
            data={},
            error_message=f"extraction failed: {exc}",
        )

    elapsed = round((time.perf_counter() - start) * 1000, 3)
    total_values = sum(len(v) for v in results.values())
    logger.info(
        "keys_extract | done",
        keys=len(body.key_names),
        total_values=total_values,
        time_ms=elapsed,
    )
    return KeysExtractResponse(
        success=True,
        time_ms=elapsed,
        data=results,
        error_message=None,
    )


# ---------------------------------------------------------------------------
# Include all routers
# ---------------------------------------------------------------------------

app.include_router(markitdown_router, prefix="/api/v1/markitdown")
app.include_router(pdf_router, prefix="/api/v1")
app.include_router(keys_router, prefix="/api/v1/keys")

# -- Mount Prometheus metrics --
app.mount("/metrics", make_asgi_app())


# ---------------------------------------------------------------------------
# Server runner
# ---------------------------------------------------------------------------

def run_server(host: str = "0.0.0.0", port: int = 7860, reload: bool = False) -> None:
    import uvicorn
    uvicorn.run("api.server:app", host=host, port=port, reload=reload)