import discord import logging import asyncio from discord import app_commands from discord.ext import commands from random import randint import data import copy import subsonic import ui from discodrome import DiscodromeClient logger = logging.getLogger(__name__) class MusicCog(commands.Cog): ''' A Cog containing music playback commands ''' bot : DiscodromeClient def __init__(self, bot: DiscodromeClient): self.bot = bot async def get_voice_client(self, interaction: discord.Interaction, *, should_connect: bool=False) -> discord.VoiceClient: ''' Returns a voice client instance for the current guild ''' # Get the voice client for the guild voice_client = discord.utils.get(self.bot.voice_clients, guild=interaction.guild) # Connect to a voice channel if voice_client is None and should_connect: try: # Check if user is in a voice channel if interaction.user.voice is None or interaction.user.voice.channel is None: await ui.ErrMsg.user_not_in_voice_channel(interaction) return None # Check if we have permission to join the voice channel permissions = interaction.user.voice.channel.permissions_for(interaction.guild.me) if not permissions.connect or not permissions.speak: logger.error("Missing permissions to connect or speak in voice channel") await ui.ErrMsg.msg(interaction, "I don't have permission to join or speak in your voice channel.") return None # Add a small delay before connecting to avoid potential race conditions await asyncio.sleep(1) # Connect with timeout and retry logic try: voice_client = await interaction.user.voice.channel.connect(timeout=10.0, reconnect=True) logger.info(f"Successfully connected to voice channel {interaction.user.voice.channel.id}") except asyncio.TimeoutError: logger.error("Timeout while connecting to voice channel") await ui.ErrMsg.msg(interaction, "Timed out while trying to connect to voice channel. Please try again.") return None except discord.ClientException as e: logger.error(f"Client exception when connecting to voice: {e}") await ui.ErrMsg.msg(interaction, f"Error connecting to voice channel: {e}") return None except AttributeError as e: logger.error(f"Attribute error when connecting to voice: {e}") await ui.ErrMsg.cannot_connect_to_voice_channel(interaction) except Exception as e: logger.error(f"Unexpected error when connecting to voice: {e}") await ui.ErrMsg.msg(interaction, f"An unexpected error occurred while connecting to voice: {e}") return voice_client @app_commands.command(name="play", description="Plays a specified track, album or playlist") @app_commands.describe(querytype="Whether what you're searching is a track, album or playlist", query="Enter a search query") @app_commands.choices(querytype=[ app_commands.Choice(name="Track", value="track"), app_commands.Choice(name="Album", value="album"), app_commands.Choice(name="Playlist", value="playlist"), ]) async def play(self, interaction: discord.Interaction, querytype: str=None, query: str=None) -> None: ''' Play a track matching the given title/artist query ''' # Check if user is in voice channel if interaction.user.voice is None: return await ui.ErrMsg.user_not_in_voice_channel(interaction) # Get a valid voice channel connection voice_client = await self.get_voice_client(interaction, should_connect=True) # Don't attempt playback if the bot is already playing if voice_client.is_playing() and query is None: return await ui.ErrMsg.already_playing(interaction) # Get the guild's player player = data.guild_data(interaction.guild_id).player # Check queue if no query is provided if query is None: # Display error if queue is empty & autoplay is disabled if player.queue == [] and data.guild_properties(interaction.guild_id).autoplay_mode == data.AutoplayMode.NONE: return await ui.ErrMsg.queue_is_empty(interaction) # Begin playback of queue await ui.SysMsg.starting_queue_playback(interaction) await player.play_audio_queue(interaction, voice_client) return # Check querytype is not blank if querytype is None: return await ui.ErrMsg.msg(interaction, "Please provide a query type.") # Check if the query is a track if querytype == "track": # Send our query to the subsonic API and retrieve a list of 1 song songs = await subsonic.search(query, artist_count=0, album_count=0, song_count=1) if songs == "Error": await ui.ErrMsg.msg(interaction, f"An api error has occurred and has been logged to console. Please contact an administrator.") return # Display an error if the query returned no results if len(songs) == 0: await ui.ErrMsg.msg(interaction, f"No track found for **{query}**.") return # Add the first result to the queue and handle queue playback player.queue.append(songs[0]) await ui.SysMsg.added_to_queue(interaction, songs[0]) elif querytype == "album": # Send query to subsonic API and retrieve a list of 1 album album = await subsonic.search_album(query) if album == None: await ui.ErrMsg.msg(interaction, f"No album found for **{query}**.") return # Add all songs from the album to the queue for song in album.songs: player.queue.append(song) await ui.SysMsg.added_album_to_queue(interaction, album) elif querytype == "playlist": # Send query to subsonic API and retrieve a list of all playlists playlists = await subsonic.get_user_playlists() if playlists == None: await ui.ErrMsg.msg(interaction, f"No playlists found.") return # Check if the specific playlist exists and get it's contents playlist = None playlist_id = None for playlist in playlists: if playlist["name"] == query: playlist_id = playlist["id"] break if playlist_id == None: await ui.ErrMsg.msg(interaction, f"No playlist found for **{query}**.") return else: playlist = await subsonic.get_playlist(playlist_id) if playlist == None: # If we end up here then the following error message doesn't really cover it... It's more likely an error in this code await ui.ErrMsg.msg(interaction, f"No playlist found for **{query}**.") return # Add all songs from the playlist to the queue for song in playlist.songs: player.queue.append(song) await ui.SysMsg.added_playlist_to_queue(interaction, playlist) await player.play_audio_queue(interaction, voice_client) @play.error async def play_error(self, ctx, error): if isinstance(error, subsonic.APIError): logging.error(f"An API error has occurred playing a track, code {error.code}: {error.message}") await ui.ErrMsg.msg(ctx, "An API error has occurred and has been logged to console. Please contact an administrator.") else: logging.error(f"An error occurred while playing a track: {error}") await ui.ErrMsg.msg(ctx, f"An unknown error has occurred and has been logged to console. Please contact an administrator. {error}") @app_commands.command(name="stop", description="Stop playing the current track") async def stop(self, interaction: discord.Interaction) -> None: ''' Disconnect from the active voice channel ''' player = data.guild_data(interaction.guild_id).player if player.current_song is None: ui.ErrMsg.not_playing(interaction) # Get the voice client instance for the current guild voice_client = await self.get_voice_client(interaction) # Check if our voice client is connected if voice_client is None: await ui.ErrMsg.bot_not_in_voice_channel(interaction) return # Stop playback voice_client.stop() player.current_song = None # Add current song back to the queue if exists player.queue.insert(0, player.current_song) # Reset bot presence to default try: await self.bot.set_default_presence() except Exception as e: logger.error(f"Failed to reset bot presence: {e}") # Display disconnect confirmation await ui.SysMsg.stopping_queue_playback(interaction) @stop.error async def stop_error(self, ctx, error): logging.error(f"An error occurred while stopping playback: {error}") await ui.ErrMsg.msg(ctx, f"An unknown error has occurred and has been logged to console. Please contact an administrator. {error}") @app_commands.command(name="queue", description="View the current queue") async def show_queue(self, interaction: discord.Interaction) -> None: ''' Show the current queue ''' # Get the audio queue for the current guild queue = data.guild_data(interaction.guild_id).player.queue # Create a string to store the output of our queue output = "" # Add currently playing song to output if available if data.guild_data(interaction.guild_id).player.current_song is not None: song = data.guild_data(interaction.guild_id).player.current_song output += f"**Now Playing:**\n{song.title} - *{song.artist}*\n{song.album} ({song.duration_printable})\n\n" # Loop over our queue, adding each song into our output string for i, song in enumerate(queue): strtoadd = f"{i+1}. **{song.title}** - *{song.artist}*\n{song.album} ({song.duration_printable})\n\n" if len(output+strtoadd) < 4083: output += strtoadd else: remaining = len(queue) - i output += f"**And {remaining} more...**" break # Check if our output string is empty & update it accordingly if output == "": output = "Queue is empty!" # Show the user their queue await ui.SysMsg.msg(interaction, "Queue", output) @show_queue.error async def show_queue_error(self, ctx, error): logging.error(f"An error occurred while displaying the queue: {error}") await ui.ErrMsg.msg(ctx, f"An unknown error has occurred and has been logged to console. Please contact an administrator. {error}") @app_commands.command(name="clear", description="Clear the current queue") async def clear_queue(self, interaction: discord.Interaction) -> None: '''Clear the queue''' queue = data.guild_data(interaction.guild_id).player.queue queue.clear() # Let the user know that the queue has been cleared await ui.SysMsg.queue_cleared(interaction) @clear_queue.error async def clear_queue_error(self, ctx, error): logging.error(f"An error occurred while clearing the queue: {error}") await ui.ErrMsg.msg(ctx, f"An unknown error has occurred and has been logged to console. Please contact an administrator. {error}") @app_commands.command(name="skip", description="Skip one or more tracks") @app_commands.describe(count="Optional number of tracks to skip (>= 1)") async def skip(self, interaction: discord.Interaction, count: int=None) -> None: ''' Skip the current track ''' # Get the voice client instance voice_client = await self.get_voice_client(interaction) # Check if the bot is connected to a voice channel if voice_client is None: await ui.ErrMsg.bot_not_in_voice_channel(interaction) return # Validate count if provided if count is not None and count < 1: await ui.ErrMsg.msg(interaction, "Skip count must be at least 1.") return # Check if the bot is playing music if not voice_client.is_playing(): await ui.ErrMsg.not_playing(interaction) return await data.guild_data(interaction.guild_id).player.skip_track(interaction, voice_client, count=count) @skip.error async def skip_error(self, ctx, error): logging.error(f"An error occurred while skipping a track: {error}") await ui.ErrMsg.msg(ctx, f"An unknown error has occurred and has been logged to console. Please contact an administrator. {error}") @app_commands.command(name="autoplay", description="Toggles autoplay") @app_commands.describe(mode="Determines the method to use when autoplaying") @app_commands.choices(mode=[ app_commands.Choice(name="None", value="none"), app_commands.Choice(name="Random", value="random"), app_commands.Choice(name="Similar", value="similar"), ]) async def autoplay(self, interaction: discord.Interaction, mode: app_commands.Choice[str]) -> None: ''' Toggles autoplay ''' logger.debug(f"Autoplay mode: {mode.value}") # Update the autoplay properties match mode.value: case "none": data.guild_properties(interaction.guild_id).autoplay_mode = data.AutoplayMode.NONE case "random": data.guild_properties(interaction.guild_id).autoplay_mode = data.AutoplayMode.RANDOM case "similar": data.guild_properties(interaction.guild_id).autoplay_mode = data.AutoplayMode.SIMILAR # Display message indicating new status of autoplay if mode.value == "none": await ui.SysMsg.msg(interaction, f"Autoplay disabled by {interaction.user.display_name}") else: await ui.SysMsg.msg(interaction, f"Autoplay enabled by {interaction.user.display_name}", f"Autoplay mode: **{mode.name}**") # If the bot is connected to a voice channel and autoplay is enabled, start queue playback voice_client = await self.get_voice_client(interaction) logger.debug(f"Voice client: {voice_client}") if voice_client: logger.debug(f"Is playing: {voice_client.is_playing()}") if voice_client is not None and not voice_client.is_playing(): player = data.guild_data(interaction.guild_id).player logger.debug(f"Queue: {player.queue}") try: logger.debug(f"Current song: {player.current_song.title}") except AttributeError: logger.debug("No current song") logger.debug("Playing audio queue...") await player.play_audio_queue(interaction, voice_client) @autoplay.error async def autoplay_error(self, ctx, error): if isinstance(error, subsonic.APIError): logging.error(f"An API error has occurred while toggling autoplay, code {error.code}: {error.message}") await ui.ErrMsg.msg(ctx, "An API error has occurred and has been logged to console. Please contact an administrator.") else: logging.error(f"An error occurred while toggling autoplay: {error}") await ui.ErrMsg.msg(ctx, f"An unknown error has occurred and has been logged to console. Please contact an administrator. {error}") @app_commands.command(name="shuffle", description="Shuffles the current queue") async def shuffle(self, interaction: discord.Interaction): ''' Randomize current queue using Fisher-Yates algorithm ''' temporaryqueue = copy.deepcopy(data.guild_data(interaction.guild_id).player.queue) shuffledqueue = [] while len(temporaryqueue) > 0: randomindex = randint(0, len(temporaryqueue) - 1) shuffledqueue.append(temporaryqueue.pop(randomindex)) data.guild_data(interaction.guild_id).player.queue = shuffledqueue await ui.SysMsg.msg(interaction, "Queue shuffled!") @shuffle.error async def shuffle_error(self, ctx, error): logging.error(f"An error occurred while shuffling the queue: {error}") await ui.ErrMsg.msg(ctx, f"An unknown error has occurred and has been logged to console. Please contact an administrator. {error}") @app_commands.command(name="disco", description="Plays the artist's entire discography") @app_commands.describe(artist="The artist to play") async def disco(self, interaction: discord.Interaction, artist: str): ''' Play the artist's entire discography''' # Get a valid voice channel connection voice_client = await self.get_voice_client(interaction, should_connect=True) # Get the guild's player player = data.guild_data(interaction.guild_id).player # Send our query to the subsonic API and retrieve list of albums in artist's discography albums = await subsonic.get_artist_discography(artist) if albums == None: await ui.ErrMsg.msg(interaction, f"No discography found for **{artist}**.") return # Add all songs from the artist's discography to the queue for album in albums: for song in album.songs: player.queue.append(song) # Display a message that discography was added to the queue await ui.SysMsg.added_discography_to_queue(interaction, artist, albums) # Begin playback of queue await player.play_audio_queue(interaction, voice_client) @disco.error async def disco_error(self, ctx, error): if isinstance(error, subsonic.APIError): logging.error(f"An API error has occurred while playing an artist's discography, code {error.code}: {error.message}") await ui.ErrMsg.msg(ctx, "An API error has occurred and has been logged to console. Please contact an administrator.") else: logging.error(f"An error occurred while playing an artist's discography: {error}") await ui.ErrMsg.msg(ctx, f"An unknown error has occurred and has been logged to console. Please contact an administrator. {error}") @app_commands.command(name="playlists", description="List all playlists") async def list_playlists(self, interaction): # Send query to subsonic API and retrieve a list of all playlists playlists = await subsonic.get_user_playlists() if playlists == None: await ui.ErrMsg.msg(interaction, f"No playlists found.") return # Create a string to store the output output = "" # Loop over the list of playlists, adding each one into our output string for i, playlist in enumerate(playlists): strtoadd = f"{i+1}. **{playlist['name']}** \n{playlist['songCount']} songs - {(playlist['duration'] // 60):02d}m {(playlist['duration'] % 60):02d}s\n\n" if len(output+strtoadd) < 4083: output += strtoadd else: remaining = len(playlists) - i output += f"**And {remaining} more...**" break # Check if our output string is empty & update it accordingly if output == "": output = "No playlists found." # Show the user their queue await ui.SysMsg.msg(interaction, "Available playlists", output) @list_playlists.error async def playlists_error(self, ctx, error): if isinstance(error, subsonic.APIError): logging.error(f"An API error has occurred while fetching playlists, code {error.code}: {error.message}") await ui.ErrMsg.msg(ctx, "An API error has occurred and has been logged to console. Please contact an administrator.") else: logging.error(f"An error occurred while fetching playlists: {error}") await ui.ErrMsg.msg(ctx, f"An unknown error has occurred and has been logged to console. Please contact an administrator. {error}") @commands.Cog.listener() async def on_voice_state_update(self, member: discord.Member, before: discord.VoiceState, after: discord.VoiceState) -> None: ''' Event called when a user's voice state changes ''' # Check if the bot is connected to a voice channel voice_client = discord.utils.get(self.bot.voice_clients, guild=member.guild) # Check if the bot is connected to a voice channel if voice_client is None: return # Check if the bot is alone in the voice channel if len(voice_client.channel.members) == 1: logger.debug("Bot is alone in voice channel, waiting 10 seconds before disconnecting...") # Wait for 10 seconds await asyncio.sleep(10) # Check again if there are still no users in the voice channel if len(voice_client.channel.members) == 1: # Disconnect the bot and clear the queue await voice_client.disconnect() player = data.guild_data(member.guild.id).player player.queue.clear() player.current_song = None # Reset bot presence to default try: await self.bot.set_default_presence() except Exception as e: logger.error(f"Failed to reset bot presence: {e}") logger.info("The bot has disconnected and cleared the queue as there are no users in the voice channel.") else: logger.debug("Bot is no longer alone in voice channel, aborting disconnect...") async def setup(bot: DiscodromeClient): ''' Setup function for the music.py cog ''' await bot.add_cog(MusicCog(bot))