# Archivo: ScoreProduccionSQL.py # Ruta: Consultas_SQL/Operaciones/Lerma/ScoreProduccion/ScoreProduccionSQL.py # Descripción: Consultas SQL para el módulo de Score de Producción # Autor: Equipo de Desarrollo IGSA # Fecha: 2026 """ Módulo de consultas SQL para Score de Producción. Este módulo maneja todas las consultas relacionadas con: - Obtención de órdenes de producción de la tabla CM_Score - Filtros por departamento, vendedor, fechas, proyecto - Detalles individuales de órdenes - Estadísticas y métricas de órdenes ESTRUCTURA DE TABLA CM_Score: ============================== Campos principales que usaremos: - OrderNum (int) - Número de orden - OrderLine (int) - Línea de orden - PartNum (nvarchar) - Número de parte - LineDesc (nvarchar) - Descripción de línea - OrderDate (date) - Fecha de orden - NeedByDate (date) - Fecha requerida - Departamento (nvarchar) - Departamento - Vendedor (nvarchar) - Vendedor - Name (nvarchar) - Nombre cliente - ProjectID (nvarchar) - ID de proyecto - JobNum2 (nvarchar) - Número de trabajo - FechaProducción (date) - Fecha de producción - AvanceProducción (numeric) - Avance de producción - AvanceDeSurtimiento (numeric) - Avance de surtimiento - EstadoFecha (nvarchar) - Estado de fecha - ComentarioProducción (nvarchar) - Comentarios """ from Consultas_SQL.conexion import get_connection import pyodbc from datetime import datetime, date from decimal import Decimal class ScoreProduccionSQL: """ Clase que contiene todas las consultas SQL para el módulo de Score de Producción. Todos los métodos son estáticos para facilitar su uso sin necesidad de instanciar la clase. """ # ======================================== # CONSULTA PRINCIPAL DE ÓRDENES # ======================================== @staticmethod def obtener_todas_ordenes(filtros=None): """ Obtiene todas las órdenes de producción con filtros opcionales. Esta es la consulta principal que alimenta la tabla del frontend. Args: filtros (dict, optional): Diccionario con filtros opcionales - departamento (str): Filtrar por departamento - vendedor (str): Filtrar por vendedor - proyecto (str): Filtrar por proyecto ID - fecha_desde (date): Filtrar desde fecha - fecha_hasta (date): Filtrar hasta fecha - part_num (str): Filtrar por número de parte - order_num (int): Filtrar por número de orden Returns: dict: { 'success': bool, 'ordenes': list de diccionarios con los datos, 'total': int (cantidad de registros), 'error': str (si falla) } """ conn = None try: conn = get_connection() cursor = conn.cursor() # ==================================================================== # CONSTRUCCIÓN DINÁMICA DE LA QUERY CON FILTROS # ==================================================================== # Query base - Seleccionamos los campos más relevantes query = """ SELECT [OrderNum&Line], OrderNum, OrderLine, PartNum, LineDesc, OrderDate, NeedByDate, Departamento, Vendedor, Name, ProjectID, JobNum2, Capacidad, Voltaje, Tipo, RevisionNum, Revision_Job, ProdCode, DecriptionProd, FechaProducción, ComentarioProducción, AvanceProducción, FechaPlaneación, EstadoFecha, ComentarioPlaneación, AvanceDeSurtimiento, AvisoDeTerminacion, FechaDeTermino, FechaDeCierre, InsumosDemandados, InsumosEmitidos, AvaceDeEmisiones, ComentarioCalidad, FechaActualización, ClienteEEUU, FechaActualización, MaterialFaltante, FechaMat, Caseta, -- Motor IssuedQty_M, OnhandQty_M, Preasignado_M, Demandado_M, Faltante_M, Comment_M, PartNum_M, Alternativa_M, En_PO_M, PO_M, NoPO_M, NoSerie_M, Estatus_M, Fecha_Llegada_M, RefCategory_M, Description_M, -- Generador IssuedQty_G, OnhandQty_G, Preasignado_G, Demandado_G, Faltante_G, PartNum_G, Alternativa_G, En_PO_G, PO_G, NoPO_G, NoSerie_G, Estatus_G, Fecha_Llegada_G, Description_G, Comment_G, EnRequisicion_G, NoRequisicion_G, En_PO_Altern_G, En_PO_Altern_M FROM CM_Score WHERE 1=1 AND PartNum LIKE 'EP%' """ # Lista de parámetros para la query parametros = [] # ==================================================================== # APLICAR FILTROS # ==================================================================== # Filtro por departamento if filtros and filtros.get('departamento'): query += " AND Departamento = ?" parametros.append(filtros['departamento']) # Filtro por vendedor if filtros and filtros.get('vendedor'): query += " AND Vendedor = ?" parametros.append(filtros['vendedor']) # Filtro por proyecto if filtros and filtros.get('proyecto'): query += " AND ProjectID LIKE ?" parametros.append(f"%{filtros['proyecto']}%") # Filtro por número de parte if filtros and filtros.get('part_num'): query += " AND PartNum LIKE ?" parametros.append(f"%{filtros['part_num']}%") # Filtro por número de orden if filtros and filtros.get('order_num'): query += " AND OrderNum = ?" parametros.append(filtros['order_num']) # Filtro por rango de fechas de orden if filtros and filtros.get('fecha_desde'): query += " AND OrderDate >= ?" parametros.append(filtros['fecha_desde']) if filtros and filtros.get('fecha_hasta'): query += " AND OrderDate <= ?" parametros.append(filtros['fecha_hasta']) # ==================================================================== # ORDENAMIENTO # ==================================================================== query += " ORDER BY OrderDate DESC, OrderNum DESC, OrderLine ASC" # ==================================================================== # EJECUTAR QUERY # ==================================================================== print(f"🔍 Ejecutando query con {len(parametros)} filtros") if parametros: cursor.execute(query, parametros) else: cursor.execute(query) # Obtener columnas columnas = [column[0] for column in cursor.description] # Obtener resultados resultados = cursor.fetchall() cursor.close() conn.close() # ==================================================================== # FORMATEAR RESULTADOS # ==================================================================== ordenes = [] for resultado in resultados: orden_dict = {} for idx, columna in enumerate(columnas): valor = resultado[idx] # Formatear tipos especiales if isinstance(valor, datetime): orden_dict[columna] = valor.isoformat() elif isinstance(valor, date): orden_dict[columna] = valor.isoformat() elif isinstance(valor, Decimal): orden_dict[columna] = float(valor) else: orden_dict[columna] = valor ordenes.append(orden_dict) print(f"✅ Se obtuvieron {len(ordenes)} órdenes") return { 'success': True, 'ordenes': ordenes, 'total': len(ordenes), 'error': None } except pyodbc.Error as e: error_msg = f'Error de base de datos al obtener órdenes: {str(e)}' print(f"❌ {error_msg}") return { 'success': False, 'ordenes': [], 'total': 0, 'error': error_msg } except Exception as e: error_msg = f'Error inesperado al obtener órdenes: {str(e)}' print(f"❌ {error_msg}") return { 'success': False, 'ordenes': [], 'total': 0, 'error': error_msg } finally: if conn: try: conn.close() except: pass # ======================================== # CONSULTA DE ORDEN INDIVIDUAL # ======================================== @staticmethod def obtener_orden_por_id(order_line_id): """ Obtiene una orden específica por su OrderNum&Line (clave primaria). Args: order_line_id (int): ID de OrderNum&Line Returns: dict: { 'success': bool, 'orden': dict con todos los datos de la orden, 'existe': bool, 'error': str (si falla) } """ conn = None try: conn = get_connection() cursor = conn.cursor() # Consulta para obtener TODOS los campos de la orden query = """ SELECT * FROM CM_Score WHERE [OrderNum&Line] = ? """ cursor.execute(query, (order_line_id,)) # Obtener columnas columnas = [column[0] for column in cursor.description] # Obtener resultado resultado = cursor.fetchone() cursor.close() conn.close() if resultado: orden_dict = {} for idx, columna in enumerate(columnas): valor = resultado[idx] # Formatear tipos especiales if isinstance(valor, datetime): orden_dict[columna] = valor.isoformat() elif isinstance(valor, date): orden_dict[columna] = valor.isoformat() elif isinstance(valor, Decimal): orden_dict[columna] = float(valor) else: orden_dict[columna] = valor return { 'success': True, 'orden': orden_dict, 'existe': True, 'error': None } else: return { 'success': False, 'orden': None, 'existe': False, 'error': f'Orden con ID {order_line_id} no encontrada' } except pyodbc.Error as e: error_msg = f'Error de base de datos al obtener orden: {str(e)}' print(f"❌ {error_msg}") return { 'success': False, 'orden': None, 'existe': False, 'error': error_msg } except Exception as e: error_msg = f'Error inesperado al obtener orden: {str(e)}' print(f"❌ {error_msg}") return { 'success': False, 'orden': None, 'existe': False, 'error': error_msg } finally: if conn: try: conn.close() except: pass # ======================================== # CONSULTAS DE CATÁLOGOS # ======================================== @staticmethod def obtener_departamentos_disponibles(): """ Obtiene todos los departamentos únicos. Returns: dict: { 'success': bool, 'departamentos': list de strings, 'error': str (si falla) } """ conn = None try: conn = get_connection() cursor = conn.cursor() query = """ SELECT DISTINCT Departamento FROM CM_Score WHERE Departamento IS NOT NULL ORDER BY Departamento ASC """ cursor.execute(query) resultados = cursor.fetchall() cursor.close() conn.close() departamentos = [resultado[0] for resultado in resultados if resultado[0]] return { 'success': True, 'departamentos': departamentos, 'error': None } except Exception as e: error_msg = f'Error al obtener departamentos: {str(e)}' print(f"❌ {error_msg}") return { 'success': False, 'departamentos': [], 'error': error_msg } finally: if conn: try: conn.close() except: pass @staticmethod def obtener_vendedores_disponibles(): """ Obtiene todos los vendedores únicos. Returns: dict: { 'success': bool, 'vendedores': list de strings, 'error': str (si falla) } """ conn = None try: conn = get_connection() cursor = conn.cursor() query = """ SELECT DISTINCT Vendedor FROM CM_Score WHERE Vendedor IS NOT NULL ORDER BY Vendedor ASC """ cursor.execute(query) resultados = cursor.fetchall() cursor.close() conn.close() vendedores = [resultado[0] for resultado in resultados if resultado[0]] return { 'success': True, 'vendedores': vendedores, 'error': None } except Exception as e: error_msg = f'Error al obtener vendedores: {str(e)}' print(f"❌ {error_msg}") return { 'success': False, 'vendedores': [], 'error': error_msg } finally: if conn: try: conn.close() except: pass # ======================================== # ESTADÍSTICAS Y MÉTRICAS # ======================================== @staticmethod def obtener_estadisticas_generales(): """ Obtiene estadísticas generales de las órdenes. Returns: dict: { 'success': bool, 'estadisticas': dict con métricas generales, 'error': str (si falla) } """ conn = None try: conn = get_connection() cursor = conn.cursor() query = """ SELECT COUNT(*) as TotalOrdenes, COUNT(DISTINCT OrderNum) as OrdenesUnicas, COUNT(DISTINCT Departamento) as TotalDepartamentos, COUNT(DISTINCT Vendedor) as TotalVendedores, COUNT(DISTINCT ProjectID) as TotalProyectos, AVG(CAST(AvanceProducción AS FLOAT)) as PromedioAvanceProduccion, AVG(CAST(AvanceDeSurtimiento AS FLOAT)) as PromedioAvanceSurtimiento, SUM(CASE WHEN AvisoDeTerminacion IS NOT NULL THEN 1 ELSE 0 END) as OrdenesTerminadas, SUM(CASE WHEN FechaDeCierre IS NOT NULL THEN 1 ELSE 0 END) as OrdenesCerradas FROM CM_Score """ cursor.execute(query) resultado = cursor.fetchone() cursor.close() conn.close() if resultado: estadisticas = { 'TotalOrdenes': resultado[0] or 0, 'OrdenesUnicas': resultado[1] or 0, 'TotalDepartamentos': resultado[2] or 0, 'TotalVendedores': resultado[3] or 0, 'TotalProyectos': resultado[4] or 0, 'PromedioAvanceProduccion': float(resultado[5]) if resultado[5] else 0.0, 'PromedioAvanceSurtimiento': float(resultado[6]) if resultado[6] else 0.0, 'OrdenesTerminadas': resultado[7] or 0, 'OrdenesCerradas': resultado[8] or 0 } return { 'success': True, 'estadisticas': estadisticas, 'error': None } else: return { 'success': False, 'estadisticas': None, 'error': 'No se pudieron obtener estadísticas' } except Exception as e: error_msg = f'Error al obtener estadísticas: {str(e)}' print(f"❌ {error_msg}") return { 'success': False, 'estadisticas': None, 'error': error_msg } finally: if conn: try: conn.close() except: pass