"""Adaptateur OCR — Azure Document Intelligence (anciennement Form Recognizer). Utilise l'API Azure Document Intelligence pour la reconnaissance de texte dans des documents historiques. Variables d'environnement requises : - ``AZURE_DOC_INTEL_KEY`` : clé API Azure - ``AZURE_DOC_INTEL_ENDPOINT`` : URL de l'endpoint (ex : https://moninstance.cognitiveservices.azure.com/) Documentation : https://learn.microsoft.com/azure/ai-services/document-intelligence/ Sprint 51 — exposition des token_confidences --------------------------------------------- La réponse Azure expose ``analyzeResult.pages[].words[]`` avec ``content`` et ``confidence`` (∈ [0, 1]). L'adapter parcourt cette hiérarchie et émet une entrée par mot au format Sprint 42. Le texte ``EngineResult.text`` est extrait depuis ``pages[].lines[]`` (préservation rétrocompat octet par octet). Les deux chemins (SDK et REST) sont normalisés vers une représentation dict unifiée. Refactor du chantier 1 (post-Sprint 97) --------------------------------------- L'adapter ne surcharge plus ``run()`` — il implémente ``_run_with_native`` et ``_extract_raw_confidences`` (les hooks factorisés dans ``BaseOCREngine``). Comportement externe et octets de sortie strictement identiques. """ from __future__ import annotations import json import logging import os import time import urllib.error import urllib.request from pathlib import Path from typing import Any, Optional from picarones.engines.base import BaseOCREngine logger = logging.getLogger(__name__) class AzureDocIntelEngine(BaseOCREngine): """Moteur OCR via Azure Document Intelligence. Configuration ------------- model_id : str Modèle Azure à utiliser. Défaut : ``"prebuilt-read"`` (lecture générique). Alternatives : ``"prebuilt-document"``, ``"prebuilt-layout"`` ou un modèle entraîné personnalisé. locale : str Paramètre de locale pour améliorer la précision (ex : ``"fr-FR"``). api_version : str Version de l'API Azure (défaut : ``"2024-02-29-preview"``). expose_confidences : bool ``True`` (défaut) : extrait ``Word.confidence`` de la réponse Azure (Sprint 51). """ @property def name(self) -> str: return "azure_doc_intel" def version(self) -> str: return self.config.get("api_version", "2024-02-29-preview") def __init__(self, config: Optional[dict] = None) -> None: super().__init__(config) self._api_key = os.environ.get("AZURE_DOC_INTEL_KEY") self._endpoint = ( os.environ.get("AZURE_DOC_INTEL_ENDPOINT", "").rstrip("/") or self.config.get("endpoint", "").rstrip("/") ) self._model_id: str = self.config.get("model_id", "prebuilt-read") self._locale: str = self.config.get("locale", "fr-FR") self._api_version: str = self.config.get("api_version", "2024-02-29-preview") def _run_ocr(self, image_path: Path) -> str: """Retourne uniquement le texte (interface ``BaseOCREngine``).""" text, _result = self._run_with_native(image_path) return text def _run_with_native( self, image_path: Path, ) -> tuple[str, Optional[dict]]: """Exécute l'OCR et retourne ``(text, analyze_result_dict)``. ``analyze_result_dict`` est la sous-structure ``analyzeResult`` (avec ``pages[].words[]`` portant les confidences) — normalisée entre les chemins SDK et REST. """ if not self._api_key: raise RuntimeError( "Clé API Azure manquante — définissez la variable d'environnement AZURE_DOC_INTEL_KEY" ) if not self._endpoint: raise RuntimeError( "Endpoint Azure manquant — définissez la variable d'environnement AZURE_DOC_INTEL_ENDPOINT" ) try: return self._run_via_sdk(image_path) except ImportError: return self._run_via_rest(image_path) def _run_via_sdk(self, image_path: Path) -> tuple[str, dict]: from azure.ai.documentintelligence import DocumentIntelligenceClient from azure.core.credentials import AzureKeyCredential client = DocumentIntelligenceClient( endpoint=self._endpoint, credential=AzureKeyCredential(self._api_key), ) with open(image_path, "rb") as f: poller = client.begin_analyze_document( model_id=self._model_id, body=f, locale=self._locale, content_type="application/octet-stream", ) result = poller.result() text = "\n".join( line.content for page in result.pages for line in (page.lines or []) ) analyze_result = self._sdk_result_to_dict(result) return text, analyze_result def _run_via_rest(self, image_path: Path) -> tuple[str, Optional[dict]]: """Appel REST direct (sans SDK Azure).""" image_bytes = image_path.read_bytes() analyze_url = ( f"{self._endpoint}/documentintelligence/documentModels/" f"{self._model_id}:analyze" f"?api-version={self._api_version}&locale={self._locale}" ) # Soumettre l'image req = urllib.request.Request( analyze_url, data=image_bytes, headers={ "Ocp-Apim-Subscription-Key": self._api_key, "Content-Type": "application/octet-stream", }, ) try: with urllib.request.urlopen(req, timeout=60) as resp: operation_url = resp.headers.get("Operation-Location", "") except urllib.error.HTTPError as exc: raise RuntimeError( f"Azure Document Intelligence erreur {exc.code}: {exc.read().decode()}" ) from exc if not operation_url: raise RuntimeError("Azure : pas d'Operation-Location dans la réponse") # Polling du résultat (Azure est asynchrone) headers = {"Ocp-Apim-Subscription-Key": self._api_key} for attempt in range(30): time.sleep(1 + attempt * 0.5) poll_req = urllib.request.Request(operation_url, headers=headers) with urllib.request.urlopen(poll_req, timeout=30) as resp: result = json.loads(resp.read().decode("utf-8")) status = result.get("status", "") if status == "succeeded": text = self._extract_text_from_result(result) analyze_result = result.get("analyzeResult") or None return text, analyze_result if status in {"failed", "canceled"}: raise RuntimeError(f"Azure Document Intelligence : analyse {status}") # status == "running" → continuer à attendre raise RuntimeError("Azure Document Intelligence : timeout — analyse trop longue") @staticmethod def _extract_text_from_result(result: dict) -> str: """Extrait le texte brut depuis la réponse JSON Azure.""" pages = result.get("analyzeResult", {}).get("pages", []) lines: list[str] = [] for page in pages: for line in page.get("lines", []): content = line.get("content", "") if content: lines.append(content) return "\n".join(lines) # ────────────────────────────────────────────────────────────────── # Conversion SDK → dict normalisé # ────────────────────────────────────────────────────────────────── @staticmethod def _sdk_result_to_dict(result: Any) -> dict: """Convertit l'objet SDK en dict ``{"pages": [{"words": [{"content", "confidence"}]}]}`` pour traitement uniforme avec le chemin REST.""" pages = [] for page in getattr(result, "pages", []) or []: words = [] for word in getattr(page, "words", []) or []: content = getattr(word, "content", "") or "" conf = getattr(word, "confidence", None) words.append({ "content": content, "confidence": float(conf) if conf is not None else None, }) pages.append({"words": words}) return {"pages": pages} # ────────────────────────────────────────────────────────────────── # Extraction des token_confidences au format Sprint 42 # ────────────────────────────────────────────────────────────────── def _extract_raw_confidences( self, native: Any, ) -> Optional[list[dict[str, Any]]]: """Parcourt ``pages[].words[]`` et émet ``{"token": str, "confidence": float}`` par mot. Filtrage cohérent avec les autres adapters : confidence None / négative ignorée, contenu vide ignoré (filtrage final assuré par ``BaseOCREngine._normalize_token_confidences``). """ if not self.config.get("expose_confidences", True): return None if not native or not isinstance(native, dict): return None out: list[dict[str, Any]] = [] for page in native.get("pages") or []: if not isinstance(page, dict): continue for word in page.get("words") or []: if not isinstance(word, dict): continue content = (word.get("content") or "").strip() conf = word.get("confidence") if not content or conf is None: continue out.append({"token": content, "confidence": conf}) return out or None