api/server.py

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

from __future__ import annotations

import asyncio
import concurrent.futures
import datetime
import os
import threading
import time
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
import uvicorn
from fastapi import FastAPI, File, Form, HTTPException, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
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

logger = get_logger(__name__)

_START_TIME = time.time()

# 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)

_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  # 30 minutes


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:
            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:
    """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)
        _ping_once()


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()


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

@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"),
    )
    _start_ping_scheduler()
    yield
    logger.info("MarkItDown API shutting down")


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

app = FastAPI(
    title="MarkItDown 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."
    ),
    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,
)

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


# ---------------------------------------------------------------------------
# Request / Response models
# ---------------------------------------------------------------------------

class ConversionMetadata(BaseModel):
    """File-level statistics attached to every successful conversion."""

    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):
    """Standard response envelope for single-item conversion endpoints."""

    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):
    """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")
    @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):
    """Request body for /batch/urls."""

    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):
    """Per-item result within a batch response."""

    filename: str
    success: bool
    time_ms: float
    content: Optional[str] = None
    error: Optional[str] = None
    metadata: Optional[ConversionMetadata] = None


class BatchResponse(BaseModel):
    """Aggregate response for batch endpoints."""

    total: int
    succeeded: int
    failed: int
    total_time_ms: float
    results: List[BatchFileResult]


# ---------------------------------------------------------------------------
# Internal 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,
        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:
    """Construct a ConversionResponse, optionally running JSON extraction."""
    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:
    """Translate a ConversionError into an appropriate HTTPException."""
    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),
    )


# ---------------------------------------------------------------------------
# System endpoints
# ---------------------------------------------------------------------------

@app.get("/", tags=["System"], summary="Root", include_in_schema=False)
async def root():
    return {"service": "reconciliation-file-processing-service", "version": "2.1.0", "status": "running"}


@app.get("/health", tags=["System"], summary="Liveness check")
async def health():
    """Return server status and uptime in seconds."""
    return {
        "success":        True,
        "status":         "ok",
        "version":        "2.1.0",
        "uptime_seconds": round(time.time() - _START_TIME, 2),
        "timestamp":      datetime.now(timezone.utc).isoformat(),
    }


@app.get("/info", tags=["System"], summary="Server and environment information")
async def info():
    """Return application version, platform details, and operational limits."""
    import platform

    return {
        "success":              True,
        "app":                  "MarkItDown API",
        "version":              "2.1.0",
        "python_version":       platform.python_version(),
        "platform":             platform.system(),
        "uptime_seconds":       round(time.time() - _START_TIME, 2),
        "max_upload_mb":        MAX_UPLOAD_BYTES // (1024 * 1024),
        "supported_extensions": len(SUPPORTED_EXTENSIONS),
        "timestamp":            datetime.now(timezone.utc).isoformat(),
    }


@app.get("/formats", tags=["System"], 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"}],
        "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()},
    }


@app.get("/spacy-labels", tags=["System"], summary="Available spaCy NER labels for field extraction")
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 {
        "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"},
        },
    }


# ---------------------------------------------------------------------------
# Convert endpoints
# ---------------------------------------------------------------------------

@app.post(
    "/convert/file",
    response_model=ConversionResponse,
    tags=["Convert"],
    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 field 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."},
        )

    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,
    )


@app.post(
    "/convert/url",
    response_model=ConversionResponse,
    tags=["Convert"],
    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}"},
            )

        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,
        )

    # 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,
    )


# ---------------------------------------------------------------------------
# Batch endpoints
# ---------------------------------------------------------------------------

@app.post(
    "/batch/files",
    response_model=BatchResponse,
    tags=["Batch"],
    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."},
        )
    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,
    )


@app.post(
    "/batch/urls",
    response_model=BatchResponse,
    tags=["Batch"],
    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)
        )

    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,
    )


# ---------------------------------------------------------------------------
# Server runner (used when invoking this module directly)
# ---------------------------------------------------------------------------

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,
    )