# Archivo: ValidarScore.py # Ruta: App\SupyCtrol_Module\IngenieroControl\ValidarScore.py # Descripción: Módulo para validar el Score descargado desde SharePoint # Autor: Equipo de Desarrollo IGSA # Fecha: 2025 import os import requests import pandas as pd from io import BytesIO from datetime import datetime from flask import jsonify from config import DevelopmentConfig, ProductionConfig from Consultas_SQL.SupYCtrol.IngDeControl.ValidarScoreSQL import (obtener_catalogo_validacion, guardar_log_validacion, obtener_cantidades_componentes) from App.SupyCtrol_Module.IngenieroControl.ValidarScoreHelpers import ValidadorTiposDatos, limpiar_dataframe # Configuración según ambiente FLASK_ENV = os.getenv('FLASK_ENV', 'development') Config = ProductionConfig if FLASK_ENV == 'production' else DevelopmentConfig # Configuración Excel url_sharepoint_global = "https://igsa1-my.sharepoint.com/personal/alexis_moreno_igsa_com_mx/_layouts/15/download.aspx?share=EStyzV5jqTRGqhyxo4jfzmIBYCfPhbrU4xMICcpWqMSqiw" nombre_hoja_global = "ScoreV2" nombre_tabla_excel_global = "ScoreV2" # CONFIGURACIÓN DE PANDAS PARA MOSTRAR DATAFRAMES COMPLETOS pd.set_option('display.max_columns', None) # Mostrar todas las columnas pd.set_option('display.max_rows', None) # Mostrar todas las filas pd.set_option('display.width', None) # Sin límite de ancho pd.set_option('display.max_colwidth', 50) # Máximo 50 caracteres por columna class ValidadorScore: """ Clase principal para manejar la validación del Score """ def __init__(self): self.url_sharepoint = url_sharepoint_global self.nombre_hoja = nombre_hoja_global self.nombre_tabla_excel = nombre_tabla_excel_global self.ruta_descarga = self._obtener_ruta_descarga() self.catalogo_validacion = None self.df_score = None self.errores_validacion = [] self.timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") def _obtener_ruta_descarga(self): """ 1.2 Configura la ruta de descarga según el ambiente Returns: str: Ruta donde se guardará el archivo """ if FLASK_ENV == 'production': # Ruta fija en Linux para producción ruta = "/var/www/elephant/temp/score_validaciones" else: # Ruta temporal para desarrollo ruta = os.path.join(os.getcwd(), "temp", "score_validaciones") # Crear directorio si no existe os.makedirs(ruta, exist_ok=True) return ruta def inicializar_catalogo(self): """ 1.3 Consulta la tabla CM_DataTypeValidator y crea el diccionario Returns: bool: True si se inicializó correctamente, False en caso contrario """ try: self.catalogo_validacion = obtener_catalogo_validacion() if not self.catalogo_validacion: print("Error: No se pudo obtener el catálogo de validación") return False print(f"Catálogo de validación cargado: {len(self.catalogo_validacion)} columnas") return True except Exception as e: print(f"Error al inicializar catálogo: {str(e)}") return False def descargar_excel(self): """ 2. Descarga el archivo Excel desde SharePoint Returns: str: Ruta del archivo descargado o None si hubo error """ try: print("Descargando archivo desde SharePoint...") response = requests.get(self.url_sharepoint, timeout=30) if response.status_code != 200: print(f"Error al descargar: código {response.status_code}") return None # 2.3 Guardar archivo con timestamp nombre_archivo = f"Score_{self.timestamp}.xlsx" ruta_completa = os.path.join(self.ruta_descarga, nombre_archivo) with open(ruta_completa, 'wb') as f: f.write(response.content) print(f"Archivo descargado: {ruta_completa}") # 2.4 Limpiar archivos antiguos (solo en producción) if FLASK_ENV == 'production': self._limpiar_archivos_antiguos() return ruta_completa except Exception as e: print(f"Error al descargar Excel: {str(e)}") return None def _limpiar_archivos_antiguos(self): """ 2.4 Elimina archivos anteriores a 3 meses """ try: fecha_limite = datetime.now() mes_actual = fecha_limite.month año_actual = fecha_limite.year if mes_actual <= 3: mes_limite = 12 - (3 - mes_actual) año_limite = año_actual - 1 else: mes_limite = mes_actual - 3 año_limite = año_actual fecha_limite_str = f"{año_limite}{mes_limite:02d}" archivos = os.listdir(self.ruta_descarga) for archivo in archivos: if archivo.startswith("Score_") and archivo.endswith(".xlsx"): try: fecha_archivo = archivo.split("_")[1][:6] if fecha_archivo < fecha_limite_str: ruta_archivo = os.path.join(self.ruta_descarga, archivo) os.remove(ruta_archivo) print(f"Archivo antiguo eliminado: {archivo}") except: continue except Exception as e: print(f"Error al limpiar archivos antiguos: {str(e)}") def _detectar_errores_excel_nativos(self, ruta_archivo): """ Detecta errores nativos de Excel usando OpenPyXL (como #VALOR!, #REF!, #DIV/0!, #N/A, #NAME?, #NULL!, #NUM!) CON PRINTS DETALLADOS PARA DEBUG Args: ruta_archivo (str): Ruta del archivo Excel Returns: list: Lista de errores encontrados en formato estándar """ from openpyxl import load_workbook from openpyxl.cell.cell import TYPE_ERROR errores_excel = [] try: print("\n" + "="*100) print("🔍 INICIANDO DETECCIÓN DE ERRORES NATIVOS DE EXCEL CON OPENPYXL") print("="*100) # ═══════════════════════════════════════════════════════════ # PASO 1: ABRIR WORKBOOK # ═══════════════════════════════════════════════════════════ print(f"\n📂 PASO 1: Abriendo archivo...") print(f" Ruta: {ruta_archivo}") print(f" Modo: data_only=False (mantiene fórmulas y detecta errores)") wb = load_workbook(ruta_archivo, data_only=True) print(f" ✅ Archivo abierto correctamente") print(f" 📊 Hojas disponibles: {wb.sheetnames}") # ═══════════════════════════════════════════════════════════ # PASO 2: VERIFICAR Y SELECCIONAR HOJA # ═══════════════════════════════════════════════════════════ print(f"\n📄 PASO 2: Seleccionando hoja '{self.nombre_hoja}'...") if self.nombre_hoja not in wb.sheetnames: print(f" ❌ ERROR: Hoja '{self.nombre_hoja}' no encontrada") print(f" Hojas disponibles: {wb.sheetnames}") wb.close() return errores_excel ws = wb[self.nombre_hoja] print(f" ✅ Hoja seleccionada correctamente") print(f" 📏 Dimensiones: {ws.dimensions}") print(f" 📊 Filas máximas: {ws.max_row}") print(f" 📊 Columnas máximas: {ws.max_column}") # ═══════════════════════════════════════════════════════════ # PASO 3: LEER ENCABEZADOS (FILA 1) # ═══════════════════════════════════════════════════════════ print(f"\n📋 PASO 3: Leyendo nombres de columnas (fila 1)...") columnas = {} for col_idx, cell in enumerate(ws[1], start=1): if cell.value: columnas[col_idx] = str(cell.value).strip() print(f" ✅ {len(columnas)} columnas detectadas") print(f"\n Primeras 10 columnas:") for idx, nombre in list(columnas.items())[:123]: print(f" Col {idx:2d}: {nombre}") if len(columnas) > 10: print(f" ... y {len(columnas) - 10} columnas más") # ═══════════════════════════════════════════════════════════ # PASO 4: RECORRER FILAS BUSCANDO ERRORES # ═══════════════════════════════════════════════════════════ print(f"\n🔎 PASO 4: Iniciando búsqueda de errores (desde fila 2)...") print(f" Se mostrarán las primeras 5 filas de datos en detalle") print("="*100) contador_errores = 0 filas_procesadas = 0 MOSTRAR_PRIMERAS_N_FILAS = 5 # Mostrar detalle de las primeras 5 filas for row_idx, row in enumerate(ws.iter_rows(min_row=2), start=2): filas_procesadas += 1 # ═══════════════════════════════════════════════════════════ # MOSTRAR DETALLE DE LAS PRIMERAS 5 FILAS # ═══════════════════════════════════════════════════════════ if filas_procesadas <= MOSTRAR_PRIMERAS_N_FILAS: print(f"\n{'─'*100}") print(f"📍 FILA {row_idx} (Fila Excel) - Índice {filas_procesadas}") print(f"{'─'*100}") celdas_con_valor = 0 celdas_vacias = 0 celdas_con_error = 0 # Mostrar primeras 10 celdas de esta fila print(f" Mostrando primeras 10 columnas:") for col_idx, cell in enumerate(row[:35], start=1): # Solo primeras 10 nombre_columna = columnas.get(col_idx, f"Col_{col_idx}") # Información de la celda valor = cell.value tipo_dato = cell.data_type coordenada = cell.coordinate # Formatear valor para mostrar if valor is None or valor == '': valor_mostrar = "[VACÍO]" celdas_vacias += 1 else: valor_mostrar = str(valor)[:50] # Limitar a 50 caracteres celdas_con_valor += 1 # Detectar si es error es_error = tipo_dato == TYPE_ERROR if es_error: celdas_con_error += 1 marcador = "❌ ERROR" else: marcador = "✓" # Mostrar información de la celda print(f" {marcador} [{coordenada}] {nombre_columna:<30} = {valor_mostrar:<40} (tipo: {tipo_dato})") # Resumen de la fila total_celdas_fila = len(row) print(f"\n 📊 Resumen fila {row_idx}:") print(f" • Total celdas: {total_celdas_fila}") print(f" • Con valor: {celdas_con_valor}") print(f" • Vacías: {celdas_vacias}") print(f" • Con ERROR: {celdas_con_error}") if celdas_con_error > 0: print(f" ⚠️ ¡FILA CON ERRORES DETECTADOS!") # ═══════════════════════════════════════════════════════════ # BUSCAR ERRORES EN TODAS LAS CELDAS DE LA FILA # ═══════════════════════════════════════════════════════════ for col_idx, cell in enumerate(row, start=1): # Verificar si la celda contiene un error if cell.data_type == TYPE_ERROR: contador_errores += 1 nombre_columna = columnas.get(col_idx, f"Columna_{col_idx}") valor_error = str(cell.value) if cell.value else "#ERROR!" coordenada = cell.coordinate # Agregar al formato estándar de errores errores_excel.append({ 'fila': row_idx, 'columna': nombre_columna, 'valor': valor_error, 'tipo_esperado': 'ERROR_EXCEL', 'error': f"🚫 Error nativo de Excel detectado: {valor_error}" }) # Print detallado del error if filas_procesadas <= MOSTRAR_PRIMERAS_N_FILAS: print(f"\n 🚨 ERROR #{contador_errores} ENCONTRADO:") print(f" Celda: {coordenada}") print(f" Fila Excel: {row_idx}") print(f" Columna: {nombre_columna}") print(f" Valor: {valor_error}") else: # Para filas posteriores, solo log simple print(f" ❌ Error #{contador_errores}: Fila {row_idx}, Columna '{nombre_columna}' [{coordenada}]: {valor_error}") # Separador cada 100 filas para no saturar la consola if filas_procesadas > MOSTRAR_PRIMERAS_N_FILAS and filas_procesadas % 100 == 0: print(f" ... Procesadas {filas_procesadas} filas ({contador_errores} errores hasta ahora)...") # ═══════════════════════════════════════════════════════════ # PASO 5: CERRAR WORKBOOK # ═══════════════════════════════════════════════════════════ print(f"\n{'='*100}") print(f"🔒 PASO 5: Cerrando workbook y liberando memoria...") wb.close() print(f" ✅ Workbook cerrado correctamente") # ═══════════════════════════════════════════════════════════ # PASO 6: RESUMEN FINAL # ═══════════════════════════════════════════════════════════ print(f"\n{'='*100}") print(f"📊 RESUMEN FINAL - DETECCIÓN DE ERRORES NATIVOS DE EXCEL") print(f"{'='*100}") print(f" 📄 Archivo: {ruta_archivo.split('/')[-1]}") print(f" 📋 Hoja: {self.nombre_hoja}") print(f" 📏 Filas procesadas: {filas_procesadas}") print(f" 📊 Columnas analizadas: {len(columnas)}") print(f" 🔍 Total celdas revisadas: {filas_procesadas * len(columnas):,}") if contador_errores > 0: print(f"\n ⚠️ ERRORES DETECTADOS: {contador_errores}") print(f"\n Detalle de errores encontrados:") # Contar tipos de error tipos_error = {} for error in errores_excel: tipo = error['valor'] tipos_error[tipo] = tipos_error.get(tipo, 0) + 1 for tipo, cantidad in sorted(tipos_error.items()): print(f" • {tipo}: {cantidad} ocurrencia(s)") print(f"\n 📋 Lista completa de errores:") for i, error in enumerate(errores_excel, 1): print(f" {i:3d}. Fila {error['fila']:4d} | {error['columna']:<30} | {error['valor']}") else: print(f"\n ✅ NO SE DETECTARON ERRORES NATIVOS DE EXCEL") print(f" 🎉 Todas las celdas están libres de errores de fórmulas") print(f"{'='*100}\n") return errores_excel except Exception as e: print(f"\n{'='*100}") print(f"❌ ERROR CRÍTICO EN DETECCIÓN DE ERRORES DE EXCEL") print(f"{'='*100}") print(f" Error: {str(e)}") print(f"\n Stack trace completo:") import traceback traceback.print_exc() print(f"{'='*100}\n") return errores_excel def leer_excel(self, ruta_archivo): """ 3. Lee el archivo Excel y carga los datos en un DataFrame Args: ruta_archivo (str): Ruta del archivo a leer Returns: bool: True si se leyó correctamente, False en caso contrario """ try: print("Leyendo archivo Excel...") excel_file = pd.ExcelFile(ruta_archivo) if self.nombre_hoja not in excel_file.sheet_names: print(f"Error: La hoja '{self.nombre_hoja}' no existe en el archivo") print(f"Hojas disponibles: {excel_file.sheet_names}") return False self.df_score = pd.read_excel(ruta_archivo, sheet_name=self.nombre_hoja, dtype=str) # Limpiar nombres de columnas self.df_score.columns = [col.strip() for col in self.df_score.columns] print(f"Archivo leído correctamente: {len(self.df_score)} filas, {len(self.df_score.columns)} columnas") # PRINTS DETALLADOS DEL DATAFRAME print("\n" + "="*150) print("📊 PRIMERAS 5 FILAS DEL DATAFRAME") print("="*150) # Mostrar primeras 5 filas fila por fila num_filas_mostrar = min(3, len(self.df_score)) for idx in range(num_filas_mostrar): fila_excel = idx + 2 # +2 porque Excel empieza en 1 y tiene encabezado print(f"\n{'='*150}") print(f"FILA NÚMERO {fila_excel} (índice DataFrame: {idx})") print(f"{'='*150}") # Transponer para mostrar columnas como filas (más legible) fila_df = self.df_score.iloc[[idx]].T fila_df.columns = ['Valor'] # Mostrar columna por columna for columna_nombre, valor in self.df_score.iloc[idx].items(): # Formatear el valor if pd.isna(valor): valor_formateado = "[NULL]" elif valor == '': valor_formateado = "[VACÍO]" else: valor_formateado = str(valor)[:100] # Limitar a 100 caracteres # Tipo de dato tipo = type(valor).__name__ print(f" {columna_nombre:<45} = {valor_formateado:<55} [{tipo}]") print("\n" + "="*150) print("FIN DE PRIMERAS 5 FILAS") print("="*150) # Resumen de tipos de datos print("\n📋 RESUMEN DE TIPOS DE DATOS:") print(self.df_score.dtypes.to_string()) # Resumen de valores nulos print("\n⚠️ RESUMEN DE VALORES NULOS:") nulos = self.df_score.isnull().sum() nulos_con_datos = nulos[nulos > 0] if len(nulos_con_datos) > 0: print(nulos_con_datos.to_string()) else: print("✓ No hay valores nulos") print("\n" + "="*150 + "\n") return True except Exception as e: print(f"Error al leer Excel: {str(e)}") import traceback traceback.print_exc() return False def validar_estructura(self): """ 4.1 y 4.2 Valida que al menos existan columnas en común YA NO valida orden ni que sean exactamente las mismas Solo valida las columnas con Origin = 'Manual' Returns: tuple: (es_valido: bool, errores: list) """ errores = [] try: columnas_excel = self.df_score.columns.tolist() columnas_catalogo = list(self.catalogo_validacion.keys()) # Encontrar columnas en común columnas_comunes = set(columnas_excel) & set(columnas_catalogo) if len(columnas_comunes) == 0: errores.append({ 'tipo': 'ESTRUCTURA', 'mensaje': 'No hay columnas en común entre el Excel y el catálogo (Origin=Manual). Verifica los nombres.' }) return False, errores # Columnas que NO coinciden columnas_solo_excel = sorted(set(columnas_excel) - set(columnas_catalogo)) columnas_solo_catalogo = sorted(set(columnas_catalogo) - set(columnas_excel)) # PRINTS DETALLADOS print("\n" + "="*80) print("ANÁLISIS DE COLUMNAS") print("="*80) print(f"\n✓ Columnas con Origin='Manual' en común para validar: {len(columnas_comunes)}") print("\nListado de columnas que se validarán:") print("-" * 80) for i, col in enumerate(sorted(columnas_comunes), 1): print(f" {i:2d}. {col}") if columnas_solo_excel: print(f"\nℹ️ Columnas en Excel que NO están en catálogo Manual: {len(columnas_solo_excel)}") print(" (Estas columnas se IGNORAN en la validación)") print("-" * 80) for i, col in enumerate(columnas_solo_excel, 1): print(f" {i:2d}. {col}") if columnas_solo_catalogo: print(f"\n⚠️ Columnas en catálogo Manual que NO están en Excel: {len(columnas_solo_catalogo)}") print(" (Estas columnas deberían desactivarse en la BD)") print("-" * 80) for i, col in enumerate(columnas_solo_catalogo, 1): tipo = self.catalogo_validacion.get(col, 'N/A') print(f" {i}. {col:<40} ({tipo})") print("\n" + "="*80) print("✓ Validación de estructura exitosa") print("="*80 + "\n") return True, [] except Exception as e: errores.append({ 'tipo': 'ESTRUCTURA', 'mensaje': f"Error al validar estructura: {str(e)}" }) return False, errores def aplicar_tratamiento_datos(self): """ 4.3 Aplica limpieza y tratamiento al DataFrame Returns: tuple: (dict estadísticas, list errores_eliminacion) """ try: filas_originales = len(self.df_score) # filas eliminadas self.df_score, filas_eliminadas_detalle = limpiar_dataframe(self.df_score) filas_finales = len(self.df_score) filas_eliminadas = filas_originales - filas_finales print(f"✓ Tratamiento aplicado: {filas_eliminadas} filas eliminadas") estadisticas = { 'filas_originales': filas_originales, 'filas_finales': filas_finales, 'filas_eliminadas': filas_eliminadas } return estadisticas, filas_eliminadas_detalle except Exception as e: print(f"Error al aplicar tratamiento: {str(e)}") return None, [] def validar_tipos_datos(self): """ 5. Valida los tipos de datos SOLO de las columnas que: - Existen en el Excel - Existen en el catálogo - Tienen Origin = 'Manual' (ya filtrado en la consulta SQL) - INCLUYE validación de campos obligatorios (NOT NULL) Returns: list: Lista de errores encontrados """ errores = [] try: print("Validando tipos de datos...") columnas_excel = set(self.df_score.columns.tolist()) columnas_catalogo = set(self.catalogo_validacion.keys()) # Solo validar columnas que existen en AMBOS columnas_a_validar = columnas_excel & columnas_catalogo print(f"Columnas a validar (Origin='Manual' y existen en Excel): {len(columnas_a_validar)}") # MOSTRAR COLUMNAS OBLIGATORIAS columnas_obligatorias_en_validacion = [ col for col in ValidadorTiposDatos.COLUMNAS_OBLIGATORIAS if col in columnas_a_validar ] if columnas_obligatorias_en_validacion: print(f"⚠️ Columnas obligatorias (NOT NULL): {columnas_obligatorias_en_validacion}") for columna in columnas_a_validar: tipo_sql = self.catalogo_validacion[columna] # Iterar por cada fila de la columna for idx, valor in enumerate(self.df_score[columna]): # PASAR EL NOMBRE DE LA COLUMNA para validar NOT NULL es_valido, mensaje_error = ValidadorTiposDatos.validar_por_tipo_sql( valor, tipo_sql, nombre_columna=columna # ← ESTO ES NUEVO ) if not es_valido: errores.append({ 'fila': idx + 2, # +2 porque Excel empieza en 1 y tiene encabezado 'columna': columna, 'valor': str(valor)[:50] if pd.notna(valor) else '[VACÍO]', 'tipo_esperado': tipo_sql, 'error': mensaje_error }) print(f"✓ Validación completada: {len(errores)} errores encontrados") return errores except Exception as e: print(f"Error al validar tipos de datos: {str(e)}") return [{ 'fila': 0, 'columna': 'SISTEMA', 'valor': '', 'tipo_esperado': '', 'error': f"Error crítico: {str(e)}" }] def generar_reporte_consolidado(self, errores_estructura, errores_datos, stats_tratamiento): """ 5.3 Genera el reporte consolidado de la validación """ try: total_errores = len(errores_estructura) + len(errores_datos) # Obtener columnas validadas (intersección) columnas_excel = set(self.df_score.columns.tolist()) columnas_catalogo = set(self.catalogo_validacion.keys()) columnas_validadas = columnas_excel & columnas_catalogo errores_por_tipo = {} for error in errores_estructura: tipo = error.get('tipo', 'ESTRUCTURA') errores_por_tipo[tipo] = errores_por_tipo.get(tipo, 0) + 1 for error in errores_datos: tipo = error.get('tipo_esperado', 'TIPO_DATO') errores_por_tipo[tipo] = errores_por_tipo.get(tipo, 0) + 1 reporte = { 'timestamp': self.timestamp, 'fecha_validacion': datetime.now().strftime("%Y-%m-%d %H:%M:%S"), 'estado': 'EXITOSO' if total_errores == 0 else 'CON_ERRORES', 'estadisticas': { 'filas_originales': stats_tratamiento.get('filas_originales', 0) if stats_tratamiento else 0, 'filas_procesadas': stats_tratamiento.get('filas_finales', 0) if stats_tratamiento else 0, 'filas_eliminadas': stats_tratamiento.get('filas_eliminadas', 0) if stats_tratamiento else 0, 'columnas_excel': len(columnas_excel), 'columnas_catalogo_manual': len(columnas_catalogo), # Solo las de Origin='Manual' 'columnas_validadas': len(columnas_validadas), 'total_errores': total_errores, 'errores_estructura': len(errores_estructura), 'errores_datos': len(errores_datos) }, 'errores_por_tipo': errores_por_tipo, 'errores': { 'estructura': errores_estructura, 'datos': errores_datos } } guardar_log_validacion( total_filas=stats_tratamiento.get('filas_finales', 0) if stats_tratamiento else 0, total_errores=total_errores, errores_por_tipo=errores_por_tipo, timestamp=datetime.now(), estado=reporte['estado'] ) return reporte except Exception as e: print(f"Error al generar reporte: {str(e)}") return None def validar_cantidades_componentes_sql(self): """ Valida que las cantidades de motores, generadores y radiadores no excedan el máximo (1) mediante consulta SQL INDEPENDIENTE del Excel Returns: list: Lista de errores encontrados donde cantidad >= 2 """ errores = [] try: print("\n" + "="*100) print("🔍 INICIANDO VALIDACIÓN DE CANTIDADES DE COMPONENTES (SQL)") print("="*100) # ═══════════════════════════════════════════════════════════ # PASO 1: EJECUTAR CONSULTA SQL # ═══════════════════════════════════════════════════════════ print("\n📊 PASO 1: Ejecutando consulta SQL...") df_cantidades = obtener_cantidades_componentes() if df_cantidades.empty: print("⚠️ La consulta SQL no devolvió resultados") return errores print(f"✅ Consulta ejecutada: {len(df_cantidades)} registros obtenidos") # Mostrar resumen de la consulta print(f"\n📋 Resumen de columnas obtenidas:") print(f" Columnas: {list(df_cantidades.columns)}") print(f" Tipos de datos:") for col in df_cantidades.columns: print(f" - {col}: {df_cantidades[col].dtype}") # ═══════════════════════════════════════════════════════════ # PASO 2: MOSTRAR PRIMERAS 5 FILAS DE LA CONSULTA # ═══════════════════════════════════════════════════════════ print(f"\n{'='*100}") print("📊 PRIMERAS 5 FILAS DEL RESULTADO SQL") print("="*100) num_filas_mostrar = min(5, len(df_cantidades)) for idx in range(num_filas_mostrar): print(f"\n{'─'*100}") print(f"📍 REGISTRO {idx + 1}") print(f"{'─'*100}") row = df_cantidades.iloc[idx] print(f" OrderNum: {row['OrderNum']}") print(f" OrderLine: {row['OrderLine']}") print(f" OrderNum&Line: {row['OrderNum&Line']}") print(f" JobNum2: {row['JobNum2']}") print(f" 🔧 CantidadDeMotores: {row['CantidadDeMotores']}") print(f" ⚡ CantidadDeGeneradores: {row['CantidadDeGeneradores']}") print(f" 🌡️ CantidadDeRadiadores: {row['CantidadDeRadiadores']}") # Marcar si hay error tiene_error = ( row['CantidadDeMotores'] >= 2 or row['CantidadDeGeneradores'] >= 2 or row['CantidadDeRadiadores'] >= 2 ) if tiene_error: print(f" ⚠️ ¡CANTIDAD EXCEDIDA DETECTADA!") print(f"\n{'='*100}\n") # ═══════════════════════════════════════════════════════════ # PASO 3: VALIDAR CADA REGISTRO # ═══════════════════════════════════════════════════════════ print("🔎 PASO 2: Validando cantidades...") print(f" Criterio: Si cantidad >= 2 → ERROR") print(f"{'='*100}\n") contador_errores = 0 for idx, row in df_cantidades.iterrows(): order_num = row['OrderNum'] order_line_full = row['OrderNum&Line'] job_num = row['JobNum2'] if pd.notna(row['JobNum2']) else 'Sin JobNum' # ─────────────────────────────────────────────────────── # VALIDAR MOTORES # ─────────────────────────────────────────────────────── cantidad_motores = row['CantidadDeMotores'] if pd.notna(cantidad_motores) and cantidad_motores >= 2: contador_errores += 1 error = { 'fila': f'OrderNum&Line: {order_line_full}', 'columna': 'CantidadDeMotores', 'valor': str(int(cantidad_motores)), 'tipo_esperado': 'CANTIDAD_MAXIMA', 'error': f'⚠️ JobNum {job_num}: {int(cantidad_motores)} motores detectados (máximo permitido: 1)' } errores.append(error) print(f"❌ Error #{contador_errores}: OrderNum {order_num} - {int(cantidad_motores)} motores") # ─────────────────────────────────────────────────────── # VALIDAR GENERADORES # ─────────────────────────────────────────────────────── cantidad_generadores = row['CantidadDeGeneradores'] if pd.notna(cantidad_generadores) and cantidad_generadores >= 2: contador_errores += 1 error = { 'fila': f'OrderNum&Line: {order_line_full}', 'columna': 'CantidadDeGeneradores', 'valor': str(int(cantidad_generadores)), 'tipo_esperado': 'CANTIDAD_MAXIMA', 'error': f'⚠️ JobNum {job_num}: {int(cantidad_generadores)} generadores detectados (máximo permitido: 1)' } errores.append(error) print(f"❌ Error #{contador_errores}: OrderNum {order_num} - {int(cantidad_generadores)} generadores") # ─────────────────────────────────────────────────────── # VALIDAR RADIADORES # ─────────────────────────────────────────────────────── cantidad_radiadores = row['CantidadDeRadiadores'] if pd.notna(cantidad_radiadores) and cantidad_radiadores >= 2: contador_errores += 1 error = { 'fila': f'OrderNum&Line: {order_line_full}', 'columna': 'CantidadDeRadiadores', 'valor': str(int(cantidad_radiadores)), 'tipo_esperado': 'CANTIDAD_MAXIMA', 'error': f'⚠️ JobNum {job_num}: {int(cantidad_radiadores)} radiadores detectados (máximo permitido: 1)' } errores.append(error) print(f"❌ Error #{contador_errores}: OrderNum {order_num} - {int(cantidad_radiadores)} radiadores") # ═══════════════════════════════════════════════════════════ # PASO 4: RESUMEN FINAL # ═══════════════════════════════════════════════════════════ print(f"\n{'='*100}") print("📊 RESUMEN FINAL - VALIDACIÓN DE CANTIDADES SQL") print("="*100) print(f" 📋 Registros analizados: {len(df_cantidades)}") print(f" ✅ Registros sin errores: {len(df_cantidades) - len(set([e['fila'] for e in errores]))}") print(f" ❌ Total errores detectados: {len(errores)}") if len(errores) > 0: print(f"\n 📋 Detalle de errores por tipo:") errores_motores = sum(1 for e in errores if e['columna'] == 'CantidadDeMotores') errores_generadores = sum(1 for e in errores if e['columna'] == 'CantidadDeGeneradores') errores_radiadores = sum(1 for e in errores if e['columna'] == 'CantidadDeRadiadores') if errores_motores > 0: print(f" • Motores excedidos: {errores_motores}") if errores_generadores > 0: print(f" • Generadores excedidos: {errores_generadores}") if errores_radiadores > 0: print(f" • Radiadores excedidos: {errores_radiadores}") print(f"\n 📋 Lista de errores:") for i, error in enumerate(errores, 1): print(f" {i:3d}. {error['fila']} | {error['columna']} = {error['valor']}") else: print(f"\n ✅ NO SE DETECTARON CANTIDADES EXCEDIDAS") print(f" 🎉 Todas las órdenes tienen cantidades válidas (0 o 1)") print(f"{'='*100}\n") return errores except Exception as e: print(f"\n{'='*100}") print(f"❌ ERROR CRÍTICO EN VALIDACIÓN DE CANTIDADES SQL") print(f"{'='*100}") print(f" Error: {str(e)}") print(f"\n Stack trace completo:") import traceback traceback.print_exc() print(f"{'='*100}\n") return errores def ejecutar_validacion_completa(self): """ Ejecuta el flujo completo de validación INCLUYE detección de errores nativos de Excel Returns: dict: Reporte completo de la validación """ try: if not self.inicializar_catalogo(): return { 'estado': 'ERROR', 'mensaje': 'No se pudo cargar el catálogo de validación' } ruta_archivo = self.descargar_excel() if not ruta_archivo: return { 'estado': 'ERROR', 'mensaje': 'No se pudo descargar el archivo de SharePoint' } if not self.leer_excel(ruta_archivo): return { 'estado': 'ERROR', 'mensaje': 'No se pudo leer el archivo Excel' } # ✨ Detectar errores de Excel nativos errores_excel_nativos = self._detectar_errores_excel_nativos(ruta_archivo) estructura_valida, errores_estructura = self.validar_estructura() if not estructura_valida: return self.generar_reporte_consolidado( errores_estructura, [], None ) # Capturar errores de filas eliminadas y validación stats_tratamiento, errores_filas_eliminadas = self.aplicar_tratamiento_datos() errores_datos = self.validar_tipos_datos() errores_cantidades_sql = self.validar_cantidades_componentes_sql() # ✨ Combinar todos los errores en una sola línea todos_los_errores_datos = errores_excel_nativos + errores_filas_eliminadas + errores_datos + errores_cantidades_sql print(f"✓ Total errores consolidados: {len(todos_los_errores_datos)}") print(f" - Errores Excel: {len(errores_excel_nativos)}") print(f" - Filas eliminadas: {len(errores_filas_eliminadas)}") print(f" - Errores de tipos: {len(errores_datos)}") reporte = self.generar_reporte_consolidado( errores_estructura, todos_los_errores_datos, stats_tratamiento ) return reporte except Exception as e: print(f"Error en validación completa: {str(e)}") import traceback traceback.print_exc() return { 'estado': 'ERROR', 'mensaje': f'Error crítico: {str(e)}' } # ======================================== # FUNCIÓN ADICIONAL PARA ValidarScore.py # ======================================== # Esta función debe agregarse AL FINAL del archivo ValidarScore.py # ANTES de la función ejecutar_validacion_score(app, mail) def validar_score_para_actualizacion(ruta_archivo=None): """ Función auxiliar para ser llamada desde ActualizarScore.py Esta función ejecuta la validación del Score y retorna un formato compatible con el módulo de actualización. Args: ruta_archivo (str, optional): Ruta del archivo Excel a validar. Si no se proporciona, descarga desde SharePoint. Returns: dict: { 'success': bool, # True si validación exitosa, False si hay errores 'status': int, # 200 si éxito, 400 si error 'mensaje': str, # Mensaje descriptivo 'ruta_archivo': str, # Ruta del archivo validado (si éxito) 'reporte': dict, # Reporte completo de validación (opcional) 'errores': dict # Errores encontrados (si falla) } Ejemplo de uso: from App.SupyCtrol_Module.IngenieroControl.ValidarScore import validar_score_para_actualizacion resultado = validar_score_para_actualizacion() if resultado['success']: ruta = resultado['ruta_archivo'] # Continuar con el proceso... else: # Manejar errores... """ try: # Crear instancia del validador validador = ValidadorScore() # Si no se proporciona ruta, descargar desde SharePoint if ruta_archivo is None: # Inicializar catálogo if not validador.inicializar_catalogo(): return { 'success': False, 'status': 400, 'mensaje': 'No se pudo cargar el catálogo de validación desde CM_DataTypeValidator' } # Descargar archivo ruta_archivo = validador.descargar_excel() if not ruta_archivo: return { 'success': False, 'status': 400, 'mensaje': 'No se pudo descargar el archivo Excel desde SharePoint' } else: # Usar archivo proporcionado if not validador.inicializar_catalogo(): return { 'success': False, 'status': 400, 'mensaje': 'No se pudo cargar el catálogo de validación' } # Verificar que el archivo existe import os if not os.path.exists(ruta_archivo): return { 'success': False, 'status': 404, 'mensaje': f'El archivo no existe: {ruta_archivo}' } # Ejecutar validación completa reporte = validador.ejecutar_validacion_completa() # Analizar resultado if reporte.get('estado') == 'ERROR': return { 'success': False, 'status': 400, 'mensaje': reporte.get('mensaje', 'Error en validación'), 'reporte': reporte } # Verificar si hay errores total_errores = reporte.get('estadisticas', {}).get('total_errores', 0) if total_errores > 0: return { 'success': False, 'status': 400, 'mensaje': f'Se encontraron {total_errores} errores de validación', 'errores': reporte.get('errores', {}), 'reporte': reporte, 'detalles': { 'total_errores': total_errores, 'errores_estructura': reporte.get('estadisticas', {}).get('errores_estructura', 0), 'errores_datos': reporte.get('estadisticas', {}).get('errores_datos', 0) } } # Validación exitosa return { 'success': True, 'status': 200, 'mensaje': 'Validación exitosa. Datos correctos.', 'ruta_archivo': ruta_archivo, 'reporte': reporte, 'estadisticas': reporte.get('estadisticas', {}) } except Exception as e: return { 'success': False, 'status': 500, 'mensaje': f'Error crítico en validación: {str(e)}' } # Función principal para registrar la ruta en Flask def ejecutar_validacion_score(app, mail): """ Registra la ruta de ejecución de validación en Flask Args: app: Instancia de Flask mail: Instancia de Flask-Mail """ # Ruta de prueba @app.route('/SyC/IngenieroControl/ModuloValidacionScore/test', methods=['GET']) def test_validacion(): """Ruta de prueba para verificar que el backend funciona""" return jsonify({ 'success': True, 'mensaje': '¡Backend funcionando correctamente!', 'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S") }) @app.route('/SyC/IngenieroControl/ModuloValidacionScore/ejecutar', methods=['POST']) def ejecutar_validacion(): """Ejecuta la validación del Score y retorna JSON""" try: print("Iniciando validación del Score...") validador = ValidadorScore() reporte = validador.ejecutar_validacion_completa() return jsonify(reporte) except Exception as e: return jsonify({ 'estado': 'ERROR', 'mensaje': str(e) }), 500