896 lines
41 KiB
Python
896 lines
41 KiB
Python
"""Procesador por lotes para la extracción de facturas.
|
|
|
|
Encapsula la lógica de iteración sobre radicados, extracción, normalización y exportación a CSV.
|
|
"""
|
|
import os
|
|
import logging
|
|
import asyncio
|
|
import openai
|
|
from typing import List, Dict, Any, Optional, Tuple, Callable, Awaitable, TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from application.service import InvoiceProcessingService
|
|
|
|
from application.profiles import get_profile
|
|
from application.prompts import PromptBuilder
|
|
from application.normalizer import normalize_to_schema
|
|
from application.costs import compute_cost_from_usage, accumulate_usage
|
|
from application.file_discovery import get_sorted_radicados, get_files_as_radicados
|
|
from application.csv_manager import CSVManager
|
|
from application.config import settings
|
|
from application.app_prefs import get_allowed_document_types
|
|
from application.tracking_manager import TrackingManager
|
|
from core.utils import collect_radicado_docs, find_first_matching_file, collect_documents_flat
|
|
from core.csv_utils import rows_from_data
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DOCUMENT_FORCE_TARGETS = {
|
|
"tipo de documento",
|
|
"documento_forzate",
|
|
}
|
|
|
|
DOCUMENT_TYPE_HINT_ALIASES = {
|
|
"PT",
|
|
"PAS",
|
|
"PASS",
|
|
}
|
|
|
|
class BatchProcessor:
|
|
def __init__(self, service: 'InvoiceProcessingService'):
|
|
"""
|
|
Args:
|
|
service: Instancia de InvoiceProcessingService para delegar la extracción individual.
|
|
"""
|
|
self.service = service
|
|
self.csv_manager = CSVManager()
|
|
self.tracking_manager = TrackingManager()
|
|
self._pause_event = asyncio.Event()
|
|
self._pause_event.set() # No pausado inicialmente
|
|
self._stop_requested = False
|
|
# Contadores acumulados para tracking web
|
|
self._track_tokens: int = 0
|
|
self._track_cost_usd: float = 0.0
|
|
self._track_success: int = 0
|
|
self._track_errors: int = 0
|
|
|
|
def process_root(
|
|
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",
|
|
retry_on_empty: bool = False,
|
|
radicado_filter: Optional[List[str]] = None,
|
|
) -> List[Dict[str, Any]]:
|
|
"""Procesa radicados con selección de campos y CSV incremental.
|
|
|
|
Delega en métodos específicos para multi-documento o single-document.
|
|
"""
|
|
# Resetear estado de pausa/parada y contadores de tracking
|
|
self._pause_event.set()
|
|
self._stop_requested = False
|
|
self._track_tokens = 0
|
|
self._track_cost_usd = 0.0
|
|
self._track_success = 0
|
|
self._track_errors = 0
|
|
|
|
# Notificar inicio a tracking si es procesamiento masivo
|
|
if fields:
|
|
# Intentar estimar total de radicados
|
|
try:
|
|
if structure_mode == "general":
|
|
temp_rads = get_files_as_radicados(root_path)
|
|
else:
|
|
temp_rads = get_sorted_radicados(root_path)
|
|
|
|
if radicado_filter:
|
|
filter_set = {r.strip() for r in radicado_filter if r.strip()}
|
|
temp_rads = [r for r in temp_rads if r in filter_set]
|
|
|
|
self.tracking_manager.start_batch(
|
|
root_path,
|
|
len(temp_rads),
|
|
config={
|
|
"structure_mode": structure_mode,
|
|
"fields_count": len(fields)
|
|
},
|
|
csv_output_path=csv_output_path,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error al iniciar tracking: {e}")
|
|
|
|
# 1. Preparación de Schemas (Fase 1 Multi-doc)
|
|
if doc_types and not fields:
|
|
return self._prepare_multidoc_schemas(root_path, doc_types, instruction_override, tiff_page_range, structure_mode)
|
|
|
|
# 2. Procesamiento Multi-doc (Fase 2) - ASYNC
|
|
if doc_types and fields:
|
|
return asyncio.run(self._process_multidoc_async(
|
|
root_path=root_path,
|
|
doc_types=doc_types,
|
|
fields=fields,
|
|
csv_output_path=csv_output_path,
|
|
csv_header_override=csv_header_override,
|
|
alias_map=alias_map,
|
|
tiff_page_range=tiff_page_range,
|
|
on_progress=on_progress,
|
|
on_usage=on_usage,
|
|
on_error_retry=on_error_retry,
|
|
structure_mode=structure_mode,
|
|
retry_on_empty=retry_on_empty,
|
|
radicado_filter=radicado_filter,
|
|
))
|
|
|
|
# 3. Procesamiento Single-doc (Retrocompatibilidad)
|
|
# Si no hay fields, es la fase de detección de esquema (secuencial, solo 1 doc)
|
|
if not fields:
|
|
return self._process_singledoc(
|
|
root_path=root_path,
|
|
keywords=keywords,
|
|
fields=fields,
|
|
csv_output_path=csv_output_path,
|
|
csv_header_override=csv_header_override,
|
|
instruction_override=instruction_override,
|
|
alias_map=alias_map,
|
|
tiff_page_range=tiff_page_range,
|
|
on_progress=on_progress,
|
|
on_usage=on_usage,
|
|
structure_mode=structure_mode,
|
|
retry_on_empty=retry_on_empty,
|
|
radicado_filter=radicado_filter,
|
|
)
|
|
else:
|
|
# Si hay fields, es procesamiento masivo -> ASYNC
|
|
return asyncio.run(self._process_singledoc_async(
|
|
root_path=root_path,
|
|
keywords=keywords,
|
|
fields=fields,
|
|
csv_output_path=csv_output_path,
|
|
csv_header_override=csv_header_override,
|
|
instruction_override=instruction_override,
|
|
alias_map=alias_map,
|
|
tiff_page_range=tiff_page_range,
|
|
on_progress=on_progress,
|
|
on_usage=on_usage,
|
|
on_error_retry=on_error_retry,
|
|
structure_mode=structure_mode,
|
|
retry_on_empty=retry_on_empty,
|
|
radicado_filter=radicado_filter,
|
|
))
|
|
|
|
def _prepare_multidoc_schemas(self, root_path, doc_types, instruction, tiff_page_range, structure_mode):
|
|
schema_by_type, first_full_by_type, first_radicado = self.service.build_schema_from_first_for_types(
|
|
root_path,
|
|
doc_types,
|
|
instruction=instruction,
|
|
tiff_page_range=tiff_page_range,
|
|
structure_mode=structure_mode
|
|
)
|
|
return [{
|
|
"_needs_fields": True,
|
|
"schema_by_type": schema_by_type,
|
|
"first_full_by_type": first_full_by_type,
|
|
"first_radicado": first_radicado,
|
|
}]
|
|
|
|
async def _process_multidoc_async(
|
|
self, root_path, doc_types, fields, csv_output_path,
|
|
csv_header_override, alias_map, tiff_page_range, on_progress, on_usage, on_error_retry=None,
|
|
structure_mode="radicados", retry_on_empty=False, radicado_filter=None
|
|
):
|
|
effective_fields = self._expand_document_force_fields(fields or [], alias_map)
|
|
effective_alias_map = self._expand_document_force_alias_map(alias_map, effective_fields)
|
|
csv_header = self._prepare_csv_header(effective_fields, effective_alias_map, csv_header_override)
|
|
|
|
if structure_mode == "general":
|
|
radicados = get_files_as_radicados(root_path)
|
|
else:
|
|
radicados = get_sorted_radicados(root_path)
|
|
|
|
if not radicados and not os.path.isdir(root_path):
|
|
raise ValueError(f"La ruta raíz no es un directorio: {root_path}")
|
|
|
|
# Aplicar filtro de radicados si está activo
|
|
if radicado_filter:
|
|
filter_set = {r.strip() for r in radicado_filter if r.strip()}
|
|
if structure_mode == "general":
|
|
radicados = [r for r in radicados if os.path.splitext(r)[0] in filter_set]
|
|
else:
|
|
radicados = [r for r in radicados if r in filter_set]
|
|
logger.info(f"Filtro activo: {len(radicados)} radicados seleccionados de {len(filter_set)} indicados")
|
|
|
|
total = len(radicados)
|
|
prof = get_profile('invoice')
|
|
|
|
# Limitar concurrencia (Configurable)
|
|
limit = settings.processing.concurrency_limit
|
|
if limit < 1: limit = 1
|
|
sem = asyncio.Semaphore(limit)
|
|
|
|
# Para mantener el orden de escritura en CSV, usamos un Lock
|
|
csv_lock = asyncio.Lock()
|
|
|
|
# Contador thread-safe para progreso
|
|
completed_count = 0
|
|
progress_lock = asyncio.Lock()
|
|
|
|
async def process_radicado(idx, rad):
|
|
nonlocal completed_count
|
|
async with sem:
|
|
attempts = 0
|
|
while True: # Bucle para permitir reintentos tras pausa
|
|
attempts += 1
|
|
should_retry_empty = False
|
|
await self._pause_event.wait()
|
|
if self._stop_requested:
|
|
return {"radicado": rad, "status": "stopped"}
|
|
|
|
try:
|
|
# Determinar ID visual y ruta
|
|
if structure_mode == "general":
|
|
rad_id = os.path.splitext(rad)[0]
|
|
rad_path = os.path.join(root_path, rad) # Es el archivo
|
|
else:
|
|
rad_id = rad
|
|
rad_path = os.path.join(root_path, rad) # Es la carpeta
|
|
|
|
logger.info(f"Procesando radicado {rad_id} (multi) [{idx}/{total}]", extra={"radicado": rad_id})
|
|
|
|
fused_data = {"radicado": rad_id}
|
|
agg_extract_usage = {"prompt_tokens": 0, "completion_tokens": 0, "prompt_tokens_details": {"cached_tokens": 0}}
|
|
agg_normalize_usage = {"prompt_tokens": 0, "completion_tokens": 0, "prompt_tokens_details": {"cached_tokens": 0}}
|
|
|
|
for dt in doc_types:
|
|
tname = str(dt.get("name") or "").strip() or "Tipo"
|
|
kw = str(dt.get("keyword") or "").strip()
|
|
|
|
match = None
|
|
if structure_mode == "general":
|
|
# En modo general, el archivo ES el documento.
|
|
# Opcional: filtrar si el nombre coincide con keyword, pero usualmente
|
|
# si el usuario elige "General" y "Tipos", quiere aplicar los tipos a ese archivo.
|
|
# Pero si tiene varios tipos (Factura, Anexo), no tiene sentido aplicarlos todos al mismo archivo
|
|
# a menos que quiera intentar extraer cosas distintas.
|
|
# Asumiremos que el archivo aplica para el tipo.
|
|
match = rad_path
|
|
else:
|
|
try:
|
|
files = sorted(await asyncio.to_thread(os.listdir, rad_path))
|
|
except Exception:
|
|
files = []
|
|
|
|
match_name = find_first_matching_file(files, kw) if kw else None
|
|
match = os.path.join(rad_path, match_name) if match_name else None
|
|
|
|
if not match:
|
|
logger.info(f"[{rad_id}] Tipo '{tname}' sin archivo coincidente (kw='{kw}')")
|
|
fused_data[tname] = {}
|
|
continue
|
|
|
|
# Extraer
|
|
selected_paths_for_type = [p.split('/', 1)[1] for p in effective_fields if p.startswith(f"{tname}/")]
|
|
try:
|
|
prompt_t = PromptBuilder.build_for_type(tname, selected_paths_for_type, {}, profile_name=prof.get('name', 'invoice'))
|
|
except Exception:
|
|
prompt_t = PromptBuilder.build(selected_paths_for_type, {}, profile_name=prof.get('name', 'invoice'))
|
|
|
|
# Llamada ASYNC - Fase: Extracción
|
|
self.tracking_manager.report_phase("extract", rad_id)
|
|
self.tracking_manager.log(f"Iniciando extracción IA", level="info", radicado=rad_id)
|
|
data, usage_ext = await self.service.extract_full_from_pdf_async(match, instruction=prompt_t, tiff_page_range=tiff_page_range)
|
|
self._accumulate_usage(agg_extract_usage, usage_ext)
|
|
|
|
# Normalizar (CPU bound, mover a thread)
|
|
raw_data = data if isinstance(data, dict) else {"data": data}
|
|
if isinstance(data, dict):
|
|
data['radicado'] = rad_id
|
|
else:
|
|
data = {"radicado": rad_id, "data": data}
|
|
|
|
# Normalización en thread aparte - Fase: Normalización
|
|
self.tracking_manager.report_phase("normalize", rad_id)
|
|
self.tracking_manager.log(f"Extracción OK — normalizando campos", level="info", radicado=rad_id)
|
|
norm = await asyncio.to_thread(
|
|
normalize_to_schema,
|
|
data,
|
|
paths=selected_paths_for_type,
|
|
profile_name=prof.get('name', 'invoice')
|
|
)
|
|
|
|
normalized_obj = data
|
|
n_usage = None
|
|
if isinstance(norm, dict):
|
|
n_usage = norm.get("_usage")
|
|
normalized_obj = norm.get("normalized") or data
|
|
|
|
if isinstance(normalized_obj, dict):
|
|
normalized_obj = self._inject_document_force_fields_from_raw(
|
|
normalized_obj,
|
|
raw_data,
|
|
selected_paths_for_type,
|
|
)
|
|
|
|
self._accumulate_usage(agg_normalize_usage, n_usage)
|
|
fused_data[tname] = normalized_obj or {}
|
|
fused_data["_source_pdf"] = match
|
|
|
|
if retry_on_empty and self._is_effectively_empty_output(fused_data, effective_fields, effective_alias_map):
|
|
if attempts < 2:
|
|
logger.info(f"[{rad_id}] Resultado vacío; reintentando extracción ({attempts + 1}/2)", extra={"radicado": rad_id})
|
|
should_retry_empty = True
|
|
else:
|
|
logger.info(f"[{rad_id}] Resultado vacío tras 2 intentos; se conserva el resultado.", extra={"radicado": rad_id})
|
|
|
|
# Report usage
|
|
if should_retry_empty:
|
|
continue
|
|
self._report_usage(on_usage, rad_id, "extract", agg_extract_usage)
|
|
self._report_usage(on_usage, rad_id, "normalize", agg_normalize_usage)
|
|
|
|
# CSV Export protegido por lock
|
|
if csv_output_path:
|
|
async with csv_lock:
|
|
await asyncio.to_thread(
|
|
self._write_csv_multidoc,
|
|
csv_output_path, fused_data, effective_fields, doc_types,
|
|
prof, csv_header, effective_alias_map, rad_id
|
|
)
|
|
|
|
return fused_data
|
|
|
|
except openai.RateLimitError as e:
|
|
if on_error_retry:
|
|
# Pausar todos los demás
|
|
self._pause_event.clear()
|
|
try:
|
|
should_retry = await on_error_retry(e)
|
|
if should_retry:
|
|
self._pause_event.set()
|
|
continue # Reintentar este radicado
|
|
else:
|
|
self._stop_requested = True
|
|
self._pause_event.set()
|
|
return self._handle_error_async(rad, e, csv_output_path, csv_header)
|
|
except Exception as e_retry:
|
|
self._pause_event.set()
|
|
return self._handle_error_async(rad, e_retry, csv_output_path, csv_header)
|
|
else:
|
|
return self._handle_error_async(rad, e, csv_output_path, csv_header)
|
|
except Exception as e:
|
|
return self._handle_error_async(rad, e, csv_output_path, csv_header)
|
|
finally:
|
|
async with progress_lock:
|
|
if not should_retry_empty:
|
|
completed_count += 1
|
|
self._notify_progress(on_progress, rad, completed_count, total)
|
|
if not should_retry_empty:
|
|
break # Salir del bucle True si no hubo RateLimitError o se manejó
|
|
|
|
tasks = [process_radicado(i, r) for i, r in enumerate(radicados, start=1)]
|
|
resultados = await asyncio.gather(*tasks)
|
|
return list(resultados)
|
|
|
|
def _process_singledoc(
|
|
self, root_path, keywords, fields, csv_output_path,
|
|
csv_header_override, instruction_override, alias_map, tiff_page_range, on_progress, on_usage, on_error_retry=None,
|
|
structure_mode="radicados", retry_on_empty=False, radicado_filter=None
|
|
):
|
|
if structure_mode == "general":
|
|
entries = collect_documents_flat(root_path, keywords)
|
|
else:
|
|
entries = collect_radicado_docs(root_path, keywords or [])
|
|
|
|
# Aplicar filtro de radicados si está activo
|
|
if radicado_filter:
|
|
filter_set = {r.strip() for r in radicado_filter if r.strip()}
|
|
entries = [e for e in entries if e.get("radicado") in filter_set]
|
|
logger.info(f"Filtro activo: {len(entries)} entradas seleccionadas de {len(filter_set)} indicadas")
|
|
|
|
if not entries:
|
|
logger.info("No se encontraron radicados con PDFs que coincidan.")
|
|
return []
|
|
|
|
# Primer documento para schema
|
|
first = entries[0]
|
|
prof = get_profile('invoice')
|
|
inst0 = instruction_override if instruction_override is not None else prof["default_instruction"]
|
|
schema, first_full = self.service.build_schema_from_first(first["path"], instruction=inst0, tiff_page_range=tiff_page_range)
|
|
|
|
selected_fields = fields or []
|
|
if not selected_fields:
|
|
return [{
|
|
"_needs_fields": True,
|
|
"schema": schema,
|
|
"first_full": first_full,
|
|
"radicado": first["radicado"],
|
|
"_source_pdf": first["path"],
|
|
}]
|
|
|
|
# Fallback a procesamiento secuencial/async si se llama directamente
|
|
return asyncio.run(self._process_singledoc_async(
|
|
root_path, keywords, fields, csv_output_path, csv_header_override,
|
|
instruction_override, alias_map, tiff_page_range, on_progress, on_usage, on_error_retry, structure_mode, retry_on_empty,
|
|
radicado_filter=radicado_filter,
|
|
))
|
|
|
|
async def _process_singledoc_async(
|
|
self, root_path, keywords, fields, csv_output_path,
|
|
csv_header_override, instruction_override, alias_map, tiff_page_range, on_progress, on_usage, on_error_retry=None,
|
|
structure_mode="radicados", retry_on_empty=False, radicado_filter=None
|
|
):
|
|
# Recolección de archivos (IO bloqueante)
|
|
if structure_mode == "general":
|
|
entries = await asyncio.to_thread(collect_documents_flat, root_path, keywords)
|
|
else:
|
|
entries = await asyncio.to_thread(collect_radicado_docs, root_path, keywords or [])
|
|
|
|
# Aplicar filtro de radicados si está activo
|
|
if radicado_filter:
|
|
filter_set = {r.strip() for r in radicado_filter if r.strip()}
|
|
entries = [e for e in entries if e.get("radicado") in filter_set]
|
|
logger.info(f"Filtro activo: {len(entries)} entradas seleccionadas de {len(filter_set)} indicadas")
|
|
|
|
if not entries:
|
|
logger.info("No se encontraron radicados con PDFs que coincidan.")
|
|
return []
|
|
|
|
selected_fields = self._expand_document_force_fields(fields or [], alias_map)
|
|
effective_alias_map = self._expand_document_force_alias_map(alias_map, selected_fields)
|
|
prompt = PromptBuilder.build(selected_fields, {})
|
|
csv_header = self._prepare_csv_header(selected_fields, effective_alias_map, csv_header_override)
|
|
|
|
prof = get_profile('invoice')
|
|
total = len(entries)
|
|
|
|
# Limitar concurrencia (Configurable)
|
|
limit = settings.processing.concurrency_limit
|
|
if limit < 1: limit = 1
|
|
sem = asyncio.Semaphore(limit)
|
|
|
|
csv_lock = asyncio.Lock()
|
|
progress_lock = asyncio.Lock()
|
|
completed_count = 0
|
|
|
|
async def process_entry(idx, entry):
|
|
nonlocal completed_count
|
|
async with sem:
|
|
attempts = 0
|
|
while True: # Bucle para permitir reintentos tras pausa
|
|
attempts += 1
|
|
should_retry_empty = False
|
|
await self._pause_event.wait()
|
|
if self._stop_requested:
|
|
return {"radicado": entry['radicado'], "status": "stopped"}
|
|
|
|
try:
|
|
logger.info(f"Procesando radicado {entry['radicado']} [{idx}/{total}]", extra={"radicado": entry['radicado'], "pdf": entry['path']})
|
|
|
|
# Async Extract - Fase: Extracción
|
|
self.tracking_manager.report_phase("extract", entry['radicado'])
|
|
self.tracking_manager.log(f"Iniciando extracción IA", level="info", radicado=entry['radicado'])
|
|
data, usage_ext = await self.service.extract_full_from_pdf_async(entry["path"], instruction=prompt, tiff_page_range=tiff_page_range)
|
|
self._report_usage(on_usage, entry['radicado'], "extract", usage_ext)
|
|
|
|
raw_data = data if isinstance(data, dict) else {"data": data}
|
|
if isinstance(data, dict):
|
|
data['radicado'] = entry['radicado']
|
|
else:
|
|
data = {"radicado": entry['radicado'], "data": data}
|
|
data["_source_pdf"] = entry["path"]
|
|
|
|
# Normalizar (Thread) - Fase: Normalización
|
|
self.tracking_manager.report_phase("normalize", entry['radicado'])
|
|
self.tracking_manager.log(f"Extracción OK — normalizando campos", level="info", radicado=entry['radicado'])
|
|
norm = await asyncio.to_thread(
|
|
normalize_to_schema,
|
|
data, paths=selected_fields, profile_name=prof.get('name', 'invoice')
|
|
)
|
|
|
|
if isinstance(norm, dict):
|
|
n_usage = norm.get("_usage")
|
|
self._report_usage(on_usage, entry['radicado'], "normalize", n_usage)
|
|
|
|
normalized_obj = norm.get("normalized")
|
|
if isinstance(normalized_obj, dict):
|
|
data = self._inject_document_force_fields_from_raw(
|
|
normalized_obj,
|
|
raw_data,
|
|
selected_fields,
|
|
)
|
|
data['radicado'] = entry['radicado']
|
|
data["_source_pdf"] = entry["path"]
|
|
|
|
if retry_on_empty and self._is_effectively_empty_output(data, selected_fields, effective_alias_map):
|
|
if attempts < 2:
|
|
logger.info(f"[{entry['radicado']}] Resultado vacío; reintentando extracción ({attempts + 1}/2)", extra={"radicado": entry['radicado'], "pdf": entry['path']})
|
|
should_retry_empty = True
|
|
else:
|
|
logger.info(f"[{entry['radicado']}] Resultado vacío tras 2 intentos; se conserva el resultado.", extra={"radicado": entry['radicado'], "pdf": entry['path']})
|
|
|
|
if should_retry_empty:
|
|
continue
|
|
if csv_output_path:
|
|
async with csv_lock:
|
|
await asyncio.to_thread(
|
|
self._write_csv_singledoc,
|
|
csv_output_path, data, selected_fields, prof, csv_header, effective_alias_map, entry['radicado']
|
|
)
|
|
|
|
return data
|
|
|
|
except openai.RateLimitError as e:
|
|
if on_error_retry:
|
|
# Pausar todos los demás
|
|
self._pause_event.clear()
|
|
try:
|
|
should_retry = await on_error_retry(e)
|
|
if should_retry:
|
|
self._pause_event.set()
|
|
continue # Reintentar este radicado
|
|
else:
|
|
self._stop_requested = True
|
|
self._pause_event.set()
|
|
return self._handle_error_async(entry['radicado'], e, csv_output_path, csv_header, entry['path'])
|
|
except Exception as e_retry:
|
|
self._pause_event.set()
|
|
return self._handle_error_async(entry['radicado'], e_retry, csv_output_path, csv_header, entry['path'])
|
|
else:
|
|
return self._handle_error_async(entry['radicado'], e, csv_output_path, csv_header, entry['path'])
|
|
except Exception as e:
|
|
return self._handle_error_async(entry['radicado'], e, csv_output_path, csv_header, entry['path'])
|
|
finally:
|
|
async with progress_lock:
|
|
if not should_retry_empty:
|
|
completed_count += 1
|
|
self._notify_progress(on_progress, entry['radicado'], completed_count, total)
|
|
if not should_retry_empty:
|
|
break # Salir del bucle si no hubo RateLimitError o se manejó
|
|
|
|
tasks = [process_entry(i, e) for i, e in enumerate(entries, start=1)]
|
|
resultados = await asyncio.gather(*tasks)
|
|
return list(resultados)
|
|
|
|
# --- Helpers ---
|
|
|
|
def _expand_alias_map(self, alias_map: dict) -> dict:
|
|
"""Expande alias con wildcard `/*/' al sub-campo equivalente."""
|
|
expanded = dict(alias_map)
|
|
for orig_key, alias_value in list(alias_map.items()):
|
|
if "/*/" in orig_key:
|
|
sub = orig_key.split("/*/")[-1]
|
|
if sub and sub not in expanded:
|
|
expanded[sub] = alias_value
|
|
return expanded
|
|
|
|
def _detect_document_force_target(self, alias_map: Optional[Dict[str, str]]) -> Optional[str]:
|
|
if not alias_map:
|
|
return None
|
|
for alias_value in alias_map.values():
|
|
target = str(alias_value or "").strip()
|
|
if target.lower() in DOCUMENT_FORCE_TARGETS:
|
|
return target
|
|
return None
|
|
|
|
def _expand_document_force_fields(self, fields: List[str], alias_map: Optional[Dict[str, str]]) -> List[str]:
|
|
"""Si existe `documento_forzate`, incluye todos los tipos documentales válidos del mismo nivel."""
|
|
target = self._detect_document_force_target(alias_map)
|
|
if not target:
|
|
return list(fields or [])
|
|
|
|
allowed_types = list(get_allowed_document_types())
|
|
expanded = list(fields or [])
|
|
seen = set(expanded)
|
|
|
|
for orig_key, alias_value in (alias_map or {}).items():
|
|
if str(alias_value or "").strip().lower() not in DOCUMENT_FORCE_TARGETS:
|
|
continue
|
|
|
|
base = str(orig_key or "").strip()
|
|
if not base:
|
|
continue
|
|
|
|
if "/" in base:
|
|
prefix = base.rsplit("/", 1)[0]
|
|
candidates = [f"{prefix}/{doc_type}" for doc_type in allowed_types]
|
|
else:
|
|
candidates = allowed_types
|
|
|
|
for candidate in candidates:
|
|
if candidate not in seen:
|
|
expanded.append(candidate)
|
|
seen.add(candidate)
|
|
|
|
return expanded
|
|
|
|
def _expand_document_force_alias_map(self, alias_map: Optional[Dict[str, str]], fields: List[str]) -> Dict[str, str]:
|
|
"""Mapea todos los tipos documentales válidos al mismo alias especial."""
|
|
expanded = dict(alias_map or {})
|
|
target = self._detect_document_force_target(expanded)
|
|
if not target:
|
|
return expanded
|
|
|
|
allowed_types = list(get_allowed_document_types())
|
|
for field in fields or []:
|
|
field_str = str(field or "").strip()
|
|
if not field_str:
|
|
continue
|
|
key_tail = field_str.split("/")[-1].upper()
|
|
if key_tail not in allowed_types:
|
|
continue
|
|
expanded[field_str] = target
|
|
|
|
return expanded
|
|
|
|
def _prepare_csv_header(self, fields, alias_map, override):
|
|
if override:
|
|
return override
|
|
header_base = ['radicado'] + list(fields or [])
|
|
if not alias_map:
|
|
return header_base
|
|
|
|
alias_map2 = self._expand_alias_map(alias_map)
|
|
raw_header = [alias_map2.get(h, h) for h in header_base]
|
|
seen: set = set()
|
|
final_header = []
|
|
for h in raw_header:
|
|
if h not in seen:
|
|
final_header.append(h)
|
|
seen.add(h)
|
|
return final_header
|
|
|
|
def _is_effectively_empty_output(self, data: Dict[str, Any], fields: List[str], alias_map: Optional[Dict[str, str]]) -> bool:
|
|
"""Verdadero cuando la salida final solo conserva el radicado y no trae datos útiles."""
|
|
try:
|
|
rows, _header = rows_from_data(data, allowed_paths=fields)
|
|
final_rows = self._apply_alias_to_rows(rows, alias_map)
|
|
except Exception:
|
|
return False
|
|
|
|
if not final_rows:
|
|
return True
|
|
|
|
for row in final_rows:
|
|
if not isinstance(row, dict):
|
|
continue
|
|
for key, value in row.items():
|
|
if str(key).strip().lower() == "radicado":
|
|
continue
|
|
if value is None:
|
|
continue
|
|
if isinstance(value, str) and not value.strip():
|
|
continue
|
|
return False
|
|
return True
|
|
|
|
def _inject_document_force_fields_from_raw(self, normalized_obj: Dict[str, Any], raw_obj: Dict[str, Any], selected_fields: List[str]) -> Dict[str, Any]:
|
|
"""Inyecta campos documentales desde la extracción cruda si el normalizador los omitió."""
|
|
result = dict(normalized_obj or {})
|
|
selected_keys = {str(p or "").split("/")[-1].strip().upper() for p in (selected_fields or [])}
|
|
candidates = self._collect_document_force_candidates(raw_obj)
|
|
|
|
for dtype, value in candidates.items():
|
|
if selected_keys and dtype not in selected_keys:
|
|
continue
|
|
if dtype not in result or result.get(dtype) in (None, ""):
|
|
result[dtype] = value
|
|
return result
|
|
|
|
def _collect_document_force_candidates(self, obj: Any) -> Dict[str, Any]:
|
|
"""Recorre el JSON crudo buscando claves tipo CC/TI/EPT/PT con sus valores."""
|
|
found: Dict[str, Any] = {}
|
|
valid_keys = set(get_allowed_document_types()) | DOCUMENT_TYPE_HINT_ALIASES
|
|
|
|
def walk(node: Any) -> None:
|
|
if isinstance(node, dict):
|
|
for key, value in node.items():
|
|
key_tail = str(key or "").split("/")[-1].strip().upper()
|
|
if key_tail in valid_keys and value not in (None, ""):
|
|
found[key_tail] = value
|
|
walk(value)
|
|
elif isinstance(node, list):
|
|
for item in node:
|
|
walk(item)
|
|
|
|
walk(obj)
|
|
return found
|
|
|
|
def _notify_progress(self, callback, rad, idx, total):
|
|
if callback:
|
|
try:
|
|
callback(rad, idx, total)
|
|
except Exception:
|
|
pass
|
|
|
|
# Notificar a tracking web con contadores reales acumulados
|
|
try:
|
|
self._track_success = idx - self._track_errors
|
|
# Log de radicado completado exitosamente
|
|
if rad and self._track_errors < idx:
|
|
self.tracking_manager.log(
|
|
f"Radicado completado ({self._track_success}/{total} exitosos)",
|
|
level="success",
|
|
radicado=str(rad)
|
|
)
|
|
if idx >= total:
|
|
self.tracking_manager.complete_batch(
|
|
success_count=self._track_success,
|
|
error_count=self._track_errors,
|
|
total_tokens=self._track_tokens,
|
|
total_cost_usd=self._track_cost_usd,
|
|
)
|
|
else:
|
|
self.tracking_manager.update_progress(
|
|
current_index=idx,
|
|
success_count=self._track_success,
|
|
error_count=self._track_errors,
|
|
total_tokens=self._track_tokens,
|
|
total_cost_usd=self._track_cost_usd,
|
|
status="processing"
|
|
)
|
|
except Exception as e:
|
|
logger.debug(f"Error al notificar progreso a tracking: {e}")
|
|
|
|
def _accumulate_usage(self, agg, new_usage):
|
|
accumulate_usage(agg, new_usage)
|
|
|
|
def _report_usage(self, callback, rad, phase, usage):
|
|
if not usage:
|
|
return
|
|
try:
|
|
model = None
|
|
if phase == "extract":
|
|
model = getattr(self.service.extractor, "model", None)
|
|
elif phase == "normalize":
|
|
model = "gpt-4.1-mini"
|
|
|
|
cost = compute_cost_from_usage(usage, phase, model=model)
|
|
|
|
# Acumular tokens y costos para el tracking web
|
|
prompt_tokens = int(usage.get("prompt_tokens") or 0)
|
|
completion_tokens = int(usage.get("completion_tokens") or 0)
|
|
self._track_tokens += prompt_tokens + completion_tokens
|
|
if isinstance(cost, dict):
|
|
self._track_cost_usd += float(cost.get("total_usd") or 0.0)
|
|
|
|
if callback:
|
|
callback(phase, rad, usage, cost)
|
|
except Exception:
|
|
pass
|
|
|
|
def _write_csv_multidoc(self, path, data, fields, doc_types, prof, header, alias_map, rad):
|
|
try:
|
|
agg_base = prof.get('aggregate_list_paths') or None
|
|
agg_paths_expanded = []
|
|
if agg_base:
|
|
for base in agg_base:
|
|
for dt in doc_types:
|
|
tname = str(dt.get("name") or "").strip() or "Tipo"
|
|
agg_paths_expanded.append(f"{tname}/{base}")
|
|
|
|
rows, computed_header = rows_from_data(
|
|
data,
|
|
aggregate_list_paths=agg_paths_expanded or None,
|
|
allowed_paths=fields,
|
|
)
|
|
final_header = header if header else computed_header
|
|
final_rows = self._apply_alias_to_rows(rows, alias_map)
|
|
self.csv_manager.append_rows(path, final_rows, final_header)
|
|
except Exception:
|
|
logger.exception(f"Error escribiendo CSV para radicado {rad}", extra={"radicado": rad})
|
|
|
|
def _write_csv_singledoc(self, path, data, fields, prof, header, alias_map, rad):
|
|
try:
|
|
rows, computed_header = rows_from_data(
|
|
data,
|
|
aggregate_list_paths=(prof.get('aggregate_list_paths') or None),
|
|
allowed_paths=fields,
|
|
)
|
|
final_header = header if header else computed_header
|
|
final_rows = self._apply_alias_to_rows(rows, alias_map)
|
|
self.csv_manager.append_rows(path, final_rows, final_header)
|
|
except Exception:
|
|
logger.exception(f"Error escribiendo CSV para radicado {rad}", extra={"radicado": rad})
|
|
|
|
def _apply_alias_to_rows(self, rows, alias_map):
|
|
if not alias_map:
|
|
return rows
|
|
alias_map2 = self._expand_alias_map(alias_map)
|
|
|
|
aliased_rows = []
|
|
for r in rows:
|
|
if isinstance(r, dict):
|
|
new_r = {}
|
|
for k, v in r.items():
|
|
new_k = alias_map2.get(k, k)
|
|
v = self._format_document_type_value(k, new_k, v)
|
|
|
|
if new_k in new_r:
|
|
# Lógica de fusión solicitada:
|
|
# 1. Si son iguales, dejar uno.
|
|
# 2. Si son diferentes, separar por coma.
|
|
prev_val = new_r[new_k]
|
|
|
|
# Normalizar a string para comparación y fusión
|
|
s_prev = str(prev_val).strip() if prev_val is not None else ""
|
|
s_curr = str(v).strip() if v is not None else ""
|
|
|
|
if s_prev == s_curr:
|
|
continue
|
|
|
|
if not s_prev:
|
|
new_r[new_k] = v
|
|
elif not s_curr:
|
|
pass
|
|
else:
|
|
new_r[new_k] = f"{s_prev}, {s_curr}"
|
|
else:
|
|
new_r[new_k] = v
|
|
aliased_rows.append(new_r)
|
|
else:
|
|
aliased_rows.append(r)
|
|
return aliased_rows
|
|
|
|
def _format_document_type_value(self, original_key, aliased_key, value):
|
|
"""Compone 'CC 123456' cuando una clave de tipo documental se exporta como 'tipo de documento'."""
|
|
if value is None:
|
|
return value
|
|
|
|
document_type_keys = set(get_allowed_document_types())
|
|
key_tail = str(original_key or "").split("/")[-1].strip().upper()
|
|
target = str(aliased_key or "").strip().lower()
|
|
if key_tail not in document_type_keys or target not in DOCUMENT_FORCE_TARGETS:
|
|
return value
|
|
|
|
text = str(value).strip()
|
|
if not text:
|
|
return value
|
|
|
|
# Si ya viene prefijado con el mismo tipo, no duplicarlo.
|
|
if text.upper().startswith(f"{key_tail} "):
|
|
return text
|
|
|
|
return f"{key_tail} {text}"
|
|
|
|
def _handle_error_async(self, rad, e, csv_path, header, pdf_path=None):
|
|
error_obj = {
|
|
"radicado": rad,
|
|
"error": str(e)
|
|
}
|
|
if pdf_path: error_obj["_source_pdf"] = pdf_path
|
|
logger.exception(f"Error procesando radicado {rad}", extra={"radicado": rad})
|
|
|
|
# Acumular error en el contador de tracking
|
|
self._track_errors += 1
|
|
error_message = str(e)
|
|
try:
|
|
self.tracking_manager.report_error(f"[{rad}] {error_message}")
|
|
except Exception:
|
|
pass
|
|
try:
|
|
self.tracking_manager.log(
|
|
message=f"Error procesando radicado: {error_message}",
|
|
level="error",
|
|
radicado=str(rad)
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
if csv_path:
|
|
header_err = header if header else ["radicado", "error"]
|
|
try:
|
|
self.csv_manager.append_rows(csv_path, [error_obj], header_err)
|
|
except Exception:
|
|
pass
|
|
return error_obj
|