Spaces:
Sleeping
Sleeping
| """ | |
| 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") |