"""Interface abstraite commune à tous les adaptateurs moteurs OCR.""" from __future__ import annotations import hashlib import time from abc import ABC, abstractmethod from dataclasses import dataclass, field from pathlib import Path from typing import Optional @dataclass class EngineResult: """Résultat brut produit par un moteur OCR sur une image.""" engine_name: str image_path: str text: str duration_seconds: float error: Optional[str] = None metadata: dict = field(default_factory=dict) @property def success(self) -> bool: return self.error is None @property def image_sha256(self) -> str: return hashlib.sha256(Path(self.image_path).read_bytes()).hexdigest() class BaseOCREngine(ABC): """Classe de base dont héritent tous les adaptateurs OCR. Chaque adaptateur doit implémenter : - ``name`` : identifiant unique du moteur - ``version()`` : retourne la version du moteur sous forme de chaîne - ``_run_ocr(image_path)`` : logique d'exécution OCR, retourne le texte brut Attribut de classe ------------------ execution_mode : ``"io"`` (défaut) ou ``"cpu"`` Indique au runner quel type d'exécuteur utiliser : - ``"io"`` → ``ThreadPoolExecutor`` (moteurs API / réseau) - ``"cpu"`` → ``ProcessPoolExecutor`` (moteurs CPU-intensifs : Tesseract, Pero, Kraken) """ execution_mode: str = "io" """``"io"`` pour ThreadPoolExecutor (défaut), ``"cpu"`` pour ProcessPoolExecutor.""" def __init__(self, config: Optional[dict] = None) -> None: self.config: dict = config or {} @property @abstractmethod def name(self) -> str: """Identifiant unique et stable du moteur.""" @abstractmethod def version(self) -> str: """Retourne la version du moteur (ex : '5.3.0').""" @abstractmethod def _run_ocr(self, image_path: Path) -> str: """Exécute l'OCR et retourne le texte brut extrait.""" def run(self, image_path: str | Path) -> EngineResult: """Point d'entrée public : exécute l'OCR et mesure le temps d'exécution.""" image_path = Path(image_path) start = time.perf_counter() try: text = self._run_ocr(image_path) error = None except Exception as exc: # noqa: BLE001 text = "" error = str(exc) duration = time.perf_counter() - start return EngineResult( engine_name=self.name, image_path=str(image_path), text=text, duration_seconds=round(duration, 4), error=error, metadata={"engine_version": self._safe_version()}, ) def _safe_version(self) -> str: try: return self.version() except Exception: # noqa: BLE001 return "unknown" def __repr__(self) -> str: return f"{self.__class__.__name__}(name={self.name!r})"