from flask import Flask from tts_utils import TTSUtils from inference import InferenceManager from huggingface_utils import HuggingFaceUtils from flow_bot import FlowBot from tunnel_manager import TunnelManager from routes import Routes from session_manager import SessionManager from data_manager import DataManager import yaml import os import webbrowser import threading import time import sys from datetime import timedelta import logging from colorama import init, Fore, Back, Style # Inicializar colorama para Windows init() def print_startup_checklist(): """Muestra el checklist de inicio con colores""" print(f"\n{Fore.CYAN}{'='*50}") print(f"{Fore.YELLOW}🤖 Iniciando Asistente Virtual...") print(f"{Fore.CYAN}{'='*50}{Style.RESET_ALL}\n") # Checklist de Modelos print(f"{Fore.MAGENTA}📋 Modelos de IA:{Style.RESET_ALL}") print(f"{Fore.GREEN}✓ {Fore.WHITE}Principal: {Fore.YELLOW}Gemini 8b 🧠") print(f"{Fore.GREEN}✓ {Fore.WHITE}Respaldo: {Fore.YELLOW}Mixtral 7b ⚡\n") # Checklist de TTS print(f"{Fore.MAGENTA}🎤 Motores TTS:{Style.RESET_ALL}") print(f"{Fore.GREEN}✓ {Fore.WHITE}EDGE (Principal)") print(f" {Fore.CYAN}└─ Voz: Jorge (MX) 📢") print(f"{Fore.GREEN}✓ {Fore.WHITE}EDGE_ES") print(f" {Fore.CYAN}└─ Voz: Álvaro (ES) 🌐") print(f"{Fore.GREEN}✓ {Fore.WHITE}VITS") print(f" {Fore.CYAN}└─ Modelo Local 🔊\n") # Checklist de Modos print(f"{Fore.MAGENTA}📋 Modos Disponibles:{Style.RESET_ALL}") print(f"{Fore.GREEN}✓ {Fore.WHITE}Créditos 💰") print(f"{Fore.GREEN}✓ {Fore.WHITE}Seguros 🛡️") print(f"{Fore.GREEN}✓ {Fore.WHITE}Cobranza 💵\n") # Estado del Sistema print(f"{Fore.MAGENTA}🔄 Estado del Sistema:{Style.RESET_ALL}") class WebChatbotApp: def __init__(self): # Mostrar checklist inicial print_startup_checklist() self.flask_app = Flask(__name__) self.flask_app.config['SECRET_KEY'] = os.urandom(24) self.flask_app.config['SESSION_TYPE'] = 'filesystem' self.flask_app.config['SESSION_FILE_DIR'] = './flask_session' self.flask_app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(hours=2) self.tunnel_url = None self.load_config() self.init_components() self.routes = Routes(self.flask_app, self) # Inicializar DataManager self.data_manager = DataManager() print(f"{Fore.GREEN}✓ {Fore.WHITE}Aplicación inicializada correctamente{Style.RESET_ALL}") def load_config(self): try: config_path = os.path.join(os.path.dirname(__file__), 'config.yaml') with open(config_path, 'r', encoding='utf-8') as f: self.config = yaml.safe_load(f) print(f"{Fore.GREEN}✓ {Fore.WHITE}Configuración cargada") except Exception as e: print(f"{Fore.RED}✗ Error cargando config.yaml: {e}{Style.RESET_ALL}") self.config = {} def init_components(self): try: print(f"{Fore.CYAN}⚙️ Iniciando componentes...{Style.RESET_ALL}") elevenlabs_key = self.config.get('ELEVENLABS_API_KEY', '') huggingface_token = self.config.get('api_keys', {}).get('HUGGINGFACE_TOKEN', '') self.tts = TTSUtils( model_name='EDGE', elevenlabs_api_key=elevenlabs_key ) self.inference = InferenceManager(self.config) self.hf_utils = HuggingFaceUtils(huggingface_token) self.flow_bot = FlowBot() self.session_manager = SessionManager() print(f"{Fore.GREEN}✓ {Fore.WHITE}Componentes iniciados correctamente{Style.RESET_ALL}") except Exception as e: print(f"{Fore.RED}✗ Error iniciando componentes: {e}{Style.RESET_ALL}") raise def reinit_components(self): try: self.load_config() elevenlabs_key = self.config.get('ELEVENLABS_API_KEY', '') huggingface_token = self.config.get('api_keys', {}).get('HUGGINGFACE_TOKEN', '') print(f"{Fore.CYAN}⚙️ Reiniciando componentes...{Style.RESET_ALL}") self.tts = TTSUtils( model_name='EDGE', elevenlabs_api_key=elevenlabs_key ) self.inference = InferenceManager(self.config) self.hf_utils = HuggingFaceUtils(huggingface_token) print(f"{Fore.GREEN}✓ {Fore.WHITE}Componentes reiniciados correctamente{Style.RESET_ALL}") except Exception as e: print(f"{Fore.RED}✗ Error reiniciando componentes: {e}{Style.RESET_ALL}") raise def procesar_mensaje(self, mensaje): """Procesa un mensaje de texto y retorna la respuesta""" try: # Obtener el modo actual de la sesión session_id = self.session_manager.get_session() current_mode = self.session_manager.get_session_data(session_id, 'mode') # Verificar si estamos en proceso de recolección de datos if self.data_manager.is_collecting_data(): respuesta = self.data_manager.handle_data_collection( mensaje, current_mode, self.flow_bot.get_data_collection_steps ) if respuesta: return respuesta # Obtener contexto del flow_bot usando el modo actual contexto = self.flow_bot.get_context(current_mode, mensaje) # Obtener respuesta usando el inference manager respuesta = self.inference.get_response( prompt=mensaje, context=contexto ) # Verificar que la respuesta no sea un mensaje de error o predeterminado mensajes_error = [ "No se proporcionó un mensaje", "Lo siento, hubo un error", "Error de autenticación", "El servicio está ocupado", "No se pudo generar una respuesta coherente", "¡Hola! Soy tu asistente virtual" ] if not respuesta or any(msg in respuesta for msg in mensajes_error): print("Respuesta no válida del modelo, usando contexto directo") respuesta = self.flow_bot.get_success_message(current_mode) # Iniciar recolección de datos si es necesario if "nombre" in respuesta.lower(): self.data_manager.start_data_collection() return respuesta except Exception as e: print(f"Error procesando mensaje: {e}") return self.flow_bot.get_negative_response(current_mode) def procesar_audio(self, audio_path): """Procesa un archivo de audio y retorna mensaje y audio de respuesta""" try: # Transcribir audio a texto mensaje = self.inference.transcribe_audio(audio_path) print(f"Audio transcrito: {mensaje}") # Obtener respuesta respuesta = self.procesar_mensaje(mensaje) print(f"Respuesta generada: {respuesta}") # Convertir respuesta a audio audio_path = self.tts.text_to_speech(respuesta) return mensaje, audio_path except Exception as e: print(f"Error procesando audio: {e}") return "Error al procesar el audio", None def open_browser(self, url): time.sleep(2) webbrowser.open(url) def run(self, host='127.0.0.1', port=5000, debug=False, use_tunnel=True): try: if use_tunnel: try: tunnel_manager = TunnelManager() # Primero intentar obtener un túnel activo self.tunnel_url = tunnel_manager.get_active_tunnel() if not self.tunnel_url: # Si no hay túnel activo, limpiar y crear uno nuevo tunnel_manager.setup_ngrok() tunnel_manager.cleanup() self.tunnel_url = tunnel_manager.start(port) if self.tunnel_url: print(f"{Fore.GREEN}✓ {Fore.WHITE}Túnel iniciado en: {Fore.CYAN}{self.tunnel_url}{Style.RESET_ALL}") threading.Thread(target=self.open_browser, args=(self.tunnel_url,)).start() else: print(f"{Fore.YELLOW}⚠️ No se pudo iniciar el túnel, continuando en modo local{Style.RESET_ALL}") except Exception as tunnel_error: print(f"{Fore.YELLOW}⚠️ Continuando en modo local{Style.RESET_ALL}") print(f"\n{Fore.GREEN}✓ {Fore.WHITE}Servidor iniciado en: {Fore.CYAN}http://{host}:{port}{Style.RESET_ALL}\n") self.flask_app.run(host=host, port=port, debug=debug, threaded=True) except Exception as e: print(f"{Fore.RED}✗ Error iniciando el servidor: {e}{Style.RESET_ALL}") raise @property def app(self): return self.flask_app if __name__ == '__main__': import os import psutil import socket def is_port_in_use(port): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: return s.connect_ex(('localhost', port)) == 0 def kill_process_on_port(port): for proc in psutil.process_iter(['pid', 'name', 'connections']): try: for conn in proc.connections(): if conn.laddr.port == port: print(f"Terminando proceso anterior en puerto {port}") proc.terminate() proc.wait() return True except (psutil.NoSuchProcess, psutil.AccessDenied): pass return False try: port = 5000 if is_port_in_use(port): print(f"Puerto {port} en uso. Intentando liberar...") if kill_process_on_port(port): print("Proceso anterior terminado") else: print(f"No se pudo liberar el puerto {port}. Por favor, cierre la aplicación anterior manualmente.") exit(1) print("Iniciando nueva instancia de la aplicación...") app = WebChatbotApp() app.run() except Exception as e: print(f"Error al iniciar la aplicación: {e}") exit(1)