Docling Production Deployment Guide

celery -A app.celery worker \
  --concurrency=4 \
  --max-memory-per-child=3500000 \  # 3.5GB per worker - prevents OOM crashes
  --max-tasks-per-child=50 \         # Restart workers after 50 tasks - clears memory leaks
  --prefetch-multiplier=1            # Prevent workers from hoarding tasks

# production_docling_celery.py
import os
import time
import traceback
from celery import Celery, Task
from celery.exceptions import SoftTimeLimitExceeded
from docling.document_converter import DocumentConverter
from docling.datamodel.base_models import InputFormat
from docling.datamodel.pipeline_options import PdfPipelineOptions, TesseractCliOcrOptions, EasyOcrOptions
from docling.backend.pypdfium2_backend import PyPdfiumDocumentBackend
from markitdown import MarkItDown
import psutil
import redis
from typing import Optional, Dict, Any

# Initialize Celery with Redis as broker and result backend
celery = Celery(
    'document_processing',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1'
)

# Celery configuration for production reliability
celery.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='UTC',
    enable_utc=True,
    task_track_started=True,
    task_time_limit=7200,  # 2 hour hard limit for large documents
    task_soft_time_limit=6900,  # 115 minute soft limit
    worker_prefetch_multiplier=1,  # One task at a time per worker
    worker_max_tasks_per_child=50,  # Restart after 50 tasks
    worker_max_memory_per_child=3500000,  # 3.5GB per worker
)

# Redis client for WebSocket real-time updates
redis_client = redis.Redis(host='localhost', port=6379, db=2, decode_responses=True)

def should_enable_ocr(file_path: str, file_type: str) -> bool:
    """
    Auto-detect if a document needs OCR processing.
    Scanned PDFs and images need OCR. Native PDFs don't.
    This saves 3-5x processing time on native PDFs.
    """
    if file_type in ['image/png', 'image/jpeg', 'image/jpg', 'image/tiff']:
        return True  # Images always need OCR

    if file_type == 'application/pdf':
        # Check if PDF has extractable text (native) or is scanned
        try:
            import PyPDF2
            with open(file_path, 'rb') as f:
                reader = PyPDF2.PdfReader(f)
                # Sample first 3 pages to determine if text exists
                text_sample = ""
                for page_num in range(min(3, len(reader.pages))):
                    text_sample += reader.pages[page_num].extract_text()

                # If less than 50 characters across 3 pages, it's likely scanned
                if len(text_sample.strip()) < 50:
                    return True  # Scanned PDF, needs OCR
                else:
                    return False  # Native PDF, skip OCR
        except Exception as e:
            print(f"Error checking PDF text: {e}. Defaulting to OCR enabled.")
            return True  # If we can't determine, enable OCR to be safe

    return False  # DOCX, PPTX don't need OCR

def publish_status_update(task_id: str, status: str, progress: int, message: str):
    """
    Publish real-time status updates via Redis pub/sub.
    Your frontend WebSocket subscribes to these updates.
    """
    update = {
        'task_id': task_id,
        'status': status,
        'progress': progress,
        'message': message,
        'timestamp': time.time()
    }
    redis_client.publish(f'task:{task_id}', str(update))

@celery.task(bind=True, max_retries=3, default_retry_delay=5)
def process_document_task(
    self: Task,
    file_path: str,
    file_type: str,
    document_id: str,
    tier: str = 'advanced'  # 'simple' or 'advanced'
) -> Dict[str, Any]:
    """
    Production-ready document processing with Docling.

    Args:
        file_path: Path to uploaded document
        file_type: MIME type (application/pdf, image/png, etc.)
        document_id: Database ID for tracking
        tier: 'simple' for small docs (under 10 pages), 'advanced' for complex docs

    Returns:
        Dict with extracted content, metadata, and processing stats
    """
    start_time = time.time()
    task_id = self.request.id

    # Update status: queued → processing
    publish_status_update(task_id, 'processing', 10, 'Starting document processing')

    try:
        # Memory baseline
        process = psutil.Process(os.getpid())
        memory_start = process.memory_info().rss / 1024 / 1024  # MB

        # Determine if OCR is needed (auto-detection saves time on native PDFs)
        enable_ocr = should_enable_ocr(file_path, file_type)

        # Smart tier routing: small docs use simple pipeline, large docs get full treatment
        if tier == 'simple':
            # Simple tier: skip OCR for speed (unless auto-detected as scanned)
            enable_ocr = enable_ocr and file_type.startswith('image/')

        publish_status_update(task_id, 'processing', 25, f'OCR {"enabled" if enable_ocr else "disabled"}')

        # Configure Docling pipeline with production-optimized settings
        pipeline_options = PdfPipelineOptions()
        pipeline_options.do_ocr = enable_ocr  # Boolean True/False (NOT string "auto")

        if enable_ocr:
            # EasyOCR for GPU, Tesseract for CPU
            if os.path.exists('/dev/nvidia0'):  # Simple GPU check
                pipeline_options.ocr_options = EasyOcrOptions(
                    lang=['en'],  # Add more languages as needed: ['en', 'es', 'fr']
                    use_gpu=True
                )
            else:
                pipeline_options.ocr_options = TesseractCliOcrOptions(
                    lang=['eng']  # Tesseract uses 3-letter codes
                )

        # Initialize DocumentConverter with optimized backend
        converter = DocumentConverter(
            format_options={
                InputFormat.PDF: pipeline_options
            },
            pdf_backend=PyPdfiumDocumentBackend  # Faster than default
        )

        publish_status_update(task_id, 'processing', 40, 'Extracting document structure')

        # Process document (this is where the magic happens)
        result = converter.convert(file_path)

        publish_status_update(task_id, 'processing', 70, 'Extracting tables and content')

        # Extract structured data
        markdown_content = result.document.export_to_markdown()
        doctags_xml = result.document.export_to_xml()  # DocTags format

        # Extract tables separately for better RAG chunking
        tables = []
        for table in result.document.tables:
            tables.append({
                'data': table.export_to_dataframe().to_dict('records'),
                'bbox': table.bbox,  # Spatial coordinates
                'page': table.page_no
            })

        # Calculate processing metrics
        end_time = time.time()
        processing_time = end_time - start_time
        memory_end = process.memory_info().rss / 1024 / 1024  # MB
        memory_used = memory_end - memory_start

        publish_status_update(task_id, 'completed', 100, 'Document processed successfully')

        return {
            'status': 'success',
            'document_id': document_id,
            'content': markdown_content,
            'doctags_xml': doctags_xml,
            'tables': tables,
            'page_count': len(result.document.pages),
            'table_count': len(tables),
            'metrics': {
                'processing_time': round(processing_time, 2),
                'memory_used_mb': round(memory_used, 2),
                'ocr_enabled': enable_ocr,
                'tier': tier
            }
        }

    except SoftTimeLimitExceeded:
        # Hit 115-minute soft limit - try MarkItDown fallback
        publish_status_update(task_id, 'processing', 75, 'Timeout - trying fallback processor')
        return fallback_to_markitdown(file_path, document_id, task_id)

    except Exception as e:
        # Log error and retry with exponential backoff (1s, 5s, 25s)
        error_msg = f"Docling error: {str(e)}"
        print(f"Task {task_id} error: {error_msg}")
        print(traceback.format_exc())

        # Retry up to 3 times with exponential backoff
        if self.request.retries < self.max_retries:
            retry_delay = 5 * (self.request.retries + 1)  # 5s, 10s, 15s
            publish_status_update(
                task_id,
                'retrying',
                50,
                f'Error occurred. Retrying in {retry_delay}s (attempt {self.request.retries + 1}/3)'
            )
            raise self.retry(exc=e, countdown=retry_delay)
        else:
            # All retries exhausted - try MarkItDown as last resort
            publish_status_update(task_id, 'processing', 80, 'Docling failed - using fallback')
            return fallback_to_markitdown(file_path, document_id, task_id)

def fallback_to_markitdown(file_path: str, document_id: str, task_id: str) -> Dict[str, Any]:
    """
    Graceful fallback to MarkItDown if Docling fails.
    MarkItDown is less accurate but more reliable for edge cases.
    """
    try:
        md = MarkItDown()
        result = md.convert(file_path)

        publish_status_update(task_id, 'completed', 100, 'Processed with fallback (lower accuracy)')

        return {
            'status': 'success_fallback',
            'document_id': document_id,
            'content': result.text_content,
            'tables': [],  # MarkItDown doesn't extract tables as structured data
            'page_count': None,
            'table_count': 0,
            'metrics': {
                'processing_time': 0,
                'memory_used_mb': 0,
                'ocr_enabled': False,
                'tier': 'fallback'
            },
            'warning': 'Processed with MarkItDown fallback - table extraction unavailable'
        }
    except Exception as e:
        publish_status_update(task_id, 'failed', 0, f'All processing methods failed: {str(e)}')
        return {
            'status': 'error',
            'document_id': document_id,
            'error': str(e)
        }

# production_ocr_config.py
import cv2
import numpy as np
from PIL import Image
import PyPDF2
from typing import Tuple, Optional
from docling.datamodel.pipeline_options import PdfPipelineOptions, TesseractCliOcrOptions, EasyOcrOptions

def detect_gpu_availability() -> bool:
    """Check if NVIDIA GPU is available for EasyOCR."""
    try:
        import torch
        return torch.cuda.is_available()
    except ImportError:
        return False

def preprocess_image_for_ocr(image_path: str, output_path: str) -> str:
    """
    Preprocess scanned images for better OCR accuracy.
    Applies deskewing, denoising, and contrast enhancement.
    This can improve OCR accuracy by 5-10%.
    """
    # Read image
    img = cv2.imread(image_path)

    # Convert to grayscale
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)

    # Deskew: rotate image to correct orientation
    # This fixes images that are slightly tilted
    coords = np.column_stack(np.where(gray > 0))
    angle = cv2.minAreaRect(coords)[-1]
    if angle < -45:
        angle = -(90 + angle)
    else:
        angle = -angle

    (h, w) = gray.shape[:2]
    center = (w // 2, h // 2)
    M = cv2.getRotationMatrix2D(center, angle, 1.0)
    deskewed = cv2.warpAffine(gray, M, (w, h), flags=cv2.INTER_CUBIC, borderMode=cv2.BORDER_REPLICATE)

    # Denoise: remove scanning artifacts and noise
    denoised = cv2.fastNlMeansDenoising(deskewed, None, h=10, templateWindowSize=7, searchWindowSize=21)

    # Adaptive threshold: enhance contrast for better character recognition
    # This works better than simple thresholding for varied lighting conditions
    enhanced = cv2.adaptiveThreshold(
        denoised,
        255,
        cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
        cv2.THRESH_BINARY,
        11,
        2
    )

    # Save preprocessed image
    cv2.imwrite(output_path, enhanced)
    return output_path

def is_pdf_scanned(pdf_path: str, sample_pages: int = 3) -> Tuple[bool, int]:
    """
    Detect if a PDF is scanned (needs OCR) or native (has text).

    Returns:
        (is_scanned, total_pages): Boolean indicating if OCR needed, total page count
    """
    try:
        with open(pdf_path, 'rb') as f:
            reader = PyPDF2.PdfReader(f)
            total_pages = len(reader.pages)

            # Sample first N pages (or all pages if document is short)
            pages_to_check = min(sample_pages, total_pages)
            total_text_length = 0

            for page_num in range(pages_to_check):
                page = reader.pages[page_num]
                text = page.extract_text()
                total_text_length += len(text.strip())

            # Heuristic: if less than 50 chars per page on average, it's scanned
            # Native PDFs typically have 1000+ chars per page
            avg_chars_per_page = total_text_length / pages_to_check
            is_scanned = avg_chars_per_page < 50

            return is_scanned, total_pages

    except Exception as e:
        print(f"Error checking PDF: {e}. Defaulting to OCR enabled.")
        return True, 0  # If we can't determine, enable OCR to be safe

def get_optimized_ocr_config(
    file_path: str,
    file_type: str,
    force_ocr: Optional[bool] = None,
    languages: list = ['en']
) -> PdfPipelineOptions:
    """
    Get production-optimized OCR configuration for Docling.

    Args:
        file_path: Path to document
        file_type: MIME type
        force_ocr: If True/False, override auto-detection. If None, auto-detect
        languages: List of language codes for OCR (e.g., ['en', 'es', 'fr'])

    Returns:
        Configured PdfPipelineOptions object
    """
    pipeline_options = PdfPipelineOptions()

    # Auto-detect if OCR is needed (unless explicitly forced)
    if force_ocr is not None:
        enable_ocr = force_ocr
    else:
        if file_type in ['image/png', 'image/jpeg', 'image/jpg', 'image/tiff']:
            enable_ocr = True  # Images always need OCR
        elif file_type == 'application/pdf':
            is_scanned, _ = is_pdf_scanned(file_path)
            enable_ocr = is_scanned
        else:
            enable_ocr = False  # DOCX, PPTX have extractable text

    # CRITICAL: do_ocr is boolean True/False, NOT string "auto"
    pipeline_options.do_ocr = enable_ocr

    if enable_ocr:
        # Choose OCR engine based on GPU availability
        has_gpu = detect_gpu_availability()

        if has_gpu:
            # EasyOCR with GPU: best accuracy (95%+), 200ms/page
            pipeline_options.ocr_options = EasyOcrOptions(
                lang=languages,  # Supports 80+ languages
                use_gpu=True,
                # Optional: adjust confidence threshold (0.4 is default)
                # Lower = more text detected but potentially noisier
                # Higher = cleaner results but might miss faint text
            )
            print(f"OCR enabled with EasyOCR (GPU) - Languages: {languages}")
        else:
            # Tesseract for CPU: faster (80ms/page), 88-92% accuracy
            # Convert language codes to Tesseract format (eng, spa, fra)
            tesseract_langs = []
            lang_map = {'en': 'eng', 'es': 'spa', 'fr': 'fra', 'de': 'deu', 'it': 'ita'}
            for lang in languages:
                tesseract_langs.append(lang_map.get(lang, lang))

            pipeline_options.ocr_options = TesseractCliOcrOptions(
                lang=tesseract_langs
            )
            print(f"OCR enabled with Tesseract (CPU) - Languages: {tesseract_langs}")
    else:
        print("OCR disabled - native PDF detected")

    return pipeline_options

# Example usage in production
if __name__ == "__main__":
    # Example 1: Auto-detect OCR for a PDF
    pdf_path = "/path/to/document.pdf"
    config = get_optimized_ocr_config(pdf_path, "application/pdf")
    print(f"OCR enabled: {config.do_ocr}")

    # Example 2: Force OCR for a scanned image
    image_path = "/path/to/scanned.png"
    config_forced = get_optimized_ocr_config(
        image_path,
        "image/png",
        force_ocr=True,
        languages=['en', 'es']  # Bilingual document
    )

    # Example 3: Preprocess image before OCR
    preprocessed_path = "/tmp/preprocessed.png"
    preprocess_image_for_ocr(image_path, preprocessed_path)
    # Now use preprocessed_path with Docling for 5-10% better accuracy