# Archivo: RecepcionCotizSQL.py # Ruta: src\Consultas_SQL\Ventas\Cotiz\RecepcionCotizSQL.py from datetime import datetime from decimal import Decimal import pyodbc import logging from typing import Optional, List, Dict, Union from Consultas_SQL.conexion import get_connection logger = logging.getLogger('recepcion_cotizacion_sql') class OpportunityAndCostingDTO: """DTO para la información unificada de Oportunidad y Encabezado de Costeo.""" def __init__(self, costing_id: str, crm_opportunity_number: str, crm_contact_name: str, crm_contact_type: str, crm_assigned_salesperson: str, crm_contact_adress: str, crm_contact_colonia: str, crm_contact_city: str, crm_contact_number: str, crm_contact_country: str, crm_contact_legal_identifier: str, crm_contact_zip: str, crm_contact_state: str, crm_contact_email: str, case_cost: str, sale_price_list: float, sale_price_min: float, discount_max_percent: float, run_time_number: int = None, run_time_type: str = None, technical_terms_and_conditions: str = None, FinancePercent: float = None, TaxCode: str = None, CurrencyCode: str = None, Q_TaxRate_FrontES: str = None, Q_TaxRate_FrontEN: str = None, Q_Currency_FrontES: str = None, Q_Currency_FrontEN: str = None): self.CostingID = costing_id self.CRM_OpportunityNumber = crm_opportunity_number self.CRM_ContactName = crm_contact_name self.CRM_ContactType = crm_contact_type self.CRM_AssignedSalesperson = crm_assigned_salesperson self.CRM_ContactAdress = crm_contact_adress self.CRM_ContactColonia = crm_contact_colonia self.CRM_ContactCity = crm_contact_city self.CRM_ContactNumber = crm_contact_number self.CRM_ContactCountry = crm_contact_country self.CRM_ContactLegalIdentifier = crm_contact_legal_identifier self.CRM_ContactZip = crm_contact_zip self.CRM_ContactState = crm_contact_state self.CRM_ContactEmail = crm_contact_email self.CaseCost = case_cost self.SalePriceList = sale_price_list self.SalePriceMin = sale_price_min self.DiscountMaxPercent = discount_max_percent self.RunTimeNumber = run_time_number self.RunTimeType = run_time_type self.TechnicalTermsAndConditions = technical_terms_and_conditions, self.FinancePercent = FinancePercent # Inicializar FinancePercent como None self.TaxCode = TaxCode, self.CurrencyCode = CurrencyCode, self.Q_TaxRate_FrontES = Q_TaxRate_FrontES, self.Q_TaxRate_FrontEN = Q_TaxRate_FrontEN, self.Q_Currency_FrontES = Q_Currency_FrontES, self.Q_Currency_FrontEN = Q_Currency_FrontEN def to_dict(self) -> dict: """Convierte el objeto a un diccionario serializable.""" return { "CostingID": self.CostingID, "CRM_OpportunityNumber": self.CRM_OpportunityNumber, "CRM_ContactName": self.CRM_ContactName, "CRM_ContactType": self.CRM_ContactType, "CRM_AssignedSalesperson": self.CRM_AssignedSalesperson, "CRM_ContactAdress": self.CRM_ContactAdress, "CRM_ContactColonia": self.CRM_ContactColonia, "CRM_ContactCity": self.CRM_ContactCity, "CRM_ContactNumber": self.CRM_ContactNumber, "CRM_ContactCountry": self.CRM_ContactCountry, "CRM_ContactLegalIdentifier": self.CRM_ContactLegalIdentifier, "CRM_ContactZip": self.CRM_ContactZip, "CRM_ContactState": self.CRM_ContactState, "CRM_ContactEmail": self.CRM_ContactEmail, "CaseCost": self.CaseCost, "SalePriceList": float(self.SalePriceList), "SalePriceMin": float(self.SalePriceMin), "DiscountMaxPercent": float(self.DiscountMaxPercent), "RunTimeNumber": self.RunTimeNumber, "RunTimeType": self.RunTimeType, "TechnicalTermsAndConditions": self.TechnicalTermsAndConditions[0], "FinancePercent": float(self.FinancePercent) if self.FinancePercent is not None else None, "TaxCode" : self.TaxCode[0], "CurrencyCode": self.CurrencyCode[0], "Q_TaxRate_FrontES": self.Q_TaxRate_FrontES[0], "Q_TaxRate_FrontEN": self.Q_TaxRate_FrontEN[0], "Q_Currency_FrontES": self.Q_Currency_FrontES[0], "Q_Currency_FrontEN": self.Q_Currency_FrontEN } class RecepcionCotizSQL: """Clase de servicio para manejar la lógica de recepción de datos de costeo y oportunidad.""" @staticmethod def get_unified_data_by_costing_num(costing_num: int) -> Optional[OpportunityAndCostingDTO]: """ Consulta unificada que obtiene los datos de Q_CostingHead y Q_OpportunityCRM en una sola llamada. """ query = """ SELECT TOP 1 Q_CostingHead.CostingID, Q_OpportunityCRM.CRM_OpportunityNumber, Q_OpportunityCRM.CRM_ContactName, Q_OpportunityCRM.CRM_ContactType, Q_OpportunityCRM.CRM_AssignedSalesperson, Q_OpportunityCRM.CRM_ContactAdress, Q_OpportunityCRM.CRM_ContactColonia, Q_OpportunityCRM.CRM_ContactCity, Q_OpportunityCRM.CRM_ContactNumber, Q_OpportunityCRM.CRM_ContactCountry, Q_OpportunityCRM.CRM_ContactLegalIdentifier, Q_OpportunityCRM.CRM_ContactZip, Q_OpportunityCRM.CRM_ContactState, Q_OpportunityCRM.CRM_ContactEmail, Q_CostingHead.CaseCost, Q_CostingHead.SalePriceList, Q_CostingHead.SalePriceMin, Q_CostingHead.DiscountMaxPercent, Q_CostingHead.RunTimeNumber, Q_CostingHead.RunTimeType, Q_CostingHead.TechnicalTermsAndConditions, Q_CostingHead.FinancePercent, Q_CostingHead.TaxCode, Q_CostingHead.CurrencyCode, Q_TaxRate.FrontES as Q_TaxRate_FrontES, Q_TaxRate.FrontEN as Q_TaxRate_FrontEN, Q_Currency.FrontES as Q_Currency_FrontES, Q_Currency.FrontEN as Q_Currency_FrontEN FROM Q_CostingHead INNER JOIN Q_OpportunityCRM ON Q_CostingHead.CRM_OpportunityID = Q_OpportunityCRM.CRM_OpportunityID LEFT JOIN Q_TaxRate ON Q_TaxRate.TaxCode = Q_CostingHead.TaxCode AND Q_TaxRate.CurrencyCode = Q_CostingHead.CurrencyCode AND Q_TaxRate.Active >= 1 LEFT JOIN Q_Currency ON Q_Currency.CurrencyCode = Q_CostingHead.CurrencyCode AND Q_Currency.Active >= 1 WHERE Q_CostingHead.CostingNum = ? ORDER BY Q_CostingHead.Version DESC; """ try: with get_connection() as conn: with conn.cursor() as cursor: cursor.execute(query, costing_num) row = cursor.fetchone() if row: return OpportunityAndCostingDTO( costing_id=row[0], crm_opportunity_number=row[1], crm_contact_name=row[2], crm_contact_type=row[3], crm_assigned_salesperson=row[4], crm_contact_adress=row[5], crm_contact_colonia=row[6], crm_contact_city=row[7], crm_contact_number=row[8], crm_contact_country=row[9], crm_contact_legal_identifier=row[10], crm_contact_zip=row[11], crm_contact_state=row[12], crm_contact_email=row[13], case_cost=row[14], sale_price_list=row[15], sale_price_min=row[16], discount_max_percent=row[17], run_time_number=row[18], run_time_type=row[19], technical_terms_and_conditions=row[20], FinancePercent=row[21], TaxCode=row[22], CurrencyCode=row[23], Q_TaxRate_FrontES = row[24], Q_TaxRate_FrontEN = row[25], Q_Currency_FrontES = row[26], Q_Currency_FrontEN = row[27] ) return None except Exception as e: print(f"Error al obtener datos unificados: {e}") return None @staticmethod def get_taxes() -> List[Dict[str, Union[str, float]]]: """ Consulta la tabla Q_TaxRate para obtener el código y la descripción de los impuestos. """ query = "SELECT TaxCode, FrontES, TaxAmount, CurrencyCode FROM Q_TaxRate WHERE Active = 1;" taxes = [] try: with get_connection() as conn: with conn.cursor() as cursor: cursor.execute(query) rows = cursor.fetchall() for row in rows: taxes.append({"TaxCode": row[0], "FrontES": row[1], "TaxAmount": float( row[2]), "CurrencyCode": row[3]}) return taxes except Exception as e: print(f"Error al obtener los impuestos: {e}") return [] @staticmethod def get_currencies() -> List[Dict[str, str]]: """ Consulta la tabla Q_Currency para obtener el código, el nombre y el símbolo de las monedas activas. """ query = "SELECT CurrencyCode, FrontES, CurrSymbol FROM Q_Currency WHERE Active = 1;" currencies = [] try: with get_connection() as conn: with conn.cursor() as cursor: cursor.execute(query) rows = cursor.fetchall() for row in rows: currencies.append( {"CurrencyCode": row[0], "FrontES": row[1], "CurrSymbol": row[2]}) return currencies except Exception as e: print(f"Error al obtener las monedas: {e}") return [] @staticmethod def get_costing_details(costing_num: int) -> List[Dict]: """ Consulta Q_CostingDetail para obtener las líneas de costeo. """ query = """ SELECT CostingLine, PartNum, PartDescription, Qty, UOMCode, UnitPrice, Amount FROM Q_CostingDetail WHERE CostingID = ( SELECT TOP 1 CostingID FROM Q_CostingHead WHERE CostingNum = ? ORDER BY Version DESC ) ORDER BY CostingLine ASC; """ details = [] try: with get_connection() as conn: with conn.cursor() as cursor: cursor.execute(query, costing_num) rows = cursor.fetchall() for row in rows: details.append({ "CostingLine": row[0], "PartNum": row[1], "PartDescription": row[2], "Qty": float(row[3]), "UOMCode": row[4], "UnitPrice": float(row[5]), "Amount": float(row[6]) }) return details except Exception as e: print(f"Error al obtener detalles de costeo: {e}") return [] # Al final de la clase Quote_Reception_Service, antes del método create_quotation @staticmethod def generate_quotation_preview(data: dict) -> str: """ Genera HTML de vista previa de la cotización SIN guardar en BD. Retorna el HTML renderizado. """ try: from flask import render_template from datetime import datetime costing_id = data.get('CostingID') # ✅ Query CORREGIDO: El vendedor es el UserID de la oportunidad query = """ SELECT TOP 1 Q_CostingHead.CostingNum, Q_CostingHead.Version, -- Cliente Q_OpportunityCRM.CRM_ContactName, Q_OpportunityCRM.CRM_ContactType, Q_OpportunityCRM.CRM_OpportunityNumber, Q_OpportunityCRM.CRM_ContactEmail, Q_OpportunityCRM.CRM_ContactNumber, Q_OpportunityCRM.CRM_ContactCity, Q_OpportunityCRM.CRM_ContactState, Q_OpportunityCRM.CRM_ContactCountry, Q_OpportunityCRM.CRM_ContactAdress, -- ✅ El vendedor es el UserID de la oportunidad TRIM(CONCAT( ISNULL(Seller.FirstName, ''), ' ', ISNULL(Seller.MiddleName, ''), ' ', ISNULL(Seller.LastName, ''), ' ', ISNULL(Seller.SecondLastName, '') )) AS VendedorNombre, ISNULL(Seller.Email, '') AS VendedorEmail, ISNULL(Seller.ContactPhone, '') AS VendedorTelefono, Q_CostingHead.TaxCode FROM Q_CostingHead INNER JOIN Q_OpportunityCRM ON Q_CostingHead.CRM_OpportunityID = Q_OpportunityCRM.CRM_OpportunityID -- ✅ JOIN directo: El vendedor es Q_OpportunityCRM.UserID LEFT JOIN Profiles AS Seller ON Q_OpportunityCRM.UserID = Seller.UserID WHERE Q_CostingHead.CostingID = ? """ with get_connection() as conn: with conn.cursor() as cursor: cursor.execute(query, costing_id) row = cursor.fetchone() taxCode = row[14] query2 = """ select TaxPercent from Q_TaxRate where TaxCode = ? """ cursor.execute(query2, (taxCode,)) tax_row = cursor.fetchone() tax_percent = tax_row[0] if tax_row else 0 if not row: print(f"❌ No se encontró CostingID: {costing_id}") return None quotation_num = row[0] current_version = row[1] if row[1] else 0 # Verificar versión siguiente cursor.execute(""" SELECT MAX(Version) FROM Q_QuotationHead WHERE QuotationNum = ? """, quotation_num) version_result = cursor.fetchone() next_version = 1 if not version_result[0] else version_result[0] + 1 # ✅ NUEVO: Obtener el nombre legible de la moneda currency_code = data.get('CurrencyCode') currency_name = currency_code # Por defecto, usar el código if currency_code: cursor.execute(""" SELECT FrontES FROM Q_Currency WHERE CurrencyCode = ? """, currency_code) currency_result = cursor.fetchone() if currency_result and currency_result[0]: currency_name = currency_result[0] print(f"💱 Moneda: {currency_code} -> {currency_name}") # ✅ DEBUG print(f"🔍 DEBUG - VendedorNombre: '{row[11]}'") print(f"🔍 DEBUG - VendedorEmail: '{row[12]}'") print(f"🔍 DEBUG - VendedorTelefono: '{row[13]}'") # Limpiar espacios y validar vendedor_nombre = row[11].strip() if row[11] else None vendedor_email = row[12].strip() if row[12] else None vendedor_telefono = row[13].strip() if row[13] else None template_data = { # Identificadores 'quotation_num': quotation_num, 'version': next_version, 'costing_id': costing_id, 'caso_costeo': data.get('CaseCost'), # Datos de terminos y condiciones 'RunTimeNumber': data.get('RunTimeNumber'), 'RunTimeType': data.get('RunTimeType'), 'TechnicalTermsAndConditions': data.get('TechnicalTermsAndConditions'), # Datos del cliente 'cliente_nombre': row[2], 'tipo_contacto': row[3], 'oportunidad_crm': row[4], 'cliente_email': row[5], 'cliente_telefono': row[6], 'cliente_ciudad': row[7], 'cliente_estado': row[8], 'cliente_pais': row[9], 'cliente_direccion': row[10], # ✅ Datos del vendedor (desde Q_OpportunityCRM.UserID → Profiles) 'vendedor_asignado': vendedor_nombre or 'No asignado', 'vendedor_email': vendedor_email or 'ventas@igsa.com', 'vendedor_telefono': vendedor_telefono or 'N/A', # Datos financieros 'precio_lista': data.get('SalePrice', 0), 'descuento_porcentaje': data.get('DiscountPercent', 0), 'precio_oferta': data.get('Amount', 0), 'impuesto_codigo': taxCode, 'impuedesto_porcentaje': tax_percent, 'precio_total': data.get('TotalAmount', 0), # Usar el nombre legible de la moneda 'moneda': currency_name, # "Peso Mexicano" en lugar de "MX" 'moneda_codigo': currency_code, # Mantener el código para referencias 'factor_sobrecosto': data.get('OvercostFactor', 1), # Líneas de cotización 'lineas': data.get('QuotationLines', []), 'total_lineas': len(data.get('QuotationLines', [])), # Datos del sistema 'fecha_actual': datetime.now().strftime('%d/%m/%Y'), 'hora_actual': datetime.now().strftime('%H:%M'), 'año_actual': datetime.now().year, 'empresa': 'IGSA', 'empresa_completa': 'Integradora de Servicios Avanzados', # Proyecto 'proyecto_nombre': data.get('proyecto_nombre', 'Por definir'), 'proyecto_requerimientos': data.get('proyecto_requerimientos', 'Ninguno'), 'caseSelected': data.get('caseSelected'), # Caso 1 'case1AnticipoPercent': data.get('case1AnticipoPercent'), 'case1FiniquitoPercent': data.get('case1FiniquitoPercent'), # Caso 2 'case2AnticipoPercent': data.get('case2AnticipoPercent'), 'case2FrequencyType': data.get('case2FrequencyType'), 'case2Periodicidad': data.get('case2Periodicidad'), 'case2Cantidad': data.get('case2Cantidad'), 'case2CantidadPeriodicidad': data.get('case2CantidadPeriodicidad'), #Caso 3 'case3FrequencyType': data.get('case3FrequencyType'), 'case3Periodicidad': data.get('case3Periodicidad'), 'case3Cantidad': data.get('case3Cantidad'), 'case3CantidadPeriodicidad': data.get('case3CantidadPeriodicidad'), # Caso 4 'case4DiasFinanciamiento': data.get('case4DiasFinanciamiento'), # Caso 5 'case5PlazoCreditoFlag': data.get('case5PlazoCreditoFlag') } print(f"✅ Vendedor final en template: {template_data['vendedor_asignado']}") html = render_template('Emails/Ventas/Cotiz/PreviewCotizacion.html', **template_data) return html except Exception as e: print(f"❌ Error al generar vista previa: {e}") import traceback traceback.print_exc() return None @staticmethod def create_quotation(data: dict) -> dict: """ Crea una nueva cotización en Q_QuotationHead y Q_QuotationDetail. Envía notificación por correo al departamento de ingeniería. """ try: with get_connection() as conn: with conn.cursor() as cursor: costing_id = data.get('CostingID') # 1. Obtener QuotationNum del CostingHead cursor.execute(""" SELECT CostingNum FROM Q_CostingHead WHERE CostingID = ? """, costing_id) costing_result = cursor.fetchone() if not costing_result: return {"success": False, "error": "CostingID no encontrado"} quotation_num = costing_result[0] # 2. Verificar si ya existe una cotización con este número cursor.execute(""" SELECT MAX(Version) FROM Q_QuotationHead WHERE QuotationNum = ? """, quotation_num) version_result = cursor.fetchone() version = 1 if not version_result[0] else version_result[0] + 1 quotation_id = f"{quotation_num}-{version}" print(f"Creando cotización: {quotation_id}") # 3. Insertar en Q_QuotationHead insert_head_query = """ INSERT INTO Q_QuotationHead ( QuotationNum, Version, CaseCost, SalePrice, DiscountPercent, OvercostFactor, Amount, TaxCode, TotalAmount, CurrencyCode, Active, caseSelected, case1AnticipoPercent, case1FiniquitoPercent, case2AnticipoPercent, case2FrequencyType, case2Periodicidad, case2Cantidad, case2CantidadPeriodicidad, case3FrequencyType, case3Periodicidad, case3Cantidad, case3CantidadPeriodicidad, case4DiasFinanciamiento, case5PlazoCreditoFlag ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """ cursor.execute(insert_head_query, ( quotation_num, version, data.get('CaseCost'), data.get('SalePrice'), data.get('DiscountPercent', 0), data.get('OvercostFactor', 1), data.get('Amount'), data.get('TaxCode'), data.get('TotalAmount'), data.get('CurrencyCode'), # ✅ Campos de casos de pago - acceso directo data.get('caseSelected'), data.get('case1AnticipoPercent'), data.get('case1FiniquitoPercent'), data.get('case2AnticipoPercent'), data.get('case2FrequencyType'), data.get('case2Periodicidad'), data.get('case2Cantidad'), data.get('case2CantidadPeriodicidad'), data.get('case3FrequencyType'), data.get('case3Periodicidad'), data.get('case3Cantidad'), data.get('case3CantidadPeriodicidad'), data.get('case4DiasFinanciamiento'), data.get('case5PlazoCreditoFlag') )) # 4. Insertar líneas en Q_QuotationDetail quotation_lines = data.get('QuotationLines', []) insert_detail_query = """ INSERT INTO Q_QuotationDetail ( QuotationID, QuotationLine, CostingLineID, PartNum, PartDescription, Qty, UOMCode, UnitPrice, Amount ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """ for line in quotation_lines: cursor.execute(insert_detail_query, ( quotation_id, line.get('QuotationLine'), line.get('CostingLineID'), line.get('PartNum'), line.get('PartDescription'), line.get('Qty'), line.get('UOMCode'), line.get('UnitPrice'), line.get('Amount') )) conn.commit() print(f"Cotización {quotation_id} creada exitosamente en BD") return { "success": True, "message": "Cotización creada exitosamente", "QuotationID": quotation_id, "Version": version } except Exception as e: print(f"Error al crear cotización: {e}") return {"success": False, "error": str(e)} @staticmethod def get_seller_info(costing_id: str) -> dict: """Obtiene información del vendedor""" try: from Consultas_SQL.conexion import get_connection with get_connection() as conn: with conn.cursor() as cursor: cursor.execute(""" SELECT TOP 1 TRIM(CONCAT( ISNULL(Seller.FirstName, ''), ' ', ISNULL(Seller.MiddleName, ''), ' ', ISNULL(Seller.LastName, ''), ' ', ISNULL(Seller.SecondLastName, '') )) AS VendedorNombre, ISNULL(Seller.Email, '') AS VendedorEmail, ISNULL(Seller.ContactPhone, '') AS VendedorTelefono FROM Q_CostingHead INNER JOIN Q_OpportunityCRM ON Q_CostingHead.CRM_OpportunityID = Q_OpportunityCRM.CRM_OpportunityID LEFT JOIN Profiles AS Seller ON Q_OpportunityCRM.UserID = Seller.UserID WHERE Q_CostingHead.CostingID = ? """, costing_id) row = cursor.fetchone() if row: return { 'nombre': row[0].strip() if row[0] else None, 'email': row[1].strip() if row[1] else None, 'telefono': row[2].strip() if row[2] else None } except Exception as e: print(f"⚠️ Error al obtener info del vendedor: {e}") return {'nombre': None, 'email': None, 'telefono': None} def json_serializer(obj): """ Convierte objetos Decimal y datetime a formatos compatibles con JSON. """ if isinstance(obj, Decimal): return float(obj) if isinstance(obj, datetime): return obj.isoformat() raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable") # --- FUNCIÓN PRINCIPAL --- def get_TermsType_JSON() -> Optional[List[Dict]]: """ Obtiene todos los registros activos del catálogo Q_TermsType y los devuelve como una lista de diccionarios serializables a JSON. Returns: List[Dict]: Lista de registros de términos y condiciones. None: Si no se encuentran registros o ocurre un error. """ query = """ SELECT TermsTypeID, FrontES, FrontEN, Active, createdAt, CreatedBy, UpdatedAt, UpdatedBy FROM Q_TermsType WHERE Active = 1 ORDER BY TermsTypeID ASC; """ conn = get_connection() if not conn: logger.error("❌ No se pudo establecer conexión con la base de datos.") return None try: cursor = conn.cursor() cursor.execute(query) columns = [col[0] for col in cursor.description] rows = cursor.fetchall() if not rows: logger.warning("⚠️ No se encontraron registros activos en Q_TermsType.") return None # Convertir filas a lista de diccionarios results = [dict(zip(columns, row)) for row in rows] # Convertir tipos no serializables (datetime, Decimal) for r in results: for key, value in r.items(): if isinstance(value, datetime): r[key] = value.isoformat() elif isinstance(value, Decimal): r[key] = float(value) elif isinstance(value, bool): r[key] = bool(value) return results except pyodbc.Error as e: logger.error(f"Error SQL al obtener términos: {str(e)}") raise except Exception as e: logger.error(f"Error inesperado en get_TermsType_JSON: {str(e)}") raise finally: conn.close() logger.info("🔒 Conexión a la base de datos cerrada.") def get_Quote_JSON_BussinessCentral(cotizacion_id: str) -> Optional[Dict]: """ Busca el encabezado y el detalle de una cotización usando su ID primario, y los devuelve en un diccionario con tipos de datos serializables a JSON. Args: cotizacion_id (str): El ID primario de la cotización (ej. "1001-1"). Returns: Un diccionario con la estructura de QuotationHead y QuotationDetail, con tipos de datos serializables a JSON, o None si no se encuentra. """ if not cotizacion_id: logger.error("Se proporcionó un cotizacion_id vacío.") return None query_head = """ SELECT QuotationID, QuotationDate, QuotationNum, Version, SalePrice, DiscountPercent, Amount, TaxCode, TotalAmount, CurrencyCode, DeliveryTime, TermsAndConditions FROM Q_QuotationHead WHERE QuotationID = ?; """ query_detail = """ SELECT QuotationLineID, QuotationID, QuotationLine, CostingLineID, PartNum, PartDescription, Qty, UOMCode, UnitPrice, Amount FROM Q_QuotationDetail WHERE QuotationID = ? ORDER BY QuotationLine ASC; """ conn = get_connection() if not conn: raise ConnectionError("No se pudo establecer conexión con la base de datos.") try: cursor = conn.cursor() # --- Obtener el encabezado (QuotationHead) --- cursor.execute(query_head, cotizacion_id) columns_head = [column[0] for column in cursor.description] row_head = cursor.fetchone() if not row_head: logger.warning(f"No se encontró encabezado de cotización para el ID: {cotizacion_id}") return None quotation_head_data = dict(zip(columns_head, row_head)) # --- Obtener el detalle (QuotationDetail) --- cursor.execute(query_detail, cotizacion_id) columns_detail = [column[0] for column in cursor.description] rows_detail = cursor.fetchall() quotation_detail_data = [dict(zip(columns_detail, row)) for row in rows_detail] # 3. Construir el diccionario final final_data = { "QuotationHead": quotation_head_data, "QuotationDetail": quotation_detail_data } # 4. Aquí aplicamos la serialización DESPUÉS de construir el diccionario # y ANTES de devolverlo. # Podríamos hacerlo al vuelo al convertir a JSON, pero hacerlo aquí # asegura que el diccionario devuelto ya tenga tipos serializables. # Serializar QuotationDate (que es un datetime) a string ISO if "QuotationDate" in final_data["QuotationHead"]: date_obj = final_data["QuotationHead"]["QuotationDate"] if isinstance(date_obj, datetime): final_data["QuotationHead"]["QuotationDate"] = date_obj.isoformat() # Serializar otros campos si son Decimal (y no han sido convertidos ya por pyodbc a float) # Esto es más robusto si pyodbc no los convierte automáticamente a float. for key, value in final_data["QuotationHead"].items(): if isinstance(value, Decimal): final_data["QuotationHead"][key] = float(value) for line in final_data["QuotationDetail"]: for key, value in line.items(): if isinstance(value, Decimal): line[key] = float(value) return final_data except pyodbc.Error as e: logger.error(f"Error de base de datos al buscar cotización '{cotizacion_id}': {str(e)}") raise except Exception as e: logger.error(f"Error inesperado en get_Quote_JSON_BussinessCentral: {str(e)}") raise finally: if conn: conn.close() logger.info("Conexión a la base de datos cerrada.") # ========================================================================================= # FUNCIÓN 1: BUSCAR COTIZACIÓN COMPLETA # ========================================================================================= def buscar_cotizacion_completa(cotizacion_id: str) -> Optional[Dict]: """ Busca los datos de la oportunidad, las líneas de ingeniería y el estado. Args: cotizacion_id (str): ID de la oportunidad (Ej: 123456-1) Returns: Dict: Datos de la oportunidad y las líneas o None. """ # 1. Dividir el ID para obtener OpportunityNumber y Version try: op_number, version = cotizacion_id.split('-') op_number = op_number.strip() version = int(version.strip()) except ValueError: logger.error(f"Formato de cotizacion_id inválido: {cotizacion_id}") return None # Consulta para obtener los datos de la Oportunidad (Q_OpportunityCRM) # y las Líneas de Cotización (Q_QuotationLines) # Esta consulta necesita ser compleja para unir las 3-4 tablas necesarias. query = """ SELECT T1.CRM_OpportunityNumber, T1.Version, T1.CRM_ContactName, T1.CRM_ContactEmail, T1.Status AS StatusVenta, -- Asumimos una tabla de estado de ingeniería T2.StatusIngenieria, T3.Partnum, T3.Description, T3.Quantity, T3.UnitPriceIngenieria, -- Precio que puso Ingeniería T3.DiscountVenta, -- Descuento que puede modificar Venta T3.FinalPrice, -- Campos financieros ya calculados (si existen) T1.PrecioLista, T1.PrecioVentaIVA, T1.TiempoEntrega, T1.UnidadTiempo, T1.TerminosIngenieria -- Términos de Ingeniería FROM Q_OpportunityCRM T1 LEFT JOIN Q_QuotationHead T2 ON T1.CRM_OpportunityID = T2.CRM_OpportunityID LEFT JOIN Q_QuotationLines T3 ON T2.QuotationID = T3.QuotationID WHERE T1.CRM_OpportunityNumber = ? AND T1.Version = ? AND T1.Active = 1 AND T2.StatusIngenieria IS NOT NULL -- Solo cotizaciones que han pasado por ingeniería ORDER BY T3.LineNum ASC """ conn = get_connection() if not conn: raise ConnectionError("No se pudo establecer conexión con la base de datos") try: with conn: cursor = conn.cursor() cursor.execute(query, (op_number, version)) columns = [column[0] for column in cursor.description] rows = cursor.fetchall() if not rows: logger.warning(f"No se encontraron datos para Cotización ID: {cotizacion_id}") return None # Procesar el resultado: un solo encabezado, múltiples líneas header = {} lines = [] for row in rows: data = dict(zip(columns, row)) if not header: # Rellenar el encabezado solo una vez header = {k: data[k] for k in data if k not in ['Partnum', 'Description', 'Quantity', 'UnitPriceIngenieria', 'DiscountVenta', 'FinalPrice']} header['cotizacion_id'] = cotizacion_id header['lineas'] = lines # Agregar la línea a la lista lines.append({ 'Partnum': data['Partnum'], 'Description': data['Description'], 'Quantity': data['Quantity'], 'UnitPriceIngenieria': data['UnitPriceIngenieria'], 'DiscountVenta': data['DiscountVenta'], 'FinalPrice': data['FinalPrice'] }) return header except pyodbc.Error as e: logger.error(f"Error de base de datos al buscar cotización: {str(e)}") raise except Exception as e: logger.error(f"Error inesperado en buscar_cotizacion_completa: {str(e)}") raise finally: if conn: conn.close() # ========================================================================================= # FUNCIÓN 2: ACTUALIZAR LÍNEAS DE COTIZACIÓN # ========================================================================================= def actualizar_lineas_cotizacion(cotizacion_id: str, lineas: List[Dict], user_id: int) -> Dict: """Actualiza campos editables (ej. Descuento) en Q_QuotationLines.""" # Simulación de la función: Asume que las líneas se actualizan en un lote if not lineas: return {'success': True, 'message': 'No hay líneas para actualizar'} try: # Obtener nombre de usuario para auditoría from .CotizEspSolicitudSQL import obtener_nombre_usuario user_name = obtener_nombre_usuario(user_id) # Lógica de actualización de lotes... logger.info(f"Actualizando {len(lineas)} líneas para {cotizacion_id} por {user_name}") # ... (código SQL para UPDATE en lote) return {'success': True, 'message': 'Líneas actualizadas'} except Exception as e: logger.error(f"Error al actualizar líneas de cotización: {str(e)}") return {'success': False, 'message': str(e)} # ========================================================================================= # FUNCIÓN 3: ACTUALIZAR CÁLCULO FINANCIERO # ========================================================================================= def actualizar_calculo_financiero(cotizacion_id: str, financieros: Dict, user_id: int) -> Dict: """Actualiza la sección financiera en Q_OpportunityCRM (o Q_QuotationHead).""" # Asume que estos campos se guardan en Q_OpportunityCRM query = """ UPDATE Q_OpportunityCRM SET DiscountPorcentaje = ?, IncludesIVA = ?, PrecioVentaIVA = ?, UpdatedBy = ?, UpdatedAt = GETDATE() WHERE CRM_OpportunityID = ? -- Asumiendo que el ID de la oportunidad es el cotizacion_id """ conn = get_connection() if not conn: return {'success': False, 'message': 'Error de conexión a BD'} try: from .CotizEspSolicitudSQL import obtener_nombre_usuario user_name = obtener_nombre_usuario(user_id) with conn: cursor = conn.cursor() cursor.execute(query, ( financieros['descuento'], financieros['incluye_iva'], financieros['precio_final_con_iva'], user_name, cotizacion_id )) conn.commit() if cursor.rowcount == 0: return {'success': False, 'message': 'Oportunidad no encontrada para actualización financiera'} logger.info(f"Datos financieros actualizados para {cotizacion_id}") return {'success': True} except pyodbc.Error as e: logger.error(f"Error de BD al actualizar financieros: {str(e)}") return {'success': False, 'message': 'Error de base de datos'} finally: if conn: conn.close() # ========================================================================================= # FUNCIÓN 4: ACTUALIZAR TIEMPO Y CONDICIONES # ========================================================================================= def actualizar_tiempo_y_condiciones(cotizacion_id: str, tiempo_condiciones: Dict, user_id: int) -> Dict: """Actualiza el tiempo de entrega y los términos técnicos en Q_OpportunityCRM.""" query = """ UPDATE Q_OpportunityCRM SET TiempoEntrega = ?, UnidadTiempo = ?, TerminosVenta = ?, -- Nuevo campo para términos de Venta UpdatedBy = ?, UpdatedAt = GETDATE() WHERE CRM_OpportunityID = ? """ conn = get_connection() if not conn: return {'success': False, 'message': 'Error de conexión a BD'} try: from .CotizEspSolicitudSQL import obtener_nombre_usuario user_name = obtener_nombre_usuario(user_id) with conn: cursor = conn.cursor() cursor.execute(query, ( tiempo_condiciones['tiempo_ejecucion'], tiempo_condiciones['unidad_tiempo'], tiempo_condiciones['terminos_venta'], user_name, cotizacion_id )) conn.commit() if cursor.rowcount == 0: return {'success': False, 'message': 'Oportunidad no encontrada para actualización de condiciones'} logger.info(f"Tiempo y condiciones actualizados para {cotizacion_id}") return {'success': True} except pyodbc.Error as e: logger.error(f"Error de BD al actualizar tiempo y condiciones: {str(e)}") return {'success': False, 'message': 'Error de base de datos'} finally: if conn: conn.close() # ========================================================================================= # FUNCIÓN 5: FINALIZAR COTIZACIÓN Y ENVIAR # ========================================================================================= def finalizar_cotizacion_y_enviar(cotizacion_id: str, user_id: int, data: Dict) -> Dict: """ Marca el estado final de la cotización, genera el PDF y registra el evento. Nota: La generación de PDF con Puppeteer/Node.js debería llamarse aquí. """ try: # 1. Actualizar estado de Oportunidad a 'ENVIADA_A_CLIENTE' # Simulación de la actualización de estado... # 2. Generación del PDF (integración con puppeteer_pdf/generate_pdf.js) # Aquí se ejecutaría el script de Node.js con los datos de la cotización pdf_url = generar_pdf_final_cotizacion(cotizacion_id, data) # 3. Registrar el evento de envío en la tabla de auditoría # Simulación de registro... logger.info(f"Cotización {cotizacion_id} marcada como ENVIADA_A_CLIENTE y PDF generado: {pdf_url}") return {'success': True, 'pdf_url': pdf_url} except Exception as e: logger.error(f"Error al finalizar cotización y generar PDF: {str(e)}") return {'success': False, 'message': str(e)} def generar_pdf_final_cotizacion(cotizacion_id: str, data: Dict) -> str: """Simula la generación de PDF final (debería llamar a Node.js/Puppeteer).""" # Lógica que llama a: # 1. Crear HTML final con datos (incluye financieras) # 2. Ejecutar comando Node.js/Puppeteer para renderizar y guardar en la ruta estática # 3. Registrar PDF en DocsManagement return f"https://sycelephant.com/static/pdfs/cotiz/{cotizacion_id}_final.pdf"