# Archivo: ActualizarScore.py
# Ruta: App\SupyCtrol_Module\IngenieroControl\ActualizarScore.py
# Descripción: Módulo para actualizar el Score desde Excel a la BD
# Autor: Equipo de Desarrollo IGSA
# Fecha: 2025
"""
Módulo completo de actualización del Score.
Maneja el flujo completo desde la validación del Excel hasta la
actualización de la tabla CM_Score en la base de datos.
"""
import os
from dotenv import load_dotenv
import pandas as pd
import ftplib
from datetime import datetime
from flask import request, jsonify, render_template, redirect
from Consultas_SQL.conexion import get_connection, get_connectionERP
from Consultas_SQL.SupYCtrol.IngDeControl.ActualizarScoreSQL import (
obtener_datos_score_erp,
obtener_estructura_validador,
verificar_tabla_cm_score,
obtener_ultima_version_history,
obtener_columnas_tabla,
crear_tabla_cm_score,
crear_tabla_scorehistory,
truncate_tabla,
insertar_bulk_score
)
# Importar función de validación del módulo ValidarScore
from App.SupyCtrol_Module.IngenieroControl.ValidarScore import validar_score_para_actualizacion
# ========================================
# CONFIGURACIÓN
# ========================================
load_dotenv()
# Configuración FTP
FTP_CONFIG = {
'host': os.getenv('FTP_HOST'),
'user': os.getenv('FTP_USER'),
'password': os.getenv('FTP_PASS'),
'remote_path': '/domains/sycelephant.com/public_html/file/SyC/IngenieroControl/RespaldoActualizarScore/' # Ajustar según tu estructura FTP
}
# Configuración OneDrive/SharePoint con Azure AD
ONEDRIVE_CONFIG = {
'enabled': False,
# Azure AD App Registration
'tenant_id': os.getenv('TENANT_ID'), # Directory (tenant) ID
'client_id': os.getenv('ONEDRIVE_CLIENT_ID'), # Application (client) ID
'client_secret': os.getenv('ONEDRIVE_CLIENT_SECRET'), # Client secret
# Redirect URI (debe coincidir con Azure AD)
'redirect_uri': 'http://localhost:5000/callback', # O tu URL de producción
# Scopes
'scopes': [
'Files.ReadWrite',
],
# Usuario y archivo (aquí revisa que realmente quieras este user)
'user_email': os.getenv('ONEDRIVE_USERNAME'), # Usuario de OneDrive
# File ID del archivo de OneDrive que quieres manipular
'file_id': 'EWDk3kJdEcdJivAlcKPi8woBTgMWG_qfQ1rYYo0Jcy5QbQ',
# URLs de MS
'authority': 'https://login.microsoftonline.com/{tenant_id}',
'graph_url': 'https://graph.microsoft.com/v1.0',
# Token cache local
'token_cache_file': 'onedrive_token_cache.json'
}
# Clasificación de columnas según origen
# Esta clasificación está basada en el archivo Tablas_llenas_de_datos.xlsx
CLASIFICACION_COLUMNAS = {
'columnas_erp': [
'OrderNum', 'Departamento', 'Vendedor', 'Name', 'TotalLines', 'OrderLine',
'PartNum', 'LineDesc', 'Capacidad', 'Voltaje', 'Tipo', 'RevisionNum',
'DecriptionProd', 'Caseta', 'OrderDate', 'NeedByDate', 'FechaAnticipo',
'LiberacionCXP', 'Terminos', 'OrderNum&Line', 'ProjectID', 'JobNum2',
'Revision_OV', 'Revision_Project', 'Revision_Job', 'FechaDeCierre',
'InsumosDemandados', 'InsumosEmitidos', 'FechaDeTermino', 'PartNum_M',
'Description_M', 'PartClass_M', 'RefCategory_M', 'QtyPer_M', 'IssuedQty_M',
'Demandado_M', 'OnhandQty_M', 'En_PO_M', 'NoPO_M', 'EnRequisicion_M',
'NoRequisicion_M', 'PartNum_G', 'Description_G', 'PartClass_G', 'RefCategory_G',
'QtyPer_G', 'IssuedQty_G', 'Demandado_G', 'OnhandQty_G', 'En_PO_G', 'NoPO_G',
'EnRequisicion_G', 'NoRequisicion_G', 'PartNum_T', 'Description_T', 'PartClass_T',
'RefCategory_T', 'QtyPer_T', 'IssuedQty_T', 'OnhandQty_T', 'PartNum_R',
'Description_R', 'PartClass_R', 'RefCategory_R', 'QtyPer_R', 'IssuedQty_R',
'OnhandQty_R', 'ComentarioLINE', 'OvEEUU', 'ClienteEEUU'
],
'columnas_manual': [
'ProdCode', 'ComentarioSyC', 'Chk', 'FechaSyC', 'ConsideradaPreasignacion',
'ConsideradaMateriales', 'FechaProducción', 'ComentarioProducción', 'AvanceProducción',
'FechaPlaneación', 'EstadoFecha', 'ComentarioPlaneación', 'FechaInicialPlaneación',
'MotivoInicial', 'FechaMat', 'AvisoDeTerminacion', 'TerminadoConFaltante',
'FechaActualización', 'MaterialFaltante', 'Auxiliar1', 'Auxiliar2', 'ComentarioCalidad',
'Preasignado_M', 'NoSerie_M', 'Req_M', 'Comment_M', 'FechaReq_M', 'PO_M',
'Cantidad_Pedida_M', 'Preasignado_G', 'NoSerie_G', 'Req_G', 'Comment_G',
'FechaReq_G', 'PO_G', 'Cantidad_Pedida_G'
],
'columnas_formulado': [
'FechaVentas', 'FechaSimulaciones', 'FechaMG', 'AvanceDeSurtimiento', 'AvaceDeEmisiones',
'Faltante_M', 'Alternativa_M', 'En_PO_Altern_M', 'NoPO_Alern_M', 'Estatus_M',
'Fecha_Llegada_M', 'Faltante_G', 'Alternativa_G', 'En_PO_Altern_G', 'NoPO_Alern_G',
'Estatus_G', 'Fecha_Llegada_G'
]
}
# ========================================
# CLASE: FTPManager
# ========================================
class FTPManager:
"""
Gestiona la conexión y operaciones con el servidor FTP
para guardar respaldos del archivo Excel
✅ Versión actualizada con navegación paso a paso
"""
def __init__(self):
self.host = FTP_CONFIG['host']
self.user = FTP_CONFIG['user']
self.password = FTP_CONFIG['password']
self.remote_path = FTP_CONFIG['remote_path']
self.ftp = None
def conectar(self):
"""
Establece conexión con el servidor FTP
Returns:
bool: True si la conexión fue exitosa
Raises:
Exception: Si hay error de conexión
"""
try:
print("Conectando a FTP...")
self.ftp = ftplib.FTP(self.host)
self.ftp.login(self.user, self.password)
print("✓ Conectado a FTP exitosamente")
return True
except Exception as e:
print(f"✗ Error al conectar a FTP: {str(e)}")
raise Exception(f"Error de conexión FTP: {str(e)}")
def crear_estructura_carpetas(self):
"""
Crea la estructura de carpetas navegando paso a paso
Se asegura de estar en el file correcto (el que tiene Formatos, Operaciones)
Returns:
bool: True si la estructura está lista
"""
try:
print("Verificando estructura de carpetas en FTP...")
# PASO 1: Ir a la raíz absoluta
try:
self.ftp.cwd('/')
print(f" ✓ En la raíz FTP")
except:
pass
# PASO 2: Navegar a domains
try:
self.ftp.cwd('/domains')
print(f" ✓ En /domains")
except Exception as e:
print(f" ✗ No se pudo acceder a /domains: {e}")
return False
# PASO 3: Navegar a sycelephant.com
try:
self.ftp.cwd('/domains/sycelephant.com')
print(f" ✓ En /domains/sycelephant.com")
except Exception as e:
print(f" ✗ No se pudo acceder a sycelephant.com: {e}")
return False
# PASO 4: Navegar a public_html
try:
self.ftp.cwd('/domains/sycelephant.com/public_html')
print(f" ✓ En /domains/sycelephant.com/public_html")
except Exception as e:
print(f" ✗ No se pudo acceder a public_html: {e}")
return False
# PASO 5: Navegar a file (el correcto)
try:
self.ftp.cwd('/domains/sycelephant.com/public_html/file')
print(f" ✓ En /domains/sycelephant.com/public_html/file")
# VERIFICAR que estamos en el file correcto
# Listar contenido para confirmar que tiene Formatos y Operaciones
items = []
self.ftp.retrlines('NLST', items.append)
print(f" 📂 Contenido actual: {', '.join(items[:5])}")
if 'Formatos' in items or 'Operaciones' in items:
print(f" ✓ Confirmado: Estamos en el file CORRECTO (tiene Formatos/Operaciones)")
else:
print(f" ⚠ ADVERTENCIA: Este file no tiene Formatos/Operaciones")
print(f" ⚠ Puede que sea el file equivocado")
except Exception as e:
print(f" ✗ No se pudo acceder a file: {e}")
return False
# PASO 6: Crear/Acceder a SyC
try:
self.ftp.cwd('/domains/sycelephant.com/public_html/file/SyC')
print(f" ✓ Carpeta existe: SyC")
except ftplib.error_perm:
try:
self.ftp.cwd('/domains/sycelephant.com/public_html/file')
self.ftp.mkd('SyC')
self.ftp.cwd('SyC')
print(f" ✓ Carpeta creada: SyC")
except Exception as e:
print(f" ✗ No se pudo crear SyC: {e}")
return False
# PASO 7: Crear/Acceder a IngenieroControl
try:
self.ftp.cwd('/domains/sycelephant.com/public_html/file/SyC/IngenieroControl')
print(f" ✓ Carpeta existe: IngenieroControl")
except ftplib.error_perm:
try:
self.ftp.cwd('/domains/sycelephant.com/public_html/file/SyC')
self.ftp.mkd('IngenieroControl')
self.ftp.cwd('IngenieroControl')
print(f" ✓ Carpeta creada: IngenieroControl")
except Exception as e:
print(f" ✗ No se pudo crear IngenieroControl: {e}")
return False
# PASO 8: Crear/Acceder a RespaldoActualizarScore
try:
self.ftp.cwd('/domains/sycelephant.com/public_html/file/SyC/IngenieroControl/RespaldoActualizarScore/')
print(f" ✓ Carpeta existe: RespaldoActualizarScore")
except ftplib.error_perm:
try:
self.ftp.cwd('/domains/sycelephant.com/public_html/file/SyC/IngenieroControl')
self.ftp.mkd('RespaldoActualizarScore')
self.ftp.cwd('RespaldoActualizarScore')
print(f" ✓ Carpeta creada: RespaldoActualizarScore")
except Exception as e:
print(f" ✗ No se pudo crear RespaldoActualizarScore: {e}")
return False
# PASO 9: Confirmar ubicación final
ubicacion_actual = self.ftp.pwd()
print(f"✓ Estructura de carpetas lista")
print(f"✓ Ubicación actual: {ubicacion_actual}")
# Verificar que estamos en la ruta correcta
if ubicacion_actual == self.remote_path:
print(f"✓ CONFIRMADO: Estamos en la ruta correcta")
else:
print(f"⚠ ADVERTENCIA: Ubicación actual no coincide con remote_path")
print(f" Esperado: {self.remote_path}")
print(f" Actual: {ubicacion_actual}")
return True
except Exception as e:
print(f"✗ Error al crear estructura de carpetas: {str(e)}")
return False
def subir_archivo(self, local_path, remote_filename):
"""
Sube un archivo al servidor FTP
Args:
local_path (str): Ruta local del archivo
remote_filename (str): Nombre del archivo en el servidor
Returns:
bool: True si se subió correctamente
Raises:
Exception: Si hay error al subir
"""
try:
if self.ftp is None:
self.conectar()
# ✅ NUEVO: Crear estructura de carpetas antes de subir
self.crear_estructura_carpetas()
print(f"\nSubiendo archivo a FTP...")
print(f" Archivo: {remote_filename}")
print(f" Destino: {self.remote_path}")
with open(local_path, 'rb') as file:
self.ftp.storbinary(f'STOR {remote_filename}', file)
print(f"✓ Archivo subido exitosamente: {remote_filename}")
print(f"✓ URL: https://sycelephantt.com/file/SyC/IngenieroControl/RespaldoActualizarScore/{remote_filename}\n")
return True
except Exception as e:
print(f"✗ Error al subir archivo a FTP: {str(e)}")
raise Exception(f"Error al subir a FTP: {str(e)}")
def desconectar(self):
"""Cierra la conexión FTP"""
try:
if self.ftp is not None:
self.ftp.quit()
print("✓ Desconectado de FTP")
except:
try:
if self.ftp is not None:
self.ftp.close()
except:
pass
# ========================================
# CLASE: OneDriveManager
# ========================================
class OneDriveManager:
"""
Gestiona actualización automática del Score en SharePoint
usando Microsoft Authentication Library (MSAL) con Azure AD
✅ Maneja refresh tokens automáticamente
✅ Guarda tokens en archivo local
✅ No requiere regenerar tokens manualmente
"""
def __init__(self):
import msal
import os
self.config = ONEDRIVE_CONFIG
self.graph_url = self.config['graph_url']
# Configurar MSAL
authority = self.config['authority'].format(
tenant_id=self.config['tenant_id']
)
self.msal_app = msal.ConfidentialClientApplication(
client_id=self.config['client_id'],
client_credential=self.config['client_secret'],
authority=authority
)
# Cargar cache de tokens (si existe)
self.token_cache_file = self.config['token_cache_file']
self.token_cache = self._load_token_cache()
def _load_token_cache(self):
"""Carga tokens guardados del archivo"""
import os
import json
if os.path.exists(self.token_cache_file):
try:
with open(self.token_cache_file, 'r') as f:
return json.load(f)
except:
return {}
return {}
def _save_token_cache(self, token_response):
"""Guarda tokens en archivo local"""
import json
try:
with open(self.token_cache_file, 'w') as f:
json.dump(token_response, f)
print(f" ✓ Tokens guardados en {self.token_cache_file}")
except Exception as e:
print(f" ⚠️ No se pudo guardar token cache: {e}")
def obtener_access_token(self):
"""
Obtiene access token (renueva automáticamente si es necesario)
Returns:
str: Access token válido
"""
# PASO 1: Intentar usar token del cache
if self.token_cache and 'access_token' in self.token_cache:
print(" ✓ Usando token del cache")
return self.token_cache['access_token']
# PASO 2: Intentar renovar con refresh token
if self.token_cache and 'refresh_token' in self.token_cache:
print(" Renovando token con refresh_token...")
result = self.msal_app.acquire_token_by_refresh_token(
refresh_token=self.token_cache['refresh_token'],
scopes=self.config['scopes']
)
if 'access_token' in result:
print(" ✓ Token renovado exitosamente")
self._save_token_cache(result)
return result['access_token']
# PASO 3: Necesitamos autorización del usuario
raise Exception(
"\n❌ NO HAY TOKEN VÁLIDO\n\n"
"Necesitas autorizar la aplicación primero.\n"
"Ejecuta el script de autorización: python autorizar_onedrive.py\n"
"O visita: http://localhost:5000/login"
)
def buscar_archivo(self, access_token):
"""Obtiene información del archivo"""
import requests
headers = {
'Authorization': f'Bearer {access_token}'
}
file_id = self.config['file_id']
url = f'{self.graph_url}/me/drive/items/{file_id}'
response = requests.get(url, headers=headers, timeout=30)
if response.status_code == 200:
data = response.json()
print(f" ✓ Archivo: {data.get('name')}")
return file_id
elif response.status_code == 404:
raise Exception(f"Archivo no encontrado. Verifica file_id y permisos.")
else:
raise Exception(f"Error al buscar archivo: {response.status_code} - {response.text}")
def actualizar_archivo(self, access_token, file_id, excel_bytes):
"""Actualiza el contenido del archivo en OneDrive"""
import requests
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
}
url = f'{self.graph_url}/me/drive/items/{file_id}/content'
response = requests.put(
url,
headers=headers,
data=excel_bytes,
timeout=120
)
if response.status_code in [200, 201]:
return True
else:
raise Exception(f"Error al actualizar archivo: {response.status_code} - {response.text}")
# ========================================
# CLASE: ScoreProcessor
# ========================================
class ScoreProcessor:
"""
Clase principal que maneja todo el proceso de actualización del Score
"""
def __init__(self):
"""Inicializa el procesador con conexiones y configuraciones"""
self.conn_erp = None
self.conn_vps = None
self.ftp_manager = FTPManager()
self.tiempo_inicio = datetime.now()
self.onedrive_manager = OneDriveManager()
# DataFrames
self.df_score_excel = None
self.df_score_erp = None
self.df_score_final = None
# Configuración
self.clasificacion_columnas = None
# Resultados
self.stats = {
'filas_excel': 0,
'filas_erp': 0,
'filas_procesadas': 0,
'filas_eliminadas': 0,
'version_history': 0,
'tabla_history': '',
'respaldo_ftp': False,
'onedrive_actualizado': False
}
def ejecutar_actualizacion_completa(self):
"""
Ejecuta el flujo completo de actualización del Score
Returns:
dict: Resultado del proceso con estadísticas
"""
try:
print("\n" + "="*80)
print("INICIANDO PROCESO DE ACTUALIZACIÓN DEL SCORE")
print("="*80 + "\n")
# PASO 1: Validar Excel con ValidarScore.py
print("PASO 1/11: Validando Excel con ValidarScore.py...")
resultado_validacion = self.validar_excel()
if not resultado_validacion['success']:
return {
'success': False,
'error': 'Validación de Excel fallida',
'detalles': resultado_validacion
}
ruta_archivo = resultado_validacion['ruta_archivo']
print(f"✓ Validación exitosa. Archivo: {ruta_archivo}\n")
# PASO 2: Guardar respaldo en FTP (NO CRÍTICO)
print("PASO 2/11: Guardando respaldo en FTP...")
try:
self.guardar_respaldo_ftp(ruta_archivo)
self.stats['respaldo_ftp'] = True
except Exception as e:
print(f"⚠ Warning: No se pudo guardar respaldo en FTP: {str(e)}")
print("Continuando proceso...\n")
# PASO 3: Leer Excel a DataFrame
print("PASO 3/11: Leyendo Excel a DataFrame...")
self.df_score_excel = self.leer_excel_a_dataframe(ruta_archivo)
self.stats['filas_excel'] = len(self.df_score_excel)
print(f"✓ Excel leído: {self.stats['filas_excel']} filas\n")
# PASO 4: Guardar respaldo en ScoreHistory_X
print("PASO 4/11: Guardando respaldo en ScoreHistory...")
self.guardar_respaldo_bd()
print(f"✓ Respaldo guardado en {self.stats['tabla_history']}\n")
# PASO 5: Obtener estructura del validador
print("PASO 5/11: Obteniendo estructura del validador...")
self.obtener_estructura_columnas()
print("✓ Estructura obtenida y clasificada\n")
# PASO 6: Obtener datos del ERP
print("PASO 6/11: Obteniendo datos del ERP...")
try:
self.df_score_erp = obtener_datos_score_erp()
self.stats['filas_erp'] = len(self.df_score_erp)
print(f"✓ Datos ERP obtenidos: {self.stats['filas_erp']} filas\n")
except Exception as e:
# Capturar el error de conexión al ERP
error_msg = str(e)
# Determinar tipo de error para el frontend
if 'ERROR DE CONEXIÓN AL ERP' in error_msg or 'ERROR CRÍTICO' in error_msg:
tipo_error = 'servidor_inaccesible'
elif 'TIMEOUT' in error_msg:
tipo_error = 'timeout'
elif 'AUTENTICACIÓN' in error_msg:
tipo_error = 'autenticacion'
else:
tipo_error = 'error_desconocido'
return {
'success': False,
'error': 'No se pudo obtener datos del ERP',
'tipo_error': 'conexion_erp_fallida',
'detalles': {
'tipo_error': tipo_error,
'error_tecnico': error_msg,
'sugerencias': [
'Verifique que el servidor SQL Server del ERP esté encendido',
'Verifique conectividad de red al servidor ERP',
'Verifique que SQL Server acepte conexiones remotas',
'Verifique que Named Pipes esté habilitado',
'Contacte al administrador del servidor ERP'
]
}
}
# PASO 7: Mezclar DataFrames
print("PASO 7/11: Mezclando DataFrames...")
self.mezclar_dataframes()
self.stats['filas_procesadas'] = len(self.df_score_final)
self.stats['filas_eliminadas'] = self.stats['filas_excel'] - self.stats['filas_procesadas']
print(f"✓ Mezcla completada: {self.stats['filas_procesadas']} filas\n")
# PASO 8: Verificar/Crear CM_Score
print("PASO 8/11: Verificando tabla CM_Score...")
self.crear_tabla_cm_score_si_no_existe()
print("✓ Tabla CM_Score lista\n")
# PASO 9: Actualizar CM_Score
print("PASO 9/11: Actualizando CM_Score...")
self.actualizar_tabla_cm_score()
print("✓ CM_Score actualizada exitosamente\n")
# PASO 10: Actualizar OneDrive
print("PASO 10/11: Actualizando archivo en OneDrive...")
if ONEDRIVE_CONFIG['enabled']:
try:
self.actualizar_onedrive()
self.stats['onedrive_actualizado'] = True
except Exception as e:
print(f"⚠ Warning: No se pudo actualizar OneDrive: {str(e)}")
print("Continuando proceso...\n")
self.stats['onedrive_actualizado'] = False
else:
print("⚠ OneDrive deshabilitado en configuración\n")
# PASO 10: Generar respuesta
print("PASO 11/11: Generando respuesta...")
respuesta = self.generar_respuesta_exitosa()
print("\n" + "="*80)
print("PROCESO COMPLETADO EXITOSAMENTE")
print("="*80 + "\n")
return respuesta
except Exception as e:
print(f"\n✗ ERROR EN EL PROCESO: {str(e)}\n")
return {
'success': False,
'error': str(e)
}
def actualizar_onedrive(self):
"""
Actualiza archivo en OneDrive con df_score_final
"""
try:
# 1. Obtener access token (se renueva automáticamente)
print(" Obteniendo access token...")
access_token = self.onedrive_manager.obtener_access_token()
print(" ✓ Token obtenido")
# 2. Buscar archivo
print(f" Buscando archivo Score.xlsx...")
file_id = self.onedrive_manager.buscar_archivo(access_token)
print(f" ✓ Archivo encontrado")
# 3. Convertir DataFrame a Excel
print(" Generando Excel actualizado...")
excel_bytes = self.dataframe_a_excel_bytes(self.df_score_final)
print(f" ✓ Excel generado ({len(excel_bytes):,} bytes)")
# 4. Actualizar archivo
print(" Subiendo archivo a OneDrive...")
self.onedrive_manager.actualizar_archivo(access_token, file_id, excel_bytes)
print(" ✓ Archivo actualizado en OneDrive")
print(f" ✓ {len(self.df_score_final)} filas escritas\n")
return True
except Exception as e:
raise Exception(f"Error al actualizar OneDrive: {str(e)}")
def dataframe_a_excel_bytes(self, df):
"""
Convierte DataFrame a bytes de Excel en memoria
"""
from io import BytesIO
output = BytesIO()
# Usar ExcelWriter para mantener formato
with pd.ExcelWriter(output, engine='openpyxl') as writer:
df.to_excel(writer, sheet_name='CM_Score', index=False)
output.seek(0)
return output.read()
def validar_excel(self):
"""
Valida el Excel usando el módulo ValidarScore.py
Returns:
dict: Resultado de la validación
"""
try:
# Invocar la función de validación
resultado = validar_score_para_actualizacion()
return resultado
except Exception as e:
return {
'success': False,
'status': 500,
'mensaje': f'Error al invocar validación: {str(e)}'
}
def guardar_respaldo_ftp(self, ruta_archivo):
"""
Guarda respaldo del Excel en el servidor FTP
Args:
ruta_archivo (str): Ruta del archivo a respaldar
"""
nombre_respaldo = generar_nombre_respaldo()
self.ftp_manager.subir_archivo(ruta_archivo, nombre_respaldo)
self.ftp_manager.desconectar()
def leer_excel_a_dataframe(self, ruta_archivo):
"""
Lee el archivo Excel y lo convierte a DataFrame
- Lee valores calculados (data_only=True) en lugar de fórmulas
Args:
ruta_archivo (str): Ruta del archivo Excel
Returns:
pd.DataFrame: DataFrame con los datos del Excel (123 columnas)
"""
from openpyxl import load_workbook
print(" Leyendo Excel (solo valores, sin fórmulas)...")
# ══════════════════════════════════════════════════════════════
# 🔧 CONFIGURACIÓN DE LA HOJA - CAMBIAR AQUÍ SI ES NECESARIO
# ══════════════════════════════════════════════════════════════
NOMBRE_HOJA_EXCEL = 'ScoreV2' # ← CAMBIAR AQUÍ si tu hoja tiene otro nombre
# ══════════════════════════════════════════════════════════════
# PASO 1: Abrir workbook con data_only=True (lee valores, no fórmulas)
wb = load_workbook(ruta_archivo, data_only=True)
# Verificar que la hoja existe
if NOMBRE_HOJA_EXCEL not in wb.sheetnames:
wb.close()
raise Exception(
f"❌ La hoja '{NOMBRE_HOJA_EXCEL}' no existe en el Excel.\n"
f" Hojas disponibles: {wb.sheetnames}\n"
f" Verifica el nombre de la hoja en ValidarScore.py y ActualizarScore.py"
)
ws = wb[NOMBRE_HOJA_EXCEL]
# PASO 2: Leer todas las filas
data = []
headers = None
for idx, row in enumerate(ws.iter_rows(values_only=True)):
if idx == 0:
# Primera fila = headers
headers = [str(cell).strip() if cell else f"Column_{i}" for i, cell in enumerate(row)]
else:
# Resto de filas = datos
data.append(row)
wb.close()
# PASO 3: Crear DataFrame
df = pd.DataFrame(data, columns=headers)
print(f" ✓ Excel leído: {len(df)} filas, {len(df.columns)} columnas")
# ═══════════════════════════════════════════════════════════════
# PASO 4: FILTRAR SOLO LAS COLUMNAS ESPERADAS
# ═══════════════════════════════════════════════════════════════
columnas_esperadas = (
CLASIFICACION_COLUMNAS['columnas_erp'] +
CLASIFICACION_COLUMNAS['columnas_manual'] +
CLASIFICACION_COLUMNAS['columnas_formulado']
)
print(f" 🔍 Columnas esperadas: {len(columnas_esperadas)}")
print(f" 🔍 Columnas en Excel: {len(df.columns)}")
# Verificar que todas las columnas esperadas existan
columnas_faltantes = set(columnas_esperadas) - set(df.columns)
if columnas_faltantes:
print(f"\n ❌ ERROR: Faltan columnas esperadas:")
for col in sorted(columnas_faltantes)[:10]: # Mostrar solo primeras 10
print(f" - {col}")
if len(columnas_faltantes) > 10:
print(f" ... y {len(columnas_faltantes) - 10} columnas más")
raise Exception(
f"El Excel no tiene todas las columnas esperadas.\n"
f"Columnas faltantes: {len(columnas_faltantes)}"
)
# Identificar columnas extra
columnas_extra = set(df.columns) - set(columnas_esperadas)
if columnas_extra:
print(f"\n ⚠️ Columnas extra detectadas ({len(columnas_extra)}):")
for col in sorted(columnas_extra)[:10]: # Mostrar solo primeras 10
print(f" - {col}")
if len(columnas_extra) > 10:
print(f" ... y {len(columnas_extra) - 10} columnas más")
print(f" ℹ️ Estas columnas serán ignoradas")
# FILTRAR SOLO LAS COLUMNAS ESPERADAS (en el orden correcto)
df = df[columnas_esperadas]
print(f" ✓ DataFrame filtrado a {len(df.columns)} columnas esperadas\n")
# ═══════════════════════════════════════════════════════════════
# PASO 5: VERIFICAR COLUMNAS FORMULADAS
# ═══════════════════════════════════════════════════════════════
columnas_formuladas = CLASIFICACION_COLUMNAS['columnas_formulado']
print(" 🔍 Verificando valores en columnas formuladas:")
for col in columnas_formuladas[:5]: # Mostrar solo primeras 5
if col in df.columns:
valores_null = df[col].isna().sum()
valores_no_null = df[col].notna().sum()
total = len(df)
if valores_null > 0:
print(f" ⚠️ {col}: {valores_no_null}/{total} valores ({valores_null} NULL)")
else:
print(f" ✓ {col}: {valores_no_null}/{total} valores")
if len(columnas_formuladas) > 5:
print(f" ... y {len(columnas_formuladas) - 5} columnas formuladas más")
# Mostrar ejemplo de FechaMG si existe
if 'FechaMG' in df.columns:
valores_ejemplo = df[df['FechaMG'].notna()]['FechaMG'].head(3)
if len(valores_ejemplo) > 0:
print(f"\n 📅 Ejemplo de valores en FechaMG:")
for idx, val in valores_ejemplo.items():
print(f" Fila {idx}: {val}")
print()
# ═══════════════════════════════════════════════════════════════
# PASO 6: LIMPIAR DATOS PARA SQL
# ═══════════════════════════════════════════════════════════════
df = limpiar_dataframe_para_sql(df)
return df
def guardar_respaldo_bd(self):
"""
Guarda respaldo del Excel en ScoreHistory_X con versionado inteligente
"""
# Obtener última versión
version_actual = obtener_ultima_version_history()
if version_actual == 0:
# No hay tablas, crear la primera
version_usar = 1
crear_nueva = True
else:
# Comparar estructura con la última versión
tabla_actual = f"ScoreHistory_{version_actual}"
columnas_history = obtener_columnas_tabla(tabla_actual)
columnas_excel = self.df_score_excel.columns.tolist()
son_iguales = comparar_estructura_columnas(columnas_excel, columnas_history)
if son_iguales:
# Usar la misma tabla
version_usar = version_actual
crear_nueva = False
else:
# Crear nueva versión
version_usar = version_actual + 1
crear_nueva = True
# Persistencia
tabla_destino = f"ScoreHistory_{version_usar}"
if crear_nueva:
# Crear nueva tabla
estructura = obtener_estructura_validador()
crear_tabla_scorehistory(version_usar, estructura)
else:
# Limpiar tabla existente
truncate_tabla(tabla_destino)
# Insertar datos
insertar_bulk_score(self.df_score_excel, tabla_destino)
# Guardar stats
self.stats['version_history'] = version_usar
self.stats['tabla_history'] = tabla_destino
def obtener_estructura_columnas(self):
"""
Obtiene y asigna la clasificación de columnas (ahora usando clasificación estática)
"""
# Usar la clasificación estática definida en la configuración
self.clasificacion_columnas = CLASIFICACION_COLUMNAS
print(f" - Columnas ERP: {len(self.clasificacion_columnas['columnas_erp'])}")
print(f" - Columnas Manual: {len(self.clasificacion_columnas['columnas_manual'])}")
print(f" - Columnas Formuladas: {len(self.clasificacion_columnas['columnas_formulado'])}")
def mezclar_dataframes(self):
"""
Mezcla df_score_excel y df_score_erp según las reglas de negocio
REGLA PRINCIPAL:
- df_score_erp "manda" → solo filas que existen en ERP
- Columnas ERP: siempre del ERP
- Columnas Manual/Formuladas: del Excel si existe, NULL si no
"""
filas_score = []
columnas_erp = self.clasificacion_columnas['columnas_erp']
columnas_manual = self.clasificacion_columnas['columnas_manual']
columnas_formulado = self.clasificacion_columnas['columnas_formulado']
print(f" Procesando {len(self.df_score_erp)} filas del ERP...")
for idx, fila_erp in self.df_score_erp.iterrows():
# Obtener OrderNum&Line del ERP
ordernum_line = fila_erp['OrderNum&Line']
# Crear diccionario para la nueva fila
fila_score = {}
# AGREGAR TODAS LAS COLUMNAS ERP
for col in columnas_erp:
if col in fila_erp.index:
fila_score[col] = fila_erp[col]
# BUSCAR COINCIDENCIA EN EXCEL
fila_excel = self.df_score_excel[
self.df_score_excel['OrderNum&Line'] == ordernum_line
]
if len(fila_excel) > 0:
# EXISTE COINCIDENCIA: agregar columnas Manual y Formuladas del Excel
fila_excel = fila_excel.iloc[0]
for col in columnas_manual:
if col in fila_excel.index:
fila_score[col] = fila_excel[col]
for col in columnas_formulado:
if col in fila_excel.index:
fila_score[col] = fila_excel[col]
else:
# NO EXISTE COINCIDENCIA: agregar columnas con NULL
for col in columnas_manual:
fila_score[col] = None
for col in columnas_formulado:
fila_score[col] = None
filas_score.append(fila_score)
# Crear DataFrame final
self.df_score_final = pd.DataFrame(filas_score)
print(f" ✓ Mezcla completada: {len(self.df_score_final)} filas resultantes")
def crear_tabla_cm_score_si_no_existe(self):
"""
Verifica si CM_Score existe y la crea si no existe
"""
existe = verificar_tabla_cm_score()
if not existe:
print(" Tabla CM_Score no existe, creándola...")
estructura = obtener_estructura_validador()
crear_tabla_cm_score(estructura)
else:
print(" Tabla CM_Score ya existe")
def actualizar_tabla_cm_score(self):
"""
Actualiza la tabla CM_Score con los datos finales
"""
# Limpiar tabla
truncate_tabla('CM_Score')
# Insertar datos
insertar_bulk_score(self.df_score_final, 'CM_Score')
def generar_respuesta_exitosa(self):
"""
Genera la respuesta JSON de éxito con estadísticas
Returns:
dict: Respuesta completa
"""
tiempo_ejecucion = calcular_tiempo_ejecucion(self.tiempo_inicio)
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
resumen = f"""
=== RESUMEN DE ACTUALIZACIÓN ===
✅ Validación exitosa
✅ Respaldo guardado en FTP: {'Sí' if self.stats['respaldo_ftp'] else 'No (sin conexión)'}
✅ Datos del ERP obtenidos
✅ Mezcla de datos completada
✅ Tabla CM_Score actualizada
# 'OneDrive actualizado: {'Sí' if self.stats['onedrive_actualizado'] else 'No'}'
📊 Estadísticas:
- Filas procesadas: {self.stats['filas_procesadas']}
- Filas del ERP: {self.stats['filas_erp']}
- Filas del Excel: {self.stats['filas_excel']}
- Filas eliminadas: {self.stats['filas_eliminadas']}
🗄️ Respaldo:
- Versión ScoreHistory: {self.stats['version_history']}
- Tabla: {self.stats['tabla_history']}
⏱️ Tiempo: {tiempo_ejecucion}
📅 Fecha: {timestamp}
"""
return {
'success': True,
'mensaje': 'Score actualizado correctamente',
'filas_procesadas': self.stats['filas_procesadas'],
'filas_erp': self.stats['filas_erp'],
'filas_excel': self.stats['filas_excel'],
'filas_eliminadas': self.stats['filas_eliminadas'],
'version_history': self.stats['version_history'],
'tabla_history': self.stats['tabla_history'],
'respaldo_ftp': self.stats['respaldo_ftp'],
'tiempo_ejecucion': tiempo_ejecucion,
'timestamp': timestamp,
'resumen': resumen
}
# ========================================
# FUNCIONES AUXILIARES
# ========================================
def generar_nombre_respaldo():
"""
Genera nombre con formato: Score_YYYY-MM-DD_HH-MM-SS.xlsx
Returns:
str: Nombre del archivo
"""
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
return f"Score_{timestamp}.xlsx"
def limpiar_dataframe_para_sql(df):
"""
Limpia DataFrame antes de insertar en SQL
✅ VERSIÓN CORREGIDA: Maneja correctamente fechas 1900-01-01
Args:
df (pd.DataFrame): DataFrame a limpiar
Returns:
pd.DataFrame: DataFrame limpio
"""
import pandas as pd
from datetime import datetime
print(" Limpiando datos para SQL...")
# PASO 1: Convertir NaN a None
df = df.where(pd.notna(df), None)
# PASO 2: Limpiar columnas de fechas (datetime)
for col in df.columns:
if df[col].dtype == 'datetime64[ns]':
print(f" Limpiando columna de fecha: {col}")
def limpiar_fecha(x):
"""Convierte fechas inválidas a None"""
# Si es None o NaN
if x is None or pd.isna(x):
return None
# Si es datetime, verificar si es 1900-01-01
if isinstance(x, pd.Timestamp):
# Convertir a datetime de Python
fecha = x.to_pydatetime()
# Si es 1900-01-01 → None
if fecha.year == 1900 and fecha.month == 1 and fecha.day == 1:
return None
# Fecha válida
return fecha
# Si es string
if isinstance(x, str):
# String vacío → None
if x.strip() == '':
return None
# String con fecha 1900 → None
if '1900-01-01' in x or '1900/01/01' in x:
return None
# Dejar como está
return x
df[col] = df[col].apply(limpiar_fecha)
# PASO 3: Limpiar strings (strip y vacíos → None)
for col in df.columns:
if df[col].dtype == 'object':
df[col] = df[col].apply(
lambda x: None if (x is None or (isinstance(x, str) and x.strip() == '')) else (x.strip() if isinstance(x, str) else x)
)
print(" ✓ Datos limpiados")
return df
def comparar_estructura_columnas(cols1, cols2):
"""
Compara dos listas de columnas
Args:
cols1 (list): Primera lista
cols2 (list): Segunda lista
Returns:
bool: True si son iguales
"""
# Ordenar y comparar
set1 = set(sorted(cols1))
set2 = set(sorted(cols2))
return set1 == set2
def calcular_tiempo_ejecucion(tiempo_inicio):
"""
Calcula tiempo transcurrido
Args:
tiempo_inicio (datetime): Tiempo de inicio
Returns:
str: Tiempo formateado (ej: "2m 34s")
"""
tiempo_fin = datetime.now()
delta = tiempo_fin - tiempo_inicio
segundos = delta.total_seconds()
minutos = int(segundos // 60)
segs = int(segundos % 60)
return f"{minutos}m {segs}s"
# ========================================
# FUNCIÓN DE REGISTRO Y RUTAS FLASK
# ========================================
def ejecutar_actualizacion_score(app, mail):
"""
Registra las rutas de actualización del Score en Flask
Args:
app: Instancia de Flask
mail: Instancia de Flask-Mail
"""
@app.route('/SyC/IngenieroControl/ActualizarScore', methods=['GET'])
def vista_actualizar_score():
"""
Renderiza la vista HTML principal
Returns:
render_template: Template HTML
"""
return render_template('SupYCtrol/IngenieroControl/ActualizarScore.html')
@app.route('/SyC/IngenieroControl/ActualizarScore/ejecutar', methods=['POST'])
def ejecutar_actualizacion():
"""
Ejecuta el proceso completo de actualización del Score
Returns:
jsonify: Respuesta JSON con resultado
"""
try:
# Crear procesador
processor = ScoreProcessor()
# Ejecutar flujo completo
resultado = processor.ejecutar_actualizacion_completa()
if resultado['success']:
return jsonify(resultado), 200
else:
return jsonify(resultado), 400
except Exception as e:
return jsonify({
'success': False,
'error': f'Error crítico: {str(e)}'
}), 500
@app.route('/onedrive/login')
def onedrive_login():
"""Inicia el flujo de autorización"""
import msal
authority = ONEDRIVE_CONFIG['authority'].format(
tenant_id=ONEDRIVE_CONFIG['tenant_id']
)
app_msal = msal.ConfidentialClientApplication(
client_id=ONEDRIVE_CONFIG['client_id'],
client_credential=ONEDRIVE_CONFIG['client_secret'],
authority=authority
)
auth_url = app_msal.get_authorization_request_url(
scopes=ONEDRIVE_CONFIG['scopes'],
redirect_uri=ONEDRIVE_CONFIG['redirect_uri']
)
return redirect(auth_url)
@app.route('/callback')
def onedrive_callback():
"""Callback que recibe el código de autorización"""
import msal
import json
code = request.args.get('code')
if not code:
return "Error: No se recibió código de autorización", 400
authority = ONEDRIVE_CONFIG['authority'].format(
tenant_id=ONEDRIVE_CONFIG['tenant_id']
)
app_msal = msal.ConfidentialClientApplication(
client_id=ONEDRIVE_CONFIG['client_id'],
client_credential=ONEDRIVE_CONFIG['client_secret'],
authority=authority
)
result = app_msal.acquire_token_by_authorization_code(
code=code,
scopes=ONEDRIVE_CONFIG['scopes'],
redirect_uri=ONEDRIVE_CONFIG['redirect_uri']
)
if 'access_token' in result:
# Guardar tokens
with open(ONEDRIVE_CONFIG['token_cache_file'], 'w') as f:
json.dump(result, f)
return """
✅ Autorización Exitosa
OneDrive ha sido autorizado correctamente.
Los tokens se han guardado y se renovarán automáticamente.
Puedes cerrar esta ventana.
"""
else:
return f"Error al obtener tokens: {result.get('error_description', result)}", 500