From 239a7510c6188dc0213c943ff45cd66140d0423d Mon Sep 17 00:00:00 2001 From: urieljareth Date: Sun, 26 Jul 2026 19:58:38 -0600 Subject: [PATCH] OCR Mistral: modelo mistral-ocr-4-0, batch + fallback con reintentos y errores etiquetados - Modelo por defecto mistral-ocr-4-0 (.env.example, config.py, docker-compose) - MistralClient: OCR en lote via API Batch (/v1/ocr) subiendo JSONL, polling y descarga de resultados - Reintentos con backoff exponencial en llamadas a Mistral (429/5xx/red); sin reintento en 401/400 - PdfProcessor: lote cuando >= umbral de paginas, fallback sincrono pagina a pagina, placeholder rastreable por pagina fallida - Errores etiquetados ([AUTH_MISTRAL], [BATCH], [NETWORK], [OCR_EMPTY], [PDF_PARSE]...) en el mensaje del job - Nuevas variables: MISTRAL_OCR_BATCH_THRESHOLD/POLL_SECONDS/TIMEOUT_SECONDS/RETRY_ATTEMPTS/RETRY_BACKOFF_BASE --- .env.example | 7 +- backend/app/config.py | 7 +- backend/app/main.py | 7 +- backend/app/services/errors.py | 85 ++++++ .../app/services/processors/pdf_processor.py | 77 ++++- .../app/services/providers/mistral_client.py | 288 ++++++++++++++++-- docker-compose.yml | 7 +- 7 files changed, 441 insertions(+), 37 deletions(-) create mode 100644 backend/app/services/errors.py diff --git a/.env.example b/.env.example index fe0065d..8ea216d 100644 --- a/.env.example +++ b/.env.example @@ -2,5 +2,10 @@ APP_ENV=development APP_DATA_DIR=./data DATABASE_PATH=./data/app.db MISTRAL_API_KEY= -MISTRAL_OCR_MODEL=mistral-ocr-latest +MISTRAL_OCR_MODEL=mistral-ocr-4-0 +MISTRAL_OCR_BATCH_THRESHOLD=3 +MISTRAL_OCR_BATCH_POLL_SECONDS=10 +MISTRAL_OCR_BATCH_TIMEOUT_SECONDS=1800 +MISTRAL_OCR_RETRY_ATTEMPTS=3 +MISTRAL_OCR_RETRY_BACKOFF_BASE=2.0 DEEPGRAM_API_KEY= diff --git a/backend/app/config.py b/backend/app/config.py index 58e1de0..dd55be4 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -8,7 +8,12 @@ class Settings(BaseSettings): app_data_dir: Path = Path("./data") database_path: Path = Path("./data/app.db") mistral_api_key: str = "" - mistral_ocr_model: str = "mistral-ocr-latest" + mistral_ocr_model: str = "mistral-ocr-4-0" + mistral_ocr_batch_threshold: int = 3 + mistral_ocr_batch_poll_seconds: int = 10 + mistral_ocr_batch_timeout_seconds: int = 1800 + mistral_ocr_retry_attempts: int = 3 + mistral_ocr_retry_backoff_base: float = 2.0 deepgram_api_key: str = "" model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore") diff --git a/backend/app/main.py b/backend/app/main.py index 3431b3f..9360a3e 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -9,6 +9,7 @@ from fastapi.responses import FileResponse, PlainTextResponse from backend.app.config import settings from backend.app.managers.knowledge_manager import DuplicateSourceError, KnowledgeManager from backend.app.models.schemas import MarkdownUpdate, SubjectCreate, SubjectUpdate, WeekCreate, WeekUpdate +from backend.app.services.errors import ProcessingError, classify_unknown_error from backend.app.services.ingestion_service import IngestionService app = FastAPI(title="Knowledge Station", version="0.1.0") @@ -284,6 +285,10 @@ def process_job(job_id: int, source_id: int, use_ocr: bool, page_ranges: str | N try: document = ingestion.process_source(source_id, use_ocr=use_ocr, page_ranges=page_ranges) manager.update_job(job_id, "completed", "Documento procesado", document_id=document["id"]) + except ProcessingError as exc: + manager.update_source_status(source_id, "failed") + manager.update_job(job_id, "failed", exc.labeled) except Exception as exc: manager.update_source_status(source_id, "failed") - manager.update_job(job_id, "failed", str(exc)) + code = classify_unknown_error(exc) + manager.update_job(job_id, "failed", f"[{code}] {exc}") diff --git a/backend/app/services/errors.py b/backend/app/services/errors.py new file mode 100644 index 0000000..33a79be --- /dev/null +++ b/backend/app/services/errors.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +import requests + + +class ProcessingError(Exception): + """Error de procesamiento con un codigo corto y rastreable. + + El atributo ``code`` es un identificador estable en MAYUSCULAS que se + antepone al mensaje del job (p.ej. ``[AUTH_MISTRAL] API Key invalida``) + para que el usuario pueda buscar y diagnosticar rapidamente el tipo de + fallo desde la UI. + """ + + def __init__(self, code: str, detail: str) -> None: + super().__init__(f"[{code}] {detail}") + self.code = code + self.detail = detail + + @property + def labeled(self) -> str: + return f"[{self.code}] {self.detail}" + + +class MistralAuthError(ProcessingError): + def __init__(self, detail: str = "API Key de Mistral invalida o no autorizada (401)") -> None: + super().__init__("AUTH_MISTRAL", detail) + + +class MistralRateLimitError(ProcessingError): + def __init__(self, detail: str = "Limite de uso de Mistral alcanzado (429)") -> None: + super().__init__("RATE_LIMIT", detail) + + +class MistralServerError(ProcessingError): + def __init__(self, status: int, body: str) -> None: + super().__init__("MISTRAL_SERVER", f"Mistral HTTP {status}: {body[:200]}") + + +class MistralBatchError(ProcessingError): + def __init__(self, detail: str) -> None: + super().__init__("BATCH", detail) + + +class MistralOCREmptyError(ProcessingError): + def __init__(self, detail: str = "OCR no devolvio contenido") -> None: + super().__init__("OCR_EMPTY", detail) + + +class NetworkError(ProcessingError): + def __init__(self, detail: str) -> None: + super().__init__("NETWORK", detail) + + +NON_RETRYABLE_CODES = { + "AUTH_MISTRAL", + "CONFIG", + "OCR_EMPTY", + "MISTRAL_CLIENT", + "BAD_PDF", + "FILE_IO", +} + + +def classify_requests_error(exc: requests.RequestException) -> ProcessingError: + if isinstance(exc, requests.Timeout): + return NetworkError(f"timeout de red con Mistral: {exc}") + if isinstance(exc, requests.ConnectionError): + return NetworkError(f"sin conexion a Mistral: {exc}") + return NetworkError(f"error de red con Mistral: {exc}") + + +def classify_unknown_error(exc: Exception) -> str: + """Etiqueta breve para excepciones no envueltas en ProcessingError.""" + name = type(exc).__name__.lower() + message = str(exc).lower() + if "fitz" in name or "pdf" in name or "pdf" in message or "muPDF" in message: + return "PDF_PARSE" + if isinstance(exc, FileNotFoundError): + return "FILE_IO" + if isinstance(exc, (PermissionError, OSError)): + return "FILE_IO" + if isinstance(exc, ValueError): + return "VALIDATION" + return "UNKNOWN" diff --git a/backend/app/services/processors/pdf_processor.py b/backend/app/services/processors/pdf_processor.py index 597b986..64c2d8a 100644 --- a/backend/app/services/processors/pdf_processor.py +++ b/backend/app/services/processors/pdf_processor.py @@ -4,8 +4,10 @@ from pathlib import Path import fitz +from backend.app.config import settings from backend.app.services.markdown_builder import title_from_path from backend.app.services.processors.base import ProcessedContent +from backend.app.services.errors import ProcessingError from backend.app.services.providers.mistral_client import MistralClient @@ -30,22 +32,77 @@ class PdfProcessor: return ProcessedContent(title=title_from_path(path), body=body, processor="pdf-text") def _process_ocr(self, path: Path) -> ProcessedContent: - doc = fitz.open(path) - client = MistralClient() - parts: list[str] = [] - temp_paths: list[Path] = [] try: - for index in self._selected_pages(len(doc)): + doc = fitz.open(path) + except Exception as exc: + raise ProcessingError("BAD_PDF", f"No se pudo abrir el PDF: {exc}") from exc + selected = self._selected_pages(len(doc)) + client = MistralClient() + page_images: dict[int, Path] = {} + try: + for index in selected: page = doc[index - 1] pix = page.get_pixmap(matrix=fitz.Matrix(2, 2), alpha=False) image_path = path.with_name(f"{path.stem}-page-{index}.jpg") pix.save(image_path) - temp_paths.append(image_path) - parts.append(f"## Pagina {index}\n\n{client.ocr_image(image_path)}") + page_images[index] = image_path + + results = self._ocr_pages(client, page_images) finally: - for temp_path in temp_paths: - temp_path.unlink(missing_ok=True) - return ProcessedContent(title=title_from_path(path), body="\n\n".join(parts), processor="mistral-ocr") + for image_path in page_images.values(): + image_path.unlink(missing_ok=True) + + parts: list[str] = [] + for index in selected: + markdown = results.get(index, "") + parts.append(f"## Pagina {index}\n\n{markdown}") + return ProcessedContent( + title=title_from_path(path), + body="\n\n".join(parts), + processor="mistral-ocr", + ) + + def _ocr_pages(self, client: MistralClient, page_images: dict[int, Path]) -> dict[int, str]: + """OCR de las paginas con lote + fallback sincrono. + + Estrategia: + 1. Si hay >= ``batch_threshold`` paginas, intenta el job en lote. + 2. Cualquier pagina no resuelta (o si el lote entero falla) cae a + OCR sincrono pagina a pagina con reintentos. + 3. Las paginas que fallen de forma definitiva quedan como un + placeholder breve con el codigo de error rastreable. + """ + indices = list(page_images.keys()) + results: dict[int, str] = {} + page_errors: dict[int, str] = {} + + if len(indices) >= settings.mistral_ocr_batch_threshold: + ordered = sorted(indices) + paths = [page_images[i] for i in ordered] + try: + batch_results = client.ocr_images_batch(paths) + for pos, index in enumerate(ordered): + if pos in batch_results: + results[index] = batch_results[pos] + except ProcessingError: + # El lote fallo; continua al fallback sincrono para todas + # las paginas no resueltas. + pass + + for index in indices: + if index in results: + continue + try: + results[index] = client.ocr_image(page_images[index]) + except ProcessingError as exc: + page_errors[index] = exc.labeled + except Exception as exc: + page_errors[index] = f"[OCR_ERROR] {exc}" + + for index in indices: + if index not in results: + results[index] = f"> ⚠️ No se pudo procesar esta pagina: {page_errors.get(index, '[UNKNOWN]')}" + return results def _selected_pages(self, total_pages: int) -> list[int]: if total_pages < 1: diff --git a/backend/app/services/providers/mistral_client.py b/backend/app/services/providers/mistral_client.py index 24e5084..fa4ed47 100644 --- a/backend/app/services/providers/mistral_client.py +++ b/backend/app/services/providers/mistral_client.py @@ -1,45 +1,287 @@ from __future__ import annotations import base64 +import json +import time from pathlib import Path import requests from backend.app.config import settings +from backend.app.services.errors import ( + MistralAuthError, + MistralBatchError, + MistralOCREmptyError, + MistralRateLimitError, + MistralServerError, + NetworkError, + NON_RETRYABLE_CODES, + ProcessingError, + classify_requests_error, +) class MistralClient: - endpoint = "https://api.mistral.ai/v1/ocr" + """Cliente Mistral para OCR. + + - ``ocr_image``: OCR sincrono de una pagina, con reintentos y backoff. + - ``ocr_images_batch``: OCR en lote via la API de Batch de Mistral + (endpoint ``/v1/ocr``), subiendo un fichero JSONL. Devuelve un dict + ``{page_index: markdown}``. Si el job falla o caduca se lanza + ``MistralBatchError`` para que el llamador caiga al modo sincrono. + """ + + base_url = "https://api.mistral.ai/v1" + ocr_url = f"{base_url}/ocr" + files_url = f"{base_url}/files" + batch_url = f"{base_url}/batch/jobs" + + def __init__(self) -> None: + self._api_key = settings.mistral_api_key + self._model = settings.mistral_ocr_model + + # ------------------------------------------------------------------ + # API publica + # ------------------------------------------------------------------ def ocr_image(self, image_path: Path) -> str: - if not settings.mistral_api_key: - raise RuntimeError("MISTRAL_API_KEY no esta configurada") + self._require_key() data_url = self._image_data_url(image_path) - response = requests.post( - self.endpoint, - headers={ - "Content-Type": "application/json", - "Authorization": f"Bearer {settings.mistral_api_key}", - }, - json={ - "model": settings.mistral_ocr_model, + return self._retry(lambda: self._ocr_sync(data_url)) + + def ocr_images_batch(self, image_paths: list[Path]) -> dict[int, str]: + """OCR en lote. Devuelve ``{posicion_en_lista: markdown}``. + + ``posicion_en_lista`` es el indice (0-based) dentro de ``image_paths``, + NO el numero de pagina del PDF; el llamador hace esa traduccion. + """ + self._require_key() + if not image_paths: + return {} + + payload = self._build_batch_payload(image_paths) + upload = self._retry(lambda: self._upload_batch_file(payload)) + file_id = upload.get("id") + if not file_id: + raise MistralBatchError("la subida del fichero batch no devolvio id") + + job = self._retry(lambda: self._create_batch_job(file_id)) + job_id = job.get("id") + if not job_id: + raise MistralBatchError("la creacion del job batch no devolvio id") + + final = self._poll_batch_job(job_id) + if final.get("status") != "SUCCESS": + reason = final.get("errors") or final.get("status") or "sin detalle" + raise MistralBatchError(f"job {job_id} finalizo como {final.get('status')}: {reason}") + + output_file_id = final.get("output_file") + if not output_file_id: + raise MistralBatchError(f"job {job_id} sin output_file") + + content = self._retry(lambda: self._download_file(output_file_id)) + return self._parse_batch_output(content) + + # ------------------------------------------------------------------ + # OCR sincrono + # ------------------------------------------------------------------ + + def _ocr_sync(self, data_url: str) -> str: + try: + response = requests.post( + self.ocr_url, + headers=self._json_headers(), + json={ + "model": self._model, + "document": {"type": "image_url", "image_url": data_url}, + "include_image_base64": False, + }, + timeout=180, + ) + except requests.RequestException as exc: + raise classify_requests_error(exc) from exc + self._raise_for_status(response) + markdown = self._extract_markdown(response.json()) + if not markdown: + raise MistralOCREmptyError() + return markdown + + # ------------------------------------------------------------------ + # Batch: construccion, subida, job, polling, descarga, parseo + # ------------------------------------------------------------------ + + def _build_batch_payload(self, image_paths: list[Path]) -> bytes: + lines: list[str] = [] + for idx, path in enumerate(image_paths): + data_url = self._image_data_url(path) + body = { "document": {"type": "image_url", "image_url": data_url}, "include_image_base64": False, - }, - timeout=180, + } + lines.append( + json.dumps( + {"custom_id": str(idx), "body": body}, + separators=(",", ":"), + ensure_ascii=False, + ) + ) + return ("\n".join(lines)).encode("utf-8") + + def _upload_batch_file(self, payload: bytes) -> dict: + try: + response = requests.post( + self.files_url, + headers={"Authorization": f"Bearer {self._api_key}"}, + data={"purpose": "batch"}, + files={"file": ("ocr-batch.jsonl", payload, "application/jsonl")}, + timeout=300, + ) + except requests.RequestException as exc: + raise classify_requests_error(exc) from exc + self._raise_for_status(response) + return response.json() + + def _create_batch_job(self, file_id: str) -> dict: + try: + response = requests.post( + self.batch_url, + headers=self._json_headers(), + json={ + "input_files": [file_id], + "model": self._model, + "endpoint": "/v1/ocr", + }, + timeout=120, + ) + except requests.RequestException as exc: + raise classify_requests_error(exc) from exc + self._raise_for_status(response) + return response.json() + + def _poll_batch_job(self, job_id: str) -> dict: + terminal = {"SUCCESS", "FAILED", "TIMEOUT_EXCEEDED", "CANCELLED"} + deadline = time.time() + max(60, settings.mistral_ocr_batch_timeout_seconds) + delay = max(2, settings.mistral_ocr_batch_poll_seconds) + last: dict = {} + while time.time() < deadline: + try: + response = requests.get( + f"{self.batch_url}/{job_id}", + headers=self._json_headers(), + timeout=60, + ) + except requests.RequestException: + time.sleep(delay) + continue + if response.ok: + last = response.json() + if last.get("status") in terminal: + return last + time.sleep(delay) + raise MistralBatchError( + f"job {job_id} excedio el timeout de polling " + f"({settings.mistral_ocr_batch_timeout_seconds}s); ultimo estado: " + f"{last.get('status') if last else 'desconocido'}" ) - if response.status_code == 401: - raise RuntimeError("API Key de Mistral invalida") - if not response.ok: - raise RuntimeError(f"Mistral error {response.status_code}: {response.text}") - data = response.json() + + def _download_file(self, file_id: str) -> bytes: + try: + response = requests.get( + f"{self.files_url}/{file_id}/content", + headers=self._json_headers(), + timeout=300, + ) + except requests.RequestException as exc: + raise classify_requests_error(exc) from exc + self._raise_for_status(response) + return response.content + + def _parse_batch_output(self, content: bytes) -> dict[int, str]: + results: dict[int, str] = {} + for raw in content.decode("utf-8", errors="replace").splitlines(): + line = raw.strip() + if not line: + continue + try: + obj = json.loads(line) + except json.JSONDecodeError: + continue + custom_id = obj.get("custom_id") + if custom_id is None: + continue + try: + idx = int(custom_id) + except (TypeError, ValueError): + continue + response = obj.get("response") or {} + status_code = response.get("status_code", 200) + body = response.get("body") or {} + if status_code >= 400 or obj.get("error"): + # Pagina fallida en lote: se omite y el llamador cae a sync. + continue + markdown = self._extract_markdown(body) + if markdown: + results[idx] = markdown + return results + + # ------------------------------------------------------------------ + # Helpers compartidos + # ------------------------------------------------------------------ + + def _require_key(self) -> None: + if not self._api_key: + raise ProcessingError("CONFIG", "MISTRAL_API_KEY no esta configurada") + + def _json_headers(self) -> dict[str, str]: + return { + "Authorization": f"Bearer {self._api_key}", + "Content-Type": "application/json", + "Accept": "application/json", + } + + def _raise_for_status(self, response: requests.Response) -> None: + if response.ok: + return + try: + body = response.text + except Exception: + body = "" + status = response.status_code + if status == 401: + raise MistralAuthError() + if status == 429: + raise MistralRateLimitError() + if status >= 500: + raise MistralServerError(status, body) + raise ProcessingError("MISTRAL_CLIENT", f"HTTP {status}: {body[:200]}") + + def _retry(self, fn): + attempts = max(1, settings.mistral_ocr_retry_attempts) + base_delay = max(0.5, settings.mistral_ocr_retry_backoff_base) + last_exc: Exception | None = None + for attempt in range(attempts): + try: + return fn() + except ProcessingError as exc: + if exc.code in NON_RETRYABLE_CODES: + raise + last_exc = exc + except Exception as exc: + last_exc = exc + if attempt < attempts - 1: + time.sleep(base_delay * (2 ** attempt)) + if last_exc is not None: + raise last_exc + raise MistralBatchError("reintentos agotados sin excepcion conocida") + + def _extract_markdown(self, data: dict) -> str: pages = data.get("pages") or [] - markdown_pages = [page.get("markdown", "").strip() for page in pages if page.get("markdown")] - if markdown_pages: - return "\n\n".join(markdown_pages) + chunks = [page.get("markdown", "").strip() for page in pages if page.get("markdown")] + if chunks: + return "\n\n".join(chunks) if data.get("markdown"): - return data["markdown"].strip() - raise RuntimeError("Mistral OCR no devolvio Markdown") + return str(data["markdown"]).strip() + return "" def _image_data_url(self, image_path: Path) -> str: encoded = base64.b64encode(image_path.read_bytes()).decode("ascii") diff --git a/docker-compose.yml b/docker-compose.yml index 267be5a..9c087da 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -11,7 +11,12 @@ services: APP_DATA_DIR: /app/data DATABASE_PATH: /app/data/app.db MISTRAL_API_KEY: ${MISTRAL_API_KEY:-} - MISTRAL_OCR_MODEL: ${MISTRAL_OCR_MODEL:-mistral-ocr-latest} + MISTRAL_OCR_MODEL: ${MISTRAL_OCR_MODEL:-mistral-ocr-4-0} + MISTRAL_OCR_BATCH_THRESHOLD: ${MISTRAL_OCR_BATCH_THRESHOLD:-3} + MISTRAL_OCR_BATCH_POLL_SECONDS: ${MISTRAL_OCR_BATCH_POLL_SECONDS:-10} + MISTRAL_OCR_BATCH_TIMEOUT_SECONDS: ${MISTRAL_OCR_BATCH_TIMEOUT_SECONDS:-1800} + MISTRAL_OCR_RETRY_ATTEMPTS: ${MISTRAL_OCR_RETRY_ATTEMPTS:-3} + MISTRAL_OCR_RETRY_BACKOFF_BASE: ${MISTRAL_OCR_RETRY_BACKOFF_BASE:-2.0} DEEPGRAM_API_KEY: ${DEEPGRAM_API_KEY:-} NODE_ENV: production PORT: 3000