254 lines
12 KiB
Python
254 lines
12 KiB
Python
import discord
|
|
import logging
|
|
import asyncio
|
|
|
|
from subsonic import Song, Album, Playlist, get_album_art_file
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
class SysMsg:
|
|
''' A class for sending system messages '''
|
|
|
|
@staticmethod
|
|
async def msg(interaction: discord.Interaction, header: str, message: str=None, thumbnail: str=None, *, ephemeral: bool=False) -> None:
|
|
''' Generic message function. Creates a message formatted as an embed '''
|
|
|
|
# Check if interaction is still valid
|
|
if interaction is None or interaction.guild is None:
|
|
logger.warning("Cannot send message: interaction is no longer valid")
|
|
return
|
|
|
|
# Handle message over character limit
|
|
if message is not None and len(message) > 4096:
|
|
message = message[:4093] + "..."
|
|
|
|
embed = discord.Embed(color=discord.Color(0x50C470), title=header, description=message)
|
|
file = discord.utils.MISSING
|
|
|
|
# Attach a thumbnail if one was provided (as a local file)
|
|
if thumbnail is not None:
|
|
try:
|
|
file = discord.File(thumbnail, filename="image.png")
|
|
embed.set_thumbnail(url="attachment://image.png")
|
|
except Exception as e:
|
|
logger.error(f"Failed to attach thumbnail: {e}")
|
|
# Continue without the thumbnail
|
|
|
|
# Attempt to send the message, up to 3 times
|
|
attempt = 0
|
|
while attempt < 3:
|
|
try:
|
|
# Check if the interaction response is already done
|
|
if interaction.response.is_done():
|
|
# Use followup
|
|
await interaction.followup.send(file=file, embed=embed, ephemeral=ephemeral)
|
|
return
|
|
else:
|
|
# Use initial response
|
|
await interaction.response.send_message(file=file, embed=embed, ephemeral=ephemeral)
|
|
return
|
|
except discord.NotFound:
|
|
logger.warning("Attempt %d at sending a system message failed (NotFound)...", attempt+1)
|
|
attempt += 1
|
|
# Short delay before retrying
|
|
await asyncio.sleep(0.5)
|
|
except discord.HTTPException as e:
|
|
logger.warning("Attempt %d at sending a system message failed (HTTPException: %s)...", attempt+1, e)
|
|
attempt += 1
|
|
await asyncio.sleep(0.5)
|
|
except Exception as e:
|
|
logger.error("Unexpected error when sending system message: %s", e)
|
|
attempt += 1
|
|
await asyncio.sleep(0.5)
|
|
|
|
# If we've exhausted all attempts, log a more detailed error
|
|
logger.error("Failed to send system message after %d attempts. Header: %s", attempt, header)
|
|
|
|
|
|
@staticmethod
|
|
async def now_playing(interaction: discord.Interaction, song: Song) -> None:
|
|
''' Sends a message containing the currently playing song '''
|
|
cover_art = await get_album_art_file(song.cover_id)
|
|
desc = f"**{song.title}** - *{song.artist}*\n{song.album} ({song.duration_printable})"
|
|
await __class__.msg(interaction, "Now Playing:", desc, cover_art)
|
|
|
|
@staticmethod
|
|
async def playback_ended(interaction: discord.Interaction) -> None:
|
|
''' Sends a message indicating playback has ended '''
|
|
await __class__.msg(interaction, "Playback ended")
|
|
|
|
@staticmethod
|
|
async def disconnected(interaction: discord.Interaction) -> None:
|
|
''' Sends a message indicating the bot disconnected from voice channel '''
|
|
await __class__.msg(interaction, "Disconnected from voice channel")
|
|
|
|
@staticmethod
|
|
async def starting_queue_playback(interaction: discord.Interaction) -> None:
|
|
''' Sends a message indicating queue playback has started '''
|
|
await __class__.msg(interaction, "Started queue playback")
|
|
|
|
@staticmethod
|
|
async def stopping_queue_playback(interaction: discord.Interaction) -> None:
|
|
''' Sends a message indicating queue playback has stopped '''
|
|
await __class__.msg(interaction, "Stopped queue playback")
|
|
|
|
@staticmethod
|
|
async def added_to_queue(interaction: discord.Interaction, song: Song) -> None:
|
|
''' Sends a message indicating the selected song was added to queue '''
|
|
desc = f"**{song.title}** - *{song.artist}*\n{song.album} ({song.duration_printable})"
|
|
cover_art = await get_album_art_file(song.cover_id)
|
|
await __class__.msg(interaction, f"{interaction.user.display_name} added track to queue", desc, cover_art)
|
|
|
|
@staticmethod
|
|
async def added_album_to_queue(interaction: discord.Interaction, album: Album) -> None:
|
|
''' Sends a message indicating the selected album was added to queue '''
|
|
desc = f"**{album.name}** - *{album.artist}*\n{album.song_count} songs ({album.duration} seconds)"
|
|
cover_art = await get_album_art_file(album.cover_id)
|
|
await __class__.msg(interaction, f"{interaction.user.display_name} added album to queue", desc, cover_art)
|
|
|
|
@staticmethod
|
|
async def added_playlist_to_queue(interaction: discord.Interaction, playlist: Playlist) -> None:
|
|
''' Sends a message indicating the selected playlist was added to queue '''
|
|
desc = f"**{playlist.name}**\n{playlist.song_count} songs ({playlist.duration} seconds)"
|
|
cover_art = await get_album_art_file(playlist.cover_id)
|
|
await __class__.msg(interaction, f"{interaction.user.display_name} added playlist to queue", desc, cover_art)
|
|
|
|
@staticmethod
|
|
async def added_discography_to_queue(interaction: discord.Interaction, artist: str, albums: list[Album]) -> None:
|
|
''' Sends a message indicating the selected artist's discography was added to queue '''
|
|
desc = f"**{artist}**\n{len(albums)} albums\n\n"
|
|
cover_art = await get_album_art_file(albums[0].cover_id)
|
|
for counter in range(len(albums)):
|
|
album = albums[counter]
|
|
desc += f"**{str(counter+1)}. {album.name}**\n{album.song_count} songs ({album.duration} seconds)\n\n"
|
|
await __class__.msg(interaction, f"{interaction.user.display_name} added discography to queue", desc, cover_art)
|
|
|
|
@staticmethod
|
|
async def queue_cleared(interaction: discord.Interaction) -> None:
|
|
''' Sends a message indicating a user cleared the queue '''
|
|
await __class__.msg(interaction, f"{interaction.user.display_name} cleared the queue")
|
|
|
|
@staticmethod
|
|
async def skipping(interaction: discord.Interaction) -> None:
|
|
''' Sends a message indicating the current song was skipped '''
|
|
await __class__.msg(interaction, "Skipped track", ephemeral=True)
|
|
|
|
|
|
class ErrMsg:
|
|
''' A class for sending error messages '''
|
|
|
|
@staticmethod
|
|
async def msg(interaction: discord.Interaction, message: str) -> None:
|
|
''' Generic message function. Creates an error message formatted as an embed '''
|
|
|
|
# Check if interaction is still valid
|
|
if interaction is None or interaction.guild is None:
|
|
logger.warning("Cannot send error message: interaction is no longer valid")
|
|
return
|
|
|
|
embed = discord.Embed(color=discord.Color(0x50C470), title="Error", description=message)
|
|
|
|
# Attempt to send the error message, up to 3 times
|
|
attempt = 0
|
|
while attempt < 3:
|
|
try:
|
|
if interaction.response.is_done():
|
|
await interaction.followup.send(embed=embed, ephemeral=True)
|
|
return
|
|
else:
|
|
await interaction.response.send_message(embed=embed, ephemeral=True)
|
|
return
|
|
except discord.NotFound:
|
|
logger.warning("Attempt %d at sending an error message failed (NotFound)...", attempt+1)
|
|
attempt += 1
|
|
# Short delay before retrying
|
|
await asyncio.sleep(0.5)
|
|
except discord.HTTPException as e:
|
|
logger.warning("Attempt %d at sending an error message failed (HTTPException: %s)...", attempt+1, e)
|
|
attempt += 1
|
|
await asyncio.sleep(0.5)
|
|
except Exception as e:
|
|
logger.error("Unexpected error when sending error message: %s", e)
|
|
attempt += 1
|
|
await asyncio.sleep(0.5)
|
|
|
|
# If we've exhausted all attempts, log a more detailed error
|
|
logger.error("Failed to send error message after %d attempts. Message: %s", attempt, message)
|
|
|
|
@staticmethod
|
|
async def user_not_in_voice_channel(interaction: discord.Interaction) -> None:
|
|
''' Sends an error message indicating user is not in a voice channel '''
|
|
await __class__.msg(interaction, "You are not connected to a voice channel.")
|
|
|
|
@staticmethod
|
|
async def bot_not_in_voice_channel(interaction: discord.Interaction) -> None:
|
|
''' Sends an error message indicating bot is connect to a voice channel '''
|
|
await __class__.msg(interaction, "Not currently connected to a voice channel.")
|
|
|
|
@staticmethod
|
|
async def cannot_connect_to_voice_channel(interaction: discord.Interaction) -> None:
|
|
''' Sends an error message indicating bot is unable to connect to a voice channel '''
|
|
await __class__.msg(interaction, "Cannot connect to voice channel.")
|
|
|
|
@staticmethod
|
|
async def queue_is_empty(interaction: discord.Interaction) -> None:
|
|
''' Sends an error message indicating the queue is empty '''
|
|
await __class__.msg(interaction, "Queue is empty.")
|
|
|
|
@staticmethod
|
|
async def already_playing(interaction: discord.Interaction) -> None:
|
|
''' Sends an error message indicating that music is already playing '''
|
|
await __class__.msg(interaction, "Already playing.")
|
|
|
|
@staticmethod
|
|
async def not_playing(interaction: discord.Interaction) -> None:
|
|
''' Sends an error message indicating nothing is playing '''
|
|
await __class__.msg(interaction, "No track is playing.")
|
|
|
|
|
|
|
|
# Methods for parsing data to Discord structures
|
|
def parse_search_as_track_selection_embed(results: list[Song], query: str, page_num: int) -> discord.Embed:
|
|
''' Takes search results obtained from the Subsonic API and parses them into a Discord embed suitable for track selection '''
|
|
|
|
options_str = ""
|
|
|
|
# Loop over the provided search results
|
|
for song in results:
|
|
|
|
# Trim displayed tags to fit neatly within the embed
|
|
tr_title = song.title
|
|
tr_artist = song.artist
|
|
tr_album = (song.album[:68] + "...") if len(song.album) > 68 else song.album
|
|
|
|
# Only trim the longest tag on the first line
|
|
top_str_length = len(song.title + " - " + song.artist)
|
|
if top_str_length > 71:
|
|
|
|
if tr_title > tr_artist:
|
|
tr_title = song.title[:(68 - top_str_length)] + '...'
|
|
else:
|
|
tr_artist = song.artist[:(68 - top_str_length)] + '...'
|
|
|
|
# Add each of the results to our output string
|
|
options_str += f"**{tr_title}** - *{tr_artist}* \n*{tr_album}* ({song.duration_printable})\n\n"
|
|
|
|
# Add the current page number to our results
|
|
options_str += f"Current page: {page_num}"
|
|
|
|
# Return an embed that displays our output string
|
|
return discord.Embed(color=discord.Color.orange(), title=f"Results for: {query}", description=options_str)
|
|
|
|
|
|
def parse_search_as_track_selection_options(results: list[Song]) -> list[discord.SelectOption]:
|
|
''' Takes search results obtained from the Subsonic API and parses them into a Discord selection list for tracks '''
|
|
|
|
select_options = []
|
|
for i, song in enumerate(results):
|
|
select_option = discord.SelectOption(label=f"{song.title}", description=f"by {song.artist}", value=i)
|
|
select_options.append(select_option)
|
|
|
|
return select_options
|