""" 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()