7. Troubleshooting Documentos

Descripción

En esta cápsula cubrimos los problemas más comunes al procesar documentos: PDFs corruptos, baja calidad de escaneo, errores de API, encoding, tablas mal extraídas y optimización de costos. Aprenderás a diagnosticar, resolver y prevenir fallos en pipelines documentales con funciones de diagnóstico, preprocesamiento y estrategias de fallback.

Por qué importa: En producción, los documentos reales son imperfectos. Saber anticipar y manejar estos casos diferencia un prototipo de un sistema robusto.

Tabla de Referencia Rápida de Errores

Problemas Comunes y Soluciones

1. PDF corrupto o dañado

Síntomas: fitz.open() lanza FileDataError, pdf2image falla.

import fitz
from pathlib import Path

def diagnose_pdf(path: str) -> dict:
    result = {"path": path, "valid": False, "pages": 0, "issues": []}
    if not Path(path).exists():
        result["issues"].append("Archivo no encontrado")
        return result
    try:
        doc = fitz.open(path)
        result["pages"] = len(doc)
        result["valid"] = True
        for i, page in enumerate(doc):
            if len(page.get_text().strip()) < 5:
                result["issues"].append(f"Página {i+1}: sin texto extraíble")
        doc.close()
    except fitz.FileDataError as e:
        result["issues"].append(f"PDF corrupto: {e}")
    return result

Fallback con pypdf (más tolerante con PDFs malformados):

from pypdf import PdfReader

def try_pypdf_fallback(path: str) -> str:
    reader = PdfReader(path)
    return "\n\n".join(p.extract_text() or "" for p in reader.pages)

2. PDF protegido con contraseña

Síntomas: fitz.open() falla con mensaje de encriptación.

def open_pdf_with_password(path: str, passwords: list[str] | None = None):
    passwords = passwords or ["", "1234", "password"]
    for pwd in passwords:
        try:
            doc = fitz.open(path)
            if doc.is_encrypted:
                if doc.authenticate(pwd):
                    return doc
                doc.close()
            else:
                return doc
        except Exception:
            continue
    return None

3. PDF escaneado sin texto (solo imágenes)

Síntomas: page.get_text() devuelve string vacío o menos de 20 caracteres.

import io
from PIL import Image
import pytesseract

def classify_pdf_type(path: str) -> str:
    doc = fitz.open(path)
    digital, scanned = 0, 0
    for page in doc:
        if len(page.get_text().strip()) > 50:
            digital += 1
        elif len(page.get_images()) > 0:
            scanned += 1
    doc.close()
    if scanned == 0: return "digital"
    if digital == 0: return "scanned"
    return "mixed"

def extract_from_scanned_page(doc, page_num: int, dpi: int = 300) -> str:
    mat = fitz.Matrix(dpi / 72, dpi / 72)
    pix = doc[page_num].get_pixmap(matrix=mat)
    img = Image.open(io.BytesIO(pix.tobytes("png")))
    return pytesseract.image_to_string(img, lang="spa")

4. Documentos escaneados con problemas de calidad

Síntomas: OCR devuelve texto con muchos errores, caracteres aleatorios.

a) Imagen borrosa:

from PIL import ImageEnhance, ImageFilter

def fix_blurry_scan(img: Image.Image) -> Image.Image:
    img = img.convert("L")
    img = ImageEnhance.Sharpness(img).enhance(2.5)
    return ImageEnhance.Contrast(img).enhance(1.8)

b) Imagen rotada o inclinada (skew):

import numpy as np

def deskew_image(img: Image.Image) -> Image.Image:
    arr = np.array(img.convert("L"))
    coords = np.column_stack(np.where((arr < 128) > 0))
    if len(coords) < 100:
        return img
    best_angle, best_score = 0, 0
    for angle in np.arange(-10, 10, 0.5):
        rad = np.radians(angle)
        proj = coords[:, 0] * np.cos(rad) + coords[:, 1] * np.sin(rad)
        hist, _ = np.histogram(proj, bins=50)
        if np.var(hist) > best_score:
            best_score = np.var(hist)
            best_angle = angle
    if abs(best_angle) > 0.5:
        return img.rotate(best_angle, fillcolor=255, expand=True)
    return img

5. Fallos en extracción de tablas

Síntomas: Tablas como texto plano, columnas mezcladas, filas perdidas.

import camelot
import re, json

def extract_tables_from_pdf(path: str, page: str = "1") -> list[dict]:
    tables = camelot.read_pdf(path, pages=page, flavor="lattice")
    if not tables:
        tables = camelot.read_pdf(path, pages=page, flavor="stream")
    return [{"accuracy": t.parsing_report.get("accuracy", 0),
             "data": t.df.to_dict(orient="records")} for t in tables]

TABLE_PROMPT = "Extrae la tabla en JSON. Array de objetos (fila=objeto). Encabezados como keys. Vacías: null."

def extract_table_with_vision(image_path: str, client) -> list[dict]:
    import base64
    with open(image_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": [
            {"type": "text", "text": TABLE_PROMPT},
            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
        ]}], temperature=0)
    text = response.choices[0].message.content
    match = re.search(r'\[[\s\S]*\]', text)
    return json.loads(match.group()) if match else []

6. Problemas de encoding

Síntomas: UnicodeDecodeError, caracteres como é en lugar de é.

import chardet

def detect_and_decode(raw_bytes: bytes) -> str:
    detection = chardet.detect(raw_bytes)
    encoding = detection.get("encoding", "utf-8")
    if detection.get("confidence", 0) < 0.5:
        for enc in ["utf-8", "latin-1", "cp1252", "iso-8859-1"]:
            try:
                return raw_bytes.decode(enc)
            except UnicodeDecodeError:
                continue
    return raw_bytes.decode(encoding, errors="replace")

def clean_extracted_text(text: str) -> str:
    text = text.replace("\x00", "").replace("\ufffd", "")
    text = re.sub(r'[\x01-\x08\x0b\x0c\x0e-\x1f]', '', text)
    return re.sub(r'\s+', ' ', text).strip()

7. PDFs muy grandes (gestión de memoria)

Síntomas: MemoryError, proceso se cuelga.

import gc

def process_large_pdf(path: str, process_fn, batch_size: int = 5) -> list:
    doc = fitz.open(path)
    results = []
    for start in range(0, len(doc), batch_size):
        for page_num in range(start, min(start + batch_size, len(doc))):
            results.append(process_fn(doc[page_num], page_num))
        gc.collect()
    doc.close()
    return results

def estimate_cost(path: str, cost_per_page: float = 0.01) -> dict:
    doc = fitz.open(path)
    pages = len(doc)
    doc.close()
    return {"pages": pages,
            "file_mb": round(Path(path).stat().st_size / 1048576, 2),
            "est_cost_usd": round(pages * cost_per_page, 2),
            "recommendation": "batch" if pages > 50 else "direct"}

8. Documentos con múltiples idiomas

Síntomas: Tesseract devuelve caracteres incorrectos en documentos bilingües.

from langdetect import detect, DetectorFactory
DetectorFactory.seed = 0

LANG_MAP = {"es": "spa", "en": "eng", "fr": "fra", "de": "deu", "pt": "por"}

def ocr_multilang(img: Image.Image, text_sample: str = "") -> str:
    lang = "es"
    if len(text_sample.strip()) > 20:
        try: lang = detect(text_sample)
        except Exception: pass
    tess_lang = LANG_MAP.get(lang, "eng")
    try:
        return pytesseract.image_to_string(img, lang=tess_lang)
    except pytesseract.TesseractError:
        return pytesseract.image_to_string(img, lang="eng")

9. Rate limits en procesamiento batch

Síntomas: 429 Too Many Requests al procesar muchos documentos.

import time
from dataclasses import dataclass, field

@dataclass
class RateLimiter:
    max_requests: int = 50
    window_seconds: float = 60.0
    _timestamps: list = field(default_factory=list)

    def wait_if_needed(self):
        now = time.time()
        self._timestamps = [t for t in self._timestamps if now - t < self.window_seconds]
        if len(self._timestamps) >= self.max_requests:
            sleep_time = self._timestamps[0] + self.window_seconds - now
            if sleep_time > 0:
                time.sleep(sleep_time)
        self._timestamps.append(time.time())

def call_with_retry(func, *args, max_retries: int = 5, **kwargs):
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            if any(k in str(e).lower() for k in ["rate", "429", "timeout"]):
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)
                    continue
            raise

10. JSON inválido en extracción estructurada

Síntomas: json.JSONDecodeError, Pydantic ValidationError.

def extract_json_from_response(text: str) -> dict | list:
    text = text.strip()
    code_block = re.search(r'```(?:json)?\s*([\s\S]*?)```', text)
    if code_block:
        text = code_block.group(1).strip()
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass
    for pattern in [r'\{[\s\S]*\}', r'\[[\s\S]*\]']:
        match = re.search(pattern, text)
        if match:
            try: return json.loads(match.group())
            except json.JSONDecodeError: continue
    raise ValueError(f"No se encontró JSON válido: {text[:200]}")

Función de Diagnóstico

Analiza un documento antes de procesarlo, detectando problemas y sugiriendo la mejor estrategia.

def full_document_diagnostic(path: str) -> dict:
    report = {"path": path, "exists": False, "file_type": None, "pages": 0,
              "pdf_type": None, "issues": [], "recommended_strategy": None}
    p = Path(path)
    if not p.exists():
        report["issues"].append("Archivo no encontrado")
        return report
    report["exists"] = True
    report["file_type"] = p.suffix.lower()

    if report["file_type"] in [".png", ".jpg", ".jpeg", ".tiff", ".bmp"]:
        report["pdf_type"] = "image"
        report["pages"] = 1
        report["recommended_strategy"] = "ocr_or_vision"
        return report
    if report["file_type"] != ".pdf":
        report["issues"].append(f"Formato no soportado: {report['file_type']}")
        return report

    try:
        doc = fitz.open(path)
    except Exception as e:
        report["issues"].append(f"No se puede abrir: {e}")
        report["recommended_strategy"] = "repair_first"
        return report

    if doc.is_encrypted:
        report["issues"].append("PDF encriptado")
        report["recommended_strategy"] = "decrypt_first"
        doc.close()
        return report

    report["pages"] = len(doc)
    digital, scanned = 0, 0
    for i in range(min(len(doc), 10)):
        text = doc[i].get_text().strip()
        if len(text) > 50: digital += 1
        elif len(doc[i].get_images()) > 0: scanned += 1
    doc.close()

    if scanned > digital: report["pdf_type"], report["recommended_strategy"] = "scanned", "ocr_pipeline"
    elif digital > scanned: report["pdf_type"], report["recommended_strategy"] = "digital", "text_extraction"
    else: report["pdf_type"], report["recommended_strategy"] = "mixed", "hybrid_pipeline"
    if report["pages"] > 50: report["issues"].append("Documento largo: usar chunking")
    return report

Preprocessing Pipeline

Pipeline configurable de preprocesamiento de imágenes para mejorar calidad de OCR.

from PIL import Image, ImageEnhance, ImageFilter

class DocumentPreprocessor:
    def __init__(self, denoise=True, deskew=True, binarize=True):
        self.denoise = denoise
        self.deskew = deskew
        self.binarize = binarize
        self.steps_applied = []

    def process(self, img: Image.Image) -> Image.Image:
        self.steps_applied = []
        img = img.convert("L"); self.steps_applied.append("grayscale")
        if self.denoise:
            img = img.filter(ImageFilter.MedianFilter(size=3)); self.steps_applied.append("denoise")
        if self.deskew:
            img = self._deskew(img)
        img = ImageEnhance.Contrast(img).enhance(2.0); self.steps_applied.append("contrast")
        if self.binarize:
            img = img.point(lambda x: 0 if x < 140 else 255); self.steps_applied.append("binarize")
        return img

    def _deskew(self, img: Image.Image) -> Image.Image:
        arr = np.array(img)
        coords = np.column_stack(np.where((arr < 128) > 0))
        if len(coords) < 100: return img
        best_angle, best_score = 0, 0
        for a in np.arange(-10, 10, 0.5):
            rad = np.radians(a)
            proj = coords[:, 0] * np.cos(rad) + coords[:, 1] * np.sin(rad)
            h, _ = np.histogram(proj, bins=50)
            if np.var(h) > best_score: best_score, best_angle = np.var(h), a
        if abs(best_angle) > 0.5:
            self.steps_applied.append(f"deskew({best_angle:.1f}°)")
            return img.rotate(best_angle, fillcolor=255, expand=True)
        return img

Estrategia de Fallback para Documentos

Cascada que intenta la opción más barata primero y escala solo si falla: texto directo → OCR → Vision API → log y skip.

import logging

logger = logging.getLogger("document_fallback")

def fallback_extract(path: str, vision_client=None) -> dict:
    result = {"path": path, "text": "", "method": "failed", "attempts": []}

    # 1. Texto directo (gratis, rápido)
    try:
        doc = fitz.open(path)
        combined = "\n".join(page.get_text() for page in doc).strip()
        doc.close()
        result["attempts"].append({"method": "text", "chars": len(combined)})
        if len(combined) > 50:
            result["text"], result["method"] = combined, "text_extraction"
            return result
    except Exception as e:
        result["attempts"].append({"method": "text", "error": str(e)})

    # 2. OCR con Tesseract + preprocesamiento (gratis, más lento)
    try:
        doc = fitz.open(path)
        preprocessor = DocumentPreprocessor()
        ocr_texts = []
        for i in range(min(len(doc), 20)):
            pix = doc[i].get_pixmap(matrix=fitz.Matrix(2, 2))
            img = preprocessor.process(Image.open(io.BytesIO(pix.tobytes("png"))))
            ocr_texts.append(pytesseract.image_to_string(img, lang="spa"))
        doc.close()
        combined = "\n".join(ocr_texts).strip()
        result["attempts"].append({"method": "ocr", "chars": len(combined)})
        if len(combined) > 50:
            result["text"], result["method"] = combined, "ocr_tesseract"
            return result
    except Exception as e:
        result["attempts"].append({"method": "ocr", "error": str(e)})

    # 3. Vision API (costo por request, más preciso)
    if vision_client:
        try:
            import base64
            doc = fitz.open(path)
            b64 = base64.b64encode(doc[0].get_pixmap(matrix=fitz.Matrix(2,2)).tobytes("png")).decode()
            doc.close()
            resp = vision_client.chat.completions.create(model="gpt-4o", messages=[{"role":"user","content":[
                {"type":"text","text":"Extrae todo el texto de este documento."},
                {"type":"image_url","image_url":{"url":f"data:image/png;base64,{b64}"}}
            ]}], temperature=0)
            text = resp.choices[0].message.content.strip()
            result["attempts"].append({"method": "vision", "chars": len(text)})
            if len(text) > 20:
                result["text"], result["method"] = text, "vision_api"
                return result
        except Exception as e:
            result["attempts"].append({"method": "vision", "error": str(e)})

    logger.error(f"Todos los métodos fallaron: {path}")
    return result

Monitoreo y Logging

Logging estructurado para pipelines documentales en producción.

import json, time, logging
from dataclasses import dataclass, field, asdict
from datetime import datetime

@dataclass
class ProcessingRecord:
    path: str
    method: str = ""
    success: bool = False
    duration: float = 0.0
    error: str = ""
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())

class DocumentPipelineMonitor:
    def __init__(self):
        self.records: list[ProcessingRecord] = []

    def track(self, func, path: str, **kwargs) -> dict:
        rec = ProcessingRecord(path=path)
        start = time.time()
        try:
            result = func(path, **kwargs)
            rec.success = True
            rec.method = result.get("method", "unknown")
            return result
        except Exception as e:
            rec.error = str(e)
            return {"error": str(e)}
        finally:
            rec.duration = round(time.time() - start, 3)
            self.records.append(rec)

    def get_summary(self) -> dict:
        total = len(self.records)
        if total == 0: return {"total": 0}
        ok = sum(1 for r in self.records if r.success)
        errors = {}
        for r in self.records:
            if r.error: errors[r.error[:80]] = errors.get(r.error[:80], 0) + 1
        return {"total": total, "successes": ok, "failures": total - ok,
                "success_rate": f"{ok/total*100:.1f}%",
                "avg_duration": round(sum(r.duration for r in self.records)/total, 3),
                "top_errors": dict(sorted(errors.items(), key=lambda x:-x[1])[:5])}

    def export_records(self, path: str):
        with open(path, "w") as f:
            json.dump([asdict(r) for r in self.records], f, indent=2, ensure_ascii=False)

Uso:

monitor = DocumentPipelineMonitor()
for pdf_path in pdf_files:
    monitor.track(fallback_extract, pdf_path, vision_client=client)
print(monitor.get_summary())

Checklist de Robustez

• Validar PDF antes de procesar (diagnóstico completo)
• Manejar PDFs protegidos con contraseña
• Clasificar tipo de PDF (digital / escaneado / mixto)
• Fallback: texto → OCR → Vision API
• Preprocesar imágenes (deskew, denoise, binarize)
• Detectar y manejar encoding correctamente
• Retry con backoff + rate limiting para APIs
• Procesamiento por lotes para PDFs grandes
• Logging estructurado y timeout configurable
• Validar JSON extraído antes de usar

Ejercicios

Ejercicio 1: Función de diagnóstico completa

Crea diagnose_document(path) que retorne tipo de archivo, validez, número de páginas, tipo de PDF (digital/escaneado/mixto), lista de problemas y estrategia recomendada.

def diagnose_document(path: str) -> dict:
    report = {"path": path, "exists": False, "file_type": None, "valid": False,
              "pages": 0, "pdf_type": None, "issues": [], "strategy": None}
    p = Path(path)
    if not p.exists():
        report["issues"].append("Archivo no encontrado")
        return report
    report["exists"] = True
    report["file_type"] = p.suffix.lower()

    if report["file_type"] in {".png", ".jpg", ".jpeg", ".tiff"}:
        return {**report, "valid": True, "pages": 1, "strategy": "ocr_or_vision"}
    if report["file_type"] != ".pdf":
        report["issues"].append(f"No soportado: {report['file_type']}")
        return report
    try:
        doc = fitz.open(path)
    except Exception as e:
        report["issues"].append(f"No se puede abrir: {e}")
        return {**report, "strategy": "repair_first"}

    report["valid"], report["pages"] = True, len(doc)
    if doc.is_encrypted:
        doc.close()
        return {**report, "issues": ["PDF encriptado"], "strategy": "decrypt_first"}

    digital, scanned = 0, 0
    for i in range(min(len(doc), 10)):
        if len(doc[i].get_text().strip()) > 50: digital += 1
        elif len(doc[i].get_images()) > 0: scanned += 1
    doc.close()

    if scanned > digital: report["pdf_type"], report["strategy"] = "scanned", "ocr_pipeline"
    elif digital > scanned: report["pdf_type"], report["strategy"] = "digital", "text_extraction"
    else: report["pdf_type"], report["strategy"] = "mixed", "hybrid_pipeline"
    return report

Ejercicio 2: Preprocessing pipeline configurable

Crea OCRPreprocessor con métodos para: escala de grises, filtro de ruido, corrección de inclinación, contraste y binarización. El método run(img) ejecuta todos los pasos y retorna (imagen, lista_de_pasos).

class OCRPreprocessor:
    def __init__(self, denoise=True, deskew=True, binarize=True, contrast=2.0):
        self.denoise = denoise
        self.deskew = deskew
        self.binarize = binarize
        self.contrast = contrast

    def run(self, img: Image.Image) -> tuple[Image.Image, list[str]]:
        steps = []
        img = img.convert("L"); steps.append("grayscale")
        if self.denoise:
            img = img.filter(ImageFilter.MedianFilter(size=3)); steps.append("denoise")
        if self.deskew:
            arr = np.array(img)
            coords = np.column_stack(np.where((arr < 128) > 0))
            if len(coords) > 100:
                best_a, best_v = 0, 0
                for a in np.arange(-10, 10, 0.5):
                    r = np.radians(a)
                    h, _ = np.histogram(coords[:,0]*np.cos(r)+coords[:,1]*np.sin(r), bins=50)
                    if np.var(h) > best_v: best_v, best_a = np.var(h), a
                if abs(best_a) > 0.5:
                    img = img.rotate(best_a, fillcolor=255, expand=True)
                    steps.append(f"deskew({best_a:.1f}°)")
        img = ImageEnhance.Contrast(img).enhance(self.contrast); steps.append("contrast")
        if self.binarize:
            img = img.point(lambda x: 0 if x < 140 else 255); steps.append("binarize")
        return img, steps

Ejercicio 3: Fallback cascade para documentos

Implementa robust_extract(path, client=None) con cascada: (1) PyMuPDF texto, (2) OCR Tesseract, (3) Vision API, (4) error con métodos intentados. Cada paso debe registrarse en attempts.

def robust_extract(path: str, client=None) -> dict:
    result = {"path": path, "text": "", "method": "none", "attempts": []}

    # Paso 1: texto directo
    try:
        doc = fitz.open(path)
        text = "\n".join(p.get_text() for p in doc).strip(); doc.close()
        result["attempts"].append({"method": "pymupdf", "chars": len(text)})
        if len(text) > 50:
            return {**result, "text": text, "method": "pymupdf"}
    except Exception as e:
        result["attempts"].append({"method": "pymupdf", "error": str(e)})

    # Paso 2: OCR
    try:
        doc = fitz.open(path)
        parts = []
        for i in range(min(len(doc), 10)):
            pix = doc[i].get_pixmap(matrix=fitz.Matrix(2, 2))
            img = ImageEnhance.Contrast(Image.open(io.BytesIO(pix.tobytes("png"))).convert("L")).enhance(2.0)
            parts.append(pytesseract.image_to_string(img, lang="spa"))
        doc.close()
        text = "\n".join(parts).strip()
        result["attempts"].append({"method": "tesseract", "chars": len(text)})
        if len(text) > 50:
            return {**result, "text": text, "method": "tesseract"}
    except Exception as e:
        result["attempts"].append({"method": "tesseract", "error": str(e)})

    # Paso 3: Vision API
    if client:
        try:
            import base64
            doc = fitz.open(path)
            b64 = base64.b64encode(doc[0].get_pixmap(matrix=fitz.Matrix(2,2)).tobytes("png")).decode()
            doc.close()
            resp = client.chat.completions.create(model="gpt-4o", messages=[{"role":"user","content":[
                {"type":"text","text":"Extrae todo el texto de este documento."},
                {"type":"image_url","image_url":{"url":f"data:image/png;base64,{b64}"}}
            ]}], temperature=0)
            text = resp.choices[0].message.content.strip()
            result["attempts"].append({"method": "vision", "chars": len(text)})
            if len(text) > 20:
                return {**result, "text": text, "method": "vision"}
        except Exception as e:
            result["attempts"].append({"method": "vision", "error": str(e)})

    return {**result, "method": "failed"}

Ejercicio 4: Batch processor con manejo de errores

Crea BatchDocumentProcessor que procese una lista de PDFs con: rate limiting (N por minuto), retry con backoff, logging, y método summary() con tasa de éxito, errores frecuentes y tiempo total.

@dataclass
class DocResult:
    path: str
    success: bool = False
    method: str = ""
    duration: float = 0.0
    error: str = ""

class BatchDocumentProcessor:
    def __init__(self, extract_fn, max_per_minute=30, max_retries=3):
        self.extract_fn = extract_fn
        self.max_per_minute = max_per_minute
        self.max_retries = max_retries
        self.results: list[DocResult] = []
        self._times: list[float] = []

    def _throttle(self):
        now = time.time()
        self._times = [t for t in self._times if now - t < 60]
        if len(self._times) >= self.max_per_minute:
            wait = 60 - (now - self._times[0])
            if wait > 0: time.sleep(wait)
        self._times.append(time.time())

    def process_batch(self, paths: list[str]) -> list[DocResult]:
        for path in paths:
            rec = DocResult(path=path)
            start = time.time()
            for attempt in range(self.max_retries):
                try:
                    self._throttle()
                    r = self.extract_fn(path)
                    rec.success, rec.method = True, r.get("method", "")
                    break
                except Exception as e:
                    rec.error = str(e)
                    if attempt < self.max_retries - 1: time.sleep(2**attempt)
            rec.duration = round(time.time() - start, 3)
            self.results.append(rec)
        return self.results

    def summary(self) -> dict:
        total = len(self.results)
        if total == 0: return {"total": 0}
        ok = sum(1 for r in self.results if r.success)
        errors = {}
        for r in self.results:
            if r.error: errors[r.error[:60]] = errors.get(r.error[:60], 0) + 1
        return {"total": total, "success": ok, "failed": total-ok,
                "rate": f"{ok/total*100:.1f}%",
                "total_time": round(sum(r.duration for r in self.results), 2),
                "top_errors": dict(sorted(errors.items(), key=lambda x:-x[1])[:3])}

Recursos adicionales

1. PyMuPDF · PIL ImageEnhance · Tenacity
2. Circuit Breaker pattern · Camelot · chardet