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)