import fnmatch import json import logging import os import re import shutil import time from typing import List, Dict, Any, Optional, Tuple, Callable import pandas as pd from application.strategies import OllamaExtractor from application.database import DatabaseManager from core.pdf_utils import document_to_images_bs64 logger = logging.getLogger(__name__) class PreclassificationService: def __init__(self, extractor: OllamaExtractor): self.extractor = extractor self.db = DatabaseManager() def export_to_excel(self, run_id: int, output_path: str) -> bool: """Exporta los resultados de una ejecución a Excel (xlsx) o CSV.""" try: results = self.db.get_run_results(run_id) if not results: return False # Pivotar para tener filas únicas por archivo # get_run_results devuelve una fila por cada key encontrada en cada archivo # Necesitamos agrupar por (radicado, file_path) data_map = {} keys_seen = set() for row in results: file_id = f"{row['radicado']}|{row['file_path']}" if file_id not in data_map: data_map[file_id] = { "Radicado": row["radicado"], "Archivo": row["file_path"], "Status": row["status"] } k = row["key"] if k: # found_value puede ser None val = row["found_value"] exists = row["exists_bool"] # Podemos poner el valor o un indicador data_map[file_id][k] = val if exists else "NO_FOUND" keys_seen.add(k) # Convertir a lista final_data = list(data_map.values()) df = pd.DataFrame(final_data) # Asegurar todas las columnas de keys for k in keys_seen: if k not in df.columns: df[k] = None if output_path.endswith(".csv"): df.to_csv(output_path, index=False, encoding="utf-8-sig", sep=";") else: df.to_excel(output_path, index=False) return True except Exception as e: logger.exception("Error exportando a Excel") raise e def organize_files(self, run_id: int, base_output_path: str) -> tuple[int, int]: """ Organiza los archivos MATCH en carpetas jerárquicas. Devuelve (total_copiados, total_errores). """ run = self.db.get_run(run_id) mode = "radicados" if run: cfg = run.get("config") or {} mode = str(cfg.get("mode", "radicados") or "radicados").strip().lower() if mode not in ("radicados", "general"): mode = "radicados" results = self.db.get_run_results(run_id) # Filtrar solo MATCH match_rows = [r for r in results if r["status"] == "MATCH"] if not match_rows: return 0, 0 # Agrupar por archivo para tener el "set" de datos de cada uno files_data = {} # { file_path: {key: val, ...}, 'radicado': ... } key_values_all = {} # { key: set(values) } for row in match_rows: fpath = row["file_path"] if not fpath: continue if fpath not in files_data: files_data[fpath] = {"radicado": row["radicado"], "keys": {}} k = row["key"] v = row["found_value"] if k and v: files_data[fpath]["keys"][k] = v if k not in key_values_all: key_values_all[k] = set() key_values_all[k].add(v) # Determinar orden de jerarquía por cardinalidad # Lista de (key, cardinalidad) hierarchy = [] for k, val_set in key_values_all.items(): hierarchy.append((k, len(val_set))) # Ordenar ascendente: primero los que tienen menos valores (ej. Categorias generales) hierarchy.sort(key=lambda x: x[1]) ordered_keys = [x[0] for x in hierarchy] logger.info(f"Jerarquía de carpetas calculada: {ordered_keys}") count_ok = 0 count_err = 0 for fpath, info in files_data.items(): try: # Construir ruta relativa # Sanitizar valores para que sean nombres de carpeta válidos path_parts = [] for k in ordered_keys: val = info["keys"].get(k, "DESCONOCIDO") safe_val = self._sanitize_filename(str(val)) # KeyName / ValueName safe_key = self._sanitize_filename(k) path_parts.append(safe_key) path_parts.append(safe_val) # Añadir radicado al final solo si NO es modo general (en general el radicado es el archivo) # Si mode es radicados: ... / Radicado / Archivo # Si mode es general: ... / Archivo if mode != "general": path_parts.append(self._sanitize_filename(info["radicado"])) dest_dir = os.path.join(base_output_path, *path_parts) os.makedirs(dest_dir, exist_ok=True) fname = os.path.basename(fpath) dest_file = os.path.join(dest_dir, fname) # Evitar sobrescrituras (especialmente en modo general sin radicado) if os.path.exists(dest_file): base, ext = os.path.splitext(fname) n = 1 while True: candidate = os.path.join(dest_dir, f"{base}_{n}{ext}") if not os.path.exists(candidate): dest_file = candidate break n += 1 # Copiar shutil.copy2(fpath, dest_file) count_ok += 1 except Exception as e: logger.error(f"Error organizando {fpath}: {e}") count_err += 1 return count_ok, count_err def _sanitize_filename(self, name: str) -> str: """Limpia caracteres inválidos para rutas de archivo.""" # Reemplazar caracteres no alfanuméricos (excepto espacio, guion, punto) clean = re.sub(r'[\\/*?:"<>|]', "", name) clean = clean.replace("\n", " ").replace("\r", "").strip() if not clean: return "UNTITLED" return clean[:50] # Limitar longitud def run_preclassification( self, root_path: str, patterns: List[Dict[str, str]], # [{key, mode, value?}, ...] target_info: Dict[str, Any], # {enabled, definition, pattern} page_range: Tuple[bool, int, int], # (enabled, start, end) on_progress: Optional[Callable[[str, int, int], None]] = None, mode: str = "radicados" # "radicados" | "general" ) -> int: """ Ejecuta el flujo de preclasificación. Devuelve el run_id. """ # 1. Crear Run en BD mode = str(mode or "radicados").strip().lower() if mode not in ("radicados", "general"): mode = "radicados" run_config = { "patterns": patterns, "target": target_info, "range": page_range, "mode": mode } run_id = self.db.create_run(root_path, run_config) # 2. Descubrir items a procesar radicados = [] if os.path.isdir(root_path): if mode == "general": # En modo general, consideramos que la raiz es el único "contenedor" # o iteramos sobre una lista dummy. # Para simplificar el loop principal, usaremos una lista con un elemento "." radicados = ["."] else: radicados = sorted([ d for d in os.listdir(root_path) if os.path.isdir(os.path.join(root_path, d)) ]) total = len(radicados) # Preparar prompt keys_to_extract = [p["key"] for p in patterns] prompt = self._build_prompt(keys_to_extract) for idx, rad in enumerate(radicados, start=1): # Si es general, rad es ".", no es util para el usuario ver "Procesando ." # Ajustaremos el reporte dentro del loop de archivos si es general if mode != "general" and on_progress: on_progress(rad, idx, total) rad_path = os.path.join(root_path, rad) # 3. Identificar archivos a procesar files_to_process = self._identify_files(rad_path, target_info, mode) logger.info(f"Radicado/Carpeta: {rad} - Archivos identificados: {len(files_to_process)}") # Ordenar archivos para consistencia files_to_process.sort() if not files_to_process: if mode == "general": # Registrar una sola vez que no había archivos self.db.add_result(run_id, "GENERAL", None, "NO_FILES", "") return run_id else: self.db.add_result(run_id, rad, None, "NO_FILES", "") continue total_files = len(files_to_process) for f_idx, file_path in enumerate(files_to_process, 1): # En modo general, el "radicado" lógico será el nombre del archivo sin extensión # En modo radicados, es el nombre de la carpeta (variable rad) current_radicado = rad # Definir radicado para modo general ANTES de cualquier lógica if mode == "general": base_name = os.path.basename(file_path) name_no_ext = os.path.splitext(base_name)[0] # Validar que no quede vacio ni sea un punto if name_no_ext and name_no_ext.strip() not in (".", ""): current_radicado = name_no_ext.strip() else: current_radicado = base_name if str(current_radicado).strip() in (".", ""): current_radicado = f"FILE_{int(time.time()*1000)}_{f_idx}" logger.info(f"Procesando archivo [{f_idx}/{total_files}]: {file_path} -> Radicado: {current_radicado}") if on_progress: try: # Asegurar que pasamos un string válido safe_rad = str(current_radicado) on_progress(safe_rad, f_idx, total_files) except Exception: pass # Evitar romper el loop si la GUI falla try: # 4. Convertir a imágenes (con soporte de rango si está habilitado) images_b64 = self._get_images(file_path, page_range) if not images_b64: self.db.add_result(run_id, rad, file_path, "ERROR_IMG", "No images generated") continue # 5. LLM response = self.extractor.extract(images_b64, prompt) raw_text = response.get("raw", "") # 6. Parseo y Validación parsed = self._parse_json_response(raw_text) # Guardar resultado cabecera result_id = self.db.add_result(run_id, current_radicado, file_path, "PROCESSED", raw_text) # Guardar detalles y calcular match global global_match = True if not parsed: global_match = False # Guardar error de parseo? else: for pattern in patterns: key = pattern["key"] pat_mode = pattern.get("mode", "key") expected_val = pattern.get("value", "").lower().strip() # Buscar en respuesta # Normalizamos claves para busqueda insensible a mayusculas found_item = None for k, v in parsed.items(): if k.lower() == key.lower(): found_item = v break exists = False found_val = None if found_item and isinstance(found_item, dict): exists = bool(found_item.get("exists", False)) found_val = str(found_item.get("value", "") or "") # Regla rule_ok = False if pat_mode == "key": rule_ok = exists elif pat_mode == "key+value": # Comparacion laxa rule_ok = exists and (expected_val in found_val.lower()) if not rule_ok: global_match = False self.db.add_match(result_id, key, expected_val if pat_mode=="key+value" else None, found_val, exists) # Actualizar status final final_status = "MATCH" if global_match else "NO_MATCH" self._update_result_status(result_id, final_status) except Exception as e: logger.exception(f"Error procesando {file_path}") self.db.add_result(run_id, current_radicado, file_path, f"ERROR: {str(e)}", "") return run_id def _identify_files(self, rad_path: str, target_info: Dict[str, Any], mode: str = "radicados") -> List[str]: """Devuelve lista de rutas de archivos a procesar.""" if not os.path.exists(rad_path): return [] # Si mode es general, rad_path es la raiz y buscamos archivos directos. # Si mode es radicados, rad_path es la subcarpeta del radicado. # La logica es la misma: listar archivos en el directorio dado. all_files = [] try: for f in os.listdir(rad_path): full_path = os.path.join(rad_path, f) if os.path.isfile(full_path) and f.lower().endswith(('.pdf', '.tif', '.tiff')): all_files.append(full_path) except Exception: pass if not target_info.get("enabled"): return all_files pattern = target_info.get("pattern", "*").lower() filtered = [f for f in all_files if fnmatch.fnmatch(os.path.basename(f).lower(), pattern)] return filtered def _get_images(self, file_path: str, page_range: Tuple[bool, int, int]) -> List[str]: """Convierte archivo a imagenes, aplicando rango si es necesario.""" enabled, start, end = page_range if enabled: from core.pdf_utils import pdf_pages_to_images_bs64 ext = os.path.splitext(file_path)[1].lower() s, e = max(1, start), max(1, end) if s > e: s, e = e, s if ext == ".pdf": return pdf_pages_to_images_bs64(file_path, s, e, dpi=300) elif ext in (".tif", ".tiff"): return document_to_images_bs64(file_path, tiff_page_range=(s, e), dpi=300) return document_to_images_bs64(file_path, dpi=300) def _build_prompt(self, keys: List[str]) -> str: keys_str = ", ".join(keys) return f"""Responde con true o false si tiene los siguientes campos, si existe extrae cuales son. Responde ÚNICAMENTE en JSON con el siguiente formato, sin bloques de código markdown: {{ "nombre_campo": {{ "exists": true, "value": "valor extraído" }}, ... }} Campos a buscar: {keys_str}""" def _parse_json_response(self, text: str) -> Dict[str, Any]: try: # Limpiar bloques markdown si el modelo los pone clean = text.replace("```json", "").replace("```", "").strip() return json.loads(clean) except Exception: # Intento de rescate simple con regex si falla json puro # Por ahora retornamos vacio para indicar fallo return {} def _update_result_status(self, result_id: int, status: str): self.db.update_result_status(result_id, status)