from openai import OpenAI import openai import json import time import re import pinecone from datetime import datetime 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'] 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/cobra/DESARROLLO/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) 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): # Conexión a la base de datos connection = conectar_db() if connection is None: return None try: cursor = connection.cursor(dictionary=True) # Consultar las instrucciones del asistente cursor.execute( """ SELECT instruccion_saludo_despedida, instruccion_personalidad, instruccion FROM asistentes_instrucciones WHERE id_asistente = %s """, (id_asistente,), ) instrucciones = cursor.fetchone() # Consultar las funciones del asistente cursor.execute( """ SELECT funcion FROM asistentes_funciones WHERE id_asistente = %s """, (id_asistente,), ) funciones = cursor.fetchall() cursor.close() connection.close() hora_actual = datetime.now().strftime("%Y-%m-%d %H:%M:%S") # Verificar si se encontraron instrucciones y funciones if instrucciones: saludo_despedida = instrucciones.get("instruccion_saludo_despedida", "") personalidad = instrucciones.get("instruccion_personalidad", "") instruccion = instrucciones.get("instruccion", "") # Concatenar las instrucciones con los títulos instrucciones_concatenadas = ( f"**Hora actual:** {hora_actual}\n\n" f"**Instrucciones de Saludo y Despedida**\n{saludo_despedida}\n\n" f"**Instrucciones de Personalidad**\n{personalidad}\n\n" f"**Instrucciones Generales**\n{instruccion}" ) # Concatenar las funciones, si existen if funciones: funciones_texto = "\n".join( [funcion["funcion"] for funcion in funciones] ) instrucciones_concatenadas += f"\n\n**Funciones**\n{funciones_texto}" else: instrucciones_concatenadas += "\n\n**Funciones**\nNo se encontraron funciones para el asistente dado." return ( instrucciones_concatenadas.strip() ) # Eliminar posibles espacios en blanco adicionales else: return "No se encontraron instrucciones para el asistente dado." except mysql.connector.Error as err: print(f"Error: {err}") return None # Definir las variables globales al inicio asistente = None thread_id = None pregunta = None archivo = None idcliente = None tokens_pregunta = None json_data = None chat_history = None instrucciones = None source = None client = None id_asistente = None voy_sin_hilo = None api_key = None human_on = None volume_up = 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 asistente, thread_id, pregunta, archivo, idcliente global tokens_pregunta, json_data, chat_history, instrucciones global source, client, id_asistente, voy_sin_hilo, api_key global human_on, volume_up # Asignamos los valores a las variables globales asistente = asistente_ thread_id = thread_id_ pregunta = pregunta_ archivo = archivo_ idcliente = idcliente_ source = source_ id_asistente = id_asistente_ voy_sin_hilo = voy_sin_hilo_ human_on = human_on_ volume_up = volume_up_ # 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_key) 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_COMFENALCO": client = OpenAI(api_key=config_manager.OPENAI_API_KEY) # 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["voy_sin_hilo"] = voy_sin_hilo json_data["Exito"] = "Exito" verificar(idcliente, thread_id, voy_sin_hilo, instrucciones, pregunta, id_asistente, volume_up, human_on) 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 # 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 agregar_mensaje(thread_id, content): try: client.beta.threads.messages.create( thread_id=thread_id, role="user", content=content ) except Exception as e: print("Error al agregar mensaje al hilo:", e) def consultar_pqrsf_tipos(): resultado = { "categorias": [], "tipos": [] } try: # Conectar a la base de datos connection = conectar_db() if connection is not None and connection.is_connected(): cursor = connection.cursor() # Consulta para obtener los registros de la tabla `pqrsf_categoria` sql_categoria = "SELECT id, categoria FROM pqrsf_categoria" cursor.execute(sql_categoria) categorias = cursor.fetchall() # Procesar los resultados de `pqrsf_categoria` for categoria in categorias: resultado["categorias"].append({ "id": categoria[0], "categoria": categoria[1] }) # Consulta para obtener los registros de la tabla `pqrsf_tipos` sql_tipos = "SELECT id, categoria, tipo_pqrsf, area_encargada, respuesta_ia, usuario_encargado FROM pqrsf_tipos" cursor.execute(sql_tipos) tipos = cursor.fetchall() # Procesar los resultados de `pqrsf_tipos` for tipo in tipos: resultado["tipos"].append({ "id": tipo[0], "categoria": tipo[1], "tipo_pqrsf": tipo[2], "area_encargada": tipo[3], "respuesta_ia": tipo[4], "usuario_encargado": tipo[5] }) except Exception as e: print("Ocurrió un error en la consulta de los datos:", e) finally: # Cerrar la conexión y el cursor si están abiertos if connection and connection.is_connected(): cursor.close() connection.close() # Retornar la estructura en formato JSON return json.dumps(resultado, indent=4) def registrar_pqr(id_categoria, id_tipo, contenido_solicitud, encargado, respuesta_ia, idcliente): # Verificar si idcliente es la palabra 'no' 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) # 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") # 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: # Construir el payload asegurando que todo es string payload = { "ciudadano": str(nombre_completo), "correos": str(correo), "categoria": "Petición", "tipo_pqrsf": "Horario de atención en taquilla", "estado": "Finalizado", "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) return f"Registro insertado exitosamente con num_radicado: {num_radicado}" except Exception as e: return f"Ocurrió un error al registrar el PQR: {e}" finally: # Cerrar la conexión y el cursor si están abiertos if connection and connection.is_connected(): cursor.close() 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): 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(documento_pasado, idcliente): # Si idcliente es 'no', retornar un mensaje de error if idcliente == 'no': return "El usuario no esta logueado. No se puede proporcionar información de predios." try: # Conectar a la base de datos connection = conectar_db() if connection is not None and connection.is_connected(): cursor = connection.cursor(dictionary=True) # Usar un cursor para devolver resultados como diccionarios # Primero, obtener el documento del cliente con el idcliente sql_query_cliente = """ SELECT `num_doc_ident` FROM `ciudadanos` WHERE `id` = %s; """ # Ejecutar la consulta para obtener el documento cursor.execute(sql_query_cliente, (idcliente,)) cliente = cursor.fetchone() # Validar si el cliente existe y si es el mismo documento if cliente: # Si se encuentra el cliente, obtener el documento documento = cliente['num_doc_ident'] # Validar si el documento es diferente al proporcionado if documento != documento_pasado: return {"status": "error", "message": "El documento proporcionado no coincide con el registrado. No se puede proporcionar información de predios."} else: # Si no se encuentra el cliente, retornar error return {"status": "error", "message": "La persona no se encuentra registrada. No se puede proporcionar información de predios.", "idcliente": idcliente} # Ahora, con el documento, proceder a consultar los predios sql_query_predios = """ SELECT p.id AS predio_id, p.total_registros, p.nro_orden, p.cod_municipio, p.mncpio_nombre, p.npn, p.sector, p.corregimiento, p.barrio, p.mnzver, p.terreno, p.nro_cons, p.ficha, p.circulo, p.matricula, p.direccion, p.numero, p.puntos, p.pisos, p.cocinas, p.habitaciones, p.locales, p.banos, p.tip_const, p.iduso, p.vr_idnuso_nombre, p.tip_nombre, p.edad, p.porc_cons, p.undprd_coeficiente, p.terreno_ha, p.area_constrm2, p.area_constr_total_m2, p.valor_constr, p.avaluo_constr_total, p.avaluo_terreno, p.avaluo_total, p.vigencia, p.cedcatas, p.caracteristica, p.tipo_predio, p.sector_nom FROM predios p INNER JOIN propietarios pr ON p.npn = pr.npn WHERE pr.documento = %s; """ # Ejecutar la consulta para obtener los predios cursor.execute(sql_query_predios, (documento,)) predios = cursor.fetchall() if predios: return {"status": "success", "data": predios} else: return {"status": "error", "message": "La persona no tiene predios registrados."} except Exception as e: return {"status": "error", "message": f"Ocurrió un error al consultar los predios: {e}"} finally: # Cerrar la conexión y el cursor si están abiertos if connection and connection.is_connected(): cursor.close() connection.close() def propietarios_del_predio(npn,idcliente): # Si idcliente es 'no', retornar un mensaje de error if idcliente == 'no': return "El usuario no esta logueado. No se puede proporcionar información de propietarios." try: # Conectar a la base de datos connection = conectar_db() if connection is not None and connection.is_connected(): cursor = connection.cursor(dictionary=True) # Usar un cursor para devolver resultados como diccionarios # Primero, obtener el documento del cliente con el idcliente sql_query_cliente = """ SELECT `num_doc_ident` FROM `ciudadanos` WHERE `id` = %s; """ # Ejecutar la consulta para obtener el documento cursor.execute(sql_query_cliente, (idcliente,)) cliente = cursor.fetchone() # Validar si el cliente existe if cliente: # Si se encuentra el cliente, obtener el documento documento = cliente['num_doc_ident'] else: # Si no se encuentra el cliente, retornar error return {"status": "error", "message": "La persona no se encuentra registrada. No se puede proporcionar información de propietarios."} # Ahora, con el NPN, proceder a consultar los propietarios asociados al predio sql_query_predio = """ SELECT pr.id AS propietario_id, pr.total_registros, pr.nro_orden, pr.cod_municipio, pr.mncpio_nombre, pr.npn, pr.ficha, pr.circulo, pr.matricula, pr.caracteristica, pr.direccion, pr.tip_doc, pr.documento, pr.genero, pr.nombre1, pr.nombre2, pr.apellido1, pr.apellido2, pr.razon_social, pr.nombre_nat_jur, pr.area_terreno_ha, pr.area_total_cons_m2, pr.destinoeconomico, pr.porc_derecho, pr.modo_adquisicion, pr.escritura, pr.fecha_escritura, pr.fecha_registro, pr.calidad_propietario, pr.coeficiente_copropiedad, pr.avaluo, pr.avaluo_terreno, pr.avaluo_construccion, pr.vigencia, pr.sector, pr.tipo_predio, pr.padre_con_inf FROM propietarios pr WHERE pr.npn = %s; """ # Ejecutar la consulta para obtener los propietarios del predio cursor.execute(sql_query_predio, (npn,)) propietarios = cursor.fetchall() if propietarios: # Formatear el resultado como texto texto_resultado = f"Propietarios asociados al predio con NPN {npn}:\n\n" for propietario in propietarios: texto_resultado += ( f"Propietario ID: {propietario['propietario_id']}\n" f"Documento: {propietario['documento']}\n" f"Nombre: {propietario.get('nombre1', '')} {propietario.get('nombre2', '')} " f"{propietario.get('apellido1', '')} {propietario.get('apellido2', '')}\n" f"Razón Social: {propietario.get('razon_social', 'N/A')}\n" f"Avalúo: {propietario['avaluo']}\n" f"Dirección: {propietario['direccion']}\n" f"Tipo Predio: {propietario['tipo_predio']}\n" "------------------------------------------\n" ) return texto_resultado else: return f"No se encontraron propietarios para el NPN proporcionado: {npn}." except Exception as e: return f"Ocurrió un error al consultar los propietarios: {e}" finally: # Cerrar la conexión y el cursor si están abiertos if connection and connection.is_connected(): cursor.close() connection.close() def consulta_radicado(num_radicado): # 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() # Función para enviar una pregunta al asistente y obtener la respuesta def hacer_pregunta(thread_id, pregunta): 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(40): 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) # Aquí vienen los argumentos que se envían a la función arguments = json.loads(arguments_json) json_data['ejecutando_funcion'] = nombre_funcion # Bandera de archivos estado flag_archivo = "no" rtamysql = None consulta = None try: # SECCION DE SEGURIDAD if nombre_funcion == "consultar_pqrsf_tipos": rtamysql = consultar_pqrsf_tipos() json_data["rtamysql"] = rtamysql flag_archivo = "no" 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["rtamysql"] = rtamysql flag_archivo = "no" if nombre_funcion == "obtener_glosario": rtamysql = obtener_glosario() json_data["rtamysql"] = rtamysql flag_archivo = "no" if nombre_funcion == "wiki_bcgs": rtamysql = wiki_bcgs(arguments["palabra_glosario"]) json_data["rtamysql"] = rtamysql flag_archivo = "no" if nombre_funcion == "pqr_horario": rtamysql = pqr_horario( arguments["pregunta"], arguments["respuesta"], idcliente, ) json_data["rtamysql"] = rtamysql flag_archivo = "no" if nombre_funcion == "predios_de_propietario": rtamysql = predios_de_propietario( arguments["documento"], idcliente, ) json_data["rtamysql"] = rtamysql flag_archivo = "no" if nombre_funcion == "propietarios_del_predio": rtamysql = propietarios_del_predio( arguments["npn"], idcliente, ) json_data["rtamysql"] = rtamysql flag_archivo = "no" if nombre_funcion == "consulta_radicado": rtamysql = consulta_radicado( arguments["numero"], ) json_data["rtamysql"] = rtamysql flag_archivo = "no" if nombre_funcion == "consulta_ciudadano": rtamysql = consulta_ciudadano( arguments["numero"], ) json_data["rtamysql"] = rtamysql flag_archivo = "no" if nombre_funcion == "ejecutar_vectorial": rtamysql = ejecutar_vectorial(arguments["nombre"], pregunta) json_data["rtamysql"] = rtamysql flag_archivo = "no" # 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": "sk_fd2d5d0e30815ea9d6283f9383058e626adf3ae03ab105be" } # Imprimir el JSON antes de enviarlo para comparar con PHP print("JSON enviado a ElevenLabs desde Python:") print(payload) # 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"/var/www/core.cic-ware.com/audios_dev/{nombre_archivo}" audio_url = f"https://core.cic-ware.com/audios_dev/{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_db() 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 = "WhatsApp" # 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, voy_sin_hilo, instrucciones, pregunta, id_asistente, volumen, audio_humano): if idcliente == "0": # Lista de posibles respuestas de registro respuestas_registro = [ "Para poder utilizar el chat, primero debe registrarse en nuestra página.", "Lo sentimos, necesita registrarse en nuestra página antes de acceder al chat.", "Por favor, regístrese en nuestra página para poder comenzar a usar el chat.", "Debe completar el registro en nuestra página para acceder al chat.", "Antes de empezar a usar el chat, le solicitamos que se registre en nuestra página.", "Necesita crear una cuenta en nuestra página para poder utilizar el chat.", "Para comenzar a utilizar el chat, es necesario registrarse en nuestro sitio web.", "Regístrese en nuestra página y luego podrá utilizar el chat sin problemas.", "Solo los usuarios registrados pueden usar el chat, por favor regístrese primero.", "Acceda al chat tras completar el registro en nuestra página." ] # Seleccionar una respuesta aleatoria de la matriz respuesta_aleatoria = random.choice(respuestas_registro) # Crear la respuesta en formato JSON json_data["Respuesta"] = respuesta_aleatoria # Imprimir el JSON con la respuesta print(json.dumps(json_data)) sys.exit() # Salir del script (esto es solo para simular el comportamiento) # Función para enviar un mensaje al asistente en caso de que no exista un hilo de trabajo y que el idcliente no sea 0 if thread_id == "nada": thread_id = crear_hilo() if thread_id: agregar_mensaje(thread_id, instrucciones) json_data["Tarea_Creada"] = thread_id if voy_sin_hilo == "si": # Lista de posibles respuestas de primer hilo respuestas_primer_hilo = [ { "saludo": "¡Hola! Bienvenido al portal de atención de Catastro Antioquia. Soy Catia, tu asistente virtual. Estoy aquí para ayudarte a obtener certificados, guiarte en tus trámites catastrales y resolver tus preguntas de forma fácil y segura. ¿En qué puedo asistirte hoy?", "video": "https://dev.ayudacatastro.co/api/crm/asistente/videos/Video_1.mp4" }, { "saludo": "¡Bienvenido a Catastro Antioquia! Soy Catia, tu aliada para facilitar tus trámites. Aquí puedes consultar cómo realizar tus procesos catastrales, obtener certificados y gestionar tus solicitudes sin complicaciones. ¡Hablemos! ¿Cómo puedo ayudarte hoy?", "video": "https://dev.ayudacatastro.co/api/crm/asistente/videos/Video_2.mp4" }, { "saludo": "¡Hola! Soy Catia, tu asistente virtual de Catastro Antioquia. En este portal encontrarás toda la información que necesitas para realizar tus trámites catastrales de manera ágil y segura. ¿Qué trámite necesitas realizar hoy? Estoy aquí para guiarte.", "video": "https://dev.ayudacatastro.co/api/crm/asistente/videos/Video_3.mp4" }, { "saludo": "¡Hola y bienvenido! Soy Catia, tu asistente para trámites en Catastro Antioquia. Desde certificados hasta orientación en trámites catastrales, estoy aquí para facilitar tu experiencia y resolver tus dudas. ¿Qué trámite necesitas hoy? ¡Comencemos juntos!", "video": "https://dev.ayudacatastro.co/api/crm/asistente/videos/Video_4.mp4" }, { "saludo": "¡Hola! Bienvenido al portal de Catastro Antioquia. Soy Catia, tu guía en este proceso. Explora servicios como la expedición de certificados y aprende cómo realizar tus trámites de manera sencilla y eficiente. Estoy aquí para ayudarte. ¿Qué necesitas hoy?", "video": "https://dev.ayudacatastro.co/api/crm/asistente/videos/Video_5.mp4" } ] # Seleccionar una respuesta aleatoria de la lista respuesta_aleatoria = random.choice(respuestas_primer_hilo) # Crear la respuesta en formato JSON json_data["Respuesta"] = respuesta_aleatoria["saludo"] json_data["Video"] = respuesta_aleatoria["video"] try: # Conectar a la base de datos connection = conectar_db() if connection is not None and connection.is_connected(): cursor = connection.cursor(dictionary=True) # Consultar el número de teléfono del cliente sql_telefono = "SELECT telefono FROM ciudadanos WHERE id = %s" cursor.execute(sql_telefono, (idcliente,)) ciudadano = cursor.fetchone() if ciudadano and "telefono" in ciudadano: telefono = ciudadano["telefono"] # Construir la URL para la solicitud remota url_remota = f"https://dev.ayudacatastro.co/api/crm/asistente/intenciones.php?cliente_tel={telefono}&rutaArchivoLocal={respuesta_aleatoria['video']}" json_data["URL_Remota"] = url_remota # Realizar la solicitud GET al archivo PHP import requests try: response = requests.get(url_remota, timeout=10) if response.status_code == 200: json_data["Resultado_Solicitud"] = response.text else: json_data["Error_Solicitud"] = f"Error {response.status_code}: {response.text}" except requests.RequestException as e: json_data["Error_Solicitud"] = f"Error en la solicitud remota: {e}" except Exception as e: json_data["Error_DB"] = f"Error al consultar la base de datos: {e}" finally: # Cerrar la conexión si está abierta if connection and connection.is_connected(): cursor.close() connection.close() # Imprimir el JSON con la respuesta print(json.dumps(json_data)) sys.exit() # Salir del script (esto es solo para simular el comportamiento) if thread_id: respuesta = hacer_pregunta(thread_id, pregunta) 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 print(json.dumps(json_data))