import argparse import platform import os import asyncio import discord from discord.ext import commands from internal.NoiseGatev2 import NoiseGate # Assuming NoiseGatev2.py is in the same directory # --- Opus Library Loading --- def load_opus(): """Loads the correct opus library for the operating system.""" try: script_dir = os.path.dirname(os.path.abspath(__file__)) if os.name == 'nt': processor = platform.machine() if processor == "AMD64": print("Loaded OPUS library for Windows AMD64") discord.opus.load_opus(os.path.join(script_dir, './opus/libopus_amd64.dll')) else: print(f"Unsupported Windows processor: {processor}. Opus may not work.") else: processor = platform.machine() print(f"Processor: {processor}") if processor == "aarch64": print("Loaded OPUS library for aarch64") discord.opus.load_opus(os.path.join(script_dir, './opus/libopus_aarcch64.so')) elif processor == "armv7l": print("Loaded OPUS library for armv7l") discord.opus.load_opus(os.path.join(script_dir, './opus/libopus_armv7l.so')) else: print(f"Attempting to load a generic opus library for {processor}") discord.opus.load_opus(os.path.join(script_dir, './opus/libopus.so.0')) if discord.opus.is_loaded(): print("Opus library loaded successfully.") return True else: print("Opus library failed to load.") return False except Exception as e: print(f"Error loading opus library: {e}") return False # --- Voice Cog for Multi-Server Management --- class VoiceCog(commands.Cog): """Cog to handle all voice-related commands and state management.""" def __init__(self, bot, device_id, ng_threshold): self.bot = bot self.device_id = device_id self.ng_threshold = ng_threshold self.voice_states = {} # { guild_id: NoiseGate_instance } @commands.Cog.listener() async def on_ready(self): print(f'Logged in as {self.bot.user} (ID: {self.bot.user.id})') print('------') # Internal API method to join a voice channel async def internal_join_voice_channel(self, guild_id: int, channel_id: int): guild = self.bot.get_guild(guild_id) if not guild: print(f"Guild with ID {guild_id} not found.") return False channel = guild.get_channel(channel_id) if not channel or not isinstance(channel, discord.VoiceChannel): print(f"Voice channel with ID {channel_id} not found in guild {guild.name}.") return False # Get voice client for this guild voice_client = discord.utils.get(self.bot.voice_clients, guild=guild) if voice_client: if voice_client.channel == channel: print(f"Already in channel: {channel.name} in guild {guild.name}.") else: await voice_client.move_to(channel) print(f"Moved to channel: {channel.name} in guild {guild.name}.") else: try: voice_client = await channel.connect(timeout=60.0, reconnect=True) print(f"Connected to channel: {channel.name} in guild {guild.name}.") except Exception as e: print(f"Failed to connect to {channel.name} in guild {guild.name}. Error: {e}") return False if discord.opus.is_loaded(): # Create and start the NoiseGate audio stream for this server stream_handler = NoiseGate( _voice_connection=voice_client, _noise_gate_threshold=self.ng_threshold, _input_device_index=self.device_id ) self.voice_states[guild.id] = stream_handler stream_handler.run() print(f"Started audio stream for server: {guild.name}") return True else: print("Opus library not loaded. Cannot start audio stream.") await voice_client.disconnect() # Disconnect if opus isn't loaded return False # Internal API method to leave a voice channel async def internal_leave_voice_channel(self, guild_id: int): if guild_id not in self.voice_states: print(f"Not currently in a voice channel on guild ID {guild_id}.") return False guild = self.bot.get_guild(guild_id) if not guild: print(f"Guild with ID {guild_id} not found.") return False voice_client = discord.utils.get(self.bot.voice_clients, guild=guild) if not voice_client: print(f"Bot not in a voice channel in guild {guild.name}, but state exists. Cleaning up.") del self.voice_states[guild.id] return True stream_handler = self.voice_states[guild.id] await stream_handler.close() # Close the NoiseGate stream del self.voice_states[guild.id] print(f"Disconnected and stopped the audio stream for guild: {guild.name}") return True # Discord command for joining (for direct user interaction) @commands.command(name='join') async def join_command(self, ctx, *, channel: discord.VoiceChannel = None): if not channel: if ctx.author.voice: channel = ctx.author.voice.channel else: await ctx.send("You are not connected to a voice channel. Please specify one to join.") return success = await self.internal_join_voice_channel(ctx.guild.id, channel.id) if success: await ctx.send(f"Connected to channel: {channel.name}.") else: await ctx.send(f"Failed to connect to {channel.name}.") # Discord command for leaving (for direct user interaction) @commands.command(name='leave') async def leave_command(self, ctx): success = await self.internal_leave_voice_channel(ctx.guild.id) if success: await ctx.send("Disconnected and stopped the audio stream.") else: await ctx.send("I am not currently in a voice channel on this server.") @join_command.before_invoke async def ensure_opus(self, ctx): if not discord.opus.is_loaded(): await ctx.send("Opus audio library is not loaded. I cannot join a voice channel.") raise commands.CommandError("Opus not loaded.") # --- Discord Bot Manager Class --- class DiscordBotManager: def __init__(self, device_id: int = 0, ng_threshold: int = 50): self.token = None self.device_id = device_id self.ng_threshold = ng_threshold self.bot = None self.voice_cog = None self._bot_task = None # To hold the running bot task for graceful stopping async def _setup_bot(self): # Define bot intents intents = discord.Intents.default() intents.message_content = True # Required for commands intents.guilds = True # Required to get guild objects by ID intents.voice_states = True # Required to get voice channel info self.bot = commands.Bot(command_prefix='!', intents=intents) self.voice_cog = VoiceCog(self.bot, self.device_id, self.ng_threshold) await self.bot.add_cog(self.voice_cog) @self.bot.event async def on_ready(): print(f'Bot fully ready: {self.bot.user}') # Set initial presence when the bot is ready await self.set_presence("Broadcasting...", discord.Game) async def start_bot(self, token: str): if self.bot and self.bot.is_ready(): print("Bot is already running.") return self.token = token if not load_opus(): print("Failed to load Opus library. Bot cannot start voice features.") return await self._setup_bot() print("Starting bot...") try: # Run the bot in a separate task so we can control it self._bot_task = asyncio.create_task(self.bot.start(self.token)) # Wait for the bot to connect (optional, useful for ensuring it's ready) await self.bot.wait_until_ready() print("Bot started and is ready.") except discord.LoginFailure: print("Failed to login to Discord. Check your client ID (token).") except Exception as e: print(f"An error occurred while starting the bot: {e}") async def stop_bot(self): if not self.bot or not self.bot.is_ready(): print("Bot is not running or not ready.") return print("Stopping bot...") # Clean up all active voice connections before stopping for guild_id in list(self.voice_cog.voice_states.keys()): await self.voice_cog.internal_leave_voice_channel(guild_id) await self.bot.close() if self._bot_task: self._bot_task.cancel() # Cancel the bot's running task try: await self._bot_task # Await cancellation to ensure cleanup except asyncio.CancelledError: pass print("Bot stopped.") self.bot = None # Reset bot instance async def join_voice_channel(self, guild_id: int, channel_id: int): if not self.bot or not self.bot.is_ready(): print("Bot is not running or ready. Cannot join voice channel.") return False return await self.voice_cog.internal_join_voice_channel(guild_id, channel_id) async def leave_voice_channel(self, guild_id: int): if not self.bot or not self.bot.is_ready(): print("Bot is not running or ready. Cannot leave voice channel.") return False return await self.voice_cog.internal_leave_voice_channel(guild_id) async def set_presence(self, name: str, activity_type: discord.Activity = discord.ActivityType.listening): if not self.bot or not self.bot.is_ready(): print("Bot is not running or ready. Cannot set presence.") return try: if activity_type == discord.Game: activity = discord.Game(name=name) elif activity_type == discord.Streaming: activity = discord.Streaming(name=name, url="https://twitch.tv/your_stream_url_here") # Replace with actual URL elif activity_type == discord.ActivityType.listening: activity = discord.Activity(type=discord.ActivityType.listening, name=name) elif activity_type == discord.ActivityType.watching: activity = discord.Activity(type=discord.ActivityType.watching, name=name) else: print(f"Invalid activity type: {activity_type}. Defaulting to Game.") activity = discord.Game(name=name) await self.bot.change_presence(activity=activity) print(f"Presence set to: {name} ({activity_type.__name__})") except Exception as e: print(f"Error setting presence: {e}") # --- Example Usage / Main Entry Point --- async def main_run(): parser = argparse.ArgumentParser(description="Discord Radio Bot Manager.") parser.add_argument("clientId", type=str, help="The Discord bot's client token.") parser.add_argument("deviceId", type=int, help="The ID of the audio input device to use.") parser.add_argument("-n", "--NGThreshold", type=int, default=50, help="The noise gate threshold (default: 50).") args = parser.parse_args() # Instantiate the bot manager bot_manager = DiscordBotManager(args.clientId, args.deviceId, args.NGThreshold) # --- Start the bot --- await bot_manager.start_bot() # --- Example of how an external system (like your websocket) would interact --- # You would replace this with your actual websocket logic print("\nBot is running. You can now use internal_api calls or Discord commands.") print("Example: Use !join in Discord.") print("Example: Use !leave in Discord.") print("Example: Simulating an external request to join a channel after 10 seconds...") await asyncio.sleep(10) # Simulate delay # Placeholder for guild_id and channel_id from your websocket server # You would receive these from your websocket server # For testing, you'll need to manually get a guild ID and channel ID where your bot is. example_guild_id = 123456789012345678 # <<< REPLACE WITH AN ACTUAL GUILD ID example_channel_id = 987654321098765432 # <<< REPLACE WITH AN ACTUAL VOICE CHANNEL ID print(f"\nAttempting to join voice channel {example_channel_id} in guild {example_guild_id} via internal API...") success_join = await bot_manager.join_voice_channel(example_guild_id, example_channel_id) print(f"Internal join successful: {success_join}") await asyncio.sleep(15) # Stay in channel for a bit print("\nAttempting to leave voice channel via internal API...") success_leave = await bot_manager.leave_voice_channel(example_guild_id) print(f"Internal leave successful: {success_leave}") await asyncio.sleep(5) print("\nChanging bot presence to 'Streaming music'...") await bot_manager.set_presence("Streaming music", discord.Streaming) await asyncio.sleep(5) print("\nChanging bot presence back to 'Idle'...") await bot_manager.set_presence("Idle", discord.Game) # Keep the bot running indefinitely until an external stop command or script exit try: while True: await asyncio.sleep(3600) # Sleep for an hour, or until interrupted except KeyboardInterrupt: print("\nKeyboardInterrupt detected. Stopping bot...") finally: await bot_manager.stop_bot() if __name__ == "__main__": asyncio.run(main_run())