interface web

web_bot_integration.py

@@ -0,0 +1,308 @@
+"""
+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()