"""Extração de texto de PDFs com fallback para OCR. A camada textual de muitos PDFs de Bíblia usa fontes sem mapeamento Unicode: o extrator devolve sequências de caracteres de controle em vez de letras. Detectamos esse caso e rasterizamos a página para o Tesseract. O resultado do OCR é cacheado por (sha256 do arquivo, página) para que reindexações não repitam horas de OCR. """ from __future__ import annotations import hashlib import logging import shutil import subprocess import tempfile import unicodedata from dataclasses import dataclass from pathlib import Path from typing import Iterator logger = logging.getLogger(__name__) try: # pragma: no cover - dependência real do ambiente import pymupdf except ImportError: # pragma: no cover - permite rodar testes sem MuPDF pymupdf = None # type: ignore[assignment] class OcrError(RuntimeError): """Falha ao executar o Tesseract.""" @dataclass(frozen=True) class PageText: """Texto extraído de uma página do PDF.""" page: int text: str failed: bool = False def legible_ratio(text: str) -> float: """Proporção de caracteres imprimíveis legíveis no texto. Um texto normal de Bíblia fica acima de 0.9; a camada corrompida do PDF real fica abaixo de 0.05. Espaços em branco não contam como denominador. """ if not text: return 0.0 meaningful = [ch for ch in text if not ch.isspace()] if not meaningful: return 0.0 legible = sum(1 for ch in meaningful if _is_legible_char(ch)) return legible / len(meaningful) def _is_legible_char(ch: str) -> bool: """Caractere que um humano consegue ler.""" if unicodedata.category(ch).startswith("C"): # C0/C1 de controle: sinal de fonte sem ToUnicode. return False return ch.isalnum() or unicodedata.category(ch).startswith(("L", "N", "P", "S", "Z")) def _is_blank_pixmap(pixmap) -> bool: """Diz se a página rasterizada está visualmente em branco. Usa as amostras do pixmap: uma página de texto tem variância alta entre pixels; uma página em branco tem todos os pixels iguais. """ try: samples = pixmap.samples stride = max(1, pixmap.n if getattr(pixmap, "n", 1) else 1) total = pixmap.width * pixmap.height if total == 0: return True step = max(1, (total * stride) // 4000) values = bytes(samples[0 : len(samples) : step]) if not values: return True return len(set(values)) <= 1 except Exception: # pragma: no cover - formatos de pixmap inesperados return False def file_sha256(path: Path, chunk_size: int = 1024 * 1024) -> str: """SHA-256 do conteúdo do arquivo.""" digest = hashlib.sha256() with path.open("rb") as handle: while chunk := handle.read(chunk_size): digest.update(chunk) return digest.hexdigest() class PdfExtractor: """Extrai texto de PDFs página a página, com OCR como fallback.""" def __init__( self, cache_dir: Path, dpi: int = 200, ocr_lang: str = "por+eng", ocr_timeout: int = 120, min_legible_ratio: float = 0.35, pymupdf_module=None, ): self.cache_dir = Path(cache_dir) self.dpi = dpi self.ocr_lang = ocr_lang self.ocr_timeout = ocr_timeout self.min_legible_ratio = min_legible_ratio self._pymupdf = pymupdf_module if pymupdf_module is not None else pymupdf def extract_document( self, path: Path, file_hash: str | None = None ) -> Iterator[PageText]: """Extrai o texto de todas as páginas do PDF.""" if self._pymupdf is None: # pragma: no cover raise OcrError("PyMuPDF não está disponível") path = Path(path) file_hash = file_hash or file_sha256(path) document = self._pymupdf.open(path) try: for index in range(len(document)): page = document[index] yield self.extract_page(page, file_hash, index + 1) finally: document.close() def extract_page(self, page, file_hash: str, page_number: int) -> PageText: """Extrai o texto de uma página, usando cache quando disponível.""" cache_file = self._cache_path(file_hash, page_number) if cache_file.exists(): return PageText(page=page_number, text=cache_file.read_text(encoding="utf-8")) try: text = page.get_text("text") or "" except Exception as exc: # pragma: no cover - PDF malformado logger.warning("Falha ao extrair texto da página %s: %s", page_number, exc) text = "" failed = False if legible_ratio(text) < self.min_legible_ratio: text, failed = self._ocr_page(page) self._write_cache(cache_file, text) return PageText(page=page_number, text=text, failed=failed) def _ocr_page(self, page) -> tuple[str, bool]: """Rasteriza a página e aplica OCR. Devolve (texto, falhou). Uma página visualmente em branco não é falha: o documento pode ter separadores. Ela é detectada pela rasterização e não gasta OCR. Falha é o OCR que não pôde ser executado. """ try: pixmap = page.get_pixmap(dpi=self.dpi) except Exception as exc: logger.warning("Falha ao rasterizar página para OCR: %s", exc) return "", True if _is_blank_pixmap(pixmap): return "", False try: text = self._run_tesseract(pixmap, self.ocr_lang, self.ocr_timeout) except OcrError as exc: logger.warning("OCR falhou: %s", exc) return "", True return text, False def _run_tesseract(self, pixmap, language: str, timeout: int) -> str: """Executa o Tesseract sobre o pixmap e devolve o texto.""" if shutil.which("tesseract") is None: raise OcrError("tesseract não está instalado no sistema") with tempfile.TemporaryDirectory() as tmp: image_path = Path(tmp) / "page.png" pixmap.save(str(image_path)) try: completed = subprocess.run( [ "tesseract", str(image_path), "stdout", "-l", language, "--psm", "6", ], capture_output=True, text=True, timeout=timeout, check=False, ) except subprocess.TimeoutExpired as exc: raise OcrError(f"tesseract excedeu o timeout de {timeout}s") from exc if completed.returncode != 0: raise OcrError(f"tesseract falhou: {completed.stderr.strip()[:200]}") return completed.stdout def _cache_path(self, file_hash: str, page_number: int) -> Path: return self.cache_dir / file_hash / f"{page_number:05d}.txt" def _write_cache(self, cache_file: Path, text: str) -> None: try: cache_file.parent.mkdir(parents=True, exist_ok=True) cache_file.write_text(text, encoding="utf-8") except OSError as exc: # pragma: no cover - disco cheio / permissões logger.warning("Não foi possível gravar cache de OCR em %s: %s", cache_file, exc)