# Archivo: quotation_cotiz_services.py import datetime from typing import Optional, List, Dict # Se asume que get_connection se conecta a la base de datos correcta (DesarrolloVPS) from Consultas_SQL.conexion import get_connection class CRM_OpportunityDTO: """DTO para la información de la oportunidad del CRM.""" def __init__(self, crm_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): self.CRM_OpportunityID = crm_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 def to_dict(self) -> dict: return { "CRM_OpportunityID": self.CRM_OpportunityID, "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, } class TaskDTO: """ Objeto de Transferencia de Datos (DTO) para la información de las tareas de cotización. Ahora incluye los datos de la oportunidad del CRM. """ def __init__( self, UserID: int, TaskID: int, SellerUserID: int, FormID: str, AssignedUserID: Optional[int], AssignedDate: Optional[datetime.datetime], Status: str, StatusDate: datetime.datetime, CRM_OpportunityID: str, QuotationID: Optional[str], QuotationDate: Optional[str], DocsID: Optional[int], UpdatedAt: datetime.datetime, UpdatedBy: str, Active: bool, # Nuevo campo para los datos del CRM crm_data: Optional[CRM_OpportunityDTO] ): self.UserID = UserID self.TaskID = TaskID self.SellerUserID = SellerUserID self.FormID = FormID self.AssignedUserID = AssignedUserID self.AssignedDate = AssignedDate self.Status = Status self.StatusDate = StatusDate self.CRM_OpportunityID = CRM_OpportunityID self.QuotationID = QuotationID self.QuotationDate = QuotationDate self.DocsID = DocsID self.UpdatedAt = UpdatedAt self.UpdatedBy = UpdatedBy self.Active = bool(Active) self.crm_data = crm_data def to_dict(self) -> dict: """Convierte el objeto a un diccionario serializable.""" base_dict = { "UserID" : self.UserID, "TaskID": self.TaskID, "SellerUserID": self.SellerUserID, "FormID": self.FormID, "AssignedUserID": self.AssignedUserID, "AssignedDate": self.AssignedDate.isoformat() if self.AssignedDate else None, "Status": self.Status, "StatusDate": self.StatusDate.isoformat() if self.StatusDate else None, "CRM_OpportunityID": self.CRM_OpportunityID, "QuotationID": self.QuotationID, "QuotationDate": self.QuotationDate, "DocsID": self.DocsID, "UpdatedAt": self.UpdatedAt.isoformat() if self.UpdatedAt else None, "UpdatedBy": self.UpdatedBy, "Active": self.Active, } # Agrega el diccionario del CRM si existe if self.crm_data: base_dict["crm_data"] = self.crm_data.to_dict() return base_dict class Quotation_Cotiz_Service: """Clase de servicio para manejar la lógica relacionada con las tareas de cotización.""" @staticmethod def getQuotationTaskById(taskId: int) -> Optional[TaskDTO]: """ Recupera una única tarea de cotización y los datos de la oportunidad del CRM por su TaskID, utilizando un JOIN. """ query = """ SELECT Q_SpQ_QuotationTasks.TaskID, --0 Q_SpQ_QuotationTasks.SellerUserID, --1 Q_SpQ_QuotationTasks.FormID, --2 Q_SpQ_QuotationTasks.AssignedUserID, --3 Q_SpQ_QuotationTasks.AssignedDate, --4 Q_SpQ_QuotationTasks.Status, --5 Q_SpQ_QuotationTasks.StatusDate, --6 Q_SpQ_QuotationTasks.CRM_OpportunityID, --7 Q_SpQ_QuotationTasks.CostingID, --8 Q_SpQ_QuotationTasks.CostingDate, --9 Q_SpQ_QuotationTasks.DocsID, --10 Q_SpQ_QuotationTasks.UpdatedAt, --11 Q_SpQ_QuotationTasks.UpdatedBy, --12 Q_SpQ_QuotationTasks.Active, --13 Q_OpportunityCRM.CRM_OpportunityNumber, --14 Q_OpportunityCRM.Version, --15 Q_OpportunityCRM.UserID, --16 Q_OpportunityCRM.CRM_ContactID, --17 Q_OpportunityCRM.CRM_ContactName, --18 Q_OpportunityCRM.CRM_ContactType, --19 Q_OpportunityCRM.CRM_AssignedSalesperson, --20 Q_OpportunityCRM.CRM_ContactAdress, --21 Q_OpportunityCRM.CRM_ContactColonia, --22 Q_OpportunityCRM.CRM_ContactCity, --23 Q_OpportunityCRM.CRM_ContactNumber, --24 Q_OpportunityCRM.CRM_ContactCountry, --25 Q_OpportunityCRM.CRM_ContactLegalIdentifier, --26 Q_OpportunityCRM.CRM_ContactZip, --27 Q_OpportunityCRM.CRM_ContactState, --28 Q_OpportunityCRM.CRM_ContactEmail --29 FROM Q_SpQ_QuotationTasks INNER JOIN Q_OpportunityCRM ON Q_SpQ_QuotationTasks.CRM_OpportunityID = Q_OpportunityCRM.CRM_OpportunityID WHERE Q_SpQ_QuotationTasks.TaskID = ?; """ try: with get_connection() as conn: with conn.cursor() as cursor: cursor.execute(query, taskId) row = cursor.fetchone() if row: # Mapear los datos de la fila a los DTOs crm_data = CRM_OpportunityDTO( # Mapear el CRM_OpportunityID del resultado del JOIN crm_id=row[7], crm_opportunity_number=row[14], crm_contact_name=row[18], crm_contact_type=row[19], crm_assigned_salesperson=row[20], crm_contact_adress=row[21], crm_contact_colonia=row[22], crm_contact_city=row[23], crm_contact_number=row[24], crm_contact_country=row[25], crm_contact_legal_identifier=row[26], crm_contact_zip=row[27], crm_contact_state=row[28], crm_contact_email=row[29] ) task_obj = TaskDTO( UserID = row[16], TaskID=row[0], SellerUserID=row[1], FormID=row[2], AssignedUserID=row[3], AssignedDate=row[4], Status=row[5], StatusDate=row[6], CRM_OpportunityID=row[7], QuotationID=row[8], QuotationDate=row[9], DocsID=row[10], UpdatedAt=row[11], UpdatedBy=row[12], Active=row[13], crm_data=crm_data ) return task_obj else: return None except Exception as e: print(f"Error en getQuotationTaskById: {e}") return None