from openai import OpenAI import openai import json import time import re from pinecone 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 tempfile import traceback import random import tiktoken import subprocess from dotenv import load_dotenv import os import logging # Cargar variables de entorno desde archivo .env current_dir = os.path.dirname(os.path.abspath(__file__)) env_file_path = os.path.join(current_dir, '.env') # Verificar que el archivo .env existe if not os.path.exists(env_file_path): logging.error(f"Archivo .env no encontrado en: {env_file_path}") raise FileNotFoundError(f"No se encontró el archivo .env en {env_file_path}") # Cargar el archivo .env load_dotenv(env_file_path) # Lista de variables requeridas REQUIRED_KEYS = [ 'OPENAI_API_KEY', 'PINECONE_API_KEY', 'OPENAI_API_VALOR_MAS', 'SERVER_ENV', 'XI_API_KEY', 'AUDIO_URL', 'RUTA_AUDIO', ] # Cargar y validar las variables de entorno requeridas missing_keys = [] config_vars = {} for key in REQUIRED_KEYS: value = os.getenv(key) if not value and key != 'SERVER_ENV': # SERVER_ENV puede estar vacío missing_keys.append(key) config_vars[key] = value if missing_keys: error_msg = ( f"Variables faltantes en .env: {', '.join(missing_keys)}\n" f"Archivo .env: {env_file_path}\n" "Por favor verifica las variables en el archivo .env." ) logging.error(error_msg) raise ValueError(error_msg) # Asignar variables a constantes globales para facilitar el acceso OPENAI_API_KEY = config_vars['OPENAI_API_KEY'] PINECONE_API_KEY = config_vars['PINECONE_API_KEY'] OPENAI_API_VALOR_MAS = config_vars['OPENAI_API_VALOR_MAS'] SERVER_ENV = config_vars['SERVER_ENV'] XI_API_KEY = config_vars['XI_API_KEY'] AUDIO_URL = config_vars['AUDIO_URL'] RUTA_AUDIO = config_vars['RUTA_AUDIO'] # Normalizar rutas terminadas en '/' if AUDIO_URL and not AUDIO_URL.endswith('/'): AUDIO_URL += '/' if RUTA_AUDIO and not RUTA_AUDIO.endswith('/'): RUTA_AUDIO += '/' # Configuraciones adicionales de validación OPENAI_API_KEY_VALIDATED = bool(OPENAI_API_KEY) PINECONE_API_KEY_VALIDATED = bool(PINECONE_API_KEY) XI_API_KEY_VALIDATED = bool(XI_API_KEY) logging.info("Variables de entorno cargadas exitosamente") logging.info(f"AUDIO_URL: {AUDIO_URL}") logging.info(f"RUTA_AUDIO: {RUTA_AUDIO}") 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 def traer_hilo(telefono): # Conexión a la base de datos connection = conectar_db2() if connection is None: return None try: cursor = connection.cursor(dictionary=True) # Consultar los datos del cliente basado en el teléfono cursor.execute( """ SELECT id, telefono, pregunta, hilo FROM conversaciones_ws WHERE telefono = %s """, (telefono,), ) cliente = cursor.fetchone() cursor.close() connection.close() if cliente: return {"voy_sin_hilo": "no" , "hilo": cliente["hilo"]} else: # Vamos a enviar dos cosas, voy_sin_hilo y la palabra nada return {"voy_sin_hilo": "si", "hilo": "nada"} except mysql.connector.Error as err: print(f"Error: {err}") return {"estado": "error", "mensaje": str(err)} def traer_ciudadano(telefono): # Conexión a la base de datos connection = conectar_db() if connection is None: return None try: cursor = connection.cursor(dictionary=True) # Consultar los datos del ciudadano basado en el teléfono cursor.execute( """ SELECT id, nombres, apellidos, tipo_doc_ident, num_doc_ident, correo, id_user_ciudadano, telefono, direccion, status FROM ciudadanos WHERE telefono = %s """, (telefono,), ) ciudadano = cursor.fetchone() cursor.close() connection.close() if ciudadano: return ciudadano["id"] else: return "0" except mysql.connector.Error as err: print(f"Error: {err}") return {"estado": "error", "mensaje": str(err)} def transcribir_ogg_a_texto(url: str, api_key: str) -> str: try: # Descargar el archivo .ogg response = requests.get(url) response.raise_for_status() with tempfile.NamedTemporaryFile(suffix=".ogg", delete=False) as tmp_ogg: tmp_ogg.write(response.content) tmp_ogg.flush() ogg_path = tmp_ogg.name # Crear ruta destino .mp3 mp3_path = ogg_path.replace(".ogg", ".mp3") # Convertir a mp3 usando ffmpeg subprocess.run([ "ffmpeg", "-y", "-i", ogg_path, mp3_path ], check=True) # Transcribir usando Whisper client = OpenAI(api_key=api_key) with open(mp3_path, "rb") as audio_file: transcription = client.audio.transcriptions.create( model="whisper-1", file=audio_file, response_format="text" ) # Limpieza os.remove(ogg_path) os.remove(mp3_path) return transcription except Exception as e: print(f"❌ Error al transcribir audio: {e}") return "Audio recibido, pero no se pudo transcribir correctamente." def main_telefono(asistente_: str, pregunta_: str, instrucciones_: str, id_asistente_: str, volume_up_: str, telefono_: str): global asistente, pregunta, idcliente global instrucciones, id_asistente, volume_up global telefono, thread_id, source, client, json_data global api_key, voy_sin_hilo asistente = asistente_ instrucciones = instrucciones_ id_asistente = id_asistente_ volume_up = volume_up_ telefono = telefono_ source = "INSTR_VALOR_MAS" # Quitar los dos primeros dígitos del número telefono = telefono[2:] # Usar las variables globales en lugar del ConfigManager api_valor_mas = OPENAI_API_VALOR_MAS api_key = OPENAI_API_KEY print("Pregunta inicial", pregunta_) # Validar si es lista de un solo elemento if isinstance(pregunta_, list) and len(pregunta_) == 1: pregunta_ = pregunta_[0] # Verificar si es un archivo de audio .ogg if isinstance(pregunta_, str) and pregunta_.strip().endswith(".ogg"): print(f"🎙️ Detectada URL de audio: {pregunta_}") pregunta = transcribir_ogg_a_texto(pregunta_, api_valor_mas) else: pregunta = pregunta_ # Asegurar que pregunta sea string (evitar error en MySQL) if isinstance(pregunta, list): pregunta = " ".join(map(str, pregunta)) if isinstance(pregunta, dict): pregunta = json.dumps(pregunta) pregunta = str(pregunta).strip() # Evitar enviar valores vacíos o inválidos a la base de datos if not pregunta: pregunta = "Consulta vacía recibida." # Flujo de datos internos idcliente = traer_ciudadano(telefono) instrucciones = traer_instrucciones(id_asistente) hilo = traer_hilo(telefono) voy_sin_hilo = hilo["voy_sin_hilo"] thread_id = hilo["hilo"] # Selección del cliente OpenAI según fuente - USANDO VARIABLES DEL .env if source == "INSTR_IA_GENERAL": client = OpenAI(api_key=os.getenv('OPENAI_API_IA_GENERAL', OPENAI_API_KEY)) elif source == "INSTR_CNX_SUMMIT": client = OpenAI(api_key=os.getenv('OPENAI_API_ON', OPENAI_API_KEY)) elif source == "INSTR_CRED_MANAGER_CLIENTE": client = OpenAI(api_key=os.getenv('OPENAI_API_LUCY', OPENAI_API_KEY)) elif source == "INSTR_VALOR_MAS": client = OpenAI(api_key=api_valor_mas) # Datos para debug o logging json_data = { "asistente": asistente, "pregunta": pregunta, "id_asistente": id_asistente, "volume_up": volume_up, "voy_sin_hilo": voy_sin_hilo, "telefono": telefono, "idcliente": idcliente, "Exito": "Exito" } print(f"Pregunta: {pregunta}") # ✅ Si usás `saludo` en verificar/insertar_logs, asegurate de sanitizarlo allí también # Ejemplo: # if isinstance(saludo, dict): saludo = json.dumps(saludo) verificar(idcliente, thread_id, instrucciones, pregunta, id_asistente, volume_up, voy_sin_hilo, telefono) 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 # 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 '' # 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, NULL, '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, 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") # Enviar correo si es necesario if respuesta_ia: # 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={categoria_nombre}&tipo_pqrsf={tipo_nombre}&estado=3&fecha_respuesta={fecha_radicado}&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: print(f"Correo enviado exitosamente a {correo}: {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 correo:", 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 = OPENAI_API_KEY # Usar variable global # Inicializa el cliente de Pinecone pc = Pinecone(api_key=PINECONE_API_KEY) # Usar variable global # 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 = OPENAI_API_KEY # Usar variable global # Inicializa el cliente de Pinecone pc = Pinecone(api_key=PINECONE_API_KEY) # Usar variable global # 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() def municipios_actualizaciones(): # 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 una lista para almacenar la información de los municipios municipios_lista = [] for row in cursor.fetchall(): municipio = row["Municipio"] direccion = row["Dirección"] operador = row["Name_Operador"] municipios_lista.append(f"Municipio: {municipio}, Dirección: {direccion}, Operador: {operador}") # Unir todos los registros en un solo párrafo municipios_texto = " | ".join(municipios_lista) 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 limpiar_runs_activos(thread_id): """ Cancela cualquier run activo en el thread antes de iniciar uno nuevo. """ try: runs = client.beta.threads.runs.list(thread_id=thread_id) for run in runs.data: if run.status in ["queued", "in_progress", "requires_action"]: client.beta.threads.runs.cancel(thread_id=thread_id, run_id=run.id) time.sleep(1) # Pequeña espera para que la cancelación se procese correctamente except Exception as e: print(f"Error al limpiar runs activos: {e}") def buscar_municipio(nombre_municipio): # 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 con un LIKE query = """ SELECT `id`, `Municipio`, `Dirección`, `Name_Operador` FROM `municipios_operadores` WHERE `Municipio` LIKE %s """ # Usamos % para buscar coincidencias parciales con el nombre del municipio cursor.execute(query, (f"%{nombre_municipio}%",)) # Crear una lista para almacenar la información de los municipios encontrados municipios_lista = [] for row in cursor.fetchall(): municipio = row["Municipio"] direccion = row["Dirección"] operador = row["Name_Operador"] municipios_lista.append(f"Municipio: {municipio}, Dirección: {direccion}, Operador: {operador}") if municipios_lista: # Unir todos los registros en un solo párrafo si se encontraron resultados municipios_texto = " | ".join(municipios_lista) return f"Por favor, dirígete al operador en el municipio correspondiente para resolver tu queja. {municipios_texto}" else: return "No se encontraron municipios que coincidan con el nombre proporcionado." 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 archivo = "no" # 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" try: if file_id != "": client.beta.threads.messages.create( thread_id, role="user", content=pregunta ) else: client.beta.threads.messages.create( 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" if nombre_funcion == "municipios_actualizaciones": rtamysql = municipios_actualizaciones() json_data["rtamysql"] = rtamysql json_data["Respuesta"] = rtamysql limpiar_runs_activos(thread_id) return if nombre_funcion == "buscar_municipio": rtamysql = buscar_municipio(arguments["municipio"]) json_data["rtamysql"] = rtamysql json_data["Respuesta"] = rtamysql limpiar_runs_activos(thread_id) return # 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): # Limpiar el texto antes de enviarlo a ElevenLabs texto_limpio = limpiar_texto(texto) # Conectar a la base de datos connection = conectar_db() if connection is None: return {"error": "Error al conectar con la base de datos."} try: cursor = connection.cursor(dictionary=True) # Consulta SQL para obtener la configuración de voz del asistente 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 use_speaker_boost en booleano use_speaker_boost = True if asistente["use_speaker_boost"] == 1 else False # Construimos los parámetros para la API de ElevenLabs payload = { "text": texto_limpio, # Texto ya limpio antes de enviarlo "voice_settings": { "similarity_boost": asistente["similarity_boost"], "stability": asistente["stability"], "style": asistente["style"], "use_speaker_boost": use_speaker_boost } } # 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" # Encabezados con API Key - USANDO VARIABLE GLOBAL headers = { "Content-Type": "application/json", "xi-api-key": XI_API_KEY } # Enviar solicitud a ElevenLabs response = requests.post(api_url, headers=headers, json=payload) if response.status_code == 200: # Generar un número aleatorio numero_aleatorio = random.randint(10000, 99999) # Crear el nombre del archivo con el número aleatorio y el ID del asistente nombre_archivo = f"audio_{numero_aleatorio}_{id_asistente}.mp3" # Definir la ruta del archivo en el servidor - USANDO VARIABLES GLOBALES ruta_audio = f"{RUTA_AUDIO}{nombre_archivo}" audio_url = f"{AUDIO_URL}{nombre_archivo}" json_data["Parametros"] = payload # Guardamos el archivo de audio en el servidor 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}" } except mysql.connector.Error as err: return {"success": False, "error": f"Error en la base de datos: {err}"} except requests.RequestException as e: return {"success": False, "error": f"Error en la solicitud a ElevenLabs: {e}"} finally: if connection.is_connected(): cursor.close() connection.close() def insertar_mensaje_ws(telefono, pregunta, thread_id): # Conexión a la base de datos connection = conectar_db2() if connection is None: return {"estado": "error", "mensaje": "No se pudo conectar a la base de datos"} try: cursor = connection.cursor() # Insertar el mensaje en la tabla conversaciones_ws cursor.execute( """ INSERT INTO conversaciones_ws (telefono, pregunta, hilo) VALUES (%s, %s, %s) """, (telefono, pregunta, thread_id), ) connection.commit() cursor.close() connection.close() return {"estado": "exito", "mensaje": "Mensaje insertado correctamente"} except mysql.connector.Error as err: print(f"Error: {err}") return {"estado": "error", "mensaje": str(err)} 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, telefono): """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() modelo = "gpt-4-turbo" llegada = "WhatsApp" # 🔒 Asegurar que pregunta y respuesta sean string serializables if isinstance(pregunta, (dict, list)): pregunta = json.dumps(pregunta, ensure_ascii=False) else: pregunta = str(pregunta).strip() if isinstance(respuesta, (dict, list)): respuesta = json.dumps(respuesta, ensure_ascii=False) else: respuesta = str(respuesta).strip() # Validar y limpiar otros campos también por seguridad id_asistente = str(id_asistente) idcliente = str(idcliente) telefono = str(telefono) # Contar tokens tokens_input = contar_tokens(pregunta, modelo) tokens_output = contar_tokens(respuesta, modelo) 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, telefono ) cursor.execute(query, valores) connection.commit() return "Registro insertado correctamente." except mysql.connector.Error as err: print(f"⚠️ Error al insertar en la base de datos: {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, voy_sin_hilo, telefono): if idcliente == "0": try: connection = conectar_db() if connection is not None and connection.is_connected(): cursor = connection.cursor(dictionary=True) # Consulta para obtener respuestas según el id_asistente query = "SELECT respuesta FROM asistente_sin_registro WHERE id_asistente = %s" cursor.execute(query, (id_asistente,)) resultados = cursor.fetchall() if resultados: # Elegir una respuesta aleatoria del resultado respuesta_aleatoria = random.choice(resultados)["respuesta"] else: # Fallback en caso de no encontrar respuestas respuesta_aleatoria = "¡Hola! 😊 Soy Catia, tu asistente virtual de Catastro Antioquia. Para chatear con nosotros, primero regístrate aquí: https://ayudacatastro.co/#/inicio. ¡Es rápido y sencillo!" # Crear la respuesta en formato JSON json_data["Respuesta"] = respuesta_aleatoria insertar_logs(id_asistente, pregunta, respuesta_aleatoria, idcliente, telefono) print(json.dumps(json_data)) except mysql.connector.Error as e: json_data["Error_DB"] = f"Error al acceder a la base de datos: {e}" print(json.dumps(json_data)) finally: if connection and connection.is_connected(): cursor.close() connection.close() return # 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": try: connection = conectar_db() if connection is not None and connection.is_connected(): cursor = connection.cursor(dictionary=True) # Consulta para obtener saludos y videos desde la tabla asistente_con_registro query = "SELECT respuesta, video_url FROM asistente_con_registro WHERE id_asistente = %s" cursor.execute(query, (id_asistente,)) resultados = cursor.fetchall() if resultados: respuesta_aleatoria = random.choice(resultados) saludo = respuesta_aleatoria["respuesta"] video = respuesta_aleatoria["video_url"] else: # Fallback en caso de no encontrar registros en la tabla 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://ayudacatastro.co/api/crm/asistente/videos/Video_1.mp4" json_data["Respuesta"] = saludo json_data["Video"] = video # 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"] json_data["Mostrar_videos"] = "si" 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() insertar_mensaje_ws(telefono, pregunta, thread_id) insertar_logs(id_asistente, pregunta, saludo, idcliente, telefono) # Imprimir el JSON con la respuesta print(json.dumps(json_data)) else: respuesta = hacer_pregunta(thread_id, pregunta) if respuesta: if volumen == "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, telefono) json_data["Respuesta"] = respuesta print(json.dumps(json_data))