# Archivo: ScoreControl.py # Ruta: App\SupyCtrol_Module\IngenieroControl\ScoreControl.py # Descripción: Controlador para el módulo de Score de Control # Autor: Equipo de Desarrollo IGSA # Fecha: 2026 """ Módulo de Score de Producción. Este módulo gestiona todo el flujo de visualización y análisis de órdenes: 1. Obtención de órdenes con filtros opcionales 2. Consulta de detalles individuales 3. Estadísticas y métricas generales 4. Exportación de datos ARQUITECTURA: ============= Este módulo sigue el patrón MVC: - Model: ScoreProduccionSQL (consultas a BD - CM_Score) - View: ScoreProduccion.html (template Jinja2) - Controller: ScoreProduccionController (lógica de negocio) """ from flask import jsonify, render_template, request from datetime import datetime, date import sys import os # Importar consultas SQL sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))) from Consultas_SQL.SupYCtrol.IngDeControl.ScoreControlSQL import ScoreProduccionSQL class ScoreProduccionController: """ Controlador principal para el módulo de Score de Producción. Esta clase maneja toda la lógica de negocio para visualizar y analizar órdenes de producción desde la tabla CM_Score. """ def __init__(self): """Inicializa el controlador.""" self.sql = ScoreProduccionSQL() # ======================================== # FUNCIÓN ORQUESTADORA PRINCIPAL # ======================================== def obtener_ordenes_produccion(self, filtros_json=None): """ 🎯 FUNCIÓN ORQUESTADORA PRINCIPAL Esta función coordina todo el flujo de obtención de órdenes de producción. FLUJO DE EJECUCIÓN: 1. Validar y procesar filtros (si existen) 2. Ejecutar consulta SQL con filtros 3. Procesar y enriquecer datos 4. Calcular estadísticas adicionales 5. Retornar respuesta estructurada Args: filtros_json (dict, optional): Diccionario con filtros opcionales - departamento (str): Filtrar por departamento - vendedor (str): Filtrar por vendedor - proyecto (str): Filtrar por proyecto ID - fechaDesde (str): Fecha desde (ISO format) - fechaHasta (str): Fecha hasta (ISO format) - partNum (str): Filtrar por número de parte - orderNum (int): Filtrar por número de orden Returns: dict: Respuesta estructurada con success, ordenes, estadísticas o errores """ try: print("\n" + "="*80) print("🎯 INICIANDO OBTENCIÓN DE ÓRDENES DE PRODUCCIÓN") print("="*80) # ==================================================================== # PASO 1: PROCESAR FILTROS # ==================================================================== print("\n🔍 PASO 1: Procesando filtros...") filtros_procesados = self._procesar_filtros(filtros_json) if filtros_procesados: print(f"✅ Filtros activos: {list(filtros_procesados.keys())}") else: print("ℹ️ Sin filtros aplicados") # ==================================================================== # PASO 2: OBTENER ÓRDENES DE BASE DE DATOS # ==================================================================== print("\n💾 PASO 2: Consultando base de datos...") resultado_ordenes = self.sql.obtener_todas_ordenes(filtros_procesados) if not resultado_ordenes['success']: print(f"❌ Error al obtener órdenes: {resultado_ordenes['error']}") return { 'success': False, 'mensaje': 'Error al obtener órdenes de producción', 'error': resultado_ordenes['error'] } ordenes = resultado_ordenes['ordenes'] print(f"✅ Órdenes obtenidas: {len(ordenes)}") # ==================================================================== # PASO 3: ENRIQUECER DATOS # ==================================================================== print("\n🔧 PASO 3: Enriqueciendo datos...") ordenes_enriquecidas = self._enriquecer_ordenes(ordenes) print("✅ Datos enriquecidos") # ==================================================================== # PASO 4: CALCULAR ESTADÍSTICAS ADICIONALES # ==================================================================== print("\n📊 PASO 4: Calculando estadísticas...") estadisticas = self._calcular_estadisticas(ordenes_enriquecidas) print("✅ Estadísticas calculadas") # ==================================================================== # PASO 5: RETORNAR RESPUESTA EXITOSA # ==================================================================== print("\n" + "="*80) print("🎉 ÓRDENES OBTENIDAS EXITOSAMENTE") print(f"📊 Total de registros: {len(ordenes_enriquecidas)}") print(f"📈 Promedio avance producción: {estadisticas.get('promedioAvanceProduccion', 0):.2f}%") print(f"📦 Promedio avance surtimiento: {estadisticas.get('promedioAvanceSurtimiento', 0):.2f}%") print("="*80 + "\n") return { 'success': True, 'ordenes': ordenes_enriquecidas, 'total': len(ordenes_enriquecidas), 'estadisticas': estadisticas, 'filtros_aplicados': filtros_procesados } except Exception as e: print(f"\n❌ ERROR CRÍTICO: {str(e)}") import traceback traceback.print_exc() return { 'success': False, 'mensaje': 'Error al procesar la solicitud', 'error': str(e) } # ======================================== # FUNCIÓN: OBTENER ORDEN INDIVIDUAL # ======================================== def obtener_detalle_orden(self, order_line_id): """ Obtiene el detalle completo de una orden específica. Args: order_line_id (int): ID de OrderNum&Line (clave primaria) Returns: dict: Respuesta con los datos de la orden o error """ try: print(f"\n🔍 Obteniendo detalle de Orden ID: {order_line_id}") resultado = self.sql.obtener_orden_por_id(order_line_id) if resultado['success'] and resultado['existe']: orden_enriquecida = resultado['orden'] print(f"✅ Orden obtenida: {orden_enriquecida.get('OrderNum')} - {orden_enriquecida.get('PartNum')}") return { 'success': True, 'orden': orden_enriquecida } else: print(f"❌ Orden no encontrada: {resultado['error']}") return { 'success': False, 'mensaje': 'Orden no encontrada', 'error': resultado['error'] } except Exception as e: print(f"❌ ERROR: {str(e)}") return { 'success': False, 'mensaje': 'Error al obtener detalle de la orden', 'error': str(e) } # ======================================== # FUNCIONES DE PROCESAMIENTO # ======================================== def _procesar_filtros(self, filtros_json): """ Procesa y valida los filtros recibidos del frontend. Convierte de camelCase (frontend) a snake_case (SQL). Args: filtros_json (dict): Filtros en formato camelCase Returns: dict: Filtros procesados en formato snake_case """ if not filtros_json: return None filtros_procesados = {} # Mapeo de nombres camelCase a snake_case mapeo_filtros = { 'departamento': 'departamento', 'vendedor': 'vendedor', 'proyecto': 'proyecto', 'partNum': 'part_num', 'orderNum': 'order_num', 'fechaDesde': 'fecha_desde', 'fechaHasta': 'fecha_hasta' } for key_camel, key_snake in mapeo_filtros.items(): if key_camel in filtros_json and filtros_json[key_camel]: valor = filtros_json[key_camel] # Convertir fechas si es necesario if 'fecha' in key_snake.lower() and isinstance(valor, str): try: valor = datetime.fromisoformat(valor.replace('Z', '')).date() except: print(f"⚠️ No se pudo convertir fecha: {valor}") continue # Convertir números si es necesario if key_snake == 'order_num' and isinstance(valor, str): try: valor = int(valor) except: print(f"⚠️ No se pudo convertir número de orden: {valor}") continue filtros_procesados[key_snake] = valor return filtros_procesados if filtros_procesados else None def _normalizar_mg(self, orden: dict) -> dict: defaults = { # Motor 'IssuedQty_M': 0, 'Preasignado_M': 0, 'Demandado_M': 0, 'PartNum_M': None, 'Alternativa_M': None, 'En_PO_M': 0, 'PO_M': None, 'NoPO_M': None, # Generador 'IssuedQty_G': 0, 'Preasignado_G': 0, 'Demandado_G': 0, 'PartNum_G': None, 'Alternativa_G': None, 'En_PO_G': 0, 'PO_G': None, 'NoPO_G': None, } for k, v in defaults.items(): if orden.get(k) is None: orden[k] = v return orden def _enriquecer_ordenes(self, ordenes): """ Enriquece los datos de órdenes con información calculada adicional. Args: ordenes (list): Lista de órdenes originales Returns: list: Lista de órdenes enriquecidas """ ordenes_enriquecidas = [] for orden in ordenes: orden_enriquecida = orden.copy() orden_enriquecida = self._normalizar_mg(orden_enriquecida) # Calcular días desde la fecha de orden if orden.get('OrderDate'): try: fecha_orden = datetime.fromisoformat(orden['OrderDate'].replace('Z', '')) dias_desde_orden = (datetime.now() - fecha_orden).days orden_enriquecida['DiasDesdOrden'] = dias_desde_orden except: orden_enriquecida['DiasDesdOrden'] = None # Calcular días hasta fecha requerida if orden.get('NeedByDate'): try: fecha_requerida = datetime.fromisoformat(orden['NeedByDate'].replace('Z', '')) dias_hasta_requerida = (fecha_requerida - datetime.now()).days orden_enriquecida['DiasHastaRequerida'] = dias_hasta_requerida # Clasificar urgencia if dias_hasta_requerida < 0: orden_enriquecida['Urgencia'] = 'Atrasada' orden_enriquecida['ColorUrgencia'] = 'danger' elif dias_hasta_requerida <= 7: orden_enriquecida['Urgencia'] = 'Urgente' orden_enriquecida['ColorUrgencia'] = 'warning' elif dias_hasta_requerida <= 30: orden_enriquecida['Urgencia'] = 'Próxima' orden_enriquecida['ColorUrgencia'] = 'info' else: orden_enriquecida['Urgencia'] = 'Normal' orden_enriquecida['ColorUrgencia'] = 'success' except: orden_enriquecida['DiasHastaRequerida'] = None orden_enriquecida['Urgencia'] = 'Desconocida' orden_enriquecida['ColorUrgencia'] = 'secondary' # ==================================================================== # MANEJO SEGURO DE VALORES NULL PARA AVANCE DE PRODUCCIÓN # ==================================================================== avance_produccion = orden.get('AvanceProducción') # Convertir None a 0 para evitar errores de comparación if avance_produccion is None: avance_produccion = 0.0 # Clasificar estado de avance de producción if avance_produccion >= 100: orden_enriquecida['EstadoProduccion'] = 'Completado' orden_enriquecida['ColorEstado'] = 'success' elif avance_produccion >= 75: orden_enriquecida['EstadoProduccion'] = 'Avanzado' orden_enriquecida['ColorEstado'] = 'primary' elif avance_produccion >= 50: orden_enriquecida['EstadoProduccion'] = 'En Proceso' orden_enriquecida['ColorEstado'] = 'info' elif avance_produccion > 0: orden_enriquecida['EstadoProduccion'] = 'Iniciado' orden_enriquecida['ColorEstado'] = 'warning' else: orden_enriquecida['EstadoProduccion'] = 'Pendiente' orden_enriquecida['ColorEstado'] = 'secondary' # ==================================================================== # MANEJO SEGURO DE VALORES NULL PARA INSUMOS # ==================================================================== insumos_demandados = orden.get('InsumosDemandados') insumos_emitidos = orden.get('InsumosEmitidos') # Convertir None a 0 if insumos_demandados is None: insumos_demandados = 0 if insumos_emitidos is None: insumos_emitidos = 0 if insumos_demandados > 0: porcentaje_emision = (insumos_emitidos / insumos_demandados) * 100 orden_enriquecida['PorcentajeEmision'] = round(porcentaje_emision, 2) if porcentaje_emision < 50: orden_enriquecida['AlertaMaterial'] = 'Crítico' elif porcentaje_emision < 80: orden_enriquecida['AlertaMaterial'] = 'Pendiente' else: orden_enriquecida['AlertaMaterial'] = 'OK' else: orden_enriquecida['PorcentajeEmision'] = 0 orden_enriquecida['AlertaMaterial'] = 'N/A' ordenes_enriquecidas.append(orden_enriquecida) return ordenes_enriquecidas def _calcular_estadisticas(self, ordenes): """ Calcula estadísticas generales de las órdenes. Args: ordenes (list): Lista de órdenes Returns: dict: Diccionario con estadísticas calculadas """ if not ordenes or len(ordenes) == 0: return { 'promedioAvanceProduccion': 0.0, 'promedioAvanceSurtimiento': 0.0, 'totalOrdenes': 0, 'ordenesUnicas': 0, 'distribucionDepartamento': {}, 'distribucionUrgencia': {}, 'distribucionEstadoProduccion': {} } # ==================================================================== # CALCULAR PROMEDIOS CON MANEJO SEGURO DE NULL # ==================================================================== suma_avance_produccion = 0 suma_avance_surtimiento = 0 count_avance_produccion = 0 count_avance_surtimiento = 0 for orden in ordenes: # Avance producción avance_prod = orden.get('AvanceProducción') if avance_prod is not None: suma_avance_produccion += avance_prod count_avance_produccion += 1 # Avance surtimiento avance_surt = orden.get('AvanceDeSurtimiento') if avance_surt is not None: suma_avance_surtimiento += avance_surt count_avance_surtimiento += 1 # Calcular promedios (evitar división por cero) promedio_avance_produccion = (suma_avance_produccion / count_avance_produccion if count_avance_produccion > 0 else 0) promedio_avance_surtimiento = (suma_avance_surtimiento / count_avance_surtimiento if count_avance_surtimiento > 0 else 0) # Contar órdenes únicas ordenes_unicas = len(set(orden.get('OrderNum') for orden in ordenes if orden.get('OrderNum'))) # Distribución por departamento distribucion_departamento = {} for orden in ordenes: dept = orden.get('Departamento') or 'Sin Departamento' distribucion_departamento[dept] = distribucion_departamento.get(dept, 0) + 1 # Distribución por urgencia distribucion_urgencia = {} for orden in ordenes: urgencia = orden.get('Urgencia', 'Desconocida') distribucion_urgencia[urgencia] = distribucion_urgencia.get(urgencia, 0) + 1 # Distribución por estado de producción distribucion_estado = {} for orden in ordenes: estado = orden.get('EstadoProduccion', 'Desconocido') distribucion_estado[estado] = distribucion_estado.get(estado, 0) + 1 return { 'promedioAvanceProduccion': round(promedio_avance_produccion, 2), 'promedioAvanceSurtimiento': round(promedio_avance_surtimiento, 2), 'totalOrdenes': len(ordenes), 'ordenesUnicas': ordenes_unicas, 'distribucionDepartamento': distribucion_departamento, 'distribucionUrgencia': distribucion_urgencia, 'distribucionEstadoProduccion': distribucion_estado } # ======================================== # FUNCIÓN PARA REGISTRAR RUTAS EN FLASK # ======================================== def ejecutar_score_control(app, mail): """ Registra las rutas de Flask para el módulo de Score de Producción. Args: app: Instancia de Flask mail: Instancia de Flask-Mail (no se usa en este módulo) """ # Instancia global del controlador controller = ScoreProduccionController() # ======================================== # RUTA: TEST DE CONEXIÓN # ======================================== @app.route('/SyC/IngenieroControl/test', methods=['GET']) def test_score_control(): """ Ruta de prueba para verificar que el backend funciona correctamente. Returns: JSON con mensaje de éxito """ return jsonify({ 'success': True, 'mensaje': '¡Backend de Score de Producción funcionando correctamente!', 'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S"), 'modulo': 'Score de Producción - Planta Lerma', 'version': '1.0', 'axios_ready': True }) # ======================================== # RUTA: OBTENER TODAS LAS ÓRDENES # ======================================== @app.route('/SyC/IngenieroControl/obtener-ordenes', methods=['GET', 'POST']) def obtener_ordenes_score_control_endpoint(): """ Endpoint para obtener todas las órdenes de producción. Acepta filtros opcionales vía query params (GET) o JSON (POST). Returns: JSON con lista de órdenes y estadísticas """ try: print("\n" + "="*80) print("📥 RECIBIENDO PETICIÓN /obtener-ordenes") print("="*80) # Obtener filtros según el método filtros = None if request.method == 'POST': filtros = request.get_json() print("📊 Filtros recibidos (POST):", filtros) elif request.method == 'GET': # Convertir query params a diccionario filtros = { 'departamento': request.args.get('departamento'), 'vendedor': request.args.get('vendedor'), 'proyecto': request.args.get('proyecto'), 'partNum': request.args.get('partNum'), 'orderNum': request.args.get('orderNum'), 'fechaDesde': request.args.get('fechaDesde'), 'fechaHasta': request.args.get('fechaHasta') } # Remover valores None filtros = {k: v for k, v in filtros.items() if v is not None} print("📊 Filtros recibidos (GET):", filtros) # Ejecutar función orquestadora resultado = controller.obtener_ordenes_produccion(filtros) # Determinar código de respuesta HTTP status_code = 200 if resultado['success'] else 400 return jsonify(resultado), status_code except Exception as e: print(f"\n❌ ERROR EN ENDPOINT: {str(e)}") import traceback traceback.print_exc() return jsonify({ 'success': False, 'mensaje': 'Error al procesar la solicitud', 'error': str(e) }), 500 # ======================================== # RUTA: OBTENER DETALLE DE ORDEN # ======================================== @app.route('/SyC/IngenieroControl/detalle/', methods=['GET']) def obtener_detalle_orden_score_control_endpoint(order_line_id): """ Endpoint para obtener el detalle de una orden específica. Args: order_line_id (int): ID de OrderNum&Line Returns: JSON con datos de la orden """ try: print(f"\n📥 GET /detalle/{order_line_id}") resultado = controller.obtener_detalle_orden(order_line_id) status_code = 200 if resultado['success'] else 404 return jsonify(resultado), status_code except Exception as e: print(f"❌ ERROR: {str(e)}") return jsonify({ 'success': False, 'mensaje': 'Error al obtener detalle de la orden', 'error': str(e) }), 500 # ======================================== # RUTA: OBTENER DEPARTAMENTOS DISPONIBLES # ======================================== @app.route('/SyC/IngenieroControl/departamentos', methods=['GET']) def obtener_departamentos_score_control_endpoint(): """ Endpoint para obtener todos los departamentos disponibles. Returns: JSON con lista de departamentos """ try: resultado = controller.sql.obtener_departamentos_disponibles() return jsonify(resultado), 200 except Exception as e: return jsonify({ 'success': False, 'mensaje': 'Error al obtener departamentos', 'error': str(e) }), 500 # ======================================== # RUTA: OBTENER VENDEDORES DISPONIBLES # ======================================== @app.route('/SyC/IngenieroControl/vendedores', methods=['GET']) def obtener_vendedores_score_control_endpoint(): """ Endpoint para obtener todos los vendedores disponibles. Returns: JSON con lista de vendedores """ try: resultado = controller.sql.obtener_vendedores_disponibles() return jsonify(resultado), 200 except Exception as e: return jsonify({ 'success': False, 'mensaje': 'Error al obtener vendedores', 'error': str(e) }), 500 # ======================================== # RUTA: OBTENER ESTADÍSTICAS GENERALES # ======================================== @app.route('/SyC/IngenieroControl/estadisticas', methods=['GET']) def obtener_estadisticas_score_control_endpoint(): """ Endpoint para obtener estadísticas generales del sistema. Returns: JSON con estadísticas """ try: resultado = controller.sql.obtener_estadisticas_generales() return jsonify(resultado), 200 except Exception as e: return jsonify({ 'success': False, 'mensaje': 'Error al obtener estadísticas', 'error': str(e) }), 500 print("✅ Rutas de Score de Producción registradas correctamente")