"""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 merge_verbatim_long_strings, 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, ) merge_verbatim_long_strings(raw_data, normalized_obj, 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, ) merge_verbatim_long_strings(raw_data, 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