gpatruno
308 lines
12 KiB
Python
308 lines
12 KiB
Python
"""
|
|
Module d'intégration entre l'interface web et le bot Twitch existant
|
|
Permet de démarrer le bot existant avec contrôle web
|
|
"""
|
|
|
|
import threading
|
|
import time
|
|
import sys
|
|
import os
|
|
from typing import Optional
|
|
|
|
# Import du bot existant
|
|
from main_auto_loop import *
|
|
from fonction.first_class import *
|
|
|
|
class WebBotIntegration:
|
|
"""Classe pour intégrer le bot existant avec l'interface web"""
|
|
|
|
def __init__(self):
|
|
self.bot_instances = {}
|
|
self.running = False
|
|
|
|
def start_bot_with_channel(self, channel_name: str, record_audio: bool = True, record_time: int = 60) -> dict:
|
|
"""
|
|
Démarre une instance du bot pour un canal spécifique
|
|
|
|
Args:
|
|
channel_name: Nom du canal Twitch
|
|
record_audio: Si True, enregistre l'audio du stream
|
|
record_time: Durée d'enregistrement par segment
|
|
|
|
Returns:
|
|
Dict avec les informations de l'instance créée
|
|
"""
|
|
try:
|
|
# Créer le dossier de travail s'il n'existe pas
|
|
work_directory = f"working_bot_{channel_name}"
|
|
if not os.path.exists(work_directory):
|
|
os.makedirs(work_directory)
|
|
|
|
# Sauvegarder le répertoire courant
|
|
original_cwd = os.getcwd()
|
|
|
|
# Changer vers le répertoire de travail
|
|
os.chdir(work_directory)
|
|
|
|
instance = {
|
|
'channel_name': channel_name,
|
|
'record_audio': record_audio,
|
|
'record_time': record_time,
|
|
'work_directory': work_directory,
|
|
'original_cwd': original_cwd,
|
|
'bots': {},
|
|
'running': True
|
|
}
|
|
|
|
try:
|
|
# Nettoyer les anciens fichiers
|
|
if os.path.exists("./storage/IA_generator.json"):
|
|
os.remove("./storage/IA_generator.json")
|
|
if os.path.exists("./storage/subtitle_data.json"):
|
|
os.remove("./storage/subtitle_data.json")
|
|
|
|
# Créer le dossier storage s'il n'existe pas
|
|
if not os.path.exists("storage"):
|
|
os.makedirs("storage")
|
|
|
|
# Démarrer l'enregistrement audio si demandé
|
|
if record_audio:
|
|
hprint("blue", f"Démarrage RecordTwitch pour {channel_name}")
|
|
record_tw = RecordTwitch(channel_name=channel_name, record_time=record_time)
|
|
instance['bots']['record'] = record_tw
|
|
|
|
# Démarrer l'enregistrement dans un thread
|
|
record_thread = threading.Thread(target=record_tw.main, daemon=True)
|
|
record_thread.start()
|
|
|
|
# Démarrer la traduction des sous-titres
|
|
hprint("blue", f"Démarrage Subtitle_translation pour {channel_name}")
|
|
sb_translation = Subtitle_translation("../config/config.json")
|
|
instance['bots']['subtitle'] = sb_translation
|
|
sb_translation.start_main_loop()
|
|
|
|
# Démarrer le générateur IA
|
|
hprint("blue", f"Démarrage IA_generator pour {channel_name}")
|
|
ask_text = IA_generator("../config/config.json")
|
|
instance['bots']['ia'] = ask_text
|
|
ask_text.start_main_loop()
|
|
|
|
# Démarrer le contrôleur de messages
|
|
controluser = messageTwitch("../config/user.json", channel_name)
|
|
instance['bots']['message'] = controluser
|
|
controluser.start_loop_respond()
|
|
|
|
# Démarrer le bot de chat (optionnel)
|
|
chat_bot = TwitchChatBot(channel_name)
|
|
instance['bots']['chat'] = chat_bot
|
|
chat_bot.start_background()
|
|
|
|
hprint("green", f"Bot démarré avec succès pour {channel_name}")
|
|
|
|
finally:
|
|
# Retourner au répertoire original
|
|
os.chdir(original_cwd)
|
|
|
|
# Stocker l'instance
|
|
self.bot_instances[channel_name] = instance
|
|
|
|
return {
|
|
'success': True,
|
|
'channel': channel_name,
|
|
'instance_id': len(self.bot_instances),
|
|
'bots_started': list(instance['bots'].keys())
|
|
}
|
|
|
|
except Exception as e:
|
|
hprint("red", f"Erreur lors du démarrage du bot pour {channel_name}: {str(e)}")
|
|
return {
|
|
'success': False,
|
|
'error': str(e),
|
|
'channel': channel_name
|
|
}
|
|
|
|
def stop_bot_for_channel(self, channel_name: str) -> dict:
|
|
"""
|
|
Arrête le bot pour un canal spécifique
|
|
|
|
Args:
|
|
channel_name: Nom du canal
|
|
|
|
Returns:
|
|
Dict avec le résultat de l'opération
|
|
"""
|
|
try:
|
|
if channel_name not in self.bot_instances:
|
|
return {'success': False, 'error': f'Aucun bot actif pour {channel_name}'}
|
|
|
|
instance = self.bot_instances[channel_name]
|
|
instance['running'] = False
|
|
|
|
# Arrêter tous les bots de cette instance
|
|
bots_stopped = []
|
|
for bot_name, bot_instance in instance['bots'].items():
|
|
try:
|
|
if hasattr(bot_instance, 'stop'):
|
|
bot_instance.stop()
|
|
bots_stopped.append(bot_name)
|
|
hprint("yellow", f"Bot {bot_name} arrêté pour {channel_name}")
|
|
except Exception as e:
|
|
hprint("red", f"Erreur lors de l'arrêt du bot {bot_name}: {str(e)}")
|
|
|
|
# Supprimer l'instance
|
|
del self.bot_instances[channel_name]
|
|
|
|
return {
|
|
'success': True,
|
|
'channel': channel_name,
|
|
'bots_stopped': bots_stopped
|
|
}
|
|
|
|
except Exception as e:
|
|
hprint("red", f"Erreur lors de l'arrêt du bot pour {channel_name}: {str(e)}")
|
|
return {'success': False, 'error': str(e)}
|
|
|
|
def get_bot_status(self, channel_name: str) -> dict:
|
|
"""
|
|
Récupère le statut du bot pour un canal
|
|
|
|
Args:
|
|
channel_name: Nom du canal
|
|
|
|
Returns:
|
|
Dict avec le statut du bot
|
|
"""
|
|
if channel_name not in self.bot_instances:
|
|
return {'active': False, 'error': 'Bot non trouvé'}
|
|
|
|
instance = self.bot_instances[channel_name]
|
|
|
|
status = {
|
|
'active': instance['running'],
|
|
'channel': channel_name,
|
|
'record_audio': instance['record_audio'],
|
|
'bots': {}
|
|
}
|
|
|
|
# Vérifier le statut de chaque bot
|
|
for bot_name, bot_instance in instance['bots'].items():
|
|
try:
|
|
if bot_name == 'record' and hasattr(bot_instance, 'running'):
|
|
status['bots'][bot_name] = {'running': bot_instance.running}
|
|
elif bot_name == 'subtitle' and hasattr(bot_instance, 'is_running'):
|
|
status['bots'][bot_name] = {'running': bot_instance.is_running}
|
|
elif bot_name == 'ia' and hasattr(bot_instance, 'ia_running'):
|
|
status['bots'][bot_name] = {'running': bot_instance.ia_running}
|
|
elif bot_name == 'message' and hasattr(bot_instance, 'message_running'):
|
|
status['bots'][bot_name] = {'running': bot_instance.message_running}
|
|
elif bot_name == 'chat' and hasattr(bot_instance, 'is_running'):
|
|
status['bots'][bot_name] = {'running': bot_instance.is_running}
|
|
else:
|
|
status['bots'][bot_name] = {'running': True} # Par défaut
|
|
except Exception as e:
|
|
status['bots'][bot_name] = {'running': False, 'error': str(e)}
|
|
|
|
return status
|
|
|
|
def list_active_bots(self) -> dict:
|
|
"""
|
|
Liste tous les bots actifs
|
|
|
|
Returns:
|
|
Dict avec la liste des bots actifs
|
|
"""
|
|
active_bots = {}
|
|
|
|
for channel_name in self.bot_instances:
|
|
active_bots[channel_name] = self.get_bot_status(channel_name)
|
|
|
|
return {
|
|
'total_instances': len(self.bot_instances),
|
|
'active_bots': active_bots
|
|
}
|
|
|
|
def send_message_to_channel(self, channel_name: str, message: str) -> dict:
|
|
"""
|
|
Envoie un message dans le chat d'un canal
|
|
|
|
Args:
|
|
channel_name: Nom du canal
|
|
message: Message à envoyer
|
|
|
|
Returns:
|
|
Dict avec le résultat de l'opération
|
|
"""
|
|
try:
|
|
if channel_name not in self.bot_instances:
|
|
return {'success': False, 'error': f'Aucun bot actif pour {channel_name}'}
|
|
|
|
instance = self.bot_instances[channel_name]
|
|
|
|
if 'message' not in instance['bots']:
|
|
return {'success': False, 'error': 'Bot de message non disponible'}
|
|
|
|
message_bot = instance['bots']['message']
|
|
message_bot.send_message(message)
|
|
|
|
return {
|
|
'success': True,
|
|
'channel': channel_name,
|
|
'message': message
|
|
}
|
|
|
|
except Exception as e:
|
|
return {'success': False, 'error': str(e)}
|
|
|
|
def generate_response_for_channel(self, channel_name: str, text: str = None) -> dict:
|
|
"""
|
|
Génère une réponse IA pour un canal
|
|
|
|
Args:
|
|
channel_name: Nom du canal
|
|
text: Texte à utiliser pour la génération (optionnel)
|
|
|
|
Returns:
|
|
Dict avec le résultat de l'opération
|
|
"""
|
|
try:
|
|
if channel_name not in self.bot_instances:
|
|
return {'success': False, 'error': f'Aucun bot actif pour {channel_name}'}
|
|
|
|
instance = self.bot_instances[channel_name]
|
|
|
|
if 'ia' not in instance['bots']:
|
|
return {'success': False, 'error': 'Bot IA non disponible'}
|
|
|
|
ia_bot = instance['bots']['ia']
|
|
|
|
# Utiliser le texte fourni ou récupérer le dernier sous-titre
|
|
if text is None:
|
|
if 'subtitle' in instance['bots']:
|
|
subtitle_bot = instance['bots']['subtitle']
|
|
text = subtitle_bot.get_lasttext()
|
|
else:
|
|
return {'success': False, 'error': 'Aucun texte disponible pour la génération'}
|
|
|
|
# Lancer la génération
|
|
ia_bot.main_ask(text)
|
|
|
|
return {
|
|
'success': True,
|
|
'channel': channel_name,
|
|
'text_used': text
|
|
}
|
|
|
|
except Exception as e:
|
|
return {'success': False, 'error': str(e)}
|
|
|
|
def stop_all_bots(self):
|
|
"""Arrête tous les bots actifs"""
|
|
channels_to_stop = list(self.bot_instances.keys())
|
|
|
|
for channel_name in channels_to_stop:
|
|
self.stop_bot_for_channel(channel_name)
|
|
|
|
hprint("blue", "Tous les bots ont été arrêtés")
|
|
|
|
# Instance globale pour l'intégration
|
|
web_bot_integration = WebBotIntegration() |