380 lines
16 KiB
Python
380 lines
16 KiB
Python
"""Servicios de alto nivel para el procesamiento de facturas.
|
|
|
|
Orquesta la extracción usando una `InvoiceExtractorStrategy`, la conversión de
|
|
documentos (PDF/TIFF) a imágenes base64 y el parseo del bloque JSON.
|
|
Incluye un flujo por lotes con selección de campos y exportación incremental a CSV.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, List, Optional, Tuple, Callable, Awaitable
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
import asyncio
|
|
from .strategies import InvoiceExtractorStrategy
|
|
from .profiles import get_profile
|
|
from .config import settings
|
|
from .logging_config import log_execution_time
|
|
from .file_discovery import get_sorted_radicados, find_file_for_type_in_radicados, get_files_as_radicados
|
|
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__)
|
|
|
|
class InvoiceProcessingService:
|
|
"""Fachada de negocio para extraer y transformar datos de facturas."""
|
|
def __init__(
|
|
self,
|
|
extractor: InvoiceExtractorStrategy,
|
|
batch_processor_factory: Callable[[InvoiceProcessingService], BatchProcessor],
|
|
estimator_factory: Callable[[InvoiceProcessingService], Estimator]
|
|
) -> None:
|
|
"""
|
|
Inicializa el servicio de procesamiento de facturas.
|
|
|
|
Args:
|
|
extractor: Estrategia para extraer datos (ej. GPT-4o, Claude).
|
|
batch_processor_factory: Fábrica para crear el procesador por lotes,
|
|
evitando dependencias circulares.
|
|
estimator_factory: Fábrica para crear el estimador de costos y tokens.
|
|
"""
|
|
self.extractor = extractor
|
|
# Componentes delegados inyectados mediante factories para evitar dependencia circular directa en construcción
|
|
self.batch_processor = batch_processor_factory(self)
|
|
self.estimator = estimator_factory(self)
|
|
|
|
# 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]:
|
|
"""
|
|
Convierte un documento (PDF o TIFF) a una lista de imágenes codificadas en base64.
|
|
|
|
Args:
|
|
pdf_path: Ruta al archivo.
|
|
dpi: Resolución de la conversión.
|
|
thread_count: Número de hilos para procesamiento paralelo (solo PDF).
|
|
tiff_pages: Páginas por defecto para TIFF si no se especifica rango.
|
|
tiff_page_range: Rango específico de páginas (inicio, fin) para TIFF.
|
|
|
|
Returns:
|
|
Lista de strings con las imágenes en base64 (formato PNG).
|
|
"""
|
|
filename = os.path.basename(pdf_path)
|
|
tiff_pages_cfg = getattr(settings.imaging, 'tiff_pages', (2, 3)) if tiff_pages is None else tiff_pages
|
|
poppler = getattr(settings.imaging, 'poppler_path', None)
|
|
try:
|
|
return document_to_images_bs64(
|
|
pdf_path, dpi=dpi, thread_count=thread_count,
|
|
tiff_pages=tiff_pages_cfg, tiff_page_range=tiff_page_range,
|
|
poppler_path=poppler,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error al convertir documento a imágenes: {filename}", extra={"pdf": pdf_path, "error": str(e)})
|
|
raise
|
|
|
|
def _apply_extractor_fallback(self, pdf_path: str) -> List[str]:
|
|
"""
|
|
Intento de recuperación ante errores de extracción usando parámetros reducidos.
|
|
|
|
Carga imágenes con DPI fijo (400) y limita a las primeras 5 páginas para
|
|
reducir el consumo de tokens y evitar límites de contexto.
|
|
|
|
Args:
|
|
pdf_path: Ruta al archivo.
|
|
|
|
Returns:
|
|
Lista de hasta 5 imágenes en base64.
|
|
"""
|
|
fb_dpi = 400
|
|
tiff_pages_fb = getattr(settings.imaging, 'tiff_pages', (2, 3))
|
|
poppler = getattr(settings.imaging, 'poppler_path', None)
|
|
images = document_to_images_bs64(pdf_path, dpi=fb_dpi, tiff_pages=tiff_pages_fb, poppler_path=poppler)
|
|
return images[:5] if len(images) > 5 else images
|
|
|
|
def _parse_model_output(self, model_output: dict, pdf_path: str, dpi_cfg: int) -> Tuple[Any, Any]:
|
|
"""
|
|
Extrae y valida el bloque JSON de la respuesta del modelo de lenguaje.
|
|
|
|
Args:
|
|
model_output: Diccionario con la respuesta cruda ("raw") y el uso de tokens ("usage").
|
|
pdf_path: Ruta del archivo original (para logging).
|
|
dpi_cfg: DPI usado (para logging).
|
|
|
|
Returns:
|
|
Tupla (datos_dict, usage_dict) con el JSON parseado y estadísticas de uso.
|
|
|
|
Raises:
|
|
ValueError: Si no se encuentra un JSON válido en la respuesta.
|
|
"""
|
|
filename = os.path.basename(pdf_path)
|
|
raw_text = model_output.get("raw", "")
|
|
usage = model_output.get("usage")
|
|
json_text = extract_json_from_text(raw_text)
|
|
if not json_text:
|
|
logger.error(f"No se pudo extraer JSON válido de la respuesta para {filename}", extra={"pdf": pdf_path})
|
|
raise ValueError(f"No se pudo extraer un bloque JSON del texto de respuesta para {filename}.")
|
|
data = json.loads(json_text)
|
|
try:
|
|
if isinstance(data, dict):
|
|
top_keys = list(data.keys())
|
|
preview = ", ".join(top_keys[:12]) + ("…" if len(top_keys) > 12 else "")
|
|
logger.debug(f"Datos extraídos de {filename} (keys={len(top_keys)}: {preview})", extra={"pdf": pdf_path, "dpi": dpi_cfg})
|
|
else:
|
|
logger.debug(f"Datos extraídos de {filename} (tipo={type(data).__name__})", extra={"pdf": pdf_path, "dpi": dpi_cfg})
|
|
except Exception:
|
|
pass
|
|
return data, usage
|
|
|
|
@log_execution_time(logger=logger)
|
|
def extract_full_from_pdf(
|
|
self,
|
|
pdf_path: str,
|
|
instruction: Optional[str] = None,
|
|
*,
|
|
tiff_page_range: Optional[Tuple[int, int]] = None,
|
|
history: Optional[List[Dict[str, Any]]] = None,
|
|
) -> Tuple[Any, Any]:
|
|
"""
|
|
Extrae un JSON completo desde un documento (PDF/TIFF) usando el extractor.
|
|
|
|
El proceso incluye:
|
|
1. Conversión del documento a imágenes base64.
|
|
2. Llamada al modelo de extracción.
|
|
3. En caso de error, reintento automático con parámetros de recuperación (DPI reducido).
|
|
4. Parseo del resultado JSON.
|
|
|
|
Args:
|
|
pdf_path: Ruta al archivo PDF o TIFF.
|
|
instruction: Instrucción personalizada para el modelo (opcional).
|
|
tiff_page_range: Rango de páginas específico si es un TIFF.
|
|
history: Historial de mensajes para el modelo (opcional).
|
|
|
|
Returns:
|
|
Tupla (datos_extraidos, estadisticas_uso).
|
|
"""
|
|
filename = os.path.basename(pdf_path)
|
|
logger.info(f"Iniciando extracción de documento: {filename}", extra={"pdf": pdf_path})
|
|
t0 = time.time()
|
|
dpi_cfg = settings.imaging.pdf_dpi
|
|
thr_cfg = settings.imaging.pdf_thread_count
|
|
tiff_pages = getattr(settings.imaging, 'tiff_pages', (2, 3))
|
|
profile = get_profile('invoice')
|
|
text_instruction = instruction if instruction is not None else profile["default_instruction"]
|
|
|
|
images_b64 = self._convert_to_images(pdf_path, dpi_cfg, thr_cfg, tiff_pages, tiff_page_range)
|
|
logger.debug(
|
|
f"Imágenes procesadas para {filename} (count={len(images_b64)}, dpi={dpi_cfg}, threads={thr_cfg})",
|
|
extra={"pdf": pdf_path, "dpi": dpi_cfg},
|
|
)
|
|
try:
|
|
model_output = self.extractor.extract(images_b64, text_instruction, history=history)
|
|
except Exception as e:
|
|
logger.warning(f"Fallo en extracción. Reintentando con DPI reducido: {filename}", extra={"pdf": pdf_path, "error": str(e)})
|
|
images_b64_fb = self._apply_extractor_fallback(pdf_path)
|
|
model_output = self.extractor.extract(images_b64_fb, text_instruction, history=history)
|
|
|
|
data, usage = self._parse_model_output(model_output, pdf_path, dpi_cfg)
|
|
logger.info(f"Extracción finalizada: {filename} (total_ms={int((time.time()-t0)*1000)})", extra={"pdf": pdf_path})
|
|
return data, usage
|
|
|
|
@log_execution_time(logger=logger)
|
|
async def extract_full_from_pdf_async(
|
|
self,
|
|
pdf_path: str,
|
|
instruction: Optional[str] = None,
|
|
*,
|
|
tiff_page_range: Optional[Tuple[int, int]] = None,
|
|
history: Optional[List[Dict[str, Any]]] = None,
|
|
) -> Tuple[Any, Any]:
|
|
"""
|
|
Versión asíncrona de extract_full_from_pdf.
|
|
|
|
Realiza la conversión de imágenes en un hilo separado para no bloquear el loop de eventos.
|
|
"""
|
|
filename = os.path.basename(pdf_path)
|
|
logger.info(f"Iniciando extracción asíncrona de documento: {filename}", extra={"pdf": pdf_path})
|
|
t0 = time.time()
|
|
dpi_cfg = settings.imaging.pdf_dpi
|
|
thr_cfg = settings.imaging.pdf_thread_count
|
|
tiff_pages = getattr(settings.imaging, 'tiff_pages', (2, 3))
|
|
profile = get_profile('invoice')
|
|
text_instruction = instruction if instruction is not None else profile["default_instruction"]
|
|
|
|
try:
|
|
images_b64 = await asyncio.to_thread(
|
|
self._convert_to_images, pdf_path, dpi_cfg, thr_cfg, tiff_pages, tiff_page_range,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error al convertir documento a imágenes: {filename}", extra={"pdf": pdf_path, "error": str(e)})
|
|
raise
|
|
|
|
logger.debug(
|
|
f"Imágenes procesadas para {filename} (count={len(images_b64)}, dpi={dpi_cfg}, threads={thr_cfg})",
|
|
extra={"pdf": pdf_path, "dpi": dpi_cfg},
|
|
)
|
|
try:
|
|
model_output = await self.extractor.extract_async(images_b64, text_instruction, history=history)
|
|
except Exception as e:
|
|
logger.warning(f"Fallo en extracción. Reintentando con DPI reducido: {filename}", extra={"pdf": pdf_path, "error": str(e)})
|
|
images_b64_fb = await asyncio.to_thread(self._apply_extractor_fallback, pdf_path)
|
|
model_output = await self.extractor.extract_async(images_b64_fb, text_instruction, history=history)
|
|
|
|
data, usage = self._parse_model_output(model_output, pdf_path, dpi_cfg)
|
|
logger.info(f"Extracción finalizada: {filename} (total_ms={int((time.time()-t0)*1000)})", extra={"pdf": pdf_path})
|
|
return data, usage
|
|
|
|
def build_schema_from_first(
|
|
self,
|
|
pdf_path: str,
|
|
instruction: Optional[str] = None,
|
|
*,
|
|
tiff_page_range: Optional[Tuple[int, int]] = None,
|
|
) -> Tuple[Dict[str, str], Dict[str, Any]]:
|
|
"""
|
|
Genera un esquema de datos (rutas de campos) basándose en el primer documento procesado.
|
|
|
|
Útil para permitir al usuario seleccionar qué campos desea exportar a CSV.
|
|
|
|
Args:
|
|
pdf_path: Ruta al documento de muestra.
|
|
instruction: Instrucción para el modelo.
|
|
tiff_page_range: Rango de páginas si es TIFF.
|
|
|
|
Returns:
|
|
Tupla (esquema_dict, datos_ejemplo_dict).
|
|
"""
|
|
data, _usage = self.extract_full_from_pdf(pdf_path, instruction=instruction, tiff_page_range=tiff_page_range)
|
|
schema = build_schema(data)
|
|
try:
|
|
logger.debug(f"Schema generado para {os.path.basename(pdf_path)} (total_paths={len(schema)})", extra={"pdf": pdf_path})
|
|
except Exception:
|
|
pass
|
|
return schema, data
|
|
|
|
def build_schema_from_first_for_types(
|
|
self,
|
|
root_path: str,
|
|
doc_types: List[Dict[str, str]],
|
|
instruction: Optional[str] = None,
|
|
*,
|
|
tiff_page_range: Optional[Tuple[int, int]] = None,
|
|
structure_mode: str = "radicados",
|
|
) -> Tuple[Dict[str, Dict[str, str]], Dict[str, Dict[str, Any]], str | None]:
|
|
"""
|
|
Genera esquemas para múltiples tipos de documentos buscando un ejemplo de cada uno.
|
|
|
|
Args:
|
|
root_path: Directorio raíz de los documentos.
|
|
doc_types: Lista de tipos con sus palabras clave de búsqueda.
|
|
instruction: Instrucción para el modelo.
|
|
tiff_page_range: Rango de páginas si es TIFF.
|
|
structure_mode: 'radicados' o 'general'.
|
|
|
|
Returns:
|
|
Tupla (esquemas_por_tipo, muestras_por_tipo, primer_radicado_encontrado).
|
|
"""
|
|
if structure_mode == "general":
|
|
radicados = get_files_as_radicados(root_path)
|
|
else:
|
|
radicados = get_sorted_radicados(root_path)
|
|
|
|
if not radicados:
|
|
raise ValueError(f"No se encontraron documentos/radicados en: {root_path}")
|
|
|
|
schema_by_type: Dict[str, Dict[str, str]] = {}
|
|
first_full_by_type: Dict[str, Dict[str, Any]] = {}
|
|
prof = get_profile('invoice')
|
|
inst0 = instruction if instruction is not None else prof["default_instruction"]
|
|
|
|
first_radicado_found = None
|
|
|
|
for dt in (doc_types or []):
|
|
tname = str(dt.get("name") or "").strip() or "Tipo"
|
|
kw = str(dt.get("keyword") or "").strip()
|
|
|
|
pdf_path, rad_used = find_file_for_type_in_radicados(root_path, radicados, kw)
|
|
|
|
if not pdf_path:
|
|
logger.info(f"No se encontró archivo para tipo '{tname}' con keyword '{kw}' en ningún radicado")
|
|
schema_by_type[tname] = {}
|
|
first_full_by_type[tname] = {}
|
|
continue
|
|
|
|
if first_radicado_found is None:
|
|
first_radicado_found = rad_used
|
|
|
|
try:
|
|
logger.info(
|
|
f"Usando archivo para schema por tipo '{tname}': {os.path.basename(pdf_path)} (Radicado: {rad_used})",
|
|
extra={"tipo": tname, "radicado": rad_used, "pdf": pdf_path},
|
|
)
|
|
sch, dat = self.build_schema_from_first(pdf_path, instruction=inst0, tiff_page_range=tiff_page_range)
|
|
schema_by_type[tname] = sch or {}
|
|
first_full_by_type[tname] = dat or {}
|
|
except Exception as e:
|
|
logger.exception(f"Error generando schema para tipo '{tname}'", extra={"tipo": tname, "radicado": rad_used})
|
|
schema_by_type[tname] = {}
|
|
first_full_by_type[tname] = {"_error": str(e)}
|
|
|
|
return schema_by_type, first_full_by_type, (first_radicado_found or radicados[0])
|
|
|
|
# --- Delegación a BatchProcessor y Estimator ---
|
|
|
|
@log_execution_time(logger=logger)
|
|
def process_root_with_field_selection(
|
|
self,
|
|
root_path: str,
|
|
keywords: List[str] | None = None,
|
|
*,
|
|
doc_types: Optional[List[Dict[str, str]]] = None,
|
|
fields: Optional[List[str]] = None,
|
|
on_progress: Optional[callable] = None,
|
|
csv_output_path: Optional[str] = None,
|
|
csv_header_override: Optional[List[str]] = None,
|
|
instruction_override: Optional[str] = None,
|
|
on_usage: Optional[Callable[[str, str, Dict[str, Any] | None, Dict[str, Any] | None], None]] = None,
|
|
on_error_retry: Optional[Callable[[Exception], Awaitable[bool]]] = None,
|
|
alias_map: Optional[Dict[str, str]] = None,
|
|
tiff_page_range: Optional[Tuple[int, int]] = None,
|
|
structure_mode: str = "radicados",
|
|
) -> List[Dict[str, Any]]:
|
|
"""Procesa radicados con selección de campos y CSV incremental. (Delegado a BatchProcessor)"""
|
|
return self.batch_processor.process_root(
|
|
root_path, keywords,
|
|
doc_types=doc_types, fields=fields, on_progress=on_progress,
|
|
csv_output_path=csv_output_path, csv_header_override=csv_header_override,
|
|
instruction_override=instruction_override, on_usage=on_usage,
|
|
on_error_retry=on_error_retry,
|
|
alias_map=alias_map, tiff_page_range=tiff_page_range,
|
|
structure_mode=structure_mode
|
|
)
|
|
|
|
def estimate_tokens_and_cost(
|
|
self,
|
|
root_path: str,
|
|
keywords: List[str] | None = None,
|
|
schema: Dict[str, str] | None = None,
|
|
selected_fields: List[str] | None = None,
|
|
*,
|
|
doc_types: Optional[List[Dict[str, str]]] = None,
|
|
schema_by_type: Optional[Dict[str, Dict[str, str]]] = None,
|
|
selected_paths: Optional[List[str]] = None,
|
|
structure_mode: str = "radicados",
|
|
) -> Dict[str, Any]:
|
|
"""Calcula estimado de tokens y costos. (Delegado a Estimator)"""
|
|
return self.estimator.estimate(
|
|
root_path, keywords, schema, selected_fields,
|
|
doc_types=doc_types, schema_by_type=schema_by_type, selected_paths=selected_paths,
|
|
structure_mode=structure_mode
|
|
)
|