From 3b01456875c072c412b4e97d6380ba9b72cb5127 Mon Sep 17 00:00:00 2001 From: erik Date: Sun, 16 Aug 2026 15:39:01 -0500 Subject: [PATCH] feat: Update default models for extraction and normalization to GPT-5.6 Luna across configuration files and documentation. Remove deprecated code and improve pricing details in the UI components. Add tests to ensure correct model settings and pricing calculations. --- .env.example | 7 +- README.md | 2 +- .../bot_descarga/audit/jsonl_logger.py | 3 - .../bot_descarga/infrastructure/browser.py | 2 +- .../infrastructure/file_manager.py | 13 - .../bot_descarga/ui/progress_tracker.py | 65 ----- .../bot_digitador/core/digitador_service.py | 53 ---- docker-compose.yml | 8 +- services/bot_descarga/api/main.py | 1 - services/bot_descarga/domain/models.py | 2 +- .../infrastructure/storage_minio.py | 6 +- .../infrastructure/storage_minio.py | 6 +- .../cleaning/domain/csv_cleaning/analyze.py | 1 - .../domain/csv_cleaning/column_hints.py | 2 - .../domain/csv_cleaning/rules/dates.py | 1 - services/cleaning/infrastructure/database.py | 11 - .../renderer/components/CountingWizard.tsx | 2 - .../renderer/components/ExtractionWizard.tsx | 16 +- .../modules/counting/CountingResults.tsx | 1 - .../extraction/JobExtractionWizard.tsx | 2 +- services/extraction/README.md | 4 +- services/extraction/api/desktop_ingest.py | 171 ++++++++++--- services/extraction/api/main.py | 91 +++---- services/extraction/core/pdf_utils.py | 82 +------ services/extraction/core/text_utils.py | 1 - services/extraction/core/utils.py | 5 - services/extraction/domain/config.py | 5 +- services/extraction/domain/costs.py | 10 +- services/extraction/domain/estimator.py | 3 +- services/extraction/domain/llm_client.py | 42 +--- services/extraction/domain/service.py | 6 - .../extraction/infrastructure/database.py | 11 - services/extraction/tests/test_config.py | 11 + services/extraction/tests/test_costs.py | 18 ++ .../extraction/tests/test_desktop_ingest.py | 228 ++++++++++++++++++ .../extraction/tests/test_worker_model.py | 114 +++++++++ services/extraction/worker/tasks.py | 31 ++- .../domain/verificador_st/core.py | 217 ----------------- 38 files changed, 615 insertions(+), 639 deletions(-) delete mode 100644 bots/bot_descarga/bot_descarga/ui/progress_tracker.py create mode 100644 services/extraction/tests/test_config.py create mode 100644 services/extraction/tests/test_costs.py create mode 100644 services/extraction/tests/test_desktop_ingest.py create mode 100644 services/extraction/tests/test_worker_model.py diff --git a/.env.example b/.env.example index c66f5ea..8fbe4c4 100644 --- a/.env.example +++ b/.env.example @@ -2,10 +2,11 @@ # ── OpenAI ──────────────────────────────────────────────── # API key usada para extracción y normalización. -# El modelo del extractor lo elige el usuario en la app (ej. gpt-4o, gpt-5.4). -# El normalizador siempre usa gpt-4o-mini (fijado en docker-compose). +# GPT-5.6 Luna es el modelo predeterminado para extracción y normalización. +# El extractor también se puede cambiar desde la aplicación. EXTRACTOR_API_KEY=sk-... -EXTRACTOR_MODEL=gpt-4o +EXTRACTOR_MODEL=gpt-5.6-luna +NORMALIZER_MODEL=gpt-5.6-luna EXTRACTOR_TIMEOUT_S=120 # ── Procesamiento de imagenes ───────────────────────────── diff --git a/README.md b/README.md index 8d2f66d..8b4b32f 100644 --- a/README.md +++ b/README.md @@ -107,7 +107,7 @@ Ver también las variables `COMPOSE_BAKE` y `COMPOSE_PARALLEL_LIMIT` en `.env.ex | `IPS_DOCUMENT_ROOT` | Carpeta local con los documentos PDF (montada en `/data`) | | `EXTRACTOR_API_KEY` | API key del LLM (OpenAI, xAI, etc.) | | `EXTRACTOR_BASE_URL` | URL base del LLM (default: OpenAI) | -| `EXTRACTOR_MODEL` | Modelo de extracción (default: `gpt-5.4-mini`) | +| `EXTRACTOR_MODEL` | Modelo de extracción (default: `gpt-5.6-luna`) | | `WORKER_REPLICAS` | Réplicas del worker de extracción (default: `2`) | Ver `.env.example` para la lista completa. diff --git a/bots/bot_descarga/bot_descarga/audit/jsonl_logger.py b/bots/bot_descarga/bot_descarga/audit/jsonl_logger.py index 7bb5f8c..9c4484b 100644 --- a/bots/bot_descarga/bot_descarga/audit/jsonl_logger.py +++ b/bots/bot_descarga/bot_descarga/audit/jsonl_logger.py @@ -53,9 +53,6 @@ class EventBus: for cb in self.callbacks: cb(event) - def add_callback(self, callback: EventCallback) -> None: - self.callbacks.append(callback) - def close(self) -> None: if self.jsonl_logger is not None: self.jsonl_logger.close() diff --git a/bots/bot_descarga/bot_descarga/infrastructure/browser.py b/bots/bot_descarga/bot_descarga/infrastructure/browser.py index 10a20fc..3379857 100644 --- a/bots/bot_descarga/bot_descarga/infrastructure/browser.py +++ b/bots/bot_descarga/bot_descarga/infrastructure/browser.py @@ -1,7 +1,7 @@ from __future__ import annotations import time -from typing import Callable, Optional, Tuple +from typing import Callable, Tuple from selenium import webdriver from selenium.common.exceptions import ( diff --git a/bots/bot_descarga/bot_descarga/infrastructure/file_manager.py b/bots/bot_descarga/bot_descarga/infrastructure/file_manager.py index 4ae9e25..a441430 100644 --- a/bots/bot_descarga/bot_descarga/infrastructure/file_manager.py +++ b/bots/bot_descarga/bot_descarga/infrastructure/file_manager.py @@ -45,16 +45,3 @@ class FileManager: os.replace(archivo_temp, ruta_final) except Exception: os.rename(archivo_temp, ruta_final) - - def remove_safe(self, ruta: str) -> None: - try: - if ruta and os.path.exists(ruta): - os.remove(ruta) - except Exception: - pass - - def list_pdfs(self) -> set: - return { - f for f in os.listdir(self.carpeta_descarga) - if f.lower().endswith(".pdf") - } diff --git a/bots/bot_descarga/bot_descarga/ui/progress_tracker.py b/bots/bot_descarga/bot_descarga/ui/progress_tracker.py deleted file mode 100644 index f30e975..0000000 --- a/bots/bot_descarga/bot_descarga/ui/progress_tracker.py +++ /dev/null @@ -1,65 +0,0 @@ -from __future__ import annotations - -from bot_descarga.audit.events import BotEvent, EventType - - -class ProgressTracker: - """Traduce BotEvent a contadores de progreso para la UI.""" - - def __init__(self): - self.total = 0 - self.procesados = 0 - self.exitosos = 0 - self.fallidos = 0 - self.skipped = 0 - self.nit_actual = "—" - self.nit_estado = "Esperando inicio del proceso..." - - def reset(self) -> None: - self.total = 0 - self.procesados = 0 - self.exitosos = 0 - self.fallidos = 0 - self.skipped = 0 - self.nit_actual = "—" - self.nit_estado = "Esperando inicio del proceso..." - - def handle(self, event: BotEvent) -> None: - if event.index is not None and event.total is not None: - self.procesados = event.index - self.total = event.total - - if event.event_type == EventType.RUN_STARTED: - if "total" in event.metadata: - self.total = int(event.metadata["total"]) - elif event.total is not None: - self.total = event.total - - elif event.event_type == EventType.NIT_STARTED: - self.nit_actual = event.nit or "—" - self.nit_estado = "Procesando radicados de este NIT..." - - elif event.event_type == EventType.NIT_FINISHED: - nit = event.nit or "" - meta = event.metadata - self.nit_estado = ( - f"Finalizó NIT {nit} | " - f"OK: {meta.get('ok', 0)} | " - f"Fallidos: {meta.get('fallidos', 0)} | " - f"Ya existían: {meta.get('omitidos', 0)}" - ) - - elif event.event_type == EventType.RADICADO_SAVED: - self.exitosos += 1 - - elif event.event_type == EventType.RADICADO_SKIPPED: - self.skipped += 1 - - elif event.event_type == EventType.RADICADO_FAILED: - if "PROCESO TERMINADO" not in event.message: - self.fallidos += 1 - - @property - def percent(self) -> float: - total = self.total if self.total > 0 else 1 - return min(self.procesados / total, 1.0) diff --git a/bots/bot_digitador/bot_digitador/core/digitador_service.py b/bots/bot_digitador/bot_digitador/core/digitador_service.py index 643bf34..a431d36 100644 --- a/bots/bot_digitador/bot_digitador/core/digitador_service.py +++ b/bots/bot_digitador/bot_digitador/core/digitador_service.py @@ -377,45 +377,6 @@ class DigitadorService: except Exception as e: print(f"Error al guardar reportes: {e}") - # ------------------------------------------------------------------ - # Clipboard (sin tkinter) - # ------------------------------------------------------------------ - def limpiar_clipboard(self) -> None: - try: - user32 = ctypes.windll.user32 - if user32.OpenClipboard(0): - user32.EmptyClipboard() - user32.CloseClipboard() - except Exception: - pass - - def leer_clipboard(self) -> str: - try: - CF_UNICODETEXT = 13 - user32 = ctypes.windll.user32 - kernel32 = ctypes.windll.kernel32 - if not user32.OpenClipboard(0): - return "" - try: - handle = user32.GetClipboardData(CF_UNICODETEXT) - if not handle: - return "" - ptr = kernel32.GlobalLock(handle) - if not ptr: - return "" - try: - return ctypes.wstring_at(ptr).strip() - finally: - kernel32.GlobalUnlock(handle) - finally: - user32.CloseClipboard() - except Exception: - return "" - - def fecha_tiene_numeros(self, texto) -> bool: - texto = str(texto).strip() - return any(c.isdigit() for c in texto) - # ------------------------------------------------------------------ # Fechas / valores / docs # ------------------------------------------------------------------ @@ -676,20 +637,6 @@ class DigitadorService: "radicado_existente.png", ) - def check_diferencias_abierto_opencv(self, window) -> bool: - return self.check_template_opencv( - window, - self.template_diferencias_path, - "flecha_diferencias_abierto.png", - ) - - def check_diferencias_cerrado_opencv(self, window) -> bool: - return self.check_template_opencv( - window, - self.template_diferencias_cerrado_path, - "flecha_diferencias_cerrado.png", - ) - def check_success_opencv(self, window) -> bool: return self.check_template_opencv( window, diff --git a/docker-compose.yml b/docker-compose.yml index ffc12e9..11d9170 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -55,10 +55,10 @@ services: ASYNC_QUEUE: ${ASYNC_QUEUE:-false} EXTRACTOR_BASE_URL: ${EXTRACTOR_BASE_URL:-https://api.openai.com/v1} EXTRACTOR_API_KEY: ${EXTRACTOR_API_KEY} - EXTRACTOR_MODEL: ${EXTRACTOR_MODEL:-gpt-5.4-mini} + EXTRACTOR_MODEL: ${EXTRACTOR_MODEL:-gpt-5.6-luna} NORMALIZER_BASE_URL: ${NORMALIZER_BASE_URL:-${EXTRACTOR_BASE_URL:-https://api.openai.com/v1}} NORMALIZER_API_KEY: ${NORMALIZER_API_KEY:-${EXTRACTOR_API_KEY}} - NORMALIZER_MODEL: ${NORMALIZER_MODEL:-gpt-5.4-mini} + NORMALIZER_MODEL: ${NORMALIZER_MODEL:-gpt-5.6-luna} EXTRACTOR_TIMEOUT_S: ${EXTRACTOR_TIMEOUT_S:-120} PDF_DPI: ${PDF_DPI:-150} PDF_THREAD_COUNT: ${PDF_THREAD_COUNT:-4} @@ -95,10 +95,10 @@ services: ASYNC_QUEUE: ${ASYNC_QUEUE:-false} EXTRACTOR_BASE_URL: ${EXTRACTOR_BASE_URL:-https://api.openai.com/v1} EXTRACTOR_API_KEY: ${EXTRACTOR_API_KEY} - EXTRACTOR_MODEL: ${EXTRACTOR_MODEL:-gpt-5.4-mini} + EXTRACTOR_MODEL: ${EXTRACTOR_MODEL:-gpt-5.6-luna} NORMALIZER_BASE_URL: ${NORMALIZER_BASE_URL:-${EXTRACTOR_BASE_URL:-https://api.openai.com/v1}} NORMALIZER_API_KEY: ${NORMALIZER_API_KEY:-${EXTRACTOR_API_KEY}} - NORMALIZER_MODEL: ${NORMALIZER_MODEL:-gpt-5.4-mini} + NORMALIZER_MODEL: ${NORMALIZER_MODEL:-gpt-5.6-luna} EXTRACTOR_TIMEOUT_S: ${EXTRACTOR_TIMEOUT_S:-120} PDF_DPI: ${PDF_DPI:-150} PDF_THREAD_COUNT: ${PDF_THREAD_COUNT:-4} diff --git a/services/bot_descarga/api/main.py b/services/bot_descarga/api/main.py index 5454da6..e8c265f 100644 --- a/services/bot_descarga/api/main.py +++ b/services/bot_descarga/api/main.py @@ -1,6 +1,5 @@ from __future__ import annotations -import os from pathlib import Path from typing import Any, Dict, List, Optional diff --git a/services/bot_descarga/domain/models.py b/services/bot_descarga/domain/models.py index b65a94a..d1daae8 100644 --- a/services/bot_descarga/domain/models.py +++ b/services/bot_descarga/domain/models.py @@ -3,7 +3,7 @@ from __future__ import annotations from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from enum import Enum -from typing import Any, Dict, List, Optional +from typing import Any, Dict, Optional def utc_now() -> str: diff --git a/services/bot_descarga/infrastructure/storage_minio.py b/services/bot_descarga/infrastructure/storage_minio.py index 0d58712..e6717e3 100644 --- a/services/bot_descarga/infrastructure/storage_minio.py +++ b/services/bot_descarga/infrastructure/storage_minio.py @@ -4,7 +4,7 @@ import json import os from dataclasses import dataclass from pathlib import Path -from typing import BinaryIO, List, Optional +from typing import List, Optional @dataclass(frozen=True) @@ -114,10 +114,6 @@ class MinioStorage: path = Path(path) return self.put_bytes(key, path.read_bytes(), content_type=content_type) - def put_stream(self, key: str, stream: BinaryIO, content_type: str = "application/octet-stream") -> str: - data = stream.read() - return self.put_bytes(key, data, content_type=content_type) - def get_bytes(self, key: str) -> bytes: obj = self.client.get_object(Bucket=self.bucket, Key=key) return obj["Body"].read() diff --git a/services/bot_digitador/infrastructure/storage_minio.py b/services/bot_digitador/infrastructure/storage_minio.py index 4149ba8..fc54b94 100644 --- a/services/bot_digitador/infrastructure/storage_minio.py +++ b/services/bot_digitador/infrastructure/storage_minio.py @@ -4,7 +4,7 @@ import json import os from dataclasses import dataclass from pathlib import Path -from typing import BinaryIO, List, Optional +from typing import List, Optional @dataclass(frozen=True) @@ -117,10 +117,6 @@ class MinioStorage: path = Path(path) return self.put_bytes(key, path.read_bytes(), content_type=content_type) - def put_stream(self, key: str, stream: BinaryIO, content_type: str = "application/octet-stream") -> str: - data = stream.read() - return self.put_bytes(key, data, content_type=content_type) - def get_bytes(self, key: str) -> bytes: obj = self.client.get_object(Bucket=self.bucket, Key=key) return obj["Body"].read() diff --git a/services/cleaning/domain/csv_cleaning/analyze.py b/services/cleaning/domain/csv_cleaning/analyze.py index ee051f4..33ff6ac 100644 --- a/services/cleaning/domain/csv_cleaning/analyze.py +++ b/services/cleaning/domain/csv_cleaning/analyze.py @@ -9,7 +9,6 @@ import pandas as pd from .models import ( TYPE_CATEGORICAL, - TYPE_DATE, TYPE_DECIMAL, TYPE_INTEGER, TYPE_TEXT, diff --git a/services/cleaning/domain/csv_cleaning/column_hints.py b/services/cleaning/domain/csv_cleaning/column_hints.py index 342b95b..230560f 100644 --- a/services/cleaning/domain/csv_cleaning/column_hints.py +++ b/services/cleaning/domain/csv_cleaning/column_hints.py @@ -2,8 +2,6 @@ from __future__ import annotations import re -from typing import Optional - import pandas as pd from .nullish import _is_nullish_scalar diff --git a/services/cleaning/domain/csv_cleaning/rules/dates.py b/services/cleaning/domain/csv_cleaning/rules/dates.py index 07d645c..f910b4e 100644 --- a/services/cleaning/domain/csv_cleaning/rules/dates.py +++ b/services/cleaning/domain/csv_cleaning/rules/dates.py @@ -10,7 +10,6 @@ from ..dates import ( _normalize_messy_date_to_compact, _parse_any_date_to_tuple, _parse_date_with_format, - _parse_date_with_text_month, detect_date_format, ) from ..nullish import _is_nullish_scalar diff --git a/services/cleaning/infrastructure/database.py b/services/cleaning/infrastructure/database.py index e32fa0e..447e79d 100644 --- a/services/cleaning/infrastructure/database.py +++ b/services/cleaning/infrastructure/database.py @@ -15,7 +15,6 @@ from sqlalchemy import ( String, Text, create_engine, - select, text, ) from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, sessionmaker @@ -156,13 +155,3 @@ def session_scope() -> Iterator[Session]: raise finally: session.close() - - -def get_extraction_rows(session: Session, batch_id: str) -> list[ExtractionRow]: - return list( - session.scalars( - select(ExtractionRow) - .where(ExtractionRow.batch_id == batch_id) - .order_by(ExtractionRow.row_index, ExtractionRow.id) - ) - ) diff --git a/services/desktop_app/src/renderer/components/CountingWizard.tsx b/services/desktop_app/src/renderer/components/CountingWizard.tsx index 01f0be5..188c81e 100644 --- a/services/desktop_app/src/renderer/components/CountingWizard.tsx +++ b/services/desktop_app/src/renderer/components/CountingWizard.tsx @@ -2,10 +2,8 @@ import { AlertCircle, ArrowRight, CheckCircle2, - Clock, FileText, FolderOpen, - Hash, Loader2, Play, RotateCcw, diff --git a/services/desktop_app/src/renderer/components/ExtractionWizard.tsx b/services/desktop_app/src/renderer/components/ExtractionWizard.tsx index 64e2e1f..5b8347c 100644 --- a/services/desktop_app/src/renderer/components/ExtractionWizard.tsx +++ b/services/desktop_app/src/renderer/components/ExtractionWizard.tsx @@ -91,10 +91,10 @@ interface ExtractionWizardProps { } const MODELS = [ - { id: "gpt-5.4", label: "GPT-5.4", tier: "Máxima precisión", priceIn: 2.50, priceOut: 15.00 }, - { id: "gpt-5.4-mini", label: "GPT-5.4 Mini", tier: "Balance óptimo", priceIn: 0.75, priceOut: 4.50 }, - { id: "gpt-5.4-nano", label: "GPT-5.4 Nano", tier: "Más económico", priceIn: 0.20, priceOut: 0.80 }, - { id: "gpt-4.1-mini", label: "GPT-4.1 Mini", tier: "Legacy rápido", priceIn: 0.40, priceOut: 1.60 }, + { id: "gpt-5.6-luna", label: "GPT-5.6 Luna", tier: "Alto volumen", priceIn: 0.20, priceCached: 0.02, priceOut: 1.20 }, + { id: "gpt-5.4", label: "GPT-5.4", tier: "Máxima precisión", priceIn: 2.50, priceCached: 0.25, priceOut: 15.00 }, + { id: "gpt-5.4-mini", label: "GPT-5.4 Mini", tier: "Balance óptimo", priceIn: 0.75, priceCached: 0.075, priceOut: 4.50 }, + { id: "gpt-5.4-nano", label: "GPT-5.4 Nano", tier: "Más económico", priceIn: 0.20, priceCached: 0.02, priceOut: 0.80 }, ]; // Tipos del resultado del estimado que devuelve el backend @@ -413,14 +413,16 @@ export function ExtractionWizard({ {m.label} {m.tier} - ${m.priceIn}/${m.priceOut} + ${m.priceIn} / ${m.priceCached} / ${m.priceOut} /1M ))}

- El normalizador siempre usa GPT-5.4 Mini ($0.75 in / $4.50 out /1M tokens). + Precios: entrada / caché / salida por 1M tokens. +
+ El normalizador usa GPT-5.6 Luna por defecto ($0.20 entrada / $0.02 caché / $1.20 salida por 1M tokens).

@@ -733,7 +735,7 @@ export function ExtractionWizard({ const total = est.estimated_total_cost; const docs = est.total_docs ?? displayDocCount; const extModel = est.extractor_model ?? form.model; - const normModel = est.normalizer_model ?? "gpt-5.4-mini"; + const normModel = est.normalizer_model ?? "gpt-5.6-luna"; const extPrompt = extUsage?.prompt_tokens ?? 0; const extCompl = extUsage?.completion_tokens ?? 0; const extCached = extUsage?.prompt_tokens_details?.cached_tokens ?? 0; diff --git a/services/desktop_app/src/renderer/modules/counting/CountingResults.tsx b/services/desktop_app/src/renderer/modules/counting/CountingResults.tsx index b0becd7..c602ec3 100644 --- a/services/desktop_app/src/renderer/modules/counting/CountingResults.tsx +++ b/services/desktop_app/src/renderer/modules/counting/CountingResults.tsx @@ -9,7 +9,6 @@ export function CountingResults({ client, task, job, - onTaskUpdated, onActivateModule, }: ModuleResultsProps) { const [files, setFiles] = useState([]); diff --git a/services/desktop_app/src/renderer/modules/extraction/JobExtractionWizard.tsx b/services/desktop_app/src/renderer/modules/extraction/JobExtractionWizard.tsx index 38a04a9..10ad739 100644 --- a/services/desktop_app/src/renderer/modules/extraction/JobExtractionWizard.tsx +++ b/services/desktop_app/src/renderer/modules/extraction/JobExtractionWizard.tsx @@ -70,7 +70,7 @@ function makeDefaultForm(job: { metadata?: Record }, fromCount tiffEnabled: false, tiffStart: "0", tiffEnd: "0", - model: localStorage.getItem("extractorModel") || "gpt-5.4", + model: localStorage.getItem("extractorModel") || "gpt-5.6-luna", }; } diff --git a/services/extraction/README.md b/services/extraction/README.md index 10303b8..3d13548 100644 --- a/services/extraction/README.md +++ b/services/extraction/README.md @@ -85,10 +85,10 @@ count-api results → POST /count-jobs → count_job_files (P | `REDIS_URL` | Redis para cola ARQ (`redis://...`) | | `EXTRACTOR_BASE_URL` | URL base del LLM (compatible OpenAI) | | `EXTRACTOR_API_KEY` | API key del LLM extractor | -| `EXTRACTOR_MODEL` | Modelo de extracción (ej. `gpt-5.4-mini`) | +| `EXTRACTOR_MODEL` | Modelo de extracción (default: `gpt-5.6-luna`) | | `NORMALIZER_BASE_URL` | URL base del LLM normalizador | | `NORMALIZER_API_KEY` | API key del LLM normalizador | -| `NORMALIZER_MODEL` | Modelo normalizador (ej. `gpt-5.4-mini`) | +| `NORMALIZER_MODEL` | Modelo normalizador (default: `gpt-5.6-luna`) | | `EXTRACTOR_TIMEOUT_S` | Timeout por documento (default `120`) | | `PDF_DPI` | DPI para renderizado PDF (default `150`) | | `PDF_THREAD_COUNT` | Hilos de renderizado PDF (default `4`) | diff --git a/services/extraction/api/desktop_ingest.py b/services/extraction/api/desktop_ingest.py index b148706..2ba6fed 100644 --- a/services/extraction/api/desktop_ingest.py +++ b/services/extraction/api/desktop_ingest.py @@ -100,59 +100,43 @@ def _service_for_model(base_service: InvoiceProcessingService, model: str | None ) -def process_desktop_ingest( - service: InvoiceProcessingService, +def persist_desktop_ingest_result( session: Session, *, task: Task, batch_id: str, - temp_path: str, radicado: str, filename: str, doc_type: str | None, - instruction: str | None, - tiff_page_range: tuple[int, int] | None, - model: str | None = None, + request: dict[str, Any], + normalized_obj: Any, ) -> dict[str, Any]: - service = _service_for_model(service, model) - request = dict(task.request or {}) + """Persiste una extracción desktop y contabiliza una vez por radicado. + + Cuando una tarea separa los campos por tipo documental, cada documento se + guarda primero en ``partials``. El radicado se materializa y ``processed`` se + incrementa únicamente cuando llegaron todos sus tipos esperados. + """ fields = list(request.get("fields") or []) alias_map = dict(request.get("alias_map") or {}) structure_mode = str(request.get("structure_mode") or "radicados") doc_types = list(request.get("doc_types") or []) - prof = get_profile("invoice") - - selected_paths = _paths_for_type(fields, doc_type if structure_mode == "radicados" and doc_types else None) - if doc_type and structure_mode == "radicados" and doc_types: - try: - prompt = PromptBuilder.build_for_type(doc_type, selected_paths, {}, profile_name=prof.get("name", "invoice")) - except Exception: - prompt = PromptBuilder.build(selected_paths, {}, profile_name=prof.get("name", "invoice")) - else: - prompt = instruction or PromptBuilder.build(selected_paths, {}, profile_name=prof.get("name", "invoice")) - - data, _usage = service.extract_full_from_pdf( - temp_path, - instruction=prompt, - tiff_page_range=tiff_page_range, - ) - raw_data = data if isinstance(data, dict) else {"data": data} - if isinstance(data, dict): - data["radicado"] = radicado - else: - data = {"radicado": radicado, "data": data} - - norm = normalize_to_schema( - data, - paths=selected_paths, - profile_name=prof.get("name", "invoice"), - ) - normalized_obj = data - if isinstance(norm, dict): - normalized_obj = norm.get("normalized") or data - result = dict(task.result or {}) - errors = list(result.get("errors") or []) + failed_radicados = {str(value) for value in result.get("failed_radicados") or []} + + # Otro documento del mismo radicado puede terminar después de que uno de sus + # hermanos haya fallado. No debe recrear parciales ni contabilizarlo otra vez. + if radicado in failed_radicados: + return { + "radicado": radicado, + "filename": filename, + "doc_type": doc_type, + "finalized": False, + "skipped": True, + "reason": "radicado_failed", + "processed": int(result.get("processed") or 0), + "total": int(result.get("total") or 0), + } if structure_mode == "radicados" and doc_types and doc_type: partials = dict(result.get("partials") or {}) @@ -216,6 +200,113 @@ def process_desktop_ingest( } +def record_desktop_ingest_error( + session: Session, + *, + task: Task, + radicado: str, + filename: str, + doc_type: str | None, + error: Exception, +) -> dict[str, Any]: + """Registra un fallo y da por procesado el radicado exactamente una vez.""" + result = dict(task.result or {}) + errors = list(result.get("errors") or []) + errors.append({ + "radicado": radicado, + "filename": filename, + "doc_type": doc_type, + "error": str(error), + }) + result["errors"] = errors + + failed_radicados = {str(value) for value in result.get("failed_radicados") or []} + first_failure = radicado not in failed_radicados + if first_failure: + failed_radicados.add(radicado) + result["processed"] = int(result.get("processed") or 0) + 1 + + # Un radicado fallido ya no puede completarse con los parciales restantes. + partials = dict(result.get("partials") or {}) + partials.pop(radicado, None) + result["partials"] = partials + result["failed_radicados"] = sorted(failed_radicados) + task.result = result + task.updated_at = utcnow() + session.flush() + + return { + "radicado": radicado, + "filename": filename, + "doc_type": doc_type, + "error": str(error), + "first_failure": first_failure, + "processed": int(result.get("processed") or 0), + "total": int(result.get("total") or 0), + } + + +def process_desktop_ingest( + service: InvoiceProcessingService, + session: Session, + *, + task: Task, + batch_id: str, + temp_path: str, + radicado: str, + filename: str, + doc_type: str | None, + instruction: str | None, + tiff_page_range: tuple[int, int] | None, + model: str | None = None, +) -> dict[str, Any]: + service = _service_for_model(service, model) + request = dict(task.request or {}) + fields = list(request.get("fields") or []) + structure_mode = str(request.get("structure_mode") or "radicados") + doc_types = list(request.get("doc_types") or []) + prof = get_profile("invoice") + + selected_paths = _paths_for_type(fields, doc_type if structure_mode == "radicados" and doc_types else None) + if doc_type and structure_mode == "radicados" and doc_types: + try: + prompt = PromptBuilder.build_for_type(doc_type, selected_paths, {}, profile_name=prof.get("name", "invoice")) + except Exception: + prompt = PromptBuilder.build(selected_paths, {}, profile_name=prof.get("name", "invoice")) + else: + prompt = instruction or PromptBuilder.build(selected_paths, {}, profile_name=prof.get("name", "invoice")) + + data, _usage = service.extract_full_from_pdf( + temp_path, + instruction=prompt, + tiff_page_range=tiff_page_range, + ) + if isinstance(data, dict): + data["radicado"] = radicado + else: + data = {"radicado": radicado, "data": data} + + norm = normalize_to_schema( + data, + paths=selected_paths, + profile_name=prof.get("name", "invoice"), + ) + normalized_obj = data + if isinstance(norm, dict): + normalized_obj = norm.get("normalized") or data + + return persist_desktop_ingest_result( + session, + task=task, + batch_id=batch_id, + radicado=radicado, + filename=filename, + doc_type=doc_type, + request=request, + normalized_obj=normalized_obj, + ) + + def mark_task_completed_if_done(session: Session, task: Task, batch_id: str) -> None: result = dict(task.result or {}) processed = int(result.get("processed") or 0) diff --git a/services/extraction/api/main.py b/services/extraction/api/main.py index 9baba48..0c9d186 100644 --- a/services/extraction/api/main.py +++ b/services/extraction/api/main.py @@ -20,7 +20,11 @@ from fastapi import BackgroundTasks, FastAPI, File, Form, HTTPException, Respons from pydantic import BaseModel, Field, model_validator from sqlalchemy import delete, select -from api.desktop_ingest import mark_task_completed_if_done, process_desktop_ingest +from api.desktop_ingest import ( + mark_task_completed_if_done, + persist_desktop_ingest_result, + record_desktop_ingest_error, +) from domain.batch_processor import BatchProcessor from domain.config import extractor_llm_config, normalizer_llm_config, settings @@ -612,10 +616,6 @@ async def _process_sync( import asyncio as _asyncio import logging as _logging import tempfile as _tempfile - from domain.service import InvoiceProcessingService - from domain.strategies import OpenAICompatibleExtractor - from domain.batch_processor import BatchProcessor - from domain.estimator import Estimator from api.desktop_ingest import _service_for_model, _paths_for_type from domain.profiles import get_profile from domain.prompts import PromptBuilder @@ -678,58 +678,61 @@ async def _process_sync( # ── Fase 2: persistir resultado (con FOR UPDATE para serializar) ─── def _persist(): - from infrastructure.serialization import flatten_extraction_result - from infrastructure.database import ExtractionRow, ExtractionBatch, utcnow - req = dict(request) - fields = list(req.get("fields") or []) - alias_map = dict(req.get("alias_map") or {}) with session_scope() as session: task = session.get(Task, task_id, with_for_update=True) if task is None: return {"error": "task_not_found"} - result = dict(task.result or {}) - fused = {"radicado": radicado, "_source_pdf": effective_name} - if doc_type: - fused[doc_type] = normalized_obj or {} - else: - fused.update( - normalized_obj if isinstance(normalized_obj, dict) - else {"data": normalized_obj} - ) - row_index = int(result.get("row_count") or 0) - for row in flatten_extraction_result(fused, fields, alias_map): - row_index += 1 - session.add(ExtractionRow( - batch_id=batch_id, - radicado=radicado, - source_pdf=effective_name, - row_index=row_index, - row_json=row, - raw_json=fused, - )) - result["row_count"] = row_index - processed_now = int(result.get("processed") or 0) + 1 - result["processed"] = processed_now - task.result = result - task.updated_at = utcnow() - session.flush() + payload = persist_desktop_ingest_result( + session, + task=task, + batch_id=batch_id, + radicado=radicado, + filename=effective_name, + doc_type=doc_type, + request=request, + normalized_obj=normalized_obj, + ) mark_task_completed_if_done(session, task, batch_id) session.flush() # persiste status=completed / finished_at si aplica + processed_now = int((task.result or {}).get("processed") or 0) _log.info( "✓ guardado radicado=%s processed=%d/%d status=%s", - radicado, processed_now, int(result.get("total") or 0), task.status, + radicado, processed_now, int((task.result or {}).get("total") or 0), task.status, ) session.refresh(task) - return { - "radicado": radicado, - "filename": effective_name, - "processed": processed_now, - "total": int(result.get("total") or 0), - "task": serialize_task(task), - } + return {**payload, "task": serialize_task(task)} return await _asyncio.to_thread(_persist) + except Exception as exc: + # La llamada HTTP conserva el error, pero el progreso debe quedar persistido + # para que un documento fallido no deje la tarea eternamente en running. + def _record_error(error: Exception) -> None: + with session_scope() as session: + task = session.get(Task, task_id, with_for_update=True) + if task is None: + return + record_desktop_ingest_error( + session, + task=task, + radicado=radicado, + filename=effective_name, + doc_type=doc_type, + error=error, + ) + mark_task_completed_if_done(session, task, batch_id) + session.flush() + + try: + await _asyncio.to_thread(_record_error, exc) + except Exception: + _log.exception( + "No se pudo registrar el error de radicado=%s filename=%s", + radicado, + effective_name, + ) + raise + finally: try: os.unlink(temp_path) diff --git a/services/extraction/core/pdf_utils.py b/services/extraction/core/pdf_utils.py index 9d59eab..c052e36 100644 --- a/services/extraction/core/pdf_utils.py +++ b/services/extraction/core/pdf_utils.py @@ -7,7 +7,7 @@ from io import BytesIO from typing import List, Tuple, Optional import time -from pdf2image import convert_from_path, pdfinfo_from_path +from pdf2image import convert_from_path from PIL import Image, ImageSequence Image.MAX_IMAGE_PIXELS = None @@ -80,86 +80,6 @@ def pdf_to_images_bs64( return resultados -def get_pdf_num_pages(pdf_path: str, poppler_path: str | None = None) -> int: - """ - Obtiene el número total de páginas de un PDF. - - Args: - pdf_path: Ruta al archivo PDF. - poppler_path: Ruta personalizada a los binarios de Poppler (opcional). - - Returns: - Número de páginas (0 si ocurre un error). - """ - info = pdfinfo_from_path(pdf_path, poppler_path=_resolve_poppler(poppler_path)) - try: - return int(info.get('Pages') or info.get('pages') or 0) - except Exception: - return 0 - - -def pdf_pages_to_images_bs64( - pdf_path: str, - first_page: int, - last_page: int, - *, - dpi: int, - grayscale: bool = True, - validate: bool = False, - thread_count: int = 10, - poppler_path: str | None = None, -) -> List[str]: - """ - Convierte un rango específico de páginas de un PDF a imágenes base64. - - Args: - pdf_path: Ruta al archivo PDF. - first_page: Índice de la primera página (1-indexed). - last_page: Índice de la última página (inclusivo). - dpi: Resolución de la imagen. - grayscale: Si es True, convierte las imágenes a escala de grises para ahorrar ancho de banda/tokens. - validate: Si es True, verifica la integridad de las imágenes generadas. - thread_count: Número de hilos para procesamiento paralelo. - poppler_path: Ruta personalizada a Poppler. - - Returns: - Lista de imágenes en base64. - """ - poppler_bin = _resolve_poppler(poppler_path) - t0 = time.time() - logger.debug( - f"Convirtiendo rango PDF→imgs (first={first_page}, last={last_page}, dpi={dpi}, gray={grayscale})", - extra={"pdf": pdf_path, "dpi": dpi}, - ) - images = convert_from_path( - pdf_path, - dpi=dpi, - thread_count=thread_count, - poppler_path=poppler_bin, - first_page=first_page, - last_page=last_page, - ) - resultados: List[str] = [] - for idx, img in enumerate(images, start=first_page): - if grayscale: - img = img.convert('L') - buffer = BytesIO() - img.save(buffer, format="PNG") - b64 = base64.b64encode(buffer.getvalue()).decode('utf-8') - if validate: - try: - raw = base64.b64decode(b64, validate=True) - with Image.open(BytesIO(raw)) as im: - im.verify() - except Exception as error: - raise ValueError(f"Imagen base64 inválida en la página {idx}: {error}") - resultados.append(b64) - logger.debug( - f"Rango convertido (count={len(resultados)}, ms={int((time.time()-t0)*1000)})", - extra={"pdf": pdf_path, "dpi": dpi}, - ) - return resultados - def tiff_to_images_bs64(tiff_path: str, validate: bool = True, pages_range: Optional[Tuple[int, int]] = (2, 3)) -> List[str]: """ Convierte un archivo TIFF multipágina a una lista de imágenes PNG en base64. diff --git a/services/extraction/core/text_utils.py b/services/extraction/core/text_utils.py index 8e4ffab..5a8a412 100644 --- a/services/extraction/core/text_utils.py +++ b/services/extraction/core/text_utils.py @@ -1,6 +1,5 @@ """Utilidades de texto para parseo de JSON embebido y construcción de schema.""" import re -import json from typing import Optional, Dict, Any diff --git a/services/extraction/core/utils.py b/services/extraction/core/utils.py index 4096c32..ad86e9a 100644 --- a/services/extraction/core/utils.py +++ b/services/extraction/core/utils.py @@ -12,11 +12,6 @@ from typing import List, Dict ALLOWED_DOC_EXTS = (".pdf", ".tif", ".tiff") -def parse_comma_separated(value: str) -> List[str]: - """Convierte una cadena coma-separada en lista normalizada (sin vacíos).""" - return [v.strip() for v in value.split(',') if v and v.strip()] - - def _matches_pattern(filename: str, pattern: str) -> bool: """Determina si `filename` coincide con `pattern` usando semántica tipo Windows. diff --git a/services/extraction/domain/config.py b/services/extraction/domain/config.py index 58be95e..97df53b 100644 --- a/services/extraction/domain/config.py +++ b/services/extraction/domain/config.py @@ -6,7 +6,6 @@ Se lee automáticamente un archivo `.env` en la raíz del proyecto si existe. from __future__ import annotations from typing import Optional, Tuple -import logging import os # Cargar variables desde .env en el entorno (para que otros módulos que usan os.environ las vean) @@ -31,12 +30,12 @@ class ApiSettings(BaseSettings): extractor_provider: str = Field("custom", validation_alias=AliasChoices("EXTRACTOR_PROVIDER")) extractor_base_url: str = Field("", validation_alias=AliasChoices("EXTRACTOR_BASE_URL")) extractor_api_key: str = Field("", validation_alias=AliasChoices("EXTRACTOR_API_KEY")) - extractor_model: str = Field("", validation_alias=AliasChoices("EXTRACTOR_MODEL")) + extractor_model: str = Field("gpt-5.6-luna", validation_alias=AliasChoices("EXTRACTOR_MODEL")) normalizer_provider: str = Field("custom", validation_alias=AliasChoices("NORMALIZER_PROVIDER")) normalizer_base_url: str = Field("", validation_alias=AliasChoices("NORMALIZER_BASE_URL")) normalizer_api_key: str = Field("", validation_alias=AliasChoices("NORMALIZER_API_KEY")) - normalizer_model: str = Field("", validation_alias=AliasChoices("NORMALIZER_MODEL")) + normalizer_model: str = Field("gpt-5.6-luna", validation_alias=AliasChoices("NORMALIZER_MODEL")) request_timeout_s: int = Field(120, validation_alias=AliasChoices("EXTRACTOR_TIMEOUT_S", "LLM_TIMEOUT_S")) normalizer_timeout_s: int = Field(0, validation_alias=AliasChoices("NORMALIZER_TIMEOUT_S")) diff --git a/services/extraction/domain/costs.py b/services/extraction/domain/costs.py index a1aa85a..95d356a 100644 --- a/services/extraction/domain/costs.py +++ b/services/extraction/domain/costs.py @@ -5,12 +5,12 @@ Define precios por modelo y funciones para estimar costos a partir del uso de to """ from typing import Dict, Any, Optional -# Precios actualizados por 1M tokens (USD) - Enero 2026 +# Precios estándar por 1M tokens (USD) - Agosto 2026 MODEL_PRICING = { - "gpt-4.1-mini": { - "input": 0.40, - "cached_input": 0.10, - "output": 1.60, + "gpt-5.6-luna": { + "input": 0.20, + "cached_input": 0.02, + "output": 1.20, }, "gpt-5.4": { "input": 2.50, diff --git a/services/extraction/domain/estimator.py b/services/extraction/domain/estimator.py index 5de6340..db890e8 100644 --- a/services/extraction/domain/estimator.py +++ b/services/extraction/domain/estimator.py @@ -2,7 +2,6 @@ Encapsula la lógica de estimación previa al procesamiento. """ -import os from typing import List, Dict, Any, Optional, TYPE_CHECKING if TYPE_CHECKING: @@ -150,7 +149,7 @@ class Estimator: # Enriquecer con modelo efectivo y conteo de páginas para el frontend effective_model = model or getattr(service.extractor, 'model', None) or getattr(self.service.extractor, 'model', None) result['extractor_model'] = effective_model - result['normalizer_model'] = settings.api.normalizer_model or 'gpt-5.4-mini' + result['normalizer_model'] = settings.api.normalizer_model or 'gpt-5.6-luna' # Estimar numero de paginas a partir de prompt_tokens de imagen: # GPT-4o cobra ~85 tok base + ~170 tok por tile 512x512. diff --git a/services/extraction/domain/llm_client.py b/services/extraction/domain/llm_client.py index 185d372..08710f4 100644 --- a/services/extraction/domain/llm_client.py +++ b/services/extraction/domain/llm_client.py @@ -207,7 +207,6 @@ class LLMClient: elapsed_ms = int((time.time() - start) * 1000) self._log_error(component=component, model=eff_model, timeout_s=eff_timeout, elapsed_ms=elapsed_ms, error=ex) raise - async def chat_text_async( self, messages: List[Dict[str, Any]], @@ -236,7 +235,6 @@ class LLMClient: elapsed_ms = int((time.time() - start) * 1000) self._log_error(component=component, model=eff_model, timeout_s=eff_timeout, elapsed_ms=elapsed_ms, error=ex) raise - def chat_json( self, messages: List[Dict[str, Any]], @@ -276,41 +274,5 @@ class LLMClient: self._log_error(component=component, model=eff_model, timeout_s=eff_timeout, elapsed_ms=elapsed_ms, error=ex) raise - async def chat_json_async( - self, - messages: List[Dict[str, Any]], - *, - model: Optional[str] = None, - response_format: str = "json_object", - timeout_s: Optional[float] = None, - component: str = "llm_client", - max_completion_tokens: Optional[int] = None, - ) -> Tuple[Dict[str, Any], Optional[Dict[str, Any]]]: - """Devuelve (objeto JSON parseado, usage) de forma asíncrona.""" - import json - eff_model = model or self.default_model - eff_timeout = float(timeout_s if timeout_s is not None else self.request_timeout_s) - self._log_start(component=component, model=eff_model, timeout_s=eff_timeout, messages_count=len(messages or [])) - start = time.time() - try: - completion_kwargs: Dict[str, Any] = { - "response_format": {"type": response_format}, - } - if max_completion_tokens is not None: - completion_kwargs["max_completion_tokens"] = max_completion_tokens - chat_completion = await self._create_completion_async( - messages=messages, - model=eff_model, - timeout_s=eff_timeout, - **completion_kwargs, - ) - elapsed_ms = int((time.time() - start) * 1000) - usage = self._extract_usage(chat_completion) - self._log_end(component=component, model=eff_model, timeout_s=eff_timeout, elapsed_ms=elapsed_ms, usage=usage) - content = chat_completion.choices[0].message.content # type: ignore[assignment] - obj = json.loads(content) - return obj, usage - except Exception as ex: - elapsed_ms = int((time.time() - start) * 1000) - self._log_error(component=component, model=eff_model, timeout_s=eff_timeout, elapsed_ms=elapsed_ms, error=ex) - raise + +__all__ = ["LLMClient"] diff --git a/services/extraction/domain/service.py b/services/extraction/domain/service.py index dd373d3..007564c 100644 --- a/services/extraction/domain/service.py +++ b/services/extraction/domain/service.py @@ -21,7 +21,6 @@ from .batch_processor import BatchProcessor from .estimator import Estimator from core.pdf_utils import document_to_images_bs64 from core.text_utils import extract_json_from_text, build_schema -from core.utils import find_first_matching_file logger = logging.getLogger(__name__) @@ -49,11 +48,6 @@ class InvoiceProcessingService: # Lock para escrituras concurrentes al mismo CSV desde múltiples hilos # (Mantenido por retrocompatibilidad si se accede directo, aunque BatchProcessor usa su propio CSVManager) - try: - import threading - self._csv_lock = threading.Lock() - except Exception: - self._csv_lock = None def _convert_to_images(self, pdf_path: str, dpi: int, thread_count: int, tiff_pages: Optional[Tuple[int, int]], tiff_page_range: Optional[Tuple[int, int]]) -> List[str]: """ diff --git a/services/extraction/infrastructure/database.py b/services/extraction/infrastructure/database.py index 46c8a01..87ce646 100644 --- a/services/extraction/infrastructure/database.py +++ b/services/extraction/infrastructure/database.py @@ -16,7 +16,6 @@ from sqlalchemy import ( String, Text, create_engine, - select, text, ) from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, sessionmaker @@ -187,13 +186,3 @@ def session_scope() -> Iterator[Session]: raise finally: session.close() - - -def get_extraction_rows(session: Session, batch_id: str) -> list[ExtractionRow]: - return list( - session.scalars( - select(ExtractionRow) - .where(ExtractionRow.batch_id == batch_id) - .order_by(ExtractionRow.row_index, ExtractionRow.id) - ) - ) diff --git a/services/extraction/tests/test_config.py b/services/extraction/tests/test_config.py new file mode 100644 index 0000000..8c05509 --- /dev/null +++ b/services/extraction/tests/test_config.py @@ -0,0 +1,11 @@ +from services.extraction.domain.config import ApiSettings + + +def test_luna_is_the_default_model_for_extraction_and_normalization(monkeypatch) -> None: + monkeypatch.delenv("EXTRACTOR_MODEL", raising=False) + monkeypatch.delenv("NORMALIZER_MODEL", raising=False) + + api = ApiSettings(_env_file=None) + + assert api.extractor_model == "gpt-5.6-luna" + assert api.normalizer_model == "gpt-5.6-luna" diff --git a/services/extraction/tests/test_costs.py b/services/extraction/tests/test_costs.py new file mode 100644 index 0000000..e3cbdcc --- /dev/null +++ b/services/extraction/tests/test_costs.py @@ -0,0 +1,18 @@ +from services.extraction.domain.costs import compute_cost_from_usage + + +def test_gpt_5_6_luna_pricing_includes_cached_tokens() -> None: + usage = { + "prompt_tokens": 2_000_000, + "completion_tokens": 1_000_000, + "prompt_tokens_details": {"cached_tokens": 1_000_000}, + } + + cost = compute_cost_from_usage(usage, model="gpt-5.6-luna") + + assert cost == { + "input_usd": 0.20, + "cached_input_usd": 0.02, + "output_usd": 1.20, + "total_usd": 1.42, + } diff --git a/services/extraction/tests/test_desktop_ingest.py b/services/extraction/tests/test_desktop_ingest.py new file mode 100644 index 0000000..e84f2ca --- /dev/null +++ b/services/extraction/tests/test_desktop_ingest.py @@ -0,0 +1,228 @@ +"""Regresiones de persistencia para ingestas desktop síncronas.""" +from __future__ import annotations + +import asyncio +from contextlib import contextmanager +from types import SimpleNamespace + +import pytest + +from api import desktop_ingest +from api import main as extraction_api + + +class FakeSession: + def __init__(self) -> None: + self.added: list[object] = [] + self.batch = SimpleNamespace(status="running", updated_at=None) + + def add(self, value: object) -> None: + self.added.append(value) + + def flush(self) -> None: + pass + + def get(self, _model: object, _identifier: str) -> object: + return self.batch + + def refresh(self, _value: object) -> None: + pass + + +def test_multiple_document_types_complete_once_per_radicado(monkeypatch) -> None: + monkeypatch.setattr( + desktop_ingest, + "flatten_extraction_result", + lambda fused, _fields, _aliases: [fused], + ) + session = FakeSession() + request = { + "structure_mode": "radicados", + "fields": ["factura/numero", "soporte/codigo"], + "doc_types": [{"name": "factura"}, {"name": "soporte"}], + "documents": [ + {"radicado": "RAD-1", "filename": "factura.pdf", "doc_type": "factura"}, + {"radicado": "RAD-1", "filename": "soporte.pdf", "doc_type": "soporte"}, + ], + } + task = SimpleNamespace( + result={"processed": 0, "total": 1, "row_count": 0, "partials": {}}, + status="running", + finished_at=None, + updated_at=None, + ) + + first = desktop_ingest.persist_desktop_ingest_result( + session, + task=task, + batch_id="batch-1", + radicado="RAD-1", + filename="factura.pdf", + doc_type="factura", + request=request, + normalized_obj={"numero": "F-1"}, + ) + desktop_ingest.mark_task_completed_if_done(session, task, "batch-1") + + assert first["finalized"] is False + assert task.result["processed"] == 0 + assert task.status == "running" + assert task.result["partials"] == {"RAD-1": {"factura": {"numero": "F-1"}}} + assert session.added == [] + + second = desktop_ingest.persist_desktop_ingest_result( + session, + task=task, + batch_id="batch-1", + radicado="RAD-1", + filename="soporte.pdf", + doc_type="soporte", + request=request, + normalized_obj={"codigo": "S-1"}, + ) + desktop_ingest.mark_task_completed_if_done(session, task, "batch-1") + + assert second["finalized"] is True + assert task.result["processed"] == 1 + assert task.result["partials"] == {} + assert task.status == "completed" + assert session.batch.status == "completed" + assert len(session.added) == 1 + row = session.added[0] + assert row.radicado == "RAD-1" + assert row.raw_json["factura"] == {"numero": "F-1"} + assert row.raw_json["soporte"] == {"codigo": "S-1"} + + +def test_document_failure_completes_radicado_and_discards_partials(monkeypatch) -> None: + monkeypatch.setattr( + desktop_ingest, + "flatten_extraction_result", + lambda fused, _fields, _aliases: [fused], + ) + session = FakeSession() + request = { + "structure_mode": "radicados", + "fields": ["factura/numero", "soporte/codigo"], + "doc_types": [{"name": "factura"}, {"name": "soporte"}], + "documents": [ + {"radicado": "RAD-1", "filename": "factura.pdf", "doc_type": "factura"}, + {"radicado": "RAD-1", "filename": "soporte.pdf", "doc_type": "soporte"}, + ], + } + task = SimpleNamespace( + result={"processed": 0, "total": 1, "row_count": 0, "partials": {}}, + status="running", + finished_at=None, + updated_at=None, + ) + + desktop_ingest.persist_desktop_ingest_result( + session, + task=task, + batch_id="batch-1", + radicado="RAD-1", + filename="factura.pdf", + doc_type="factura", + request=request, + normalized_obj={"numero": "F-1"}, + ) + failure = desktop_ingest.record_desktop_ingest_error( + session, + task=task, + radicado="RAD-1", + filename="soporte.pdf", + doc_type="soporte", + error=RuntimeError("backend no disponible"), + ) + desktop_ingest.mark_task_completed_if_done(session, task, "batch-1") + + assert failure["first_failure"] is True + assert task.result["processed"] == 1 + assert task.result["partials"] == {} + assert task.result["failed_radicados"] == ["RAD-1"] + assert task.result["errors"] == [{ + "radicado": "RAD-1", + "filename": "soporte.pdf", + "doc_type": "soporte", + "error": "backend no disponible", + }] + assert task.status == "completed" + assert session.added == [] + + late_success = desktop_ingest.persist_desktop_ingest_result( + session, + task=task, + batch_id="batch-1", + radicado="RAD-1", + filename="soporte.pdf", + doc_type="soporte", + request=request, + normalized_obj={"codigo": "S-1"}, + ) + duplicate_failure = desktop_ingest.record_desktop_ingest_error( + session, + task=task, + radicado="RAD-1", + filename="factura.pdf", + doc_type="factura", + error=RuntimeError("otro fallo tardío"), + ) + + assert late_success["skipped"] is True + assert duplicate_failure["first_failure"] is False + assert task.result["processed"] == 1 + assert task.result["partials"] == {} + assert session.added == [] + + +def test_process_sync_persists_failure_before_reraising(monkeypatch) -> None: + task = SimpleNamespace( + result={"processed": 0, "total": 1, "row_count": 0, "partials": {}}, + status="running", + finished_at=None, + updated_at=None, + ) + session = FakeSession() + + def get(model: object, _identifier: str, **_kwargs: object) -> object: + if model is extraction_api.Task: + return task + return session.batch + + session.get = get # type: ignore[method-assign] + + @contextmanager + def fake_session_scope(): + yield session + + class FailingService: + extractor = SimpleNamespace(model="test-model") + + def extract_full_from_pdf(self, *_args: object, **_kwargs: object) -> object: + raise RuntimeError("fallo simulado") + + monkeypatch.setattr(extraction_api, "session_scope", fake_session_scope) + monkeypatch.setattr(extraction_api, "processing_service", FailingService()) + + with pytest.raises(RuntimeError, match="fallo simulado"): + asyncio.run(extraction_api._process_sync( + "task-1", + "batch-1", + b"pdf", + "factura.pdf", + "RAD-1", + "factura", + { + "fields": ["factura/numero"], + "doc_types": [{"name": "factura"}], + "documents": [ + {"radicado": "RAD-1", "filename": "factura.pdf", "doc_type": "factura"}, + ], + }, + )) + + assert task.result["processed"] == 1 + assert task.result["failed_radicados"] == ["RAD-1"] + assert task.result["errors"][0]["error"] == "fallo simulado" + assert task.status == "completed" diff --git a/services/extraction/tests/test_worker_model.py b/services/extraction/tests/test_worker_model.py new file mode 100644 index 0000000..97a1485 --- /dev/null +++ b/services/extraction/tests/test_worker_model.py @@ -0,0 +1,114 @@ +"""Regresiones de selección de modelo en trabajos ARQ.""" +from __future__ import annotations + +import asyncio +from contextlib import contextmanager +from types import SimpleNamespace + +from worker import tasks + + +class FakeSession: + def __init__(self, task: object) -> None: + self.task = task + + def get(self, _model: object, _identifier: str, **_kwargs: object) -> object: + return self.task + + def add(self, _value: object) -> None: + pass + + def flush(self) -> None: + pass + + +class FakeRateLimiter: + async def acquire(self, *, estimated_tokens: int) -> None: + assert estimated_tokens >= 2000 + + +class FakeService: + def __init__(self, model: str) -> None: + self.extractor = SimpleNamespace(model=model) + self.calls = 0 + + def extract_full_from_pdf(self, *_args: object, **_kwargs: object): + self.calls += 1 + return {"numero": "F-1"}, {} + + +def test_process_document_uses_requested_model(monkeypatch, tmp_path) -> None: + task = SimpleNamespace( + request={ + "model": "modelo-elegido", + "fields": ["numero"], + "structure_mode": "general", + "doc_types": [], + }, + result={"processed": 0, "total": 1, "row_count": 0}, + updated_at=None, + ) + session = FakeSession(task) + + @contextmanager + def fake_session_scope(): + yield session + + base_service = FakeService("modelo-global") + selected_service = FakeService("modelo-elegido") + selected_models: list[str] = [] + + def fake_service_for_model(_base: object, model: str): + selected_models.append(model) + return selected_service + + monkeypatch.setattr(tasks, "session_scope", fake_session_scope) + monkeypatch.setattr(tasks, "get_rate_limiter", lambda: FakeRateLimiter()) + monkeypatch.setattr(tasks, "_service_for_model", fake_service_for_model) + monkeypatch.setattr(tasks, "normalize_to_schema", lambda data, **_kwargs: {"normalized": data}) + monkeypatch.setattr(tasks, "flatten_extraction_result", lambda *_args, **_kwargs: []) + monkeypatch.setattr(tasks, "flag_modified", lambda *_args, **_kwargs: None) + monkeypatch.setattr(tasks, "mark_task_completed_if_done", lambda *_args, **_kwargs: None) + + queue_file = tmp_path / "factura.pdf" + queue_file.write_bytes(b"pdf") + ctx = {"service": base_service} + + asyncio.run(tasks.process_document( + ctx, + task_id="task-1", + batch_id="batch-1", + queue_path=str(queue_file), + radicado="RAD-1", + filename="factura.pdf", + doc_type=None, + instruction=None, + tiff_page_range=None, + )) + + assert selected_models == ["modelo-elegido"] + assert selected_service.calls == 1 + assert base_service.calls == 0 + assert ctx["services_by_model"]["modelo-elegido"] is selected_service + + +def test_service_for_request_reuses_cached_model(monkeypatch) -> None: + base_service = FakeService("modelo-global") + selected_service = FakeService("modelo-elegido") + builds = 0 + + def fake_service_for_model(_base: object, _model: str): + nonlocal builds + builds += 1 + return selected_service + + monkeypatch.setattr(tasks, "_service_for_model", fake_service_for_model) + ctx = {"service": base_service} + request = {"model": "modelo-elegido"} + + first = tasks._service_for_request(ctx, request) + second = tasks._service_for_request(ctx, request) + + assert first is selected_service + assert second is selected_service + assert builds == 1 diff --git a/services/extraction/worker/tasks.py b/services/extraction/worker/tasks.py index 252ea70..ff00230 100644 --- a/services/extraction/worker/tasks.py +++ b/services/extraction/worker/tasks.py @@ -17,6 +17,7 @@ from infrastructure.serialization import flatten_extraction_result from api.desktop_ingest import ( _finalize_radicado_if_ready, _paths_for_type, + _service_for_model, mark_task_completed_if_done, ) from domain.profiles import get_profile @@ -27,6 +28,27 @@ from domain.rate_limiter import get_rate_limiter logger = logging.getLogger(__name__) +def _service_for_request( + ctx: dict[str, Any], request: dict[str, Any] +) -> Any: + """Resuelve y reutiliza el servicio correspondiente al modelo solicitado.""" + base_service = ctx["service"] + requested_model = str(request.get("model") or "").strip() or None + if not requested_model: + return base_service + + current_model = getattr(getattr(base_service, "extractor", None), "model", None) + if current_model == requested_model: + return base_service + + services_by_model = ctx.setdefault("services_by_model", {}) + service = services_by_model.get(requested_model) + if service is None: + service = _service_for_model(base_service, requested_model) + services_by_model[requested_model] = service + return service + + async def process_document( ctx: dict[str, Any], *, @@ -52,7 +74,6 @@ async def process_document( instruction: Instrucción de extracción personalizada. tiff_page_range: Rango de páginas TIFF [inicio, fin]. """ - service = ctx["service"] rate_limiter = get_rate_limiter() tiff_range = tuple(tiff_page_range) if tiff_page_range else None @@ -72,6 +93,14 @@ async def process_document( logger.warning("Tarea %s no encontrada — saltando %s", task_id, filename) return {"skipped": True, "reason": "task_not_found"} + service = _service_for_request(ctx, request) + effective_model = getattr(getattr(service, "extractor", None), "model", "?") + logger.info( + "Modelo resuelto para task=%s: %s", + task_id, + effective_model, + ) + fields: list[str] = list(request.get("fields") or []) alias_map: dict[str, str] = dict(request.get("alias_map") or {}) structure_mode: str = str(request.get("structure_mode") or "radicados") diff --git a/services/poblacion_apta/domain/verificador_st/core.py b/services/poblacion_apta/domain/verificador_st/core.py index e68b0e2..85a59b7 100644 --- a/services/poblacion_apta/domain/verificador_st/core.py +++ b/services/poblacion_apta/domain/verificador_st/core.py @@ -3,9 +3,6 @@ from __future__ import annotations import re -from pathlib import Path -from typing import Any - import pandas as pd COLUMNAS_FINAL_IT = [ @@ -164,48 +161,6 @@ def normalizar_nombre_columna(nombre): return s -def detectar_delimitador(path): - try: - with open(path, "r", encoding="utf-8-sig", errors="ignore") as f: - muestra = f.read(4096) - delimitadores = ["|", ";", ",", "\t"] - conteos = {d: muestra.count(d) for d in delimitadores} - mejor = max(conteos, key=conteos.get) - return mejor if conteos[mejor] > 0 else "," - except Exception: - return "," - - -def leer_csv_flexible(path): - sep = detectar_delimitador(path) - intentos = [ - {"sep": sep, "encoding": "utf-8-sig"}, - {"sep": sep, "encoding": "latin1"}, - {"sep": ";", "encoding": "utf-8-sig"}, - {"sep": ",", "encoding": "utf-8-sig"}, - {"sep": "|", "encoding": "utf-8-sig"}, - ] - - ultimo_error = None - for opt in intentos: - try: - df = pd.read_csv( - path, - sep=opt["sep"], - encoding=opt["encoding"], - dtype=str, - engine="python", - keep_default_na=False, - on_bad_lines="skip", - ) - df.columns = [str(c).strip() for c in df.columns] - return df - except Exception as e: - ultimo_error = e - - raise RuntimeError(f"No se pudo leer CSV {path.name}: {ultimo_error}") - - def encontrar_columna(df, campo): cols_norm = {normalizar_nombre_columna(c): c for c in df.columns} equivalencias = DICCIONARIO_COLUMNAS.get(campo, []) @@ -758,7 +713,6 @@ def construir_base_normal(df, archivo_nombre): col_cop_mod = encontrar_columna(df, "cop_mod") col_neto = encontrar_columna(df, "neto") - n = len(df) base = pd.DataFrame(index=df.index) base["archivo_origen"] = archivo_nombre base["tipo_archivo"] = "NORMAL" @@ -1019,160 +973,6 @@ def normalizar_registros(base): return pd.DataFrame(registros), pd.DataFrame(errores) -def separar_duplicados(df_ok): - """ - Lógica tipo CONTAR.SI del usuario: - - Si un radicado aparece 2 o más veces: - → TODOS esos registros van a DUPLICADOS.xlsx - → NINGUNO pasa al CSV_FINAL_IT - """ - if df_ok.empty: - return df_ok, pd.DataFrame() - - # Limpiar radicado temporalmente - temp = df_ok.copy() - temp["_radicado_clean"] = temp["radicado"].astype(str).str.strip() - - # Conteo tipo CONTAR.SI - conteo = temp.groupby("_radicado_clean")["_radicado_clean"].transform("count") - - # Duplicado = aparece 2 o más veces - dup_mask = (conteo >= 2) & temp["_radicado_clean"].ne("") - - duplicados = temp[dup_mask].copy() - unicos = temp[~dup_mask].copy() - - if not duplicados.empty: - duplicados["cantidad_duplicados"] = conteo[dup_mask] - duplicados["observacion"] = ( - "Radicado duplicado detectado por CONTAR.SI. " - "Cantidad: " + duplicados["cantidad_duplicados"].astype(str) - ) - - # Quitar columnas temporales - for df_ in [duplicados, unicos]: - if "_radicado_clean" in df_.columns: - df_.drop(columns=["_radicado_clean"], inplace=True, errors="ignore") - - return unicos, duplicados - - - -def clasificar_errores_por_hoja(df_errores): - """ - Separa los errores en hojas por categoría. - Un mismo registro puede aparecer en varias hojas si tiene varios errores. - """ - if df_errores.empty or "observacion" not in df_errores.columns: - return {} - - categorias = { - "Autorizaciones": [ - "autorización", "autorizacion" - ], - "Fechas": [ - "fecha ingreso", "fecha" - ], - "Documentos": [ - "documento inválido", "documento invalido", "longitud inválida", - "longitud invalida", "tipo de documento" - ], - "Total_Bruto": [ - "total cargos", "bruto" - ], - "Tipo_Documento": [ - "tipo de documento vacío", "tipo de documento vacio", - "tipo de documento no parametrizado" - ], - "Otros": [] - } - - hojas = {} - obs_norm = df_errores["observacion"].astype(str).str.lower() - - usados = pd.Series(False, index=df_errores.index) - - for hoja, palabras in categorias.items(): - if hoja == "Otros": - continue - - mask = pd.Series(False, index=df_errores.index) - for palabra in palabras: - mask = mask | obs_norm.str.contains(palabra, na=False, regex=False) - - if mask.any(): - hojas[hoja] = df_errores[mask].copy() - usados = usados | mask - - otros = df_errores[~usados].copy() - if not otros.empty: - hojas["Otros"] = otros - - return hojas - - -def escribir_errores_por_hojas(writer, df_err_total): - """ - Escribe ERRORES_NORMALIZACION.xlsx con: - - Hoja Todos - - Hojas por categoría: Autorizaciones, Fechas, Documentos, Total_Bruto, Tipo_Documento, Otros - """ - columnas_base = [ - "archivo_origen", "tipo_archivo", - *COLUMNAS_FINAL_IT, - "observacion" - ] - - if df_err_total.empty: - pd.DataFrame(columns=columnas_base).to_excel(writer, index=False, sheet_name="Todos") - return - - df_err_total.to_excel(writer, index=False, sheet_name="Todos") - - hojas = clasificar_errores_por_hoja(df_err_total) - for nombre_hoja, df_hoja in hojas.items(): - # Excel permite máximo 31 caracteres en nombre de hoja - nombre_seguro = str(nombre_hoja)[:31] - df_hoja.to_excel(writer, index=False, sheet_name=nombre_seguro) - - - - - -def leer_archivo_flexible(path): - """Lee CSV o Excel manteniendo todo como texto.""" - path = Path(path) - ext = path.suffix.lower() - if ext == ".csv": - return leer_csv_flexible(path) - if ext in {".xlsx", ".xlsm", ".xls"}: - return pd.read_excel(path, dtype=str, keep_default_na=False) - raise RuntimeError(f"Formato no soportado: {path.name}. Usa CSV o Excel.") - - -def listar_archivos_entrada(carpeta): - carpeta = Path(carpeta) - archivos = [] - for patron in ("*.csv", "*.xlsx", "*.xlsm", "*.xls"): - archivos.extend(carpeta.glob(patron)) - - # Evitar leer salidas anteriores o temporales. - archivos_limpios = [] - for a in sorted(archivos): - nombre = a.name.upper() - if a.parent.name.upper() == "SALIDA_NORMALIZADOR_IT": - continue - if nombre.startswith("CONSOLIDADO_BOT_IT"): - continue - if "ORIGEN_SERVICIO" in nombre: - continue - if nombre.startswith("~$"): - continue - archivos_limpios.append(a) - return archivos_limpios - - def normalizar_clave_radicado(valor): """ Normaliza el radicado para poder cruzarlo contra Origen Servicio. @@ -1293,23 +1093,6 @@ def cargar_origen_servicio_desde_bytes(content: bytes, filename: str = "origen.x return cargar_origen_servicio_desde_df(df_serv) -def cargar_origen_servicio(ruta_archivo): - """ - Carga el Excel/CSV seleccionado por el usuario y arma un diccionario: - radicado -> tipo_servicio. - """ - if not ruta_archivo: - return {} - - archivo_servicio = Path(ruta_archivo) - - if not archivo_servicio.exists(): - return {} - - df_serv = leer_archivo_flexible(archivo_servicio) - return cargar_origen_servicio_desde_df(df_serv) - - def aplicar_origen_servicio(base: pd.DataFrame, mapa_servicios: dict[str, str]) -> tuple[pd.DataFrame, int, int]: """ Rellena tipo_servicio desde el mapa Origen Servicio usando radicado.