ok

.gitignore

@@ -0,0 +1,127 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+dist/
+downloads/
+develop-eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# Virtual environments
+.env
+.venv
+venv/
+ENV/
+env/
+env.bak/
+venv.bak/
+
+# FastAPI / Uvicorn
+*.log
+uvicorn.log
+
+# Environment variables
+.env.*
+!.env.example
+
+# IDEs and editors
+.vscode/
+.idea/
+*.swp
+*.swo
+
+# OS files
+.DS_Store
+Thumbs.db
+
+# Testing
+.pytest_cache/
+.coverage
+coverage.xml
+htmlcov/
+.tox/
+.nox/
+
+# Type checking
+.mypy_cache/
+.pyre/
+.pytype/
+
+# Ruff / Lint
+.ruff_cache/
+
+# Jupyter Notebook
+.ipynb_checkpoints/
+
+# PyInstaller
+*.manifest
+*.spec
+
+# SQLAlchemy / SQLite
+*.db
+*.sqlite3
+
+# Logs
+logs/
+*.log
+
+# Cache
+.cache/
+
+# Temporary files
+tmp/
+temp/
+
+# Docker
+docker-compose.override.yml
+
+# Node (if frontend exists)
+node_modules/
+
+# Mac/Linux hidden files
+.*.swp
+*.pid
+
+# Secrets
+secrets/
+credentials.json
+service-account.json
+
+# Alembic
+alembic/versions/*.pyc
+
+# Static build outputs
+staticfiles/
+
+# PyCharm
+.idea/
+
+# VSCode
+.history/
+
+# Poetry
+poetry.lock
+
+# Pipenv
+Pipfile.lock
+run_docker.py
+docs
+.claude
+start_docker.txt

Dockerfile

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

README.md

@@ -0,0 +1,9 @@
+---
+title: LLM Ready Data API — v2.1.0
+emoji: ⚡
+colorFrom: green
+colorTo: purple
+sdk: docker
+pinned: false
+app_port: 7860
+---

__init__.py

@@ -0,0 +1,8 @@
+from .core import ConversionError, ConversionResult, DocumentConverter, SUPPORTED_EXTENSIONS
+
+__all__ = [
+    "ConversionError",
+    "ConversionResult",
+    "DocumentConverter",
+    "SUPPORTED_EXTENSIONS",
+]

api/__init__.py

@@ -0,0 +1,3 @@
+from .server import app, run_server
+
+__all__ = ["app", "run_server"]

api/server.py

@@ -0,0 +1,675 @@
+"""
+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
+from fastapi import FastAPI, File, Form, HTTPException, UploadFile, status
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.middleware.gzip import GZipMiddleware
+from fastapi.responses import PlainTextResponse
+from pydantic import BaseModel, field_validator
+
+from core import ConversionError, ConversionResult, DocumentConverter, SUPPORTED_EXTENSIONS
+from extraction.generic_json_extractor import extract
+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", "http://localhost:7860/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("/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,
+    )

core/__init__.py

@@ -0,0 +1,15 @@
+from .converter import ConversionError, ConversionResult, DocumentConverter, SUPPORTED_EXTENSIONS
+from .batch import BatchProcessor, BatchReport
+from .output import OutputWriter
+from .ocr_engine import ocr_image
+
+__all__ = [
+    "ConversionError",
+    "ConversionResult",
+    "DocumentConverter",
+    "SUPPORTED_EXTENSIONS",
+    "BatchProcessor",
+    "BatchReport",
+    "OutputWriter",
+    "ocr_image",
+]

core/batch.py

@@ -0,0 +1,135 @@
+"""
+Batch processing utilities for the MarkItDown API.
+
+Provides BatchProcessor for converting multiple files concurrently using
+a shared DocumentConverter instance, and BatchReport for aggregating results.
+
+These classes are used internally by the CLI and server batch endpoints.
+"""
+
+from __future__ import annotations
+
+import concurrent.futures
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Callable, Optional, Sequence
+
+from .converter import ConversionError, ConversionResult, DocumentConverter, SUPPORTED_EXTENSIONS
+from logger import get_logger
+
+logger = get_logger(__name__)
+
+
+@dataclass
+class BatchReport:
+    """Aggregated results from a batch file conversion run."""
+
+    total: int
+    succeeded: int
+    failed: int
+    results: list[ConversionResult]
+    errors: list[ConversionError]
+    total_chars: int
+    total_words: int
+    total_duration_ms: float
+
+    @property
+    def success_rate(self) -> float:
+        """Percentage of files converted successfully."""
+        return (self.succeeded / self.total * 100) if self.total else 0.0
+
+
+class BatchProcessor:
+    """Convert multiple files concurrently using a thread pool.
+
+    Parameters
+    ----------
+    converter:
+        Shared DocumentConverter instance.
+    max_workers:
+        Number of threads in the pool. Defaults to min(8, cpu_count + 4).
+    """
+
+    def __init__(
+        self,
+        converter: DocumentConverter,
+        max_workers: int = min(8, (os.cpu_count() or 1) + 4),
+    ) -> None:
+        self._converter = converter
+        self._max_workers = max_workers
+
+    def process_files(
+        self,
+        paths: Sequence[str | Path],
+        progress_callback: Optional[Callable[[int, int, str], None]] = None,
+    ) -> BatchReport:
+        """Convert all files in *paths* and return a BatchReport.
+
+        Parameters
+        ----------
+        paths:
+            Iterable of file paths to convert.
+        progress_callback:
+            Optional callable invoked after each file completes.
+            Receives ``(completed_count, total_count, source_path)``.
+        """
+        results: list[ConversionResult] = []
+        errors: list[ConversionError] = []
+        total = len(paths)
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=self._max_workers) as executor:
+            future_to_path = {
+                executor.submit(self._converter.convert_file, p): p for p in paths
+            }
+            completed = 0
+            for future in concurrent.futures.as_completed(future_to_path):
+                completed += 1
+                outcome = future.result()
+                source = str(future_to_path[future])
+                if isinstance(outcome, ConversionResult):
+                    results.append(outcome)
+                else:
+                    errors.append(outcome)
+                if progress_callback:
+                    progress_callback(completed, total, source)
+
+        logger.info(
+            "batch_processor | done | total=%d | succeeded=%d | failed=%d",
+            total, len(results), len(errors),
+        )
+        return BatchReport(
+            total=total,
+            succeeded=len(results),
+            failed=len(errors),
+            results=results,
+            errors=errors,
+            total_chars=sum(r.char_count for r in results),
+            total_words=sum(r.word_count for r in results),
+            total_duration_ms=sum(r.duration_ms for r in results),
+        )
+
+    def discover_files(
+        self,
+        directory: str | Path,
+        recursive: bool = True,
+        extensions: Optional[set[str]] = None,
+    ) -> list[Path]:
+        """Return all convertible files under *directory*.
+
+        Parameters
+        ----------
+        directory:
+            Root directory to scan.
+        recursive:
+            When True, scan subdirectories as well.
+        extensions:
+            Set of extensions to include. Defaults to SUPPORTED_EXTENSIONS.
+        """
+        root = Path(directory).resolve()
+        exts = extensions or SUPPORTED_EXTENSIONS
+        glob_pattern = "**/*" if recursive else "*"
+        return [
+            p for p in root.glob(glob_pattern)
+            if p.is_file() and p.suffix.lower() in exts
+        ]

core/converter.py

@@ -0,0 +1,259 @@
+"""
+Document converter for the MarkItDown API.
+
+Wraps Microsoft MarkItDown and RapidOCR to provide a unified conversion
+interface that accepts file paths, raw byte streams, and public URLs.
+
+Supported extensions are declared in SUPPORTED_EXTENSIONS and imported by
+the API layer for format listing and validation.
+
+Public classes
+--------------
+    ConversionResult
+        Immutable dataclass holding the converted Markdown and file statistics.
+
+    ConversionError
+        Immutable dataclass holding error details when conversion fails.
+
+    DocumentConverter
+        Main converter class. All convert_* methods return either a
+        ConversionResult or a ConversionError — they do not raise.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import mimetypes
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Optional
+from urllib.parse import urlparse
+
+from markitdown import MarkItDown
+
+from .ocr_engine import ocr_image, ocr_pdf
+from logger import get_logger
+
+logger = get_logger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Supported formats
+# ---------------------------------------------------------------------------
+
+SUPPORTED_EXTENSIONS = {
+    ".pdf", ".docx", ".doc", ".pptx", ".ppt",
+    ".xlsx", ".xls", ".csv", ".json", ".xml",
+    ".html", ".htm", ".txt", ".md", ".rst",
+    ".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp", ".tiff",
+    ".mp3", ".wav", ".ogg", ".flac",
+    ".zip", ".epub",
+}
+
+# Extensions that route through RapidOCR rather than MarkItDown.
+IMAGE_EXTENSIONS = {
+    ".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp", ".tiff",
+}
+
+IMAGE_MIME_PREFIXES = {"image/"}
+
+
+def _is_image(ext: str, mime: str) -> bool:
+    """Return True when the input should be routed through RapidOCR."""
+    return ext.lower() in IMAGE_EXTENSIONS or any(
+        mime.startswith(p) for p in IMAGE_MIME_PREFIXES
+    )
+
+
+# ---------------------------------------------------------------------------
+# Result and error types
+# ---------------------------------------------------------------------------
+
+@dataclass(frozen=True)
+class ConversionResult:
+    """Successful conversion output."""
+
+    source: str
+    markdown: str
+    char_count: int
+    word_count: int
+    line_count: int
+    duration_ms: float
+    file_size_bytes: int
+    mime_type: str
+    content_hash: str
+    metadata: dict = field(default_factory=dict)
+
+    @property
+    def token_estimate(self) -> int:
+        """Rough LLM token estimate based on word count (4/3 words per token)."""
+        return max(1, self.word_count * 4 // 3)
+
+
+@dataclass(frozen=True)
+class ConversionError:
+    """Conversion failure details."""
+
+    source: str
+    error_type: str
+    message: str
+    duration_ms: float
+
+
+# ---------------------------------------------------------------------------
+# Converter
+# ---------------------------------------------------------------------------
+
+class DocumentConverter:
+    """Converts documents from various formats to Markdown.
+
+    All public methods return either a ConversionResult or a ConversionError
+    and never raise exceptions to callers.
+    """
+
+    def __init__(self, enable_plugins: bool = False) -> None:
+        self._engine = MarkItDown(enable_plugins=enable_plugins)
+
+    # ------------------------------------------------------------------
+    # Public conversion methods
+    # ------------------------------------------------------------------
+
+    def convert_file(self, path: str | Path) -> ConversionResult | ConversionError:
+        """Convert a local file identified by *path*."""
+        path = Path(path).resolve()
+        start = time.perf_counter()
+
+        if not path.exists():
+            return ConversionError(
+                source=str(path),
+                error_type="FileNotFoundError",
+                message=f"File does not exist: {path}",
+                duration_ms=0.0,
+            )
+
+        file_size = path.stat().st_size
+        mime_type, _ = mimetypes.guess_type(str(path))
+        mime_type = mime_type or "application/octet-stream"
+
+        try:
+            if _is_image(path.suffix, mime_type):
+                markdown = ocr_image(str(path))
+            else:
+                markdown = self._engine.convert(str(path)).text_content
+                if not markdown.strip() and path.suffix.lower() == ".pdf":
+                    logger.info(
+                        "convert_file | no text from PDF, falling back to OCR | file=%s",
+                        path.name,
+                    )
+                    markdown = ocr_pdf(str(path))
+
+            elapsed = (time.perf_counter() - start) * 1000
+            return self._build_result(str(path), markdown, file_size, mime_type, elapsed)
+
+        except Exception as exc:
+            elapsed = (time.perf_counter() - start) * 1000
+            logger.error("convert_file | exception | file=%s | error=%s", path, exc, exc_info=True)
+            return ConversionError(
+                source=str(path),
+                error_type=type(exc).__name__,
+                message=str(exc),
+                duration_ms=elapsed,
+            )
+
+    def convert_url(self, url: str) -> ConversionResult | ConversionError:
+        """Fetch and convert a public HTTP/HTTPS URL."""
+        parsed = urlparse(url)
+        if parsed.scheme not in {"http", "https"}:
+            return ConversionError(
+                source=url,
+                error_type="ValueError",
+                message=f"Unsupported URL scheme: {parsed.scheme!r}",
+                duration_ms=0.0,
+            )
+
+        start = time.perf_counter()
+        try:
+            url_ext = Path(urlparse(url).path).suffix.lower()
+            if url_ext in IMAGE_EXTENSIONS:
+                markdown = ocr_image(url)
+                mime_type = mimetypes.guess_type(url)[0] or "image/jpeg"
+            else:
+                result = self._engine.convert(url)
+                markdown = result.text_content
+                mime_type = "text/html"
+
+            elapsed = (time.perf_counter() - start) * 1000
+            return self._build_result(url, markdown, 0, mime_type, elapsed)
+
+        except Exception as exc:
+            elapsed = (time.perf_counter() - start) * 1000
+            logger.error("convert_url | exception | url=%s | error=%s", url, exc, exc_info=True)
+            return ConversionError(
+                source=url,
+                error_type=type(exc).__name__,
+                message=str(exc),
+                duration_ms=elapsed,
+            )
+
+    def convert_stream(self, data: bytes, filename: str) -> ConversionResult | ConversionError:
+        """Convert raw bytes identified by *filename* (used for upload payloads)."""
+        import io
+
+        start = time.perf_counter()
+        mime_type, _ = mimetypes.guess_type(filename)
+        mime_type = mime_type or "application/octet-stream"
+        ext = Path(filename).suffix.lower()
+
+        try:
+            if _is_image(ext, mime_type):
+                markdown = ocr_image(data)
+            else:
+                result = self._engine.convert_stream(io.BytesIO(data), file_extension=ext)
+                markdown = result.text_content
+                if not markdown.strip() and ext == ".pdf":
+                    logger.info(
+                        "convert_stream | no text from PDF stream, falling back to OCR | filename=%s",
+                        filename,
+                    )
+                    markdown = ocr_pdf(data)
+
+            elapsed = (time.perf_counter() - start) * 1000
+            return self._build_result(filename, markdown, len(data), mime_type, elapsed)
+
+        except Exception as exc:
+            elapsed = (time.perf_counter() - start) * 1000
+            logger.error("convert_stream | exception | filename=%s | error=%s", filename, exc, exc_info=True)
+            return ConversionError(
+                source=filename,
+                error_type=type(exc).__name__,
+                message=str(exc),
+                duration_ms=elapsed,
+            )
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _build_result(
+        source: str,
+        markdown: str,
+        file_size: int,
+        mime_type: str,
+        elapsed: float,
+    ) -> ConversionResult:
+        lines = markdown.splitlines()
+        words = markdown.split()
+        content_hash = hashlib.sha256(markdown.encode()).hexdigest()
+        return ConversionResult(
+            source=source,
+            markdown=markdown,
+            char_count=len(markdown),
+            word_count=len(words),
+            line_count=len(lines),
+            duration_ms=elapsed,
+            file_size_bytes=file_size,
+            mime_type=mime_type,
+            content_hash=content_hash,
+        )

core/ocr_engine.py

@@ -0,0 +1,197 @@
+"""
+OCR utilities for the MarkItDown API.
+
+Two engines are provided:
+
+    ocr_image(source)
+        RapidOCR singleton for raster images (JPEG, PNG, WEBP, etc.).
+        Accepts bytes, a local file path string, an HTTP/HTTPS URL string,
+        a numpy.ndarray, or a PIL.Image instance.
+
+    ocr_pdf(source, dpi=150)
+        Scanned-PDF fallback. Renders each page with pypdfium2, then feeds
+        the rendered PIL image through ocr_image. Returns all pages joined
+        with double newlines.
+
+Both functions return a plain string and never raise; errors are logged and
+an empty string is returned on failure.
+"""
+
+from __future__ import annotations
+
+import io
+import threading
+from typing import Union
+from urllib.parse import urlparse
+
+import numpy as np
+
+from logger import get_logger
+
+logger = get_logger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# RapidOCR singleton
+# ---------------------------------------------------------------------------
+
+_lock = threading.Lock()
+_engine = None
+
+
+def _get_engine():
+    """Return the shared RapidOCR instance, initialising it on first call."""
+    global _engine
+    if _engine is None:
+        with _lock:
+            if _engine is None:
+                from rapidocr_onnxruntime import RapidOCR
+                _engine = RapidOCR(
+                    Det={"use_cuda": False, "use_dml": False},
+                    Cls={"use_cuda": False, "use_dml": False},
+                    Rec={"use_cuda": False, "use_dml": False},
+                    print_verbose=False,
+                )
+                logger.info("RapidOCR engine initialised")
+    return _engine
+
+
+# ---------------------------------------------------------------------------
+# Input normalisation
+# ---------------------------------------------------------------------------
+
+def _to_numpy(source) -> Union[np.ndarray, str]:
+    """Normalise *source* to a numpy array or a local file path string.
+
+    Accepted input types:
+        PIL.Image   — converted directly to ndarray.
+        bytes       — decoded via PIL then converted to ndarray.
+        str         — HTTP/HTTPS URL fetched then decoded; local paths returned as-is.
+        np.ndarray  — returned unchanged.
+    """
+    from PIL import Image
+
+    def _pil_to_array(img: Image.Image) -> np.ndarray:
+        if img.mode not in ("RGB", "L", "RGBA"):
+            img = img.convert("RGB")
+        return np.array(img)
+
+    if isinstance(source, Image.Image):
+        return _pil_to_array(source)
+
+    if isinstance(source, (bytes, bytearray)):
+        return _pil_to_array(Image.open(io.BytesIO(source)))
+
+    if isinstance(source, str):
+        parsed = urlparse(source)
+        if parsed.scheme in {"http", "https"}:
+            import httpx
+            resp = httpx.get(source, follow_redirects=True, timeout=30)
+            resp.raise_for_status()
+            return _pil_to_array(Image.open(io.BytesIO(resp.content)))
+        # Local file path — RapidOCR accepts it directly.
+        return source
+
+    if isinstance(source, np.ndarray):
+        return source
+
+    raise TypeError(
+        f"ocr_image expects bytes, str (URL or path), numpy.ndarray, or PIL.Image; "
+        f"received {type(source).__name__!r}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def ocr_image(
+    source,
+    *,
+    use_det: bool = True,
+    use_cls: bool = True,
+    use_rec: bool = True,
+    text_score: float = 0.5,
+) -> str:
+    """Extract text from an image using RapidOCR.
+
+    Parameters
+    ----------
+    source:
+        Input image — bytes, URL string, local path string, numpy.ndarray,
+        or PIL.Image.
+    use_det, use_cls, use_rec:
+        RapidOCR pipeline stages (detection, classification, recognition).
+    text_score:
+        Minimum confidence threshold for accepted text lines.
+
+    Returns
+    -------
+    str
+        Recognised text lines joined by newlines, or an empty string when
+        no text is detected.
+    """
+    engine = _get_engine()
+    img = _to_numpy(source)
+
+    result, _ = engine(
+        img,
+        use_det=use_det,
+        use_cls=use_cls,
+        use_rec=use_rec,
+        text_score=text_score,
+    )
+
+    if not result:
+        return ""
+
+    return "\n".join(item[1] for item in result if len(item) > 1 and item[1])
+
+
+def ocr_pdf(source: Union[str, bytes], *, dpi: int = 150) -> str:
+    """Extract text from a scanned (image-only) PDF using pypdfium2 and RapidOCR.
+
+    Each page is rendered to a PIL image in memory (no temporary files are
+    written), then passed through ocr_image. All page outputs are joined
+    with double newlines.
+
+    Parameters
+    ----------
+    source:
+        Local file path (str) or raw PDF bytes.
+    dpi:
+        Rendering resolution. 150 balances speed and OCR quality for most
+        document types. Increase to 200-300 for small or dense text.
+
+    Returns
+    -------
+    str
+        Concatenated OCR text from all pages, or an empty string on failure.
+    """
+    try:
+        import pypdfium2 as pdfium
+    except ImportError:
+        logger.error("ocr_pdf | pypdfium2 not installed; run: pip install pypdfium2")
+        return ""
+
+    try:
+        pdf = pdfium.PdfDocument(source)
+        scale = dpi / 72.0  # pypdfium2 native resolution is 72 dpi
+        page_texts: list[str] = []
+
+        for page_index in range(len(pdf)):
+            page = pdf[page_index]
+            bitmap = page.render(scale=scale, rotation=0)
+            pil_image = bitmap.to_pil()
+
+            logger.debug("ocr_pdf | processing page %d/%d", page_index + 1, len(pdf))
+            page_text = ocr_image(pil_image)
+            if page_text:
+                page_texts.append(page_text)
+
+        pdf.close()
+        return "\n\n".join(page_texts)
+
+    except Exception as exc:
+        logger.error("ocr_pdf | failed | error=%s", exc, exc_info=True)
+        return ""

core/output.py

@@ -0,0 +1,127 @@
+"""
+Output writer for the MarkItDown API.
+
+Writes conversion results to disk in Markdown, plain text, or JSON format.
+Intended for use by the CLI batch pipeline; the API server handles output
+directly via HTTP responses and does not use this module.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Literal
+
+from .converter import ConversionResult
+from logger import get_logger
+
+logger = get_logger(__name__)
+
+
+OutputFormat = Literal["markdown", "json", "txt"]
+
+
+class OutputWriter:
+    """Write ConversionResult objects to files in a specified format.
+
+    Parameters
+    ----------
+    output_dir:
+        Destination directory. Created automatically if it does not exist.
+    format:
+        Output format: ``"markdown"`` (default), ``"json"``, or ``"txt"``.
+    """
+
+    def __init__(self, output_dir: str | Path, format: OutputFormat = "markdown") -> None:
+        self._output_dir = Path(output_dir)
+        self._format = format
+        self._output_dir.mkdir(parents=True, exist_ok=True)
+
+    def write(self, result: ConversionResult) -> Path:
+        """Serialise *result* and write it to the output directory.
+
+        Collisions are resolved by appending an incrementing counter to the stem.
+
+        Returns
+        -------
+        Path
+            The path of the written file.
+        """
+        stem = Path(result.source).stem if not result.source.startswith("http") else "web_content"
+        suffix = ".md" if self._format == "markdown" else f".{self._format}"
+        output_path = self._resolve_collision(self._output_dir / f"{stem}{suffix}")
+
+        content = self._render(result)
+        output_path.write_text(content, encoding="utf-8")
+        logger.debug("write | path=%s | chars=%d", output_path, len(content))
+        return output_path
+
+    def write_batch_report(self, report, output_path: str | Path) -> None:
+        """Write a BatchReport summary as JSON to *output_path*."""
+        path = Path(output_path)
+        data = {
+            "summary": {
+                "total":             report.total,
+                "succeeded":         report.succeeded,
+                "failed":            report.failed,
+                "success_rate_pct":  round(report.success_rate, 2),
+                "total_chars":       report.total_chars,
+                "total_words":       report.total_words,
+                "total_duration_ms": round(report.total_duration_ms, 2),
+            },
+            "results": [
+                {
+                    "source":       r.source,
+                    "char_count":   r.char_count,
+                    "word_count":   r.word_count,
+                    "line_count":   r.line_count,
+                    "duration_ms":  round(r.duration_ms, 2),
+                    "content_hash": r.content_hash,
+                    "mime_type":    r.mime_type,
+                }
+                for r in report.results
+            ],
+            "errors": [
+                {
+                    "source":      e.source,
+                    "error_type":  e.error_type,
+                    "message":     e.message,
+                    "duration_ms": round(e.duration_ms, 2),
+                }
+                for e in report.errors
+            ],
+        }
+        path.write_text(json.dumps(data, indent=2), encoding="utf-8")
+        logger.debug("write_batch_report | path=%s | total=%d", path, report.total)
+
+    def _render(self, result: ConversionResult) -> str:
+        """Serialise *result* to the configured output format."""
+        if self._format == "json":
+            return json.dumps(
+                {
+                    "source":   result.source,
+                    "markdown": result.markdown,
+                    "meta": {
+                        "char_count":   result.char_count,
+                        "word_count":   result.word_count,
+                        "line_count":   result.line_count,
+                        "duration_ms":  round(result.duration_ms, 2),
+                        "mime_type":    result.mime_type,
+                        "content_hash": result.content_hash,
+                    },
+                },
+                indent=2,
+            )
+        return result.markdown
+
+    @staticmethod
+    def _resolve_collision(path: Path) -> Path:
+        """Return *path* if it does not exist, otherwise append a counter to the stem."""
+        if not path.exists():
+            return path
+        counter = 1
+        while True:
+            candidate = path.with_stem(f"{path.stem}_{counter}")
+            if not candidate.exists():
+                return candidate
+            counter += 1

extraction/generic_json_extractor.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Dict, Optional, Union
+
+from logger import get_logger
+from .label_mapper import validate_mappings
+from .spacy_extractor import extract_fields
+
+logger = get_logger(__name__)
+
+TABULAR_EXTENSIONS = {".csv", ".xls", ".xlsx"}
+
+
+def is_tabular(filename: Union[str, Path]) -> bool:
+    return Path(filename).suffix.lower() in TABULAR_EXTENSIONS
+
+
+def extract(
+    filename: Union[str, Path],
+    markdown_text: str,
+    mappings: Optional[Dict[str, Dict[str, Any]]],
+    file_data: Optional[bytes] = None,
+) -> Dict[str, Any]:
+    ext = Path(filename).suffix.lower()
+
+    if ext in TABULAR_EXTENSIONS:
+        from .json_extractor import extract_json_from_file
+        result = extract_json_from_file(filename, file_data)
+        if "error" not in result:
+            result["extractor"] = "pandas"
+        return result
+
+    if not mappings:
+        return {
+            "error": (
+                f"Cannot extract JSON from '{ext}' files without field mappings. "
+                "Provide a 'mappings' object with field extraction rules."
+            ),
+            "file_type": ext,
+        }
+
+    valid, mapper_error = validate_mappings(mappings)
+    if not valid:
+        return {
+            "error": "invalid_spacy_labels",
+            "label_mapper": mapper_error,
+            "file_type": ext,
+        }
+
+    try:
+        data = extract_fields(markdown_text, mappings)
+        logger.info("spaCy extraction completed for %s: %d fields", ext, len(data))
+        return {
+            "success": True,
+            "extractor": "spacy",
+            "file_type": ext,
+            "data": data,
+        }
+    except Exception as exc:
+        logger.exception("spaCy extraction failed for %s", ext)
+        return {
+            "error": f"spaCy extraction failed: {exc}",
+            "file_type": ext,
+            "exception_type": type(exc).__name__,
+        }

extraction/json_extractor.py

@@ -0,0 +1,176 @@
+"""
+JSON extractor for structured data files.
+
+Supports only CSV, XLS, XLSX file types for JSON extraction.
+Returns error for unsupported file types.
+"""
+
+import io
+import warnings
+from pathlib import Path
+from typing import Any, Dict, Optional, Union
+
+import pandas as pd
+
+from logger import get_logger
+
+logger = get_logger(__name__)
+
+# Supported file extensions for JSON extraction
+SUPPORTED_EXTENSIONS = {'.csv', '.xls', '.xlsx'}
+
+# Resource limits to prevent abuse
+MAX_FILE_SIZE_BYTES = 100 * 1024 * 1024  # 100 MB
+MAX_CSV_ROWS = 100000  # Limit rows for CSV
+MAX_EXCEL_ROWS = 50000  # Limit rows for Excel
+MAX_MEMORY_ROWS = 100000  # Global memory guard
+
+
+def _validate_file_size(size: int) -> Optional[str]:
+    if size > MAX_FILE_SIZE_BYTES:
+        return f"File size {size} bytes exceeds limit of {MAX_FILE_SIZE_BYTES} bytes"
+    return None
+
+
+def _check_memory_usage(rows: int, cols: int) -> Optional[str]:
+    approx_mb = (rows * cols * 50) / (1024 * 1024)
+    if rows * cols > MAX_MEMORY_ROWS * 20:
+        return f"Data size too large (approx {approx_mb:.1f} MB). Too many cells: {rows}x{cols}"
+    return None
+
+
+def extract_json_from_file(
+    file_path: Union[str, Path],
+    file_data: Optional[bytes] = None
+) -> Dict[str, Any]:
+    """
+    Extract JSON data from structured files (CSV, XLS, XLSX).
+
+    Parameters
+    ----------
+    file_path : Union[str, Path]
+        Path to the file or filename with extension
+    file_data : Optional[bytes]
+        Raw file data (for stream processing)
+
+    Returns
+    -------
+    Dict[str, Any]
+        Extracted data or error information
+    """
+    ext = Path(file_path).suffix.lower()
+
+    if ext not in SUPPORTED_EXTENSIONS:
+        return {
+            "error": f"Unsupported file type: {ext}. Supported: {', '.join(sorted(SUPPORTED_EXTENSIONS))}",
+            "file_type": ext
+        }
+
+    # Validate file size if we have raw data
+    if file_data is not None:
+        size_error = _validate_file_size(len(file_data))
+        if size_error:
+            return {"error": size_error, "file_type": ext}
+    elif Path(file_path).exists():
+        size_error = _validate_file_size(Path(file_path).stat().st_size)
+        if size_error:
+            return {"error": size_error, "file_type": ext}
+
+    try:
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", UserWarning)
+
+            if ext == '.csv':
+                if file_data:
+                    stream = io.BytesIO(file_data)
+                    # Peek for encoding detection
+                    sample = stream.read(1024)
+                    stream.seek(0)
+                    df = pd.read_csv(stream, nrows=MAX_CSV_ROWS + 1, low_memory=False)
+                else:
+                    df = pd.read_csv(file_path, nrows=MAX_CSV_ROWS + 1, low_memory=False)
+            else:
+                if file_data:
+                    df = pd.read_excel(io.BytesIO(file_data), engine='openpyxl' if ext == '.xlsx' else 'xlrd')
+                else:
+                    df = pd.read_excel(file_path, engine='openpyxl' if ext == '.xlsx' else 'xlrd')
+
+        # Enforce row limits to prevent memory exhaustion
+        max_rows = MAX_EXCEL_ROWS if ext != '.csv' else MAX_CSV_ROWS
+        if len(df) > max_rows:
+            return {
+                "error": f"File contains {len(df)} rows, exceeds limit of {max_rows}",
+                "file_type": ext,
+                "row_count": len(df)
+            }
+
+        # Additional memory guard
+        mem_error = _check_memory_usage(len(df), len(df.columns))
+        if mem_error:
+            return {"error": mem_error, "file_type": ext}
+
+        # Efficient conversion to JSON-serializable format
+        result = {
+            "success": True,
+            "file_type": ext,
+            "data": {
+                "columns": list(df.columns),
+                "rows": df.where(pd.notnull(df), None).to_dict(orient='records'),
+                "shape": [len(df), len(df.columns)],
+                "dtypes": {col: str(dtype) for col, dtype in df.dtypes.items()}
+            }
+        }
+
+        logger.info("Extracted JSON from %s: %d rows, %d cols", ext, len(df), len(df.columns))
+        return result
+
+    except pd.errors.EmptyDataError:
+        return {
+            "error": f"File is empty or has no data",
+            "file_type": ext
+        }
+    except MemoryError:
+        return {
+            "error": "Out of memory processing file",
+            "file_type": ext
+        }
+    except Exception as exc:
+        logger.exception("JSON extraction failed for %s", ext)
+        return {
+            "error": f"Processing failed: {str(exc)}",
+            "file_type": ext,
+            "exception_type": type(exc).__name__
+        }
+
+
+
+
+
+def is_supported_file_type(file_path: Union[str, Path]) -> bool:
+    """
+    Check if file type is supported for JSON extraction.
+    
+    Parameters
+    ----------
+    file_path : Union[str, Path]
+        Path to the file or filename with extension
+        
+    Returns
+    -------
+    bool
+        True if supported, False otherwise
+    """
+    extension = Path(file_path).suffix.lower()
+    return extension in SUPPORTED_EXTENSIONS
+
+
+def get_supported_extensions() -> list:
+    """
+    Get list of supported file extensions for JSON extraction.
+    
+    Returns
+    -------
+    list
+        Supported file extensions
+    """
+    return sorted(SUPPORTED_EXTENSIONS)

extraction/label_mapper.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional, Tuple
+
+from .spacy_extractor import VALID_SPACY_LABELS
+
+
+def validate_mappings(fields: Dict[str, Dict[str, Any]]) -> Tuple[bool, Optional[Dict]]:
+    invalid: List[str] = []
+    for field_name, rule in fields.items():
+        if rule.get("source_type") == "entity":
+            label = rule.get("label", "")
+            if label not in VALID_SPACY_LABELS:
+                invalid.append(label)
+
+    if not invalid:
+        return True, None
+
+    return False, {
+        "invalid_labels": invalid,
+        "valid_labels": VALID_SPACY_LABELS,
+        "suggestion": (
+            "Use source_type 'regex' for custom patterns not covered by spaCy NER labels."
+        ),
+    }

extraction/spacy_extractor.py

@@ -0,0 +1,707 @@
+from __future__ import annotations
+
+import re
+import threading
+from typing import Any, Callable, Dict, List, Optional, Tuple, Union
+
+from logger import get_logger
+
+logger = get_logger(__name__)
+
+# ---------------------------------------------------------------------------
+# Type aliases
+# ---------------------------------------------------------------------------
+
+# A "schema node" is one of:
+#   • a leaf rule dict  → has a "source_type" key  (str leaf)
+#   • an array rule     → has "type": "array"
+#   • an object rule    → has "type": "object"
+# Results mirror the shape: str | list | dict | None at any depth.
+
+SchemaNode = Dict[str, Any]
+ResultNode = Union[str, List[Any], Dict[str, Any], None]
+
+VALID_SPACY_LABELS: Dict[str, str] = {
+    "ORG":         "Companies, agencies, institutions",
+    "PERSON":      "People, including fictional",
+    "DATE":        "Absolute or relative dates or periods",
+    "MONEY":       "Monetary values, including unit",
+    "GPE":         "Countries, cities, states",
+    "LOC":         "Non-GPE locations, mountain ranges, bodies of water",
+    "PRODUCT":     "Objects, vehicles, foods, etc.",
+    "EVENT":       "Named hurricanes, battles, wars, sports events",
+    "CARDINAL":    "Numerals that do not fall under another type",
+    "PERCENT":     "Percentage, including '%'",
+    "QUANTITY":    "Measurements, as of weight or distance",
+    "TIME":        "Times smaller than a day",
+    "NORP":        "Nationalities or religious or political groups",
+    "FAC":         "Buildings, airports, highways, bridges",
+    "WORK_OF_ART": "Titles of books, songs, etc.",
+    "LAW":         "Named documents made into laws",
+    "LANGUAGE":    "Any named language",
+    "ORDINAL":     "'first', 'second', etc.",
+}
+
+
+# ---------------------------------------------------------------------------
+# spaCy singleton
+# ---------------------------------------------------------------------------
+
+_nlp_lock = threading.Lock()
+_nlp      = None
+
+
+def _get_nlp():
+    global _nlp
+    if _nlp is not None:
+        return _nlp
+    with _nlp_lock:
+        if _nlp is None:
+            import spacy
+            _nlp = spacy.load(
+                "en_core_web_sm",
+                exclude=["tagger", "parser", "lemmatizer", "attribute_ruler"],
+            )
+            logger.info("spaCy en_core_web_sm loaded (singleton)")
+    return _nlp
+
+
+def clean_text(text: str) -> str:
+    return text
+
+
+# ---------------------------------------------------------------------------
+# Normalizer registry
+# ---------------------------------------------------------------------------
+
+_norm_lock  = threading.Lock()
+_NORMALIZERS: Dict[str, Callable[[str], str]] = {
+    "strip":               lambda s: s.strip(),
+    "upper":               lambda s: s.upper(),
+    "lower":               lambda s: s.lower(),
+    "remove_commas":       lambda s: s.replace(",", ""),
+    "remove_spaces":       lambda s: s.replace(" ", ""),
+    "remove_newlines":     lambda s: s.replace("\n", " ").replace("\r", ""),
+    "collapse_whitespace": lambda s: re.sub(r"\s+", " ", s).strip(),
+    "remove_currency":     lambda s: re.sub(r"[$€£¥₹]", "", s),
+    "remove_non_numeric":  lambda s: re.sub(r"[^\d.]", "", s),
+    "normalize_date_sep":  lambda s: re.sub(r"[/.]", "-", s),
+}
+
+
+def register_normalizer(name: str, fn: Callable[[str], str]) -> None:
+    """Register a custom normalizer. Thread-safe, overwrites silently."""
+    with _norm_lock:
+        _NORMALIZERS[name] = fn
+
+
+def _apply_normalizers(value: Optional[str], normalize: Any) -> Optional[str]:
+    if not isinstance(value, str):
+        return None
+    if not normalize:
+        return value
+    if isinstance(normalize, str):
+        normalize = [normalize]
+    for key in normalize:
+        with _norm_lock:
+            fn = _NORMALIZERS.get(key)
+        if fn is None:
+            logger.warning("Unknown normalizer %r — skipped", key)
+            continue
+        try:
+            value = fn(value)
+        except Exception as exc:
+            logger.error("Normalizer %r raised on value %r: %s", key, value, exc)
+    return value if value else None
+
+
+# ---------------------------------------------------------------------------
+# Resolver registry
+# ---------------------------------------------------------------------------
+
+_resolver_lock = threading.Lock()
+_RESOLVERS: Dict[str, Callable[[Dict[str, Any], Any, str], Optional[str]]] = {}
+
+
+def register_resolver(
+    source_type: str,
+    fn: Callable[[Dict[str, Any], Any, str], Optional[str]],
+) -> None:
+    """Register a custom resolver for a source_type. Thread-safe."""
+    with _resolver_lock:
+        _RESOLVERS[source_type] = fn
+
+
+# ---------------------------------------------------------------------------
+# Regex resolver
+# ---------------------------------------------------------------------------
+
+def _build_flags(rule: Dict[str, Any]) -> re.RegexFlag:
+    flags = re.RegexFlag(0)
+    for name in rule.get("flags", []):
+        obj = getattr(re, name.upper(), None)
+        if obj is None:
+            logger.warning("Unknown re flag %r — skipped", name)
+            continue
+        flags |= obj
+    return flags
+
+
+def _try_group(match: re.Match, capture_group: Any) -> Tuple[bool, Optional[str]]:
+    try:
+        return True, match.group(capture_group)
+    except (IndexError, re.error):
+        logger.warning(
+            "Group %r does not exist in pattern %r",
+            capture_group, match.re.pattern,
+        )
+        return False, None
+
+
+def _resolve_regex(rule: Dict[str, Any], text: str) -> Optional[str]:
+    primary = rule.get("pattern", "")
+    if not primary:
+        logger.warning("Regex rule missing 'pattern': %s", rule)
+        return None
+
+    flags         = _build_flags(rule)
+    capture_group = rule.get("capture_group", 0)
+    match_index   = rule.get("match_index", 0)
+    normalize     = rule.get("normalize", "")
+    strip_chars   = rule.get("strip_chars", "")
+    fallbacks     = rule.get("fallback_patterns", [])
+
+    for pat in [primary, *fallbacks]:
+        try:
+            matches = list(re.finditer(pat, text, flags))
+        except re.error as exc:
+            logger.error("Invalid regex %r: %s", pat, exc)
+            continue
+
+        if not matches:
+            continue
+
+        try:
+            target_matches = [matches[match_index]]
+        except IndexError:
+            target_matches = [matches[-1]]
+
+        for m in target_matches:
+            exists, result = _try_group(m, capture_group)
+            if not exists:
+                break
+            if result is None:
+                break
+            result = _apply_normalizers(result, normalize)
+            if result is None:
+                break
+            result = result.strip(strip_chars) if strip_chars else result.strip()
+            return result or None
+
+    return None
+
+
+# ---------------------------------------------------------------------------
+# Regex-array resolver  (all matches of a pattern → list of strings)
+# ---------------------------------------------------------------------------
+
+def _resolve_regex_all(rule: Dict[str, Any], text: str) -> List[Optional[str]]:
+    """
+    Like _resolve_regex but returns ALL matches as a list instead of one.
+
+    Extra rule keys versus the scalar regex rule:
+        max_items   int   Cap the number of results (default: unlimited).
+    """
+    primary = rule.get("pattern", "")
+    if not primary:
+        logger.warning("Regex-array rule missing 'pattern': %s", rule)
+        return []
+
+    flags         = _build_flags(rule)
+    capture_group = rule.get("capture_group", 0)
+    normalize     = rule.get("normalize", "")
+    strip_chars   = rule.get("strip_chars", "")
+    max_items     = rule.get("max_items")
+
+    try:
+        matches = list(re.finditer(primary, text, flags))
+    except re.error as exc:
+        logger.error("Invalid regex %r: %s", primary, exc)
+        return []
+
+    results: List[Optional[str]] = []
+    for m in matches:
+        exists, result = _try_group(m, capture_group)
+        if not exists or result is None:
+            continue
+        result = _apply_normalizers(result, normalize)
+        if result is None:
+            continue
+        result = result.strip(strip_chars) if strip_chars else result.strip()
+        if result:
+            results.append(result)
+        if max_items is not None and len(results) >= max_items:
+            break
+
+    return results
+
+
+# ---------------------------------------------------------------------------
+# Entity resolver
+# ---------------------------------------------------------------------------
+
+def _resolve_entity(rule: Dict[str, Any], doc: Any) -> Optional[str]:
+    if doc is None:
+        logger.warning("Entity resolver received None doc — skipping")
+        return None
+
+    labels = rule.get("label")
+    if isinstance(labels, str):
+        labels = [labels]
+    labels = set(labels or [])
+
+    match_index   = rule.get("match_index", 0)
+    min_length    = rule.get("min_length", 1)
+    exclude_pat   = rule.get("exclude_pattern", "")
+    exclude_flags = _build_flags({"flags": rule.get("exclude_flags", [])})
+    normalize     = rule.get("normalize", "")
+
+    candidates = [
+        ent.text for ent in doc.ents
+        if ent.label_ in labels
+        and len(ent.text) >= min_length
+        and not (exclude_pat and re.search(exclude_pat, ent.text, exclude_flags))
+    ]
+
+    if not candidates:
+        return None
+
+    try:
+        result = candidates[match_index]
+    except IndexError:
+        result = candidates[-1]
+
+    return _apply_normalizers(result, normalize)
+
+
+# ---------------------------------------------------------------------------
+# Entity-array resolver  (all matching entities → list)
+# ---------------------------------------------------------------------------
+
+def _resolve_entity_all(rule: Dict[str, Any], doc: Any) -> List[Optional[str]]:
+    """
+    Returns ALL entities matching the label filter as a list.
+
+    Extra rule key:
+        max_items   int   Cap the number of results (default: unlimited).
+        unique      bool  Deduplicate while preserving order (default: False).
+    """
+    if doc is None:
+        logger.warning("Entity-array resolver received None doc — skipping")
+        return []
+
+    labels = rule.get("label")
+    if isinstance(labels, str):
+        labels = [labels]
+    labels = set(labels or [])
+
+    min_length    = rule.get("min_length", 1)
+    exclude_pat   = rule.get("exclude_pattern", "")
+    exclude_flags = _build_flags({"flags": rule.get("exclude_flags", [])})
+    normalize     = rule.get("normalize", "")
+    max_items     = rule.get("max_items")
+    unique        = rule.get("unique", False)
+
+    results: List[str] = []
+    seen: set = set()
+
+    for ent in doc.ents:
+        if ent.label_ not in labels:
+            continue
+        if len(ent.text) < min_length:
+            continue
+        if exclude_pat and re.search(exclude_pat, ent.text, exclude_flags):
+            continue
+
+        value = _apply_normalizers(ent.text, normalize)
+        if not value:
+            continue
+        if unique:
+            if value in seen:
+                continue
+            seen.add(value)
+
+        results.append(value)
+        if max_items is not None and len(results) >= max_items:
+            break
+
+    return results
+
+
+# ---------------------------------------------------------------------------
+# Token-attribute resolver
+# ---------------------------------------------------------------------------
+
+def _resolve_token_attr(rule: Dict[str, Any], doc: Any) -> Optional[str]:
+    if doc is None:
+        logger.warning("Token-attr resolver received None doc — skipping")
+        return None
+
+    attr        = rule.get("attr", "")
+    match_index = rule.get("match_index", 0)
+    normalize   = rule.get("normalize", "")
+
+    candidates = [t.text for t in doc if getattr(t, attr, False)]
+    if not candidates:
+        return None
+
+    try:
+        result = candidates[match_index]
+    except IndexError:
+        result = candidates[-1]
+
+    return _apply_normalizers(result, normalize)
+
+
+# ---------------------------------------------------------------------------
+# Built-in resolver registration
+# ---------------------------------------------------------------------------
+
+register_resolver("regex",        lambda rule, doc, text: _resolve_regex(rule, text))
+register_resolver("entity",       lambda rule, doc, text: _resolve_entity(rule, doc))
+register_resolver("token_attr",   lambda rule, doc, text: _resolve_token_attr(rule, doc))
+# Array-producing leaf resolvers (used internally by the array node path):
+register_resolver("regex_all",    lambda rule, doc, text: _resolve_regex_all(rule, text))
+register_resolver("entity_all",   lambda rule, doc, text: _resolve_entity_all(rule, doc))
+
+
+# ---------------------------------------------------------------------------
+# Scalar field dispatcher  (returns str | None)
+# ---------------------------------------------------------------------------
+
+def _resolve_scalar_field(
+    rule: Dict[str, Any],
+    doc: Any,
+    text: str,
+) -> Optional[str]:
+    src = rule.get("source_type")
+    with _resolver_lock:
+        fn = _RESOLVERS.get(src)
+    if fn is None:
+        logger.warning("Unknown source_type %r — no resolver registered", src)
+        return None
+    return fn(rule, doc, text)
+
+
+# ---------------------------------------------------------------------------
+# Generic nested schema resolver
+# ---------------------------------------------------------------------------
+#
+# Schema node shapes
+# ──────────────────
+#
+# 1. LEAF  (scalar string)
+#    {
+#      "source_type": "regex" | "entity" | "token_attr" | <custom>,
+#      ...resolver-specific keys...
+#    }
+#
+# 2. OBJECT  (nested dict of named fields)
+#    {
+#      "type": "object",
+#      "fields": {
+#        "field_a": <schema_node>,
+#        "field_b": <schema_node>,
+#        ...
+#      }
+#    }
+#
+# 3. ARRAY  (repeated items)
+#    {
+#      "type": "array",
+#
+#      # --- how to split the text into per-item segments (optional) ---
+#      # If omitted the whole text is the single segment (useful when
+#      # the item schema itself fans out via regex_all / entity_all).
+#      "split_pattern": "<regex>",   # splits text; each piece → one item
+#      "split_flags":   ["DOTALL"],  # re flags for split_pattern
+#
+#      # --- what each item looks like ---
+#      "items": <schema_node>
+#      # Can be a leaf, an object, or even another array (any depth).
+#    }
+#
+# Results
+# ───────
+#   LEAF   → str | None
+#   OBJECT → {field: result, ...}   (all keys always present, value may be None)
+#   ARRAY  → [result, ...]          (may be empty; each element mirrors item schema)
+
+
+def _resolve_node(node: SchemaNode, doc: Any, text: str) -> ResultNode:
+    """
+    Recursively resolve a schema node against `text` / `doc`.
+    Dispatches on node["type"] or falls back to scalar leaf resolution.
+    """
+    node_type = node.get("type")
+
+    if node_type == "object":
+        return _resolve_object_node(node, doc, text)
+
+    if node_type == "array":
+        return _resolve_array_node(node, doc, text)
+
+    # No "type" key → treat as a scalar leaf rule
+    return _resolve_scalar_field(node, doc, text)
+
+
+def _resolve_object_node(
+    node: SchemaNode,
+    doc: Any,
+    text: str,
+) -> Dict[str, ResultNode]:
+    """
+    Resolve every field in node["fields"] and return a dict.
+    Each field may itself be a leaf, object, or array — fully recursive.
+    """
+    fields: Dict[str, SchemaNode] = node.get("fields", {})
+    result: Dict[str, ResultNode] = {}
+
+    for field_name, child_node in fields.items():
+        try:
+            result[field_name] = _resolve_node(child_node, doc, text)
+        except Exception as exc:
+            logger.error(
+                "Object field %r raised unexpectedly: %s", field_name, exc,
+                exc_info=True,
+            )
+            result[field_name] = None
+
+    return result
+
+
+def _resolve_array_node(
+    node: SchemaNode,
+    doc: Any,
+    text: str,
+) -> List[ResultNode]:
+    """
+    Resolve an array node:
+
+    Two operating modes, selected by whether "split_pattern" is present:
+
+    MODE A — split_pattern present
+        Split the text into N segments; resolve item schema against each
+        segment with its own spaCy doc.  Good for table rows, repeated
+        blocks, delimited records, etc.
+
+    MODE B — no split_pattern
+        Resolve item schema against the full text once.
+        If the item schema is a leaf with source_type in {regex_all,
+        entity_all} it returns a list natively.
+        If the item schema returns a list → that IS the array.
+        If it returns a scalar → wrap in [scalar].
+        This handles "give me all ORG entities" without needing a split.
+    """
+    item_schema: SchemaNode  = node.get("items", {})
+    split_pat: Optional[str] = node.get("split_pattern")
+    split_flags_rule          = {"flags": node.get("split_flags", [])}
+    max_items: Optional[int]  = node.get("max_items")
+
+    results: List[ResultNode] = []
+
+    if split_pat:
+        # ── MODE A: segment-per-item ────────────────────────────────────
+        try:
+            flags    = _build_flags(split_flags_rule)
+            segments = re.split(split_pat, text, flags=flags)
+        except re.error as exc:
+            logger.error("Invalid split_pattern %r: %s", split_pat, exc)
+            return []
+
+        nlp = _get_nlp()
+
+        try:
+            segment_docs = list(nlp.pipe(segments))
+        except Exception as exc:
+            logger.error("spaCy pipe failed on array segments: %s", exc, exc_info=True)
+            segment_docs = [None] * len(segments)
+
+        for seg_doc, seg_text in zip(segment_docs, segments):
+            if not seg_text.strip():
+                continue
+            try:
+                item_result = _resolve_node(item_schema, seg_doc, seg_text)
+            except Exception as exc:
+                logger.error(
+                    "Array item resolve raised: %s", exc, exc_info=True
+                )
+                item_result = None
+
+            results.append(item_result)
+            if max_items is not None and len(results) >= max_items:
+                break
+
+    else:
+        # ── MODE B: whole-text, native fan-out ─────────────────────────
+        try:
+            raw = _resolve_node(item_schema, doc, text)
+        except Exception as exc:
+            logger.error("Array item resolve raised: %s", exc, exc_info=True)
+            return []
+
+        if isinstance(raw, list):
+            results = raw
+        elif raw is not None:
+            results = [raw]
+
+        if max_items is not None:
+            results = results[:max_items]
+
+    return results
+
+
+# ---------------------------------------------------------------------------
+# Safe wrappers
+# ---------------------------------------------------------------------------
+
+def _safe_resolve_node(
+    path: str,
+    node: SchemaNode,
+    doc: Any,
+    text: str,
+) -> ResultNode:
+    """Resolve a node; isolate crashes so siblings still complete."""
+    try:
+        return _resolve_node(node, doc, text)
+    except Exception as exc:
+        logger.error(
+            "Schema path %r raised unexpectedly: %s", path, exc, exc_info=True
+        )
+        return None
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def extract_fields(
+    text: str,
+    fields: Dict[str, SchemaNode],
+) -> Dict[str, ResultNode]:
+    """
+    Extract fields from a single text string.
+
+    `fields` is a flat dict of {name: schema_node}.  Each schema node may be
+    a scalar leaf, an object node, or an array node — nested to any depth.
+
+    Returns {field_name: result_or_None}.
+
+    Backward-compatible: callers that pass flat scalar rules unchanged still work.
+    """
+    nlp     = _get_nlp()
+    cleaned = clean_text(text)
+
+    try:
+        doc = next(iter(nlp.pipe([cleaned])))
+    except Exception as exc:
+        logger.error("spaCy pipe failed: %s", exc, exc_info=True)
+        doc = None
+
+    return {
+        field: _safe_resolve_node(field, node, doc, cleaned)
+        for field, node in fields.items()
+    }
+
+
+def extract_schema(
+    text: str,
+    schema: SchemaNode,
+) -> ResultNode:
+    """
+    Resolve a *single* schema node (which may be a leaf, object, or array)
+    against `text`.
+
+    Useful when the top-level result should itself be a list or a structured
+    object rather than a flat dict of fields.
+
+    Example
+    -------
+    schema = {
+        "type": "array",
+        "split_pattern": r"\\n\\n+",
+        "items": {
+            "type": "object",
+            "fields": {
+                "date":   {"source_type": "entity", "label": "DATE"},
+                "amount": {"source_type": "regex",  "pattern": r"\\$[\\d,]+"},
+            }
+        }
+    }
+    result = extract_schema(invoice_text, schema)
+    # → [{"date": "Jan 2024", "amount": "$1,200"}, ...]
+    """
+    nlp     = _get_nlp()
+    cleaned = clean_text(text)
+
+    try:
+        doc = next(iter(nlp.pipe([cleaned])))
+    except Exception as exc:
+        logger.error("spaCy pipe failed: %s", exc, exc_info=True)
+        doc = None
+
+    return _safe_resolve_node("<root>", schema, doc, cleaned)
+
+
+def extract_fields_batch(
+    texts: List[str],
+    fields: Dict[str, SchemaNode],
+) -> List[Dict[str, ResultNode]]:
+    """
+    Extract fields from a list of texts in a single top-level spaCy pipe pass.
+
+    Returns one result dict per input text, in the same order.
+
+    Note: array nodes with split_pattern trigger their own inner pipe call per
+    text; the outer pass still processes the top-level texts efficiently.
+    """
+    nlp     = _get_nlp()
+    cleaned = [clean_text(t) for t in texts]
+
+    try:
+        docs = list(nlp.pipe(cleaned))
+    except Exception as exc:
+        logger.error("spaCy pipe failed: %s", exc, exc_info=True)
+        docs = [None] * len(cleaned)
+
+    return [
+        {
+            field: _safe_resolve_node(field, node, doc, text)
+            for field, node in fields.items()
+        }
+        for doc, text in zip(docs, cleaned)
+    ]
+
+
+def extract_schema_batch(
+    texts: List[str],
+    schema: SchemaNode,
+) -> List[ResultNode]:
+    """
+    Like extract_schema but processes a list of texts efficiently.
+
+    Returns one ResultNode per input text, in the same order.
+    """
+    nlp     = _get_nlp()
+    cleaned = [clean_text(t) for t in texts]
+
+    try:
+        docs = list(nlp.pipe(cleaned))
+    except Exception as exc:
+        logger.error("spaCy pipe failed: %s", exc, exc_info=True)
+        docs = [None] * len(cleaned)
+
+    return [
+        _safe_resolve_node(f"<root>[{i}]", schema, doc, text)
+        for i, (doc, text) in enumerate(zip(docs, cleaned))
+    ]

logger.py

@@ -0,0 +1,136 @@
+"""
+Logging configuration for the MarkItDown API.
+
+Provides a structured plaintext formatter suitable for production log
+aggregation pipelines (stdout + optional log file). No ANSI colour codes
+are emitted, keeping output clean for log collectors and CI environments.
+
+Public surface
+--------------
+    get_logger(name)  ->  logging.Logger
+        Returns the named logger, bootstrapping from the root application
+        logger if the logger has not been configured yet.
+
+    setup_logger(...)  ->  logging.Logger
+        Full configuration path used once at application startup.
+
+    app_logger
+        Pre-configured root logger instance used by all modules.
+"""
+
+from __future__ import annotations
+
+import logging
+import sys
+from pathlib import Path
+from typing import Optional
+
+
+# ---------------------------------------------------------------------------
+# Formatter
+# ---------------------------------------------------------------------------
+
+class PlainFormatter(logging.Formatter):
+    """Structured pipe-delimited log line.
+
+    Example output::
+        2024-01-15 12:34:56 | INFO     | llm_ready_data_extractor.api.server | message text
+    """
+
+    def format(self, record: logging.LogRecord) -> str:
+        timestamp = self.formatTime(record, self.datefmt)
+        line = (
+            f"{timestamp} | {record.levelname:<8} | "
+            f"{record.name} | {record.getMessage()}"
+        )
+        if record.exc_info:
+            line = f"{line}\n{self.formatException(record.exc_info)}"
+        return line
+
+
+# ---------------------------------------------------------------------------
+# Handler factories
+# ---------------------------------------------------------------------------
+
+def _build_console_handler() -> logging.StreamHandler:
+    """Return a StreamHandler writing structured lines to stdout."""
+    handler = logging.StreamHandler(sys.stdout)
+    handler.setFormatter(
+        PlainFormatter(datefmt="%Y-%m-%d %H:%M:%S")
+    )
+    return handler
+
+
+def _build_file_handler(log_file: str) -> logging.FileHandler:
+    """Return a FileHandler writing to *log_file*, creating parent dirs as needed."""
+    log_path = Path(log_file)
+    log_path.parent.mkdir(parents=True, exist_ok=True)
+    handler = logging.FileHandler(log_file, encoding="utf-8")
+    handler.setFormatter(
+        PlainFormatter(datefmt="%Y-%m-%d %H:%M:%S")
+    )
+    return handler
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def setup_logger(
+    name: str,
+    level: str = "INFO",
+    log_file: Optional[str] = None,
+    enable_console: bool = True,
+) -> logging.Logger:
+    """Configure and return a named logger.
+
+    Parameters
+    ----------
+    name:
+        Logger name — pass ``__name__`` from the calling module.
+    level:
+        Minimum log level string, e.g. ``"DEBUG"``, ``"INFO"``, ``"WARNING"``.
+    log_file:
+        Optional path to a persistent log file. When supplied, records are
+        written to both stdout and the file.
+    enable_console:
+        Set to ``False`` to suppress stdout output (useful in test environments).
+    """
+    logger = logging.getLogger(name)
+    logger.setLevel(getattr(logging, level.upper(), logging.INFO))
+    logger.handlers.clear()
+    logger.propagate = False
+
+    if enable_console:
+        logger.addHandler(_build_console_handler())
+
+    if log_file:
+        logger.addHandler(_build_file_handler(log_file))
+
+    return logger
+
+
+def get_logger(name: str) -> logging.Logger:
+    """Return the named logger, configuring it with defaults if not yet set up.
+
+    Intended usage in application modules::
+
+        from logger import get_logger
+        logger = get_logger(__name__)
+    """
+    existing = logging.getLogger(name)
+    if existing.handlers:
+        return existing
+    return setup_logger(name)
+
+
+# ---------------------------------------------------------------------------
+# Root application logger — configured once at import time
+# ---------------------------------------------------------------------------
+
+app_logger = setup_logger(
+    "llm_ready_data_extractor",
+    level="INFO",
+    log_file="logs/markitdown.log",
+    enable_console=True,
+)

main.py

@@ -0,0 +1,27 @@
+"""
+MarkItDown API — entry point.
+
+Run directly::
+    python main.py
+
+Or via start.sh for production (environment variable overrides supported).
+"""
+from __future__ import annotations
+
+import uvicorn
+
+from logger import get_logger
+
+logger = get_logger(__name__)
+
+if __name__ == "__main__":
+    logger.info("Starting MarkItDown API server")
+    uvicorn.run(
+        "api.server:app",
+        host="0.0.0.0",
+        port=7860,
+        reload=False,
+        workers=1,
+        log_level="info",
+        access_log=True,
+    )

requirements.txt

@@ -0,0 +1,21 @@
+# MarkItDown API — Dependencies (Lightweight)
+markitdown[all]>=0.1.5
+fastapi>=0.111.0
+uvicorn[standard]>=0.30.0
+pydantic>=2.7.0
+python-multipart>=0.0.9
+httpx>=0.27.0
+
+# OCR — RapidOCR with onnxruntime CPU backend (English + Chinese)
+rapidocr-onnxruntime>=1.4.4
+onnxruntime>=1.18.0
+pillow>=10.0.0
+
+# Scanned-PDF fallback (page rendering before OCR)
+pypdfium2>=4.30.0
+
+# Data processing for JSON extraction
+pandas>=2.0.0
+
+# spaCy NER extraction for non-tabular file formats
+spacy>=3.7.0

self_ping.py

@@ -0,0 +1,19 @@
+#!/usr/bin/env python3
+"""Self-ping script that hits a URL."""
+
+import datetime
+import os
+import urllib.request
+
+PING_URL = os.environ.get("PING_URL", "https://your-url.com/health")  
+
+def main():
+    timestamp = datetime.datetime.now().isoformat()
+    try:
+        urllib.request.urlopen(PING_URL, timeout=10)
+        print(f"[{timestamp}] Pinged {PING_URL} - OK")
+    except Exception as e:
+        print(f"[{timestamp}] Ping failed: {e}")
+
+if __name__ == "__main__":
+    main()

start.sh

@@ -0,0 +1,86 @@
+#!/usr/bin/env bash
+# MarkItDown API - Production Startup Script
+#
+# Environment variable overrides:
+#   LOG_DIR   - directory for persistent log files  (default: /app/logs)
+#   HOST      - bind address                         (default: 0.0.0.0)
+#   PORT      - listen port                          (default: 7860)
+#   WORKERS   - uvicorn worker count                 (default: 1)
+#   LOG_LEVEL - uvicorn log level                    (default: info)
+
+set -euo pipefail
+
+# ---------------------------------------------------------------------------
+# Logging helpers
+# ---------------------------------------------------------------------------
+LOG_DIR="${LOG_DIR:-/app/logs}"
+mkdir -p "$LOG_DIR"
+LOG_FILE="$LOG_DIR/startup.log"
+
+_log() {
+    local level="$1"
+    shift
+    local ts
+    ts="$(date -u '+%Y-%m-%dT%H:%M:%SZ')"
+    local line="[$ts] [$level] $*"
+    echo "$line"
+    echo "$line" >> "$LOG_FILE"
+}
+
+log_info()  { _log "INFO " "$@"; }
+log_ok()    { _log "OK   " "$@"; }
+log_warn()  { _log "WARN " "$@"; }
+log_err()   { _log "ERROR" "$@"; }
+log_step()  { echo ""; _log "STEP " "---- $* ----"; }
+
+# ---------------------------------------------------------------------------
+# Step 1: Verify core Python dependencies
+# ---------------------------------------------------------------------------
+log_step "1/2  Verifying Python dependencies"
+
+if python -c "import markitdown, fastapi, httpx" 2>/dev/null; then
+    log_ok "Core dependencies verified."
+else
+    log_err "Missing core dependencies. Ensure requirements.txt has been installed."
+    exit 1
+fi
+
+# ---------------------------------------------------------------------------
+# Step 2: Smoke-test imports
+# ---------------------------------------------------------------------------
+log_step "2/2  Running import smoke test"
+
+python >> "$LOG_FILE" 2>&1 << PYEOF
+import sys
+try:
+    import markitdown, fastapi, httpx, pandas
+    print("[smoke-test] All core imports successful.")
+except ImportError as exc:
+    print(f"[smoke-test] Import error: {exc}")
+    sys.exit(1)
+PYEOF
+
+if [ $? -eq 0 ]; then
+    log_ok "Import smoke test passed."
+else
+    log_warn "Import smoke test failed. Check $LOG_FILE for details. Continuing startup."
+fi
+
+# ---------------------------------------------------------------------------
+# Start uvicorn
+# ---------------------------------------------------------------------------
+log_step "Starting uvicorn"
+
+HOST="${HOST:-0.0.0.0}"
+PORT="${PORT:-7860}"
+WORKERS="${WORKERS:-1}"
+LOG_LEVEL="${LOG_LEVEL:-info}"
+
+log_info "Binding to $HOST:$PORT | workers=$WORKERS | log_level=$LOG_LEVEL"
+
+exec python -m uvicorn api.server:app \
+    --host "$HOST" \
+    --port "$PORT" \
+    --workers "$WORKERS" \
+    --log-level "$LOG_LEVEL" \
+    --access-log