""" MIT License Autor: Fernando Lafuente Proyecto: Ejercicio opcional NLP - Chunking y Recuperación Descripción: Configuración de sistema de logging y utilidades generales Este módulo proporciona utilidades para logging estructurado y funciones auxiliares que son utilizadas por múltiples componentes del sistema. Centraliza la configuración de logging para mantener consistencia en toda la aplicación. """ import logging import sys from datetime import datetime from typing import Any, Optional import settings def setup_logging( level: str = settings.LOG_LEVEL, format_string: str = settings.LOG_FORMAT ) -> logging.Logger: """ Configura el sistema de logging de la aplicación con formato estándar. Esta función establece un logger principal que será utilizado por todos los módulos del sistema para mantener un formato consistente y facilitar el debugging y monitoreo de operaciones. Args: level: Nivel de logging (DEBUG, INFO, WARNING, ERROR) format_string: Formato de los mensajes de log Returns: logging.Logger: Logger configurado para la aplicación Example: logger = setup_logging() logger.info("Sistema de chunking inicializado correctamente") """ # Convertir string de nivel a constante de logging numeric_level = getattr(logging, level.upper(), None) if not isinstance(numeric_level, int): raise ValueError(f"Nivel de logging inválido: {level}") # Configurar formato del logger formatter = logging.Formatter(format_string) # Configurar handler para consola console_handler = logging.StreamHandler(sys.stdout) console_handler.setFormatter(formatter) # Crear y configurar logger principal logger = logging.getLogger("chunking_system") logger.setLevel(numeric_level) logger.addHandler(console_handler) # Evitar duplicación de mensajes logger.propagate = False return logger def log_operation_time(func): """ Decorador que mide y registra el tiempo de ejecución de una función. Este decorador es útil para monitorear el rendimiento de operaciones críticas como chunking, generación de embeddings y búsquedas vectoriales. Args: func: Función a decorar Returns: Función decorada que registra su tiempo de ejecución Example: @log_operation_time def process_document(document): # procesar documento return result """ def wrapper(*args, **kwargs): logger = logging.getLogger("chunking_system") start_time = datetime.now() logger.info(f"Iniciando operación: {func.__name__}") try: result = func(*args, **kwargs) end_time = datetime.now() duration = (end_time - start_time).total_seconds() logger.info(f"Operación completada: {func.__name__} " f"(duración: {duration:.2f} segundos)") return result except Exception as e: end_time = datetime.now() duration = (end_time - start_time).total_seconds() logger.error(f"Error en operación: {func.__name__} " f"(duración: {duration:.2f} segundos) - {str(e)}") raise return wrapper def safe_division(numerator: float, denominator: float, default: float = 0.0) -> float: """ Realiza división segura evitando errores por división entre cero. Esta función es especialmente útil al calcular métricas de similitud y normalización de vectores, donde pueden aparecer casos límite. Args: numerator: Numerador de la división denominator: Denominador de la división default: Valor por defecto si el denominador es cero Returns: float: Resultado de la división o valor por defecto Example: # Cálculo seguro de similitud coseno similarity = safe_division(dot_product, norm_product, 0.0) """ if abs(denominator) < 1e-10: # Evitar divisiones por números muy pequeños return default return numerator / denominator def clean_text_basic(text: str) -> str: """ Aplica limpieza básica a un texto eliminando caracteres problemáticos. Esta función preprocesa el texto para mejorar la calidad del chunking y la generación de embeddings, eliminando caracteres que pueden introducir ruido en el procesamiento posterior. Args: text: Texto a limpiar Returns: str: Texto limpio y normalizado Example: texto_limpio = clean_text_basic("Texto con\n\nsaltos\t\tde línea") # Resultado: "Texto con saltos de línea" """ if not text or not isinstance(text, str): return "" # Eliminar caracteres especificados en configuración for char in settings.CHARS_TO_REMOVE: text = text.replace(char, "") # Reemplazar caracteres especificados for old_char, new_char in settings.CHARS_TO_REPLACE.items(): text = text.replace(old_char, new_char) # Normalizar espacios múltiples while " " in text: text = text.replace(" ", " ") # Eliminar espacios al inicio y final text = text.strip() return text def validate_text_length(text: str, min_length: int = settings.MIN_TEXT_LENGTH) -> bool: """ Valida que un texto tenga longitud suficiente para ser procesado. Esta función evita procesar chunks demasiado cortos que no aportan información semántica útil y pueden introducir ruido en la búsqueda. Args: text: Texto a validar min_length: Longitud mínima requerida Returns: bool: True si el texto es válido, False en caso contrario Example: if validate_text_length(chunk): # Procesar chunk válido embeddings = generate_embeddings(chunk) """ if not text or not isinstance(text, str): return False # Limpiar texto antes de validar longitud clean_text = clean_text_basic(text) return len(clean_text) >= min_length def format_similarity_score(score: float, decimals: int = 4) -> str: """ Formatea una puntuación de similitud para mostrar al usuario. Args: score: Puntuación de similitud (0-1) decimals: Número de decimales a mostrar Returns: str: Puntuación formateada como string Example: formatted = format_similarity_score(0.85739) # Resultado: "0.8574" """ if not isinstance(score, (int, float)): return "N/A" return f"{score:.{decimals}f}" def truncate_text(text: str, max_length: int = 100, suffix: str = "...") -> str: """ Trunca un texto a una longitud máxima agregando sufijo si es necesario. Útil para mostrar previews de chunks en logs o resultados sin saturar la consola con textos muy largos. Args: text: Texto a truncar max_length: Longitud máxima permitida suffix: Sufijo a agregar si se trunca Returns: str: Texto truncado o original si no excede la longitud Example: preview = truncate_text(long_chunk, 50) logger.info(f"Chunk procesado: {preview}") """ if not text or len(text) <= max_length: return text return text[:max_length - len(suffix)] + suffix def get_current_timestamp() -> str: """ Obtiene timestamp actual en formato legible para logs. Returns: str: Timestamp formateado Example: timestamp = get_current_timestamp() # Resultado: "2024-01-15 14:30:45" """ return datetime.now().strftime("%Y-%m-%d %H:%M:%S")