449 lines
18 KiB
Python
449 lines
18 KiB
Python
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:
|
|
"""
|
|
Servicio para la preclasificación masiva de documentos.
|
|
|
|
Permite identificar si un documento cumple con ciertos criterios (patterns)
|
|
y extraer información clave para organizar los archivos físicamente en carpetas.
|
|
"""
|
|
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 de preclasificación a Excel o CSV.
|
|
|
|
Args:
|
|
run_id: ID de la ejecución en la base de datos.
|
|
output_path: Ruta del archivo de salida (.xlsx o .csv).
|
|
|
|
Returns:
|
|
True si la exportación fue exitosa, False si no hay datos.
|
|
"""
|
|
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]:
|
|
"""
|
|
Copia físicamente los archivos que hicieron 'MATCH' a una nueva estructura de carpetas.
|
|
La jerarquía se determina automáticamente basada en los valores extraídos.
|
|
|
|
Args:
|
|
run_id: ID de la ejecución.
|
|
base_output_path: Directorio donde se creará la nueva estructura.
|
|
|
|
Returns:
|
|
Tupla (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 en Windows/Linux."""
|
|
# 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 completo de preclasificación sobre un directorio.
|
|
|
|
Args:
|
|
root_path: Directorio con los documentos.
|
|
patterns: Reglas de validación (ej. buscar NIT, buscar Fecha).
|
|
target_info: Filtro de archivos por nombre/patrón.
|
|
page_range: Rango de páginas a procesar por documento.
|
|
on_progress: Callback para reportar avance a la GUI.
|
|
mode: 'radicados' (carpetas) o 'general' (archivos sueltos).
|
|
|
|
Returns:
|
|
ID de la ejecución (run_id) generada.
|
|
"""
|
|
# 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)
|