ocr-service/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── config.py
│   ├── models.py
│   ├── services/
│   │   ├── __init__.py
│   │   ├── ocr.py
│   │   └── preprocessing.py
│   ├── api/
│   │   ├── __init__.py
│   │   └── routes.py
│   └── utils/
│       ├── __init__.py
│       └── logging.py
├── tests/
├── docker-compose.yml
├── Dockerfile
├── requirements.txt
└── .env.example

fastapi==0.104.1
uvicorn[standard]==0.24.0
python-multipart==0.0.6
pytesseract==0.3.10
opencv-python-headless==4.8.1.78
Pillow==10.1.0
numpy==1.26.2
redis==5.0.1
sqlalchemy==2.0.23
asyncpg==0.29.0
pydantic==2.5.0
pydantic-settings==2.1.0
celery==5.3.4
prometheus-client==0.19.0
structlog==23.2.0

# app/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    # Application
    app_name: str = "OCR Production Service"
    debug: bool = False
    workers: int = 4

    # OCR Engine
    tesseract_path: str = "/usr/bin/tesseract"
    tesseract_datapath: str = "/usr/share/tesseract-ocr/5/tessdata"
    ocr_languages: str = "eng"
    ocr_psm: int = 3  # Page segmentation mode
    ocr_oem: int = 3  # OCR engine mode

    # Redis
    redis_host: str = "localhost"
    redis_port: int = 6379
    redis_db: int = 0
    redis_password: str = ""

    # Database
    database_url: str
    pool_size: int = 20
    max_overflow: int = 10

    # Performance
    max_upload_size: int = 10 * 1024 * 1024  # 10MB
    preprocessing_timeout: int = 30
    ocr_timeout: int = 120

    # Monitoring
    prometheus_port: int = 9090
    log_level: str = "INFO"

    class Config:
        env_file = ".env"
        case_sensitive = False

@lru_cache()
def get_settings() -> Settings:
    return Settings()

# app/models.py
from pydantic import BaseModel, Field, validator
from typing import Optional, List, Dict
from datetime import datetime
from enum import Enum

class OCRLanguage(str, Enum):
    ENGLISH = "eng"
    SPANISH = "spa"
    FRENCH = "fra"
    GERMAN = "deu"
    CHINESE_SIMPLIFIED = "chi_sim"
    CHINESE_TRADITIONAL = "chi_tra"

class ProcessingStatus(str, Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"

class OCRRequest(BaseModel):
    language: OCRLanguage = OCRLanguage.ENGLISH
    preprocess: bool = True
    deskew: bool = True
    remove_noise: bool = True
    enhance_contrast: bool = True
    psm: int = Field(default=3, ge=0, le=13)

    @validator('psm')
    def validate_psm(cls, v):
        if v not in range(0, 14):
            raise ValueError('PSM must be between 0 and 13')
        return v

class OCRResponse(BaseModel):
    job_id: str
    status: ProcessingStatus
    text: Optional[str] = None
    confidence: Optional[float] = None
    processing_time: Optional[float] = None
    word_count: Optional[int] = None
    metadata: Dict = {}
    error: Optional[str] = None
    created_at: datetime
    completed_at: Optional[datetime] = None

# app/services/preprocessing.py
import cv2
import numpy as np
from typing import Tuple
import logging

logger = logging.getLogger(__name__)

class ImagePreprocessor:
    """Handles [image preprocessing](/articles/preprocessing-techniques) for improved [OCR accuracy](/articles/character-recognition-accuracy)."""

    @staticmethod
    def resize_image(image: np.ndarray, max_width: int = 2000) -> np.ndarray:
        """Resize image maintaining aspect ratio."""
        height, width = image.shape[:2]
        if width > max_width:
            ratio = max_width / width
            new_height = int(height * ratio)
            image = cv2.resize(image, (max_width, new_height),
                             interpolation=cv2.INTER_AREA)
        return image

    @staticmethod
    def deskew(image: np.ndarray) -> Tuple[np.ndarray, float]:
        """Detect and correct skew in document images."""
        # Convert to grayscale if needed
        if len(image.shape) == 3:
            gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
        else:
            gray = image

        # Detect edges
        edges = cv2.Canny(gray, 50, 150, apertureSize=3)

        # Detect lines using Hough transform
        lines = cv2.HoughLines(edges, 1, np.pi / 180, 200)

        if lines is None:
            return image, 0.0

        # Calculate dominant angle
        angles = []
        for rho, theta in lines[:, 0]:
            angle = np.rad2deg(theta) - 90
            angles.append(angle)

        median_angle = np.median(angles)

        # Rotate image
        if abs(median_angle) > 0.5:
            h, w = image.shape[:2]
            center = (w // 2, h // 2)
            rotation_matrix = cv2.getRotationMatrix2D(center, median_angle, 1.0)
            rotated = cv2.warpAffine(image, rotation_matrix, (w, h),
                                    flags=cv2.INTER_CUBIC,
                                    borderMode=cv2.BORDER_REPLICATE)
            logger.info(f"Deskewed image by {median_angle:.2f} degrees")
            return rotated, median_angle

        return image, 0.0

    @staticmethod
    def remove_noise(image: np.ndarray) -> np.ndarray:
        """Remove noise using morphological operations."""
        # Convert to grayscale
        if len(image.shape) == 3:
            gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
        else:
            gray = image

        # Apply morphological opening to remove noise
        kernel = np.ones((1, 1), np.uint8)
        opening = cv2.morphologyEx(gray, cv2.MORPH_OPEN, kernel, iterations=1)

        # Apply Gaussian blur
        denoised = cv2.GaussianBlur(opening, (3, 3), 0)

        return denoised

    @staticmethod
    def enhance_contrast(image: np.ndarray) -> np.ndarray:
        """Enhance image contrast using CLAHE."""
        if len(image.shape) == 3:
            gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
        else:
            gray = image

        # Apply CLAHE (Contrast Limited Adaptive Histogram Equalization)
        clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
        enhanced = clahe.apply(gray)

        return enhanced

    @staticmethod
    def binarize(image: np.ndarray) -> np.ndarray:
        """Apply [adaptive thresholding for binarization](/articles/image-binarization-methods)."""
        if len(image.shape) == 3:
            gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
        else:
            gray = image

        # Otsu's thresholding
        _, binary = cv2.threshold(gray, 0, 255,
                                 cv2.THRESH_BINARY + cv2.THRESH_OTSU)

        return binary

    def preprocess(self, image: np.ndarray,
                   deskew: bool = True,
                   remove_noise: bool = True,
                   enhance_contrast: bool = True) -> np.ndarray:
        """Apply full preprocessing pipeline."""
        processed = image.copy()

        # Resize if too large
        processed = self.resize_image(processed)

        # Deskew if requested
        if deskew:
            processed, angle = self.deskew(processed)

        # Remove noise
        if remove_noise:
            processed = self.remove_noise(processed)

        # Enhance contrast
        if enhance_contrast:
            processed = self.enhance_contrast(processed)

        # Binarize
        processed = self.binarize(processed)

        return processed

# app/services/ocr.py
import pytesseract
import cv2
import numpy as np
from PIL import Image
from typing import Dict, Optional
import asyncio
import time
import logging
from concurrent.futures import ThreadPoolExecutor

from app.config import get_settings
from app.services.preprocessing import ImagePreprocessor
from app.models import OCRRequest

logger = logging.getLogger(__name__)
settings = get_settings()

class OCRService:
    """Production-grade OCR service with preprocessing and error handling."""

    def __init__(self):
        self.preprocessor = ImagePreprocessor()
        self.executor = ThreadPoolExecutor(max_workers=settings.workers)
        pytesseract.pytesseract.tesseract_cmd = settings.tesseract_path

    async def process_image(self,
                          image_bytes: bytes,
                          config: OCRRequest) -> Dict:
        """Process image bytes and extract text."""
        start_time = time.time()

        try:
            # Convert bytes to numpy array
            nparr = np.frombuffer(image_bytes, np.uint8)
            image = cv2.imdecode(nparr, cv2.IMREAD_COLOR)

            if image is None:
                raise ValueError("Failed to decode image")

            # Preprocess if requested
            if config.preprocess:
                image = self.preprocessor.preprocess(
                    image,
                    deskew=config.deskew,
                    remove_noise=config.remove_noise,
                    enhance_contrast=config.enhance_contrast
                )

            # Convert to PIL Image for Tesseract
            pil_image = Image.fromarray(image)

            # Run OCR in thread pool
            loop = asyncio.get_event_loop()
            result = await loop.run_in_executor(
                self.executor,
                self._run_tesseract,
                pil_image,
                config
            )

            processing_time = time.time() - start_time

            return {
                "text": result["text"],
                "confidence": result["confidence"],
                "processing_time": processing_time,
                "word_count": len(result["text"].split()),
                "metadata": {
                    "language": config.language.value,
                    "psm": config.psm,
                    "preprocessed": config.preprocess
                }
            }

        except Exception as e:
            logger.error(f"OCR processing failed: {str(e)}", exc_info=True)
            raise

    def _run_tesseract(self, image: Image, config: OCRRequest) -> Dict:
        """Run Tesseract OCR (blocking operation)."""
        custom_config = f"--psm {config.psm} --oem {settings.ocr_oem}"

        # Extract text
        text = pytesseract.image_to_string(
            image,
            lang=config.language.value,
            config=custom_config
        )

        # Get detailed data for confidence
        data = pytesseract.image_to_data(
            image,
            lang=config.language.value,
            config=custom_config,
            output_type=pytesseract.Output.DICT
        )

        # Calculate average confidence
        confidences = [int(conf) for conf in data['conf'] if int(conf) > 0]
        avg_confidence = sum(confidences) / len(confidences) if confidences else 0.0

        return {
            "text": text.strip(),
            "confidence": avg_confidence
        }

    async def close(self):
        """Cleanup resources."""
        self.executor.shutdown(wait=True)

# app/main.py
from fastapi import FastAPI, File, UploadFile, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, Response
from prometheus_client import Counter, Histogram, generate_latest
from prometheus_client import CONTENT_TYPE_LATEST
import structlog
import uuid
from contextlib import asynccontextmanager

from app.config import get_settings
from app.models import OCRRequest, OCRResponse, ProcessingStatus
from app.services.ocr import OCRService

settings = get_settings()
logger = structlog.get_logger()

# Prometheus metrics
ocr_requests_total = Counter(
    'ocr_requests_total',
    'Total number of OCR requests',
    ['status', 'language']
)
ocr_processing_time = Histogram(
    'ocr_processing_seconds',
    'Time spent processing OCR requests',
    ['language']
)

# Global OCR service instance
ocr_service: Optional[OCRService] = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Lifespan context manager for startup/shutdown."""
    global ocr_service
    logger.info("Starting OCR service")
    ocr_service = OCRService()
    yield
    logger.info("Shutting down OCR service")
    if ocr_service:
        await ocr_service.close()

app = FastAPI(
    title=settings.app_name,
    version="1.0.0",
    lifespan=lifespan
)

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/health")
async def health_check():
    """Health check endpoint."""
    return {"status": "healthy", "service": settings.app_name}

@app.get("/metrics")
async def metrics():
    """Prometheus metrics endpoint."""
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

@app.post("/ocr", response_model=OCRResponse)
async def process_document(
    file: UploadFile = File(...),
    language: str = "eng",
    preprocess: bool = True,
    deskew: bool = True,
    remove_noise: bool = True,
    enhance_contrast: bool = True,
    psm: int = 3
):
    """Process document and extract text via OCR."""
    job_id = str(uuid.uuid4())

    logger.info("ocr_request_received", job_id=job_id, filename=file.filename)

    try:
        # Validate file size
        contents = await file.read()
        if len(contents) > settings.max_upload_size:
            raise HTTPException(
                status_code=413,
                detail=f"File too large. Max size: {settings.max_upload_size} bytes"
            )

        # Create OCR config
        config = OCRRequest(
            language=language,
            preprocess=preprocess,
            deskew=deskew,
            remove_noise=remove_noise,
            enhance_contrast=enhance_contrast,
            psm=psm
        )

        # Process with timeout
        with ocr_processing_time.labels(language=language).time():
            result = await asyncio.wait_for(
                ocr_service.process_image(contents, config),
                timeout=settings.ocr_timeout
            )

        ocr_requests_total.labels(status="success", language=language).inc()

        response = OCRResponse(
            job_id=job_id,
            status=ProcessingStatus.COMPLETED,
            text=result["text"],
            confidence=result["confidence"],
            processing_time=result["processing_time"],
            word_count=result["word_count"],
            metadata=result["metadata"],
            created_at=datetime.utcnow(),
            completed_at=datetime.utcnow()
        )

        logger.info("ocr_request_completed", job_id=job_id,
                   word_count=result["word_count"],
                   confidence=result["confidence"])

        return response

    except asyncio.TimeoutError:
        ocr_requests_total.labels(status="timeout", language=language).inc()
        logger.error("ocr_timeout", job_id=job_id)
        raise HTTPException(status_code=504, detail="OCR processing timeout")

    except Exception as e:
        ocr_requests_total.labels(status="error", language=language).inc()
        logger.error("ocr_error", job_id=job_id, error=str(e), exc_info=True)
        raise HTTPException(status_code=500, detail=f"OCR processing failed: {str(e)}")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "app.main:app",
        host="0.0.0.0",
        port=8000,
        workers=settings.workers,
        log_level=settings.log_level.lower()
    )

# Dockerfile
FROM python:3.11-slim as base

# Install system dependencies
RUN apt-get update && apt-get install -y \
    tesseract-ocr \
    tesseract-ocr-eng \
    tesseract-ocr-spa \
    tesseract-ocr-fra \
    tesseract-ocr-deu \
    libsm6 \
    libxext6 \
    libxrender-dev \
    libgomp1 \
    && rm -rf /var/lib/apt/lists/*

WORKDIR /app

# Copy requirements
COPY requirements.txt .

# Install Python dependencies
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY app/ ./app/

# Create non-root user
RUN useradd -m -u 1000 ocruser && chown -R ocruser:ocruser /app
USER ocruser

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'

services:
  ocr-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://postgres:postgres@db:5432/ocr
      - REDIS_HOST=redis
      - REDIS_PORT=6379
      - LOG_LEVEL=INFO
    depends_on:
      - db
      - redis
    volumes:
      - ./app:/app/app
    restart: unless-stopped

  db:
    image: postgres:16-alpine
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=ocr
    volumes:
      - postgres_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus_data:/prometheus
    ports:
      - "9090:9090"
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'

volumes:
  postgres_data:
  redis_data:
  prometheus_data:

gunicorn app.main:app \
    --workers 4 \
    --worker-class uvicorn.workers.UvicornWorker \
    --bind 0.0.0.0:8000 \
    --timeout 120 \
    --access-logfile - \
    --error-logfile - \
    --log-level info

# prometheus.yml
global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'ocr-service'
    static_configs:
      - targets: ['ocr-api:8000']

curl -X POST http://localhost:8000/ocr \
  -F "[email protected]" \
  -F "language=eng" \
  -F "preprocess=true" \
  -F "psm=3"

import requests

with open("document.jpg", "rb") as f:
    response = requests.post(
        "http://localhost:8000/ocr",
        files={"file": f},
        data={
            "language": "eng",
            "preprocess": True,
            "deskew": True,
            "psm": 3
        }
    )

result = response.json()
print(f"Text: {result['text']}")
print(f"Confidence: {result['confidence']}%")