import anthropic from anthropic import Anthropic 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 traceback import random import tiktoken from dotenv import load_dotenv import os class ConfigManager: _instance = None REQUIRED_KEYS = ['OPENAI_API_KEY', 'ANTHROPIC_API_KEY', 'PINECONE_API_KEY', 'OPENAI_API_VALOR_MAS', 'SERVER_ENV', 'XI_API_KEY', 'AUDIO_URL', 'RUTA_AUDIO'] def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._load_config() return cls._instance def _load_config(self): # Primero intenta cargar desde variables de entorno del sistema self.SERVER_ENV = os.getenv('SERVER_ENV') # Si no está definida en las variables de entorno del sistema, busca en .env if not self.SERVER_ENV: env_path = os.getenv('ENV_PATH', '.env') if os.path.exists(env_path): load_dotenv(env_path) self.SERVER_ENV = os.getenv('SERVER_ENV') # Si aún no tenemos SERVER_ENV, usa un valor por defecto if not self.SERVER_ENV: self.SERVER_ENV = '/var/www/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.ANTHROPIC_API_KEY_VALIDATED = bool(self.ANTHROPIC_API_KEY) self.PINECONE_API_KEY_VALIDATED = bool(self.PINECONE_API_KEY) self.XI_API_KEY_VALIDATED = bool(self.XI_API_KEY) self.AUDIO_URL = self.AUDIO_URL if self.AUDIO_URL.endswith('/') else self.AUDIO_URL + '/' self.RUTA_AUDIO = self.RUTA_AUDIO if self.RUTA_AUDIO.endswith('/') else self.RUTA_AUDIO + '/' try: config = ConfigManager() except Exception as e: print(f"Error crítico de configuración: {str(e)}") raise SystemExit(1) def traer_instrucciones(id_asistente): # 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 # Cliente Anthropic global client = None chat_history = [] 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, chat_history # Asignación de variables (igual que original) 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_ config_manager = ConfigManager() # Obtener instrucciones desde la base de datos instrucciones = traer_instrucciones(id_asistente) # Configurar cliente Anthropic según source if source == "INSTR_IA_GENERAL": api_key = config_manager.ANTHROPIC_API_KEY client = Anthropic(api_key=api_key) # Inicializar JSON (igual que original) json_data = { "asistente": asistente, "thread_id": thread_id, "pregunta": pregunta, "archivo": archivo, "idcliente": idcliente, "source": source, "id_asistente": id_asistente, "voy_sin_hilo": voy_sin_hilo, "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 def conectar_db2(): # traerse la variable de conifg del archivo de conexiones.py para conectarse al localhost config2 = conexiones.config2 try: connection = mysql.connector.connect(**config2) if connection.is_connected(): return connection else: return None except mysql.connector.Error as err: print(f"Error: {err}") return None def 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(pregunta): global chat_history, client try: if not chat_history: chat_history = [] chat_history.append({"role": "user", "content": pregunta}) tools = obtener_herramientas() tool_choice = {"type": "auto"} if tools else None # Primera llamada message = client.beta.messages.create( model=asistente, max_tokens=4096, messages=chat_history, tools=tools, tool_choice=tool_choice ) respuesta = "" tool_outputs = [] tool_uses = [] # Procesar contenido manteniendo toda la estructura assistant_content = [] for content in message.content: if content.type == "text": respuesta += content.text assistant_content.append({"type": "text", "text": content.text}) elif content.type == "tool_use": output = ejecutar_funcion(content.name, content.input) tool_outputs.append({ "tool_use_id": content.id, "content": output }) assistant_content.append({ "type": "tool_use", "id": content.id, # Campo crítico añadido "name": content.name, "input": content.input }) # Actualizar historial con respuesta completa chat_history.append({ "role": "assistant", "content": assistant_content }) # Manejar tool outputs if tool_outputs: # Construir respuestas de herramientas tool_responses = [{ "type": "tool_result", "tool_use_id": out["tool_use_id"], "content": out["content"] } for out in tool_outputs] # Agregar respuestas como mensaje de usuario chat_history.append({ "role": "user", "content": tool_responses }) # Segunda llamada second_message = client.beta.messages.create( model=asistente, max_tokens=4096, messages=chat_history ) # Procesar respuesta final final_response = "" for content in second_message.content: if content.type == "text": final_response += content.text # Actualizar historial con respuesta limpia chat_history.append({"role": "assistant", "content": final_response}) respuesta = final_response else: respuesta = message.content[0].text # Mantener formato consistente chat_history[-1]["content"] = respuesta return respuesta except Exception as e: error_msg = f"Error al hacer la pregunta: {str(e)}" print(error_msg) return error_msg def obtener_herramientas(): return [ { "name": "consultar_pqrsf_tipos", "description": "Consultar tipos de PQRSF disponibles", "input_schema": { "type": "object", "properties": {} } }, { "name": "registrar_pqr", "description": "Registrar una nueva PQR en el sistema", "input_schema": { "type": "object", "properties": { "id_categoria": {"type": "number"}, "id_tipo": {"type": "number"}, "contenido_solicitud": {"type": "string"}, "encargado": {"type": "string"}, "respuesta_ia": {"type": "string"} } } }, { "name": "obtener_glosario", "description": "Obtener listado completo del glosario catastral", "input_schema": { "type": "object", "properties": {} } }, { "name": "wiki_bcgs", "description": "Buscar en la base de conocimiento de términos catastrales", "input_schema": { "type": "object", "properties": { "palabra_glosario": {"type": "string"} } } }, { "name": "pqr_horario", "description": "Registrar consulta sobre horarios de atención", "input_schema": { "type": "object", "properties": { "pregunta": {"type": "string"}, "respuesta": {"type": "string"} } } }, { "name": "predios_de_propietario", "description": "Consultar predios asociados a un propietario", "input_schema": { "type": "object", "properties": { "documento": {"type": "string"} } } }, { "name": "propietarios_del_predio", "description": "Consultar propietarios de un predio específico", "input_schema": { "type": "object", "properties": { "npn": {"type": "string"} } } }, { "name": "consulta_radicado", "description": "Consultar estado de un radicado PQRSF", "input_schema": { "type": "object", "properties": { "numero": {"type": "string"} } } }, { "name": "consulta_ciudadano", "description": "Consultar información de un ciudadano registrado", "input_schema": { "type": "object", "properties": { "numero": {"type": "string"} } } }, { "name": "ejecutar_vectorial", "description": "Realizar búsqueda vectorial en documentos catastrales", "input_schema": { "type": "object", "properties": { "nombre": {"type": "string"}, "pregunta": {"type": "string"} } } } ] def ejecutar_funcion(nombre_funcion, arguments): global chat_history try: # Validación básica de argumentos if not isinstance(arguments, dict): return "Error: Formato de parámetros inválido" # Mapeo de funciones con validación de parámetros func_map = { "consultar_pqrsf_tipos": { "function": consultar_pqrsf_tipos, "required_params": [] }, "registrar_pqr": { "function": lambda args: registrar_pqr( args.get("id_categoria"), args.get("id_tipo"), args.get("contenido_solicitud"), args.get("encargado"), args.get("respuesta_ia"), idcliente ), "required_params": ["id_categoria", "id_tipo", "contenido_solicitud"] }, "obtener_glosario": { "function": obtener_glosario, "required_params": [] }, "pqr_horario": { "function": lambda args: pqr_horario( args.get("pregunta"), args.get("respuesta"), idcliente ), "required_params": ["pregunta", "respuesta"] }, "predios_de_propietario": { "function": lambda args: predios_de_propietario( args.get("documento"), idcliente ), "required_params": ["documento"] }, "propietarios_del_predio": { "function": lambda args: propietarios_del_predio( args.get("npn"), idcliente ), "required_params": ["npn"] }, "consulta_radicado": { "function": lambda args: consulta_radicado(args.get("numero")), "required_params": ["numero"] }, "consulta_ciudadano": { "function": lambda args: consulta_ciudadano(args.get("numero")), "required_params": ["numero"] } } # Verificar si la función existe if nombre_funcion not in func_map: return f"Función {nombre_funcion} no implementada" # Validar parámetros requeridos func_info = func_map[nombre_funcion] missing_params = [param for param in func_info["required_params"] if param not in arguments] if missing_params: return f"Error: Parámetros faltantes: {', '.join(missing_params)}" # Ejecutar función y manejar diferentes tipos de retorno try: result = func_info["function"](arguments) # Convertir diferentes tipos de datos a string if isinstance(result, dict): return json.dumps(result) if isinstance(result, list): return "\n".join(str(item) for item in result) return str(result) except KeyError as e: return f"Error: Parámetro inválido - {str(e)}" except TypeError as e: return f"Error: Tipo de parámetro incorrecto - {str(e)}" except Exception as e: return f"Error en ejecución: {str(e)}" except Exception as e: error_msg = f"Error crítico en {nombre_funcion}: {str(e)}" print(f"[ERROR] {error_msg}") return error_msg def limpiar_texto(texto): # Mapeo de caracteres erróneos a caracteres correctos con acentos reemplazos = { "á": "á", "é": "é", "í": "í", "ó": "ó", "ú": "ú", "ñ": "ñ", "Ñ": "Ñ", "ü": "ü", "É": "É", "Ú": "Ú", "Ó": "Ó", "Ã�": "Í", "À": "À" } # Reemplazar caracteres mal codificados for incorrecto, correcto in reemplazos.items(): texto = texto.replace(incorrecto, correcto) # Eliminar saltos de línea y espacios innecesarios texto = re.sub(r"\s+", " ", texto).strip() return texto def obtener_configuracion_voz_y_generar_audio(id_asistente, texto): texto_limpio = limpiar_texto(texto) try: connection = conectar_db() if connection is None: raise Exception("Error al conectar con la base de datos.") except Exception as e: return {"error": f"Error al conectar con la base de datos: {str(e)}"} try: cursor = connection.cursor(dictionary=True) query_voz = """ SELECT asistentes.id_voz, voces.nombre_voz, voces.id_eleven, asistentes_voces.similarity_boost, asistentes_voces.stability, asistentes_voces.style, asistentes_voces.use_speaker_boost FROM asistentes INNER JOIN asistentes_voces ON asistentes.id_voz = asistentes_voces.id INNER JOIN voces ON asistentes_voces.id_modelo_voz = voces.id WHERE asistentes.id = %s """ cursor.execute(query_voz, (id_asistente,)) asistente = cursor.fetchone() if not asistente: return {"error": "No se encontró un asistente con el ID proporcionado."} # Convertimos los valores para replicar PHP use_speaker_boost = "true" if asistente["use_speaker_boost"] == 1 else "false" # Construimos el JSON manualmente igual que en PHP voice_settings = ( f"\"similarity_boost\": {asistente['similarity_boost']}, " f"\"stability\": {asistente['stability']}, " f"\"style\": {asistente['style']}, " f"\"use_speaker_boost\": {use_speaker_boost}" ) payload = ( "{\n" f" \"model_id\": \"eleven_multilingual_v2\",\n" f" \"text\": \"{texto_limpio}\",\n" f" \"voice_settings\": {{\n {voice_settings}\n }}\n" "}" ) # URL de ElevenLabs - api_url = f"https://api.elevenlabs.io/v1/text-to-speech/{asistente['id_eleven']}?optimize_streaming_latency=0&output_format=mp3_44100_128" headers = { "Content-Type": "application/json", "xi-api-key": config.XI_API_KEY } # Imprimir el JSON antes de enviarlo para comparar con P H P 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"{config.RUTA_AUDIO}{nombre_archivo}" # ruta_audio = f"/var/www/catia.ayudacatastro.co/valormas/audios/{nombre_archivo}" audio_url = f"{config.AUDIO_URL}{nombre_archivo}" # audio_url = f"https://catia.ayudacatastro.co/valormas/audios/{nombre_archivo}" with open(ruta_audio, "wb") as audio_file: audio_file.write(response.content) return audio_url else: return { "success": False, "error": f"Error en ElevenLabs: {response.status_code} - {response.text}" } finally: if connection.is_connected(): cursor.close() connection.close() # client = anthropic.Anthropic() # def contar_tokens(texto): # return client.messages.count_tokens(texto) def insertar_logs(id_asistente, pregunta, respuesta, idcliente): """Registra la interacción en la base de datos con el cálculo de tokens usando GPT-4 Turbo.""" # Conexión a la base de datos connection = conectar_db2() if connection is None: return "Error al conectar con la base de datos." try: cursor = connection.cursor() # Especificamos el modelo GPT-4 Turbo modelo = "claude-3-5-haiku-20241022" # Definir la llegada como "WhatsApp" llegada = "WEB" # Contar tokens correctamente # tokens_input = contar_tokens(pregunta) # tokens_output = contar_tokens(respuesta) input_tokens = 200 output_tokens = 200 # 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, input_tokens, output_tokens ,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): global chat_history, json_data, client try: # 1. Inicialización del historial de chat if not chat_history or thread_id.lower() in ["nada", "no", ""]: chat_history = [{"role": "assistant", "content": instrucciones}] # 2. Manejo de usuarios no registrados if str(idcliente) == "0": 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." ] json_data["Respuesta"] = random.choice(respuestas_registro) print(json.dumps(json_data)) sys.exit() # 3. Manejo de nuevo hilo de conversación if thread_id.lower() == "nada" and voy_sin_hilo == "si": 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" } ] respuesta_aleatoria = random.choice(respuestas_primer_hilo) json_data.update({ "Respuesta": respuesta_aleatoria["saludo"], "Video": respuesta_aleatoria["video"] }) try: # Lógica de base de datos connection = conectar_db() if connection and connection.is_connected(): with connection.cursor(dictionary=True) as cursor: cursor.execute("SELECT telefono FROM ciudadanos WHERE id = %s", (idcliente,)) ciudadano = cursor.fetchone() if ciudadano and "telefono" in ciudadano: url_remota = f"https://dev.ayudacatastro.co/api/crm/asistente/intenciones.php?cliente_tel={ciudadano['telefono']}&rutaArchivoLocal={respuesta_aleatoria['video']}" json_data["URL_Remota"] = url_remota # Realizar solicitud GET response = requests.get(url_remota, timeout=10) json_data["Resultado_Solicitud"] = response.text if response.ok else f"Error {response.status_code}" except Exception as e: json_data["Error_DB"] = str(e) finally: if connection and connection.is_connected(): connection.close() print(json.dumps(json_data)) sys.exit() # 4. Procesamiento de mensajes regulares chat_history.append({"role": "user", "content": pregunta}) # Primera llamada a la API message = client.beta.messages.create( model=asistente, max_tokens=4096, messages=chat_history, tools=obtener_herramientas(), tool_choice={"type": "auto"} ) respuesta = "" tool_outputs = [] assistant_content = [] # Procesar contenido del mensaje for content in message.content: if content.type == "text": respuesta += content.text assistant_content.append({"type": "text", "text": content.text}) elif content.type == "tool_use": output = ejecutar_funcion(content.name, content.input) tool_outputs.append({ "tool_use_id": content.id, "content": output }) assistant_content.append({ "type": "tool_use", "id": content.id, # Campo crítico añadido "name": content.name, "input": content.input }) # Actualizar historial chat_history.append({"role": "assistant", "content": assistant_content}) # Manejo de respuestas de herramientas if tool_outputs: tool_responses = [{ "type": "tool_result", "tool_use_id": out["tool_use_id"], "content": out["content"] } for out in tool_outputs] chat_history.append({"role": "user", "content": tool_responses}) # Segunda llamada a la API second_message = client.beta.messages.create( model=asistente, max_tokens=4096, messages=chat_history, tools=obtener_herramientas() ) # Procesar respuesta final final_response = "".join([c.text for c in second_message.content if c.type == "text"]) chat_history.append({"role": "assistant", "content": final_response}) respuesta = final_response # Manejo de audio y logs if volumen == "si" and audio_humano == "si": json_data["ruta_audio"] = obtener_configuracion_voz_y_generar_audio(id_asistente, respuesta) insertar_logs(id_asistente, pregunta, respuesta, idcliente) json_data["Respuesta"] = respuesta print(json.dumps(json_data)) except Exception as e: json_data["Error"] = f"Error en el proceso: {str(e)}" print(json.dumps(json_data))