Effi/effi-ocr-server/main.py

"""
Effi-OCR : Service OCR pour Effi
API FastAPI qui expose Tesseract via une interface REST
"""

from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import JSONResponse
from PIL import Image
import pytesseract
import io
import logging
from typing import Optional

# Configuration logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(
    title="Effi-OCR",
    description="Service OCR pour extraction de texte d'images",
    version="1.0.0"
)


@app.get("/health")
async def health_check():
    """Endpoint de santé pour monitoring"""
    try:
        # Vérifier que Tesseract est accessible
        import subprocess
        result = subprocess.run(['tesseract', '--version'], capture_output=True, text=True)
        version_line = result.stdout.split('\n')[0] if result.stdout else "unknown"
        tesseract_version = str(version_line)  # Forcer en string
    except Exception as e:
        tesseract_version = f"error: {str(e)}"
    
    return {
        "status": "healthy",
        "service": "effi-ocr",
        "version": "1.0.0",
        "tesseract_version": tesseract_version
    }


@app.post("/ocr")
async def ocr_image(
    file: UploadFile = File(...),
    lang: str = "fra",
    preprocess: bool = True
):
    """
    Extraire le texte d'une image
    
    Args:
        file: Image à analyser (jpg, png, etc.)
        lang: Code langue (fra, eng, fra+eng, etc.)
        preprocess: Appliquer un prétraitement pour améliorer la qualité
    
    Returns:
        JSON avec le texte extrait et les métadonnées
    """
    logger.info(f"OCR demandé - lang: {lang}, fichier: {file.filename}")
    
    # Déterminer le type MIME (avec fallback depuis l'extension)
    content_type = file.content_type
    if content_type is None:
        import mimetypes
        content_type, _ = mimetypes.guess_type(file.filename)
        content_type = content_type or 'application/octet-stream'
    
    allowed_types = ['image/jpeg', 'image/png', 'image/gif', 'image/bmp', 'image/tiff', 'image/webp']
    if content_type not in allowed_types:
        raise HTTPException(
            status_code=400,
            detail=f"Type de fichier non supporté: {content_type}. Types acceptés: {', '.join(allowed_types)}"
        )
    
    try:
        # Lire l'image
        contents = await file.read()
        image = Image.open(io.BytesIO(contents))
        
        # Convertir en RGB si nécessaire (pour les images avec transparence)
        if image.mode in ('RGBA', 'LA', 'P'):
            image = image.convert('RGB')
        
        # Prétraitement optionnel pour améliorer la qualité OCR
        if preprocess:
            # Convertir en niveaux de gris pour meilleure détection texte
            image = image.convert('L')
            # Augmenter le contraste
            from PIL import ImageEnhance
            enhancer = ImageEnhance.Contrast(image)
            image = enhancer.enhance(2.0)
        
        # OCR avec Tesseract
        text = pytesseract.image_to_string(image, lang=lang)
        
        # Nettoyer le texte
        text = text.strip()
        
        # Informations sur l'image
        width, height = image.size
        
        logger.info(f"OCR terminé - {len(text)} caractères extraits")
        
        return JSONResponse(content={
            "success": True,
            "text": text,
            "language": lang,
            "filename": file.filename,
            "image_size": {"width": width, "height": height},
            "characters_count": len(text),
            "words_count": len(text.split()) if text else 0
        })
        
    except Exception as e:
        logger.error(f"Erreur OCR: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Erreur lors de l'OCR: {str(e)}")


@app.post("/ocr/base64")
async def ocr_base64(
    image_base64: str,
    lang: str = "fra",
    preprocess: bool = True
):
    """
    Extraire le texte d'une image en base64
    
    Args:
        image_base64: Image encodée en base64
        lang: Code langue (fra, eng, fra+eng, etc.)
        preprocess: Appliquer un prétraitement
    
    Returns:
        JSON avec le texte extrait
    """
    import base64
    
    logger.info(f"OCR base64 demandé - lang: {lang}")
    
    try:
        # Décoder le base64
        image_data = base64.b64decode(image_base64)
        image = Image.open(io.BytesIO(image_data))
        
        # Convertir en RGB si nécessaire
        if image.mode in ('RGBA', 'LA', 'P'):
            image = image.convert('RGB')
        
        # Prétraitement
        if preprocess:
            image = image.convert('L')
            from PIL import ImageEnhance
            enhancer = ImageEnhance.Contrast(image)
            image = enhancer.enhance(2.0)
        
        # OCR
        text = pytesseract.image_to_string(image, lang=lang)
        text = text.strip()
        
        width, height = image.size
        
        logger.info(f"OCR base64 terminé - {len(text)} caractères")
        
        return {
            "success": True,
            "text": text,
            "language": lang,
            "image_size": {"width": width, "height": height},
            "characters_count": len(text),
            "words_count": len(text.split()) if text else 0
        }
        
    except Exception as e:
        logger.error(f"Erreur OCR base64: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Erreur lors de l'OCR: {str(e)}")


@app.get("/languages")
async def list_languages():
    """Lister les langues disponibles pour l'OCR"""
    try:
        # Commande tesseract pour lister les langues
        import subprocess
        result = subprocess.run(
            ['tesseract', '--list-langs'],
            capture_output=True,
            text=True
        )
        
        # Parser la sortie
        lines = result.stdout.strip().split('\n')
        # La première ligne est "List of available languages (xxx):"
        languages = lines[1:] if len(lines) > 1 else []
        
        return {
            "available_languages": languages,
            "count": len(languages)
        }
    except Exception as e:
        logger.error(f"Erreur listage langues: {str(e)}")
        # Fallback avec les langues communes
        return {
            "available_languages": ["fra", "eng", "fra+eng", "deu", "spa", "ita"],
            "count": 6,
            "note": "Liste par défaut, erreur lors de la récupération"
        }


if __name__ == "__main__":
    import uvicorn
    import os
    port = int(os.getenv("PORT", "8301"))
    uvicorn.run(app, host="0.0.0.0", port=port)