from openai import OpenAI import openai import json import time import re from pinecone import Pinecone from datetime import datetime, timedelta from azure.storage.blob import BlobClient import base64 import ast import conexiones import json import a_env_vars import mysql.connector from mysql.connector import Error import sys import requests import traceback import random import tiktoken from dotenv import load_dotenv import os class ConfigManager: _instance = None REQUIRED_KEYS = ['OPENAI_API_KEY', 'PINECONE_API_KEY', 'OPENAI_API_VALOR_MAS', 'SERVER_ENV', 'XI_API_KEY', 'AUDIO_URL', 'RUTA_AUDIO'] def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._load_config() return cls._instance def _load_config(self): # Primero intenta cargar desde variables de entorno del sistema self.SERVER_ENV = os.getenv('SERVER_ENV') # Si no está definida en las variables de entorno del sistema, busca en .env if not self.SERVER_ENV: env_path = os.getenv('ENV_PATH', '.env') if os.path.exists(env_path): load_dotenv(env_path) self.SERVER_ENV = os.getenv('SERVER_ENV') # Si aún no tenemos SERVER_ENV, usa un valor por defecto if not self.SERVER_ENV: self.SERVER_ENV = '/var/www/dev.catia.catastroantioquia-mas.com/valormas/.env' # Verificar si el archivo .env existe en la ubicación especificada if not os.path.exists(self.SERVER_ENV): raise FileNotFoundError(f"Archivo .env no encontrado en {self.SERVER_ENV}") # Cargar el archivo .env definitivo load_dotenv(self.SERVER_ENV) # Cargar y validar todas las claves requeridas missing_keys = [] for key in self.REQUIRED_KEYS: value = os.getenv(key) if not value and key != 'SERVER_ENV': missing_keys.append(key) setattr(self, key, value) if missing_keys: raise ValueError( f"Claves faltantes en .env: {', '.join(missing_keys)}\n" f"Archivo .env encontrado en: {self.SERVER_ENV}\n" "Por favor verifica que las claves estén configuradas correctamente." ) # Configuraciones adicionales self.OPENAI_API_KEY_VALIDATED = bool(self.OPENAI_API_KEY) self.PINECONE_API_KEY_VALIDATED = bool(self.PINECONE_API_KEY) self.XI_API_KEY_VALIDATED = bool(self.XI_API_KEY) self.AUDIO_URL = self.AUDIO_URL if self.AUDIO_URL.endswith('/') else self.AUDIO_URL + '/' self.RUTA_AUDIO = self.RUTA_AUDIO if self.RUTA_AUDIO.endswith('/') else self.RUTA_AUDIO + '/' try: config = ConfigManager() except Exception as e: print(f"Error crítico de configuración: {str(e)}") raise SystemExit(1) def traer_instrucciones(id_asistente): connection = conectar_db() if connection is None: return None try: cursor = connection.cursor(dictionary=True) # Consulta principal de instrucciones incluyendo restricciones cursor.execute( """ SELECT instruccion_saludo_despedida, instruccion_personalidad, instruccion, restricciones FROM asistentes_instrucciones WHERE id_asistente = %s """, (id_asistente,) ) instrucciones = cursor.fetchone() # Consulta de funciones adicionales cursor.execute( """ SELECT funcion FROM asistentes_funciones WHERE id_asistente = %s """, (id_asistente,) ) funciones = cursor.fetchall() cursor.close() connection.close() if not instrucciones: return "No se encontraron instrucciones para el asistente dado." # Variables bien nombradas y limpias hora_actual = datetime.now().strftime("%Y-%m-%d %H:%M:%S") saludo_despedida = instrucciones.get("instruccion_saludo_despedida", "").strip() personalidad = instrucciones.get("instruccion_personalidad", "").strip() instruccion_general = instrucciones.get("instruccion", "").strip() restricciones = instrucciones.get("restricciones", "").strip() funciones_texto = ( "\n".join(f"- {funcion['funcion'].strip()}" for funcion in funciones) if funciones else "Ninguna función registrada." ) # Construcción del bloque final de instrucciones instrucciones_final = ( f"# 🧠 Instrucciones para el Asistente Catastro Antioquia\n\n" f"**🕒 Hora actual:** {hora_actual}\n\n" f"## 👋 Saludo y Despedida\n{saludo_despedida or 'No definido.'}\n\n" f"## 💡 Personalidad del Asistente\n{personalidad or 'No definido.'}\n\n" f"## 📜 Instrucciones Generales\n{instruccion_general or 'No definido.'}\n\n" f"## 🚫 Restricciones y Límites\n{restricciones or 'No definido.'}\n\n" f"## 🛠️ Funciones Disponibles\n{funciones_texto}" ) return instrucciones_final.strip() except mysql.connector.Error as err: print(f"Error: {err}") return None def main(asistente: str, thread_id: str, pregunta: str, archivo: str, idcliente: int, source: str, id_asistente: str, voy_sin_hilo: str, human_on: str, volume_up: str): global client global api_key # Cargar configuración usando ConfigManager config_manager = ConfigManager() # Obtener las claves necesarias desde el ConfigManager api_valor_mas = config_manager.OPENAI_API_VALOR_MAS api_key = config_manager.OPENAI_API_KEY # Verificar asistentes disponibles con esta API Key try: client_test = OpenAI(api_key=api_valor_mas) assistants = client_test.beta.assistants.list() for a in assistants.data: pass except Exception as e: pass # Obtener instrucciones desde la base de datos instrucciones = traer_instrucciones(id_asistente) # Crear cliente según el origen del asistente if source == "INSTR_IA_GENERAL": client = OpenAI(api_key=config_manager.OPENAI_API_IA_GENERAL) elif source == "INSTR_CNX_SUMMIT": client = OpenAI(api_key=config_manager.OPENAI_API_ON) elif source == "INSTR_OLIVO": client = OpenAI(api_key=config_manager.OPENAI_API_OLIVO) elif source == "INSTR_CRED_MANAGER_CLIENTE": client = OpenAI(api_key=config_manager.OPENAI_API_LUCY) elif source == "INSTR_VALOR_MAS": client = OpenAI(api_key=api_valor_mas) elif source == "INSTR_CREDIR_MASTER": client = OpenAI(api_key=config_manager.OPENAI_API_CREDIR_MASTER) # Crear una variable JSON en memoria json_data = {} json_data["asistente"] = asistente json_data["thread_id"] = thread_id json_data["pregunta"] = pregunta json_data["archivo"] = archivo json_data["idcliente"] = idcliente json_data["source"] = source json_data["id_asistente"] = id_asistente json_data["Exito"] = "Exito" verificar(idcliente, thread_id, instrucciones, pregunta, id_asistente, volume_up, human_on, json_data, archivo, asistente) return json_data # función para conectar a la base de datos en local def conectar_db(): # traerse la variable de conifg del archivo de conexiones.py para conectarse al localhost config = conexiones.config try: connection = mysql.connector.connect(**config) if connection.is_connected(): return connection else: return None except mysql.connector.Error as err: print(f"Error: {err}") return None def conectar_db2(): # traerse la variable de conifg del archivo de conexiones.py para conectarse al localhost config2 = conexiones.config2 try: connection = mysql.connector.connect(**config2) if connection.is_connected(): return connection else: return None except mysql.connector.Error as err: print(f"Error: {err}") return None def conectar_db3(): # traerse la variable de conifg del archivo de conexiones.py para conectarse al localhost config3 = conexiones.config3 try: connection = mysql.connector.connect(**config3) if connection.is_connected(): return connection else: return None except mysql.connector.Error as err: print(f"Error: {err}") return None # Función para subir un archivo al hilo de trabajo def subir_archivo(archivo): file = client.files.create(file=open(archivo, "rb"), purpose="assistants") return file.id, file.bytes # Función para crear un hilo de trabajo def crear_hilo(): try: # Crear el hilo sin 'instructions' ni 'messages' ya que estos parecen ser incorrectos thread_response = client.beta.threads.create() return thread_response.id except Exception as e: print("Error al crear el hilo:", e) return None def consultar_pqrsf_tipos(): resultado = { "tipos": [] } try: connection = conectar_db() if connection is not None and connection.is_connected(): cursor = connection.cursor() # Consulta SQL con JOIN para obtener nombre de categoría sql_tipos = """ SELECT t.id, t.categoria, c.categoria AS nombre_categoria, t.tipo_pqrsf, t.area_encargada, t.respuesta_ia, t.usuario_encargado FROM pqrsf_tipos t INNER JOIN pqrsf_categoria c ON t.categoria = c.id WHERE t.status = 1 """ cursor.execute(sql_tipos) tipos = cursor.fetchall() for tipo in tipos: resultado["tipos"].append({ "id_tipo": tipo[0], # ID único de este registro de tipo PQRSF "id_categoria": tipo[1], # ID de la categoría (clave foránea) "nombre_categoria": tipo[2], # Nombre descriptivo de la categoría "nombre_tipo_pqrsf": tipo[3], # Tipo de PQRSF (ej. Solicitud, Queja) "respuesta_ia": tipo[5], # Respuesta automática sugerida "id_usuario_encargado": tipo[6] # Usuario responsable de atenderlo }) except Exception as e: print("Ocurrió un error en la consulta de los tipos:", e) finally: if connection and connection.is_connected(): cursor.close() connection.close() return json.dumps(resultado, indent=4) def obtener_credenciales_blob(): try: # Solicitar datos (base64) r = requests.get("https://dev.core.ayudacatastro.co:5002/ask/configCuentaServicios/1", timeout=10) r.raise_for_status() base64_data = r.text # Decodificar base64 decoded_data = base64.b64decode(base64_data).decode() # Evaluar el string como un diccionario data = ast.literal_eval(decoded_data) if not data.get("success"): raise ValueError("No se pudo obtener la configuración de blob storage") cuentas = data.get("data", []) if not cuentas or len(cuentas) < 2: raise ValueError("No se encontraron suficientes datos en la configuración") blob_service_url = cuentas[0]["valor"] sas_token = cuentas[1]["valor"] return blob_service_url, sas_token except Exception as e: print(f"❌ Error al obtener credenciales del Blob Storage: {e}") return None, None def descargar_blob_a_local(ruta_blob): try: blob_service_url, sas_token = obtener_credenciales_blob() if not blob_service_url or not sas_token: raise ValueError("No se pudieron obtener las credenciales para el Blob Storage") container, blob_path = ruta_blob.split('/', 1) # Construir la URL completa blob_url = f"{blob_service_url}/{container}/{blob_path}?{sas_token}" # Crear el cliente Blob blob_client = BlobClient.from_blob_url(blob_url) # Descargar al directorio temporal local_path = f"/tmp/{os.path.basename(blob_path)}" with open(local_path, "wb") as f: f.write(blob_client.download_blob().readall()) print(f"✅ Archivo descargado correctamente en: {local_path}") return local_path except Exception as e: print(f"❌ Error en descarga de archivo desde Blob Storage: {e}") return None def registrar_pqr(id_categoria, id_tipo, contenido_solicitud, encargado, respuesta_ia, idcliente, json_data): json_data["idcliente"] = idcliente # Verificar si idcliente es la palabra 'no' if idcliente == 'no': return ("El usuario no está autenticado. Se procederá a brindar orientación e información general disponible.") connection = None cursor = None try: # Conectar a la base de datos connection = conectar_db() if connection is not None and connection.is_connected(): cursor = connection.cursor(dictionary=True) # Validación de envío reciente sql_check = """ SELECT fecha_radicado FROM pqrsf WHERE id_ciudadano = %s ORDER BY fecha_radicado DESC LIMIT 1 """ cursor.execute(sql_check, (idcliente,)) registro_existente = cursor.fetchone() if registro_existente: ultima_fecha_str = registro_existente["fecha_radicado"] ultima_fecha = datetime.strptime(str(ultima_fecha_str), "%Y-%m-%d %H:%M:%S") ahora = datetime.now() if ahora - ultima_fecha < timedelta(minutes=1): return "Ya se registró una solicitud recientemente. Por favor, espera al menos 1 minuto antes de enviar otra." # Definir la fecha actual para `fecha_radicado` fecha_radicado = datetime.now().strftime('%Y-%m-%d %H:%M:%S') # Determinar el valor de medio_respuesta basado en respuesta_ia medio_respuesta = 'Correo' if respuesta_ia else '' # Evaluar si se debe asignar una fecha de respuesta fecha_respuesta = fecha_radicado if respuesta_ia else None # Determinar el valor de status basado en medio_respuesta status = 3 if medio_respuesta else 1 # Definir la consulta SQL para insertar el registro sql_insert = """ INSERT INTO pqrsf ( num_radicado, fecha_radicado, id_categoria_pqrs, id_tipo_pqrs, contenido_solicid, id_ciudadano, id_predio, id_encargado, tipo_encargado, respuesta_encargado, fecha_respuesta, medio_llegada, medio_respuesta, status ) VALUES ( '', %s, %s, %s, %s, %s, 0, %s, 'Asistente', %s, %s, 'Catia', %s, %s ) """ # Ejecutar la consulta de inserción cursor.execute(sql_insert, ( fecha_radicado, id_categoria, id_tipo, contenido_solicitud, idcliente, encargado, respuesta_ia, fecha_respuesta, medio_respuesta, status )) connection.commit() # Obtener el último id insertado last_id = cursor.lastrowid # Generar `num_radicado` en el formato especificado fecha_num = datetime.now().strftime('%Y%m%d') num_radicado = f"{fecha_num}{idcliente}{last_id}" # Realizar el UPDATE para asignar el num_radicado sql_update = "UPDATE pqrsf SET num_radicado = %s WHERE id = %s" cursor.execute(sql_update, (num_radicado, last_id)) connection.commit() # Obtener los nombres de la categoría y tipo sql_categoria = "SELECT categoria FROM pqrsf_categoria WHERE id = %s" cursor.execute(sql_categoria, (id_categoria,)) categoria_nombre = cursor.fetchone().get("categoria") sql_tipo = "SELECT tipo_pqrsf FROM pqrsf_tipos WHERE id = %s" cursor.execute(sql_tipo, (id_tipo,)) tipo_nombre = cursor.fetchone().get("tipo_pqrsf") # Obtener la ruta del archivo desde las tablas relacionadas sql_fuente = """ SELECT fd.ubicacion FROM pqrsf_tipos_fuentes tf JOIN fuentes_datos fd ON tf.id_fuente_datos = fd.id WHERE tf.id_pqrsf_tipo = %s """ cursor.execute(sql_fuente, (id_tipo,)) ruta_blob = cursor.fetchone() if not ruta_blob or not ruta_blob.get("ubicacion"): print("⚠️ No se encontró ruta de archivo para este tipo de PQR.") else: json_data["Ruta_Blob"] = ruta_blob["ubicacion"] # Consultar datos del ciudadano sql_ciudadano = "SELECT nombres, apellidos, correo FROM ciudadanos WHERE id = %s" cursor.execute(sql_ciudadano, (idcliente,)) ciudadano = cursor.fetchone() if status == 3: status_nombre = "Finalizado" else: status_nombre = "Pendiente" if id_categoria != "6": nombres = ciudadano["nombres"] apellidos = ciudadano["apellidos"] nombre_completo = f"{nombres} {apellidos}" correo = ciudadano["correo"] try: # Construir el payload asegurando que todo es string payload = { "ciudadano": str(nombre_completo), "correos": str(correo), "categoria": str(categoria_nombre), "tipo_pqrsf": str(tipo_nombre), "estado": str(status_nombre), "fecha_respuesta": str(fecha_radicado), "num_radicado": str(num_radicado) } response = requests.post( "https://core.ayudacatastro.co:5004/ask/sendEmailPQR", json=payload, headers={"Content-Type": "application/json"}, timeout=10 ) if response.status_code == 200: json_data["Correo"] = "Correo enviado con éxito." else: json_data["Error_Correo"] = response.text except requests.RequestException as e: print("❌ Error en la solicitud HTTP:", e) json_data["Error_Correo"] = str(e) if id_categoria == "6": return respuesta_ia else: return f"Este es el numero del radicado de la solicitud, informaselo al usuario: {num_radicado}" except Exception as e: return f"Ocurrió un error al registrar el PQR: {e}" finally: if cursor: cursor.close() if connection and connection.is_connected(): connection.close() def obtener_glosario(): # Conexión a la base de datos connection = conectar_db() if connection is None: return "Error al conectar con la base de datos." try: cursor = connection.cursor(dictionary=True) # Consulta para obtener los títulos de la tabla query = "SELECT titulo FROM wiki_bcgs ORDER BY titulo" cursor.execute(query) # Crear un diccionario para organizar las páginas por su primer carácter glosario = {} for row in cursor.fetchall(): titulo = row["titulo"] first_char = titulo[0].upper() if first_char not in glosario: glosario[first_char] = [] glosario[first_char].append(titulo) # Construir el glosario en formato de texto glosario_texto = "" for letter in sorted(glosario.keys()): glosario_texto += f"{letter}:\n" for title in sorted(glosario[letter]): glosario_texto += f" - {title}\n" glosario_texto += "\n" return glosario_texto except mysql.connector.Error as err: return f"Error al consultar la base de datos: {err}" finally: if connection.is_connected(): cursor.close() connection.close() def wiki_bcgs(pregunta): # Inicializa el cliente de OpenAI openai.api_key = config.OPENAI_API_KEY # Inicializa el cliente de Pinecone pc = Pinecone(api_key=config.PINECONE_API_KEY) # Nombre del índice en Pinecone index_name = 'wiki' # Conecta con el índice index = pc.Index(index_name) def generar_embedding(texto): response = openai.embeddings.create( input=texto, model='text-embedding-ada-002' ) return response.data[0].embedding # Genera el embedding de la pregunta embedding_pregunta = generar_embedding(pregunta) # Determina si la pregunta es específica para un speaker # (Por ejemplo, buscando un nombre en la pregunta) # Aquí se asume que los nombres están en el formato "Nombre Apellido" patron_nombre = re.compile(r'\b[A-Z][a-z]+ [A-Z][a-z]+\b') coincidencias = patron_nombre.findall(pregunta) if coincidencias: # Si se encuentra un nombre, se asume que la pregunta es específica nombre_speaker = coincidencias[0] filtro = {'Instruction': {'$eq': nombre_speaker}} else: # Si no se encuentra un nombre, se asume que la pregunta es general filtro = {} # Realiza la consulta en el índice de Pinecone resultados = index.query( vector=embedding_pregunta, top_k=50, # Número de resultados a recuperar include_metadata=True, filter=filtro ) # Procesa y devuelve los metadatos de los resultados metadatos = [match['metadata'] for match in resultados['matches']] return metadatos def ejecutar_vectorial(indexe, pregunta): # Inicializa el cliente de OpenAI openai.api_key = config.OPENAI_API_KEY # Inicializa el cliente de Pinecone pc = Pinecone(api_key=config.PINECONE_API_KEY) # Nombre del índice en Pinecone index_name = indexe # Conecta con el índice index = pc.Index(index_name) def generar_embedding(texto): response = openai.embeddings.create( input=texto, model='text-embedding-ada-002' ) return response.data[0].embedding # Genera el embedding de la pregunta embedding_pregunta = generar_embedding(pregunta) # Determina si la pregunta es específica para un speaker # (Por ejemplo, buscando un nombre en la pregunta) # Aquí se asume que los nombres están en el formato "Nombre Apellido" patron_nombre = re.compile(r'\b[A-Z][a-z]+ [A-Z][a-z]+\b') coincidencias = patron_nombre.findall(pregunta) if coincidencias: # Si se encuentra un nombre, se asume que la pregunta es específica nombre_speaker = coincidencias[0] filtro = {'Instruction': {'$eq': nombre_speaker}} else: # Si no se encuentra un nombre, se asume que la pregunta es general filtro = {} # Realiza la consulta en el índice de Pinecone resultados = index.query( vector=embedding_pregunta, top_k=50, # Número de resultados a recuperar include_metadata=True, filter=filtro ) # Procesa y devuelve los metadatos de los resultados metadatos = [match['metadata'] for match in resultados['matches']] return metadatos def pqr_horario(pregunta, respuesta, idcliente, json_data): if idcliente == 'no': return "ID de cliente no proporcionado porque el usuario no está logueado." try: # Conectar a la base de datos connection = conectar_db() if connection is not None and connection.is_connected(): cursor = connection.cursor(dictionary=True) # Fecha actual para `fecha_radicado` y `fecha_respuesta` fecha_radicado = datetime.now().strftime('%Y-%m-%d %H:%M:%S') fecha_respuesta = datetime.now().strftime('%Y-%m-%d') # Solo la fecha # Insertar registro en `pqrsf` sql_insert_pqrsf = """ INSERT INTO pqrsf ( num_radicado, fecha_radicado, id_categoria_pqrs, id_tipo_pqrs, contenido_solicid, id_ciudadano, id_predio, id_encargado, tipo_encargado, respuesta_encargado, fecha_respuesta, medio_llegada, medio_respuesta, status ) VALUES ( '', %s, 1, 13, %s, %s, 0, 1, 'Asistente', %s, %s, 'Catia', 'Correo', 3 ) """ cursor.execute(sql_insert_pqrsf, (fecha_radicado, pregunta, idcliente, respuesta, fecha_respuesta)) connection.commit() # Obtener el último id insertado last_id = cursor.lastrowid # Generar `num_radicado` en el formato especificado fecha_num = datetime.now().strftime('%Y%m%d') num_radicado = f"{fecha_num}{idcliente}{last_id}" # Actualizar `num_radicado` sql_update = "UPDATE pqrsf SET num_radicado = %s WHERE id = %s" cursor.execute(sql_update, (num_radicado, last_id)) connection.commit() # Registrar log: Inicio de respuesta fecha_hora = datetime.now().strftime('%Y-%m-%d %H:%M:%S') log_inicio = """ INSERT INTO pqrsf_logs (id_pqrsf, fecha_hora, evento, descripcion, id_usuario, leida) VALUES (%s, %s, 'Inicio de respuesta', 'Inicio del proceso de respuesta', 1, 0) """ cursor.execute(log_inicio, (last_id, fecha_hora)) connection.commit() # Registrar log: Solución PQR log_solucion = """ INSERT INTO pqrsf_logs (id_pqrsf, fecha_hora, evento, descripcion, id_usuario, leida) VALUES (%s, %s, 'Solución PQR', 'Se ha dado una respuesta al PQR', 1, 0) """ cursor.execute(log_solucion, (last_id, fecha_hora)) connection.commit() # Consultar datos del ciudadano sql_ciudadano = """ SELECT nombres, apellidos, correo FROM ciudadanos WHERE id = %s """ cursor.execute(sql_ciudadano, (idcliente,)) ciudadano = cursor.fetchone() if ciudadano: nombres = ciudadano["nombres"] apellidos = ciudadano["apellidos"] nombre_completo = f"{nombres} {apellidos}" correo = ciudadano["correo"] try: # Definir la URL con los parámetros GET url_php = f"https://dev.ayudacatastro.co/api/crm/pqrsf/mailing_pqrsf.php?correos={correo}&ciudadano={nombre_completo}&categoria=Petición&tipo_pqrsf=Horario de atención en taquilla&estado=3&fecha_respuesta={fecha_respuesta}&num_radicado={num_radicado}" # Realizar la solicitud GET al archivo PHP response = requests.get(url_php) # Verificar la respuesta del servidor if response.status_code == 200: json_data["Whatsapp"] = url_php else: print(f"Error en la llamada al PHP: {response.status_code}, {response.text}") except Exception as e: print("Ocurrió un error al intentar enviar el WhatsApp:", e) return f"Horario de atención en taquilla PETICION De lunes a jueves a 7:30 am a 12:00 m y de 1:15 pm a 5:30 pm y los viernes de 7:30 am a 12:00 m y de 1:30 pm a 4:00 pm." except Exception as e: return f"Ocurrió un error al registrar el PQR y logs: {e}" finally: # Cerrar la conexión y el cursor si están abiertos if connection and connection.is_connected(): cursor.close() connection.close() def predios_de_propietario(idcliente): if idcliente == 'no': return "No se puede entregar la información porque el usuario no está logiado. Debe iniciar sesión para consultar sus predios." try: # Conexión 1: Base de datos principal (ciudadanos) connection = conectar_db() if connection is None or not connection.is_connected(): return {"status": "error", "message": "No se pudo conectar a la base de datos principal."} cursor = connection.cursor(dictionary=True) try: # Limpieza de resultados pendientes por seguridad while cursor.nextset(): cursor.fetchall() # Consulta del documento del ciudadano cursor.execute(""" SELECT `num_doc_ident` FROM `ciudadanos` WHERE `id` = %s; """, (idcliente,)) cliente = cursor.fetchone() if not cliente: return { "status": "error", "message": "La persona no se encuentra registrada. No se puede proporcionar información de predios.", "idcliente": idcliente } documento = cliente['num_doc_ident'] finally: cursor.close() connection.close() # Conexión 2: Base de datos de propiedades connection2 = conectar_db3() if connection2 is None or not connection2.is_connected(): return {"status": "error", "message": "No se pudo conectar a la base de datos de propiedades."} cursor2 = connection2.cursor(dictionary=True) try: while cursor2.nextset(): cursor2.fetchall() # Consulta en registro_1 para obtener FICHA cursor2.execute(""" SELECT `FICHA` FROM `registro_1` WHERE `DOCUMENTO` = %s; """, (documento,)) registro1 = cursor2.fetchone() if not registro1: return { "status": "error", "message": "No se encontraron registros en 'registro_1' para el documento proporcionado." } ficha = registro1['FICHA'] # Limpieza por si quedan resultados antes de siguiente consulta while cursor2.nextset(): cursor2.fetchall() # Consulta de predios en registro_2 usando la ficha cursor2.execute(""" SELECT my_row_id, TOTAL_REGISTROS, NRO_ORDEN, COD_MUNICIPIO, MNCPIO_NOMBRE, NPN, SECTOR, CORREGIMIENTO, BARRIO, MNZVER, TERRENO, NRO_CONS, FICHA, CIRULO, MATRICULA, DIRECCION, NUMERO, PUNTOS, PISOS, COCINAS, HABITACIONES, LOCALES, BAÑOS, TIP_CONST, IDUSO, VR_IDNUSO_NOMBRE, TIP_NOMBRE, EDAD, PORC_CONS, UNDPRD_COEFICIENTE, TERRENO_HA, AREA_CONSTRM2, AREA_CONSTR_TOTAL_M2, VALOR_CONSTR, AVALUO_CONSTR_TOTAL, AVALUO_TERRENO, AVALUO_TOTAL, VIGENCIA, CEDCATAS, CARACTERISTICA, TIPO_PREDIO, SECTOR2 FROM `registro_2` WHERE `FICHA` = %s; """, (ficha,)) predios = cursor2.fetchall() if predios: print(f"✅ Se encontraron {len(predios)} predios asociados a la ficha {ficha}.") return {"status": "success", "data": predios} else: print(f"⚠️ No se encontraron predios asociados a la ficha {ficha}.") return {"status": "error", "message": "No se encontraron predios con la ficha proporcionada."} finally: cursor2.close() connection2.close() except Exception as e: print(f"❌ Error general al consultar los predios: {e}") return {"status": "error", "message": f"Ocurrió un error al consultar los predios: {e}"} def propietarios_del_predio(idcliente): if idcliente == 'no': return "No se puede entregar la información porque el usuario no está logiado. Debe iniciar sesión para consultar sus predios." connection = None connection2 = None cursor = None cursor2 = None try: # Paso 1: Obtener el documento del ciudadano connection = conectar_db() if connection is None or not connection.is_connected(): return {"status": "error", "message": "No se pudo conectar a la base de datos principal."} cursor = connection.cursor(dictionary=True) while cursor.nextset(): cursor.fetchall() cursor.execute(""" SELECT num_doc_ident FROM ciudadanos WHERE id = %s; """, (idcliente,)) cliente = cursor.fetchone() if not cliente: return {"status": "error", "message": "El ciudadano no está registrado."} documento = cliente['num_doc_ident'] cursor.close() connection.close() # Paso 2: Obtener todas las fichas del ciudadano desde registro_1 connection2 = conectar_db3() if connection2 is None or not connection2.is_connected(): return {"status": "error", "message": "No se pudo conectar a la base de datos de propiedades."} cursor2 = connection2.cursor(dictionary=True) while cursor2.nextset(): cursor2.fetchall() cursor2.execute(""" SELECT FICHA FROM registro_1 WHERE DOCUMENTO = %s; """, (documento,)) fichas = cursor2.fetchall() if not fichas: return {"status": "error", "message": "No se encontraron fichas prediales asociadas al ciudadano."} resultado = "" for ficha_row in fichas: ficha = ficha_row["FICHA"] while cursor2.nextset(): cursor2.fetchall() # Consulta unida para obtener propietarios por ficha cursor2.execute(""" SELECT r2.FICHA, r2.MATRICULA, r2.DIRECCION, r2.AREA_CONSTR_TOTAL_M2, r2.AVALUO_TERRENO, r1.NOMBRE1, r1.NOMBRE2, r1.APELLIDO1, r1.APELLIDO2, r1.RAZON_SOCIAL, r1.PORC_DERECHO, r1.MODO_ADQUISICION, r1.FECHA_REGISTRO FROM registro_2 r2 INNER JOIN registro_1 r1 ON r2.NPN = r1.NPN WHERE r2.FICHA = %s; """, (ficha,)) propietarios = cursor2.fetchall() if propietarios: resultado += f"\n📌 Propietarios para ficha predial {ficha}:\n\n" for p in propietarios: nombre_completo = ( f"{p.get('NOMBRE1', '')} {p.get('NOMBRE2', '')} " f"{p.get('APELLIDO1', '')} {p.get('APELLIDO2', '')}".strip() ) nombre_final = nombre_completo if nombre_completo else p.get('RAZON_SOCIAL', 'N/A') resultado += ( f"🧾 Ficha: {p['FICHA']}\n" f"📜 Matrícula: {p['MATRICULA']}\n" f"📍 Dirección: {p['DIRECCION']}\n" f"📐 Área construida total (m²): {p['AREA_CONSTR_TOTAL_M2']}\n" f"🌍 Avalúo terreno: {p['AVALUO_TERRENO']}\n" f"👤 Propietario: {nombre_final}\n" f"🔢 Porcentaje derecho: {p.get('PORC_DERECHO', 'N/A')}\n" f"📝 Modo de adquisición: {p.get('MODO_ADQUISICION', 'N/A')}\n" f"📅 Fecha de registro: {p.get('FECHA_REGISTRO', 'N/A')}\n" "--------------------------------------------\n" ) else: resultado += f"\n⚠️ No se encontraron propietarios para la ficha {ficha}.\n" if resultado.strip() == "": return {"status": "error", "message": "No se encontraron datos de propietarios para ninguna ficha del ciudadano."} return resultado.strip() except Exception as e: return {"status": "error", "message": f"Ocurrió un error al consultar los propietarios: {e}"} finally: if cursor2: cursor2.close() if connection2 and connection2.is_connected(): connection2.close() def consulta_radicado(num_radicado, idcliente): # Verificar si idcliente es la palabra 'no' if idcliente == 'no': return ("El usuario no está autenticado. Se procederá a brindar orientación e información general disponible.") # Conexión a la base de datos connection = conectar_db() if connection is None: return "Error al conectar con la base de datos." try: cursor = connection.cursor(dictionary=True) # Consulta con INNER JOIN para traer nombres en lugar de IDs query = """ SELECT pq.id AS id_radicado, pq.num_radicado, pq.fecha_radicado, cat.categoria AS categoria_pqrs, tipos.tipo_pqrsf AS tipo_pqrs, tipos.area_encargada, pq.contenido_solicid, ciu.nombres AS nombre_ciudadano, ciu.apellidos AS apellido_ciudadano, emp.nombres AS nombre_encargado, emp.apellidos AS apellido_encargado, tipos.respuesta_ia, pq.respuesta_encargado, pq.fecha_respuesta, est.status AS estado FROM pqrsf pq INNER JOIN pqrsf_categoria cat ON pq.id_categoria_pqrs = cat.id INNER JOIN pqrsf_tipos tipos ON pq.id_tipo_pqrs = tipos.id INNER JOIN ciudadanos ciu ON pq.id_ciudadano = ciu.id INNER JOIN empleados emp ON pq.id_encargado = emp.id INNER JOIN pqrsf_status est ON pq.status = est.id WHERE pq.num_radicado = %s """ cursor.execute(query, (num_radicado,)) # Obtener el contenido del radicado result = cursor.fetchone() if result: # Formatear el resultado como texto result_text = ( f"ID del Radicado: {result['id_radicado']}\n" f"Número de Radicado: {result['num_radicado']}\n" f"Fecha de Radicado: {result['fecha_radicado']}\n" f"Categoría PQRs: {result['categoria_pqrs']}\n" f"Tipo PQRs: {result['tipo_pqrs']}\n" f"Área Encargada: {result['area_encargada']}\n" f"Contenido de la Solicitud: {result['contenido_solicid']}\n" f"Ciudadano: {result['nombre_ciudadano']} {result['apellido_ciudadano']}\n" f"Encargado: {result['nombre_encargado']} {result['apellido_encargado']}\n" f"Respuesta IA: {result['respuesta_ia']}\n" f"Respuesta del Encargado: {result['respuesta_encargado']}\n" f"Fecha de Respuesta: {result['fecha_respuesta']}\n" f"Estado: {result['estado']}" ) return result_text else: return "No se encontraron resultados para el número de radicado." except mysql.connector.Error as err: return f"Error al consultar la base de datos: {err}" finally: if connection.is_connected(): cursor.close() connection.close() def consulta_ciudadano(id_ciudadano): # Conexión a la base de datos connection = conectar_db() if connection is None: return "Error al conectar con la base de datos." try: cursor = connection.cursor(dictionary=True) # Consulta para obtener los datos del ciudadano query_ciudadano = """ SELECT id, nombres, apellidos, tipo_doc_ident, num_doc_ident, correo, telefono, direccion, status FROM ciudadanos WHERE id = %s """ cursor.execute(query_ciudadano, (id_ciudadano,)) ciudadano = cursor.fetchone() if not ciudadano: return "No se encontró información para el ciudadano con el ID proporcionado." # Consulta para obtener los registros asociados en la tabla pqrsf query_pqrsf = """ SELECT pq.id AS id_radicado, pq.num_radicado, pq.fecha_radicado, cat.categoria AS categoria_pqrs, tipos.categoria AS tipo_pqrs, pq.contenido_solicid, pq.respuesta_encargado, pq.fecha_respuesta, est.status AS estado FROM pqrsf pq INNER JOIN pqrsf_categoria cat ON pq.id_categoria_pqrs = cat.id INNER JOIN pqrsf_tipos tipos ON pq.id_tipo_pqrs = tipos.id INNER JOIN pqrsf_status est ON pq.status = est.id WHERE pq.id_ciudadano = %s """ cursor.execute(query_pqrsf, (id_ciudadano,)) registros_pqrsf = cursor.fetchall() # Formatear los datos del ciudadano resultado = ( f"Datos del Ciudadano:\n" f"ID: {ciudadano['id']}\n" f"Nombre Completo: {ciudadano['nombres']} {ciudadano['apellidos']}\n" f"Tipo de Documento: {ciudadano['tipo_doc_ident']}\n" f"Número de Documento: {ciudadano['num_doc_ident']}\n" f"Correo: {ciudadano['correo']}\n" f"Teléfono: {ciudadano['telefono']}\n" f"Dirección: {ciudadano['direccion']}\n" f"Estado: {ciudadano['status']}\n\n" ) # Agregar los registros de PQRSF si existen if registros_pqrsf: resultado += "Registros Asociados en PQRSF:\n" for registro in registros_pqrsf: resultado += ( f"\nID del Radicado: {registro['id_radicado']}\n" f"Número de Radicado: {registro['num_radicado']}\n" f"Fecha de Radicado: {registro['fecha_radicado']}\n" f"Categoría PQRs: {registro['categoria_pqrs']}\n" f"Tipo PQRs: {registro['tipo_pqrs']}\n" f"Contenido de la Solicitud: {registro['contenido_solicid']}\n" f"Respuesta del Encargado: {registro['respuesta_encargado']}\n" f"Fecha de Respuesta: {registro['fecha_respuesta']}\n" f"Estado: {registro['estado']}\n" ) else: resultado += "No se encontraron registros asociados en PQRSF." return resultado except mysql.connector.Error as err: return f"Error al consultar la base de datos: {err}" finally: if connection.is_connected(): cursor.close() connection.close() def municipios_actualizacion_catastral(): # Conexión a la base de datos connection = conectar_db() if connection is None: return "Error al conectar con la base de datos." try: cursor = connection.cursor(dictionary=True) # Consulta para obtener los registros de la tabla municipios_operadores query = "SELECT `id`, `Municipio`, `Dirección`, `Name_Operador` FROM `municipios_operadores` WHERE 1" cursor.execute(query) # Crear un diccionario para organizar los municipios por su nombre municipios = {} for row in cursor.fetchall(): municipio = row["Municipio"] municipios[municipio] = { "ID": row["id"], "Dirección": row["Dirección"], "Operador": row["Name_Operador"] } # Construir el texto con la información de los municipios municipios_texto = "" for municipio, info in municipios.items(): municipios_texto += f"Municipio: {municipio}\n" municipios_texto += f" - ID: {info['ID']}\n" municipios_texto += f" - Dirección: {info['Dirección']}\n" municipios_texto += f" - Operador: {info['Operador']}\n\n" return municipios_texto except mysql.connector.Error as err: return f"Error al consultar la base de datos: {err}" finally: if connection.is_connected(): cursor.close() connection.close() def municipios_atencion(): # Conexión a la base de datos connection = conectar_db() if connection is None: return "Error al conectar con la base de datos." try: cursor = connection.cursor(dictionary=True) # Consulta a la tabla municipios_atencion query = "SELECT id, nombre_municipio FROM municipios_atencion" cursor.execute(query) # Crear un diccionario para organizar por nombre del municipio municipios = {} for row in cursor.fetchall(): municipio = row["nombre_municipio"] municipios[municipio] = { "ID": row["id"] } # Construir el texto de salida municipios_texto = "" for municipio, info in municipios.items(): municipios_texto += f"Municipio: {municipio}\n" municipios_texto += f" - ID: {info['ID']}\n\n" return municipios_texto except mysql.connector.Error as err: return f"Error al consultar la base de datos: {err}" finally: if connection.is_connected(): cursor.close() connection.close() # Función para enviar una pregunta al asistente y obtener la respuesta def hacer_pregunta(thread_id, pregunta, json_data, archivo, asistente, instrucciones, idcliente): global tamaño_archivo # si se adjunta un archivo se sube al modelo y se guarda su id y su tamaño para los logs if archivo == "no": file_id = "" tamaño_archivo = 0 else: file_id = "si" if file_id != "": file_id = subir_archivo(archivo) files_id = file_id[0] tamaño_archivo = file_id[1] json_data["tamaño_archivo"] = tamaño_archivo json_data["file_id"] = files_id try: if file_id != "": client.beta.threads.messages.create( thread_id=thread_id, role="user", content=pregunta, attachments=[{ "file_id": files_id, "tools": [{"type": "file_search"}] }] ) else: client.beta.threads.messages.create( thread_id=thread_id, role="user", content=pregunta ) run = client.beta.threads.runs.create( thread_id=thread_id, assistant_id=asistente, instructions=instrucciones ) # Esperar respuesta for _ in range(60): run_status = client.beta.threads.runs.retrieve( thread_id=thread_id, run_id=run.id ) if run_status.status == "completed": break if run_status.status == "requires_action": json_data["requires_action"] = "requires_action" required_action = run_status.required_action tool_calls = required_action.submit_tool_outputs.tool_calls # Aquí recolectamos todas las salidas de las herramientas tools_outputs = [] for tool_call in tool_calls: nombre_funcion = tool_call.function.name arguments_json = tool_call.function.arguments global arguments arguments = json.loads(arguments_json) json_data['ejecutando_funcion'] = nombre_funcion # Bandera de archivos estado rtamysql = None try: # SECCION DE SEGURIDAD if nombre_funcion == "consultar_pqrsf_tipos": rtamysql = consultar_pqrsf_tipos() json_data["rtamysql"] = rtamysql if nombre_funcion == "registrar_pqr": rtamysql = registrar_pqr( arguments["id_categoria"], arguments["id_tipo"], arguments["contenido_solicitud"], arguments["encargado"], arguments["respuesta_ia"], idcliente, json_data, ) json_data["rtamysql"] = rtamysql if nombre_funcion == "obtener_glosario": rtamysql = obtener_glosario() json_data["rtamysql"] = rtamysql if nombre_funcion == "wiki_bcgs": rtamysql = wiki_bcgs(arguments["palabra_glosario"]) json_data["rtamysql"] = rtamysql if nombre_funcion == "pqr_horario": rtamysql = pqr_horario( arguments["pregunta"], arguments["respuesta"], idcliente, json_data, ) json_data["rtamysql"] = rtamysql if nombre_funcion == "predios_de_propietario": rtamysql = predios_de_propietario( idcliente, ) json_data["rtamysql"] = rtamysql if nombre_funcion == "propietarios_del_predio": rtamysql = propietarios_del_predio( idcliente, ) json_data["rtamysql"] = rtamysql if nombre_funcion == "consulta_radicado": rtamysql = consulta_radicado( arguments["numero"], idcliente, ) json_data["rtamysql"] = rtamysql if nombre_funcion == "consulta_ciudadano": rtamysql = consulta_ciudadano( arguments["numero"], ) json_data["rtamysql"] = rtamysql if nombre_funcion == "ejecutar_vectorial": rtamysql = ejecutar_vectorial(arguments["nombre"], pregunta) json_data["rtamysql"] = rtamysql if nombre_funcion == "municipios_actualizacion_catastral": rtamysql = municipios_actualizacion_catastral() json_data["rtamysql"] = rtamysql if nombre_funcion == "municipios_atencion": rtamysql = municipios_atencion() json_data["rtamysql"] = rtamysql # Añadir salida al array de tools_outputs tools_outputs.append({ "tool_call_id": tool_call.id, "output": str(rtamysql) # Convertirlo siempre a cadena }) except Exception as e: print("Error al enviar la respuesta al asistente 3:", e) # Salto de línea para mejorar la legibilidad print print ("Argumentos:", arguments) print ("Json hasta el momento:", json.dumps(json_data)) break # Romper el bucle for inmediatamente # Enviar todas las salidas de herramientas al mismo tiempo if tools_outputs: try: run = client.beta.threads.runs.submit_tool_outputs( thread_id=thread_id, run_id=run.id, tool_outputs=tools_outputs ) except Exception as e: print("Error al enviar la respuesta al asistente 2:", e) print("Tipo de excepción:", type(e).__name__) traceback.print_exc() # Cancelar el run cancelar = client.beta.threads.runs.cancel( thread_id=thread_id, run_id=run.id ) print("Run cancelado:", cancelar.status) return "Parece que hay intermitencia en la red, por favor envia tu pregunta nuevamente." time.sleep(1) json_data["run_status"] = run_status.status.replace("in_progress", "Pensando...") else: print("Tiempo de espera excedido para la ejecución del 'run'.") return None messages = client.beta.threads.messages.list(thread_id=thread_id) for message in messages.data: if message.role == "assistant": respuesta = message.content[0].text.value return respuesta except Exception as e: print("Error al hacer la pregunta:", e) return None def limpiar_texto(texto): # Mapeo de caracteres erróneos a caracteres correctos con acentos reemplazos = { "á": "á", "é": "é", "í": "í", "ó": "ó", "ú": "ú", "ñ": "ñ", "Ñ": "Ñ", "ü": "ü", "É": "É", "Ú": "Ú", "Ó": "Ó", "Ã�": "Í", "À": "À" } # Reemplazar caracteres mal codificados for incorrecto, correcto in reemplazos.items(): texto = texto.replace(incorrecto, correcto) # Eliminar saltos de línea y espacios innecesarios texto = re.sub(r"\s+", " ", texto).strip() return texto def obtener_configuracion_voz_y_generar_audio(id_asistente, texto): texto_limpio = limpiar_texto(texto) try: connection = conectar_db() if connection is None: raise Exception("Error al conectar con la base de datos.") except Exception as e: return {"error": f"Error al conectar con la base de datos: {str(e)}"} try: cursor = connection.cursor(dictionary=True) query_voz = """ SELECT asistentes.id_voz, voces.nombre_voz, voces.id_eleven, asistentes_voces.similarity_boost, asistentes_voces.stability, asistentes_voces.style, asistentes_voces.use_speaker_boost FROM asistentes INNER JOIN asistentes_voces ON asistentes.id_voz = asistentes_voces.id INNER JOIN voces ON asistentes_voces.id_modelo_voz = voces.id WHERE asistentes.id = %s """ cursor.execute(query_voz, (id_asistente,)) asistente = cursor.fetchone() if not asistente: return {"error": "No se encontró un asistente con el ID proporcionado."} # Convertimos los valores para replicar PHP use_speaker_boost = "true" if asistente["use_speaker_boost"] == 1 else "false" # Construimos el JSON manualmente igual que en PHP voice_settings = ( f"\"similarity_boost\": {asistente['similarity_boost']}, " f"\"stability\": {asistente['stability']}, " f"\"style\": {asistente['style']}, " f"\"use_speaker_boost\": {use_speaker_boost}" ) payload = ( "{\n" f" \"model_id\": \"eleven_multilingual_v2\",\n" f" \"text\": \"{texto_limpio}\",\n" f" \"voice_settings\": {{\n {voice_settings}\n }}\n" "}" ) # URL de ElevenLabs - api_url = f"https://api.elevenlabs.io/v1/text-to-speech/{asistente['id_eleven']}?optimize_streaming_latency=0&output_format=mp3_44100_128" headers = { "Content-Type": "application/json", "xi-api-key": config.XI_API_KEY } # Enviar solicitud con `data=` en vez de `json= ` response = requests.post(api_url, headers=headers, data=payload) if response.status_code == 200: numero_aleatorio = random.randint(10000, 99999) nombre_archivo = f"audio_{numero_aleatorio}_{id_asistente}.mp3" ruta_audio = f"{config.RUTA_AUDIO}{nombre_archivo}" # ruta_audio = f"/var/www/catia.ayudacatastro.co/valormas/audios/{nombre_archivo}" audio_url = f"{config.AUDIO_URL}{nombre_archivo}" # audio_url = f"https://catia.ayudacatastro.co/valormas/audios/{nombre_archivo}" with open(ruta_audio, "wb") as audio_file: audio_file.write(response.content) return audio_url else: return { "success": False, "error": f"Error en ElevenLabs: {response.status_code} - {response.text}" } finally: if connection.is_connected(): cursor.close() connection.close() def contar_tokens(texto, modelo="gpt-3.5-turbo"): """Calcula el número de tokens consumidos por un texto.""" encoding = tiktoken.encoding_for_model(modelo) return len(encoding.encode(texto)) def insertar_logs(id_asistente, pregunta, respuesta, idcliente): """Registra la interacción en la base de datos con el cálculo de tokens usando GPT-4 Turbo.""" # Conexión a la base de datos connection = conectar_db2() if connection is None: return "Error al conectar con la base de datos." try: cursor = connection.cursor() # Especificamos el modelo GPT-4 Turbo modelo = "gpt-4-turbo" # Definir la llegada como "WhatsApp" llegada = "WEB" # Contar tokens de entrada y salida tokens_input = contar_tokens(pregunta, modelo) tokens_output = contar_tokens(respuesta, modelo) # Consulta SQL para insertar el log con el nuevo campo `telefono` query = """ INSERT INTO conversaciones (id_asistente, modelo, llegada, fecha_hora, pregunta, respuesta, input, output, usuario, telefono) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ valores = (id_asistente, modelo, llegada, datetime.now(), pregunta, respuesta, tokens_input, tokens_output, idcliente, "0") cursor.execute(query, valores) connection.commit() return "Registro insertado correctamente." except mysql.connector.Error as err: return f"Error al insertar en la base de datos: {err}" finally: if connection.is_connected(): cursor.close() connection.close() def verificar(idcliente, thread_id, instrucciones, pregunta, id_asistente, volumen, audio_humano, json_data, archivo, asistente): if thread_id == "nada": thread_id = crear_hilo() if thread_id: json_data["Tarea_Creada"] = thread_id json_data["thread_id"] = thread_id try: # 🧠 Ejecutar la función manualmente resultado_funcion = consultar_pqrsf_tipos() json_data["funcion_inicial_contexto"] = "se paso el contexto entero" # 📩 Agregar mensaje de contexto como si fuera del sistema client.beta.threads.messages.create( thread_id=thread_id, role="user", content=( "Este es el contexto inicial que debes tener en cuenta antes de responder cualquier pregunta que pueda tener el usuario:\n" f"{resultado_funcion}" ) ) except Exception as e: json_data["error_funcion_inicial"] = f"Error ejecutando manualmente consultar_pqrsf_tipos: {e}" return # Si ya hay hilo, proceder con la pregunta if thread_id: respuesta = hacer_pregunta(thread_id, pregunta, json_data, archivo, asistente, instrucciones, idcliente) if respuesta: if volumen == "si" and audio_humano == "si": ruta_audio = obtener_configuracion_voz_y_generar_audio(id_asistente, respuesta) json_data["ruta_audio"] = ruta_audio insertar_logs(id_asistente, pregunta, respuesta, idcliente) json_data["Respuesta"] = respuesta