deadcxap/YandexMusicDiscordBot
1
0
Fork 0
mirror of https://github.com/deadcxap/YandexMusicDiscordBot.git synced 2026-01-10 02:51:45 +03:00
Code Issues Packages Projects Releases Wiki Activity

impr: Add more error handling.

Browse Source
This commit is contained in:
Lemon4ksan
2025-02-27 22:42:19 +03:00
parent 48fe0b894b
commit 3c1b0ec266
8 changed files with 204 additions and 237 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
MusicBot/cogs/general.py
View File

@@ -18,7 +18,7 @@ def setup(bot):
bot.add_cog(General(bot)) bot.add_cog(General(bot))
async def get_search_suggestions(ctx: discord.AutocompleteContext) -> list[str]: async def get_search_suggestions(ctx: discord.AutocompleteContext) -> list[str]:
if not ctx.interaction.user or not ctx.value or len(ctx.value) < 2: if not ctx.interaction.user or not ctx.value or not (100 > len(ctx.value) > 2):
return [] return []
uid = ctx.interaction.user.id uid = ctx.interaction.user.id
@@ -41,6 +41,10 @@ async def get_search_suggestions(ctx: discord.AutocompleteContext) -> list[str]:
logging.debug(f"[GENERAL] Searching for '{ctx.value}' for user {uid}") logging.debug(f"[GENERAL] Searching for '{ctx.value}' for user {uid}")
if content_type not in ('Трек', 'Альбом', 'Артист', 'Плейлист'):
logging.error(f"[GENERAL] Invalid content type '{content_type}' for user {uid}")
return []
if content_type == 'Трек' and search.tracks is not None: if content_type == 'Трек' and search.tracks is not None:
res = [f"{item.title} {f"({item.version})" if item.version else ''} - {", ".join(item.artists_name())}" for item in search.tracks.results] res = [f"{item.title} {f"({item.version})" if item.version else ''} - {", ".join(item.artists_name())}" for item in search.tracks.results]
elif content_type == 'Альбом' and search.albums is not None: elif content_type == 'Альбом' and search.albums is not None:
@@ -50,13 +54,13 @@ async def get_search_suggestions(ctx: discord.AutocompleteContext) -> list[str]:
elif content_type == 'Плейлист' and search.playlists is not None: elif content_type == 'Плейлист' and search.playlists is not None:
res = [f"{item.title}" for item in search.playlists.results] res = [f"{item.title}" for item in search.playlists.results]
else: else:
logging.warning(f"[GENERAL] Invalid content type '{content_type}' for user {uid}") logging.info(f"[GENERAL] Failed to get content type '{content_type}' with name '{ctx.value}' for user {uid}")
return [] return []
return res[:100] return res[:100]
async def get_user_playlists_suggestions(ctx: discord.AutocompleteContext) -> list[str]: async def get_user_playlists_suggestions(ctx: discord.AutocompleteContext) -> list[str]:
if not ctx.interaction.user or not ctx.value or len(ctx.value) < 2: if not ctx.interaction.user or not ctx.value or not (100 > len(ctx.value) > 2):
return [] return []
uid = ctx.interaction.user.id uid = ctx.interaction.user.id
@@ -70,21 +74,25 @@ async def get_user_playlists_suggestions(ctx: discord.AutocompleteContext) -> li
except UnauthorizedError: except UnauthorizedError:
logging.info(f"[GENERAL] User {uid} provided invalid token") logging.info(f"[GENERAL] User {uid} provided invalid token")
return [] return []
logging.debug(f"[GENERAL] Searching for '{ctx.value}' for user {uid}")
playlists_list = await client.users_playlists_list() logging.debug(f"[GENERAL] Searching for '{ctx.value}' for user {uid}")
try:
playlists_list = await client.users_playlists_list()
except Exception as e:
logging.error(f"[GENERAL] Failed to get playlists for user {uid}: {e}")
return []
return [playlist.title for playlist in playlists_list if playlist.title and ctx.value in playlist.title][:100] return [playlist.title for playlist in playlists_list if playlist.title and ctx.value in playlist.title][:100]
class General(Cog): class General(Cog):
def __init__(self, bot: discord.Bot): def __init__(self, bot: discord.Bot):
self.bot = bot self.bot = bot
self.db = BaseGuildsDatabase() self.db = BaseGuildsDatabase()
self.users_db = users_db self.users_db = users_db
account = discord.SlashCommandGroup("account", "Команды, связанные с аккаунтом.") account = discord.SlashCommandGroup("account", "Команды, связанные с аккаунтом.")
@discord.slash_command(description="Получить информацию о командах YandexMusic.") @discord.slash_command(description="Получить информацию о командах YandexMusic.")
@discord.option( @discord.option(
"command", "command",
@@ -208,10 +216,11 @@ class General(Cog):
await ctx.respond("❌ Укажите токен через /account login.", delete_after=15, ephemeral=True) await ctx.respond("❌ Укажите токен через /account login.", delete_after=15, ephemeral=True)
return return
client = await YMClient(token).init() try:
if not client.me or not client.me.account or not client.me.account.uid: client = await YMClient(token).init()
logging.warning(f"Failed to fetch user info for user {ctx.user.id}") except UnauthorizedError:
await ctx.respond('❌ Что-то пошло не так. Повторите попытку позже.', delete_after=15, ephemeral=True) logging.info(f"[GENERAL] Invalid token for user {ctx.user.id}")
await ctx.respond('❌ Неверный токен. Укажите новый через /account login.', delete_after=15, ephemeral=True)
return return
likes = await client.users_likes_tracks() likes = await client.users_likes_tracks()
@@ -256,7 +265,7 @@ class General(Cog):
client = await YMClient(token).init() client = await YMClient(token).init()
except UnauthorizedError: except UnauthorizedError:
logging.info(f"[GENERAL] User {ctx.user.id} provided invalid token") logging.info(f"[GENERAL] User {ctx.user.id} provided invalid token")
await ctx.respond("❌ Недействительный токен. Если это не так, попробуйте ещё раз.", delete_after=15, ephemeral=True) await ctx.respond('❌ Неверный токен. Укажите новый через /account login.', delete_after=15, ephemeral=True)
return return
search = await client.search(content_type, type_='playlist') search = await client.search(content_type, type_='playlist')
@@ -287,7 +296,7 @@ class General(Cog):
autocomplete=discord.utils.basic_autocomplete(get_user_playlists_suggestions) autocomplete=discord.utils.basic_autocomplete(get_user_playlists_suggestions)
) )
async def playlist(self, ctx: discord.ApplicationContext, name: str) -> None: async def playlist(self, ctx: discord.ApplicationContext, name: str) -> None:
logging.info(f"[GENERAL] Playlists command invoked by user {ctx.user.id} in guild {ctx.guild_id}") logging.info(f"[GENERAL] Playlist command invoked by user {ctx.user.id} in guild {ctx.guild_id}")
token = await self.users_db.get_ym_token(ctx.user.id) token = await self.users_db.get_ym_token(ctx.user.id)
if not token: if not token:
@@ -299,10 +308,15 @@ class General(Cog):
client = await YMClient(token).init() client = await YMClient(token).init()
except UnauthorizedError: except UnauthorizedError:
logging.info(f"[GENERAL] User {ctx.user.id} provided invalid token") logging.info(f"[GENERAL] User {ctx.user.id} provided invalid token")
await ctx.respond("❌ Недействительный токен. Если это не так, попробуйте ещё раз.", delete_after=15, ephemeral=True) await ctx.respond('❌ Неверный токен. Укажите новый через /account login.', delete_after=15, ephemeral=True)
return return
playlists = await client.users_playlists_list() try:
playlists = await client.users_playlists_list()
except UnauthorizedError:
logging.warning(f"[GENERAL] Unknown token error for user {ctx.user.id}")
await ctx.respond("❌ Произошла неизвестная ошибка при попытке получения плейлистов. Пожалуйста, сообщите об этом разработчику.", delete_after=15, ephemeral=True)
return
playlist = next((playlist for playlist in playlists if playlist.title == name), None) playlist = next((playlist for playlist in playlists if playlist.title == name), None)
if not playlist: if not playlist:

275
MusicBot/cogs/utils/embeds.py
View File

@@ -1,5 +1,6 @@
import logging import logging
from typing import cast from functools import lru_cache
from typing import cast, Final
from math import ceil from math import ceil
from os import getenv from os import getenv
@@ -10,29 +11,35 @@ from PIL import Image
from yandex_music import Track, Album, Artist, Playlist, Label from yandex_music import Track, Album, Artist, Playlist, Label
from discord import Embed from discord import Embed
explicit_eid: Final[str | None] = getenv('EXPLICIT_EID')
if not explicit_eid:
raise ValueError('You must specify explicit emoji id in your enviroment (EXPLICIT_EID).')
async def generate_item_embed(item: Track | Album | Artist | Playlist | list[Track], vibing: bool = False) -> Embed: async def generate_item_embed(item: Track | Album | Artist | Playlist | list[Track], vibing: bool = False) -> Embed:
"""Generate item embed. list[Track] is used for likes. If vibing is True, add vibing image. """Generate item embed. list[Track] is used for likes. If vibing is True, add vibing image.
Args: Args:
item (yandex_music.Track | yandex_music.Album | yandex_music.Artist | yandex_music.Playlist): Item to be processed. item (Track | Album | Artist | Playlist | list[Track]): Item to be processed.
vibing (bool, optional): Add vibing image. Defaults to False.
Returns: Returns:
discord.Embed: Item embed. discord.Embed: Item embed.
""" """
logging.debug(f"[EMBEDS] Generating embed for type: '{type(item).__name__}'") logging.debug(f"[EMBEDS] Generating embed for type: '{type(item).__name__}'")
if isinstance(item, Track): match item:
embed = await _generate_track_embed(item) case Track():
elif isinstance(item, Album): embed = await _generate_track_embed(item)
embed = await _generate_album_embed(item) case Album():
elif isinstance(item, Artist): embed = await _generate_album_embed(item)
embed = await _generate_artist_embed(item) case Artist():
elif isinstance(item, Playlist): embed = await _generate_artist_embed(item)
embed = await _generate_playlist_embed(item) case Playlist():
elif isinstance(item, list): embed = await _generate_playlist_embed(item)
embed = _generate_likes_embed(item) case list():
else: embed = _generate_likes_embed(item)
raise ValueError(f"Unknown item type: {type(item).__name__}") case _:
raise ValueError(f"Unknown item type: {type(item).__name__}")
if vibing: if vibing:
embed.set_image( embed.set_image(
@@ -41,13 +48,12 @@ async def generate_item_embed(item: Track | Album | Artist | Playlist | list[Tra
return embed return embed
def _generate_likes_embed(tracks: list[Track]) -> Embed: def _generate_likes_embed(tracks: list[Track]) -> Embed:
track_count = len(tracks)
cover_url = "https://avatars.yandex.net/get-music-user-playlist/11418140/favorit-playlist-cover.bb48fdb9b9f4/300x300" cover_url = "https://avatars.yandex.net/get-music-user-playlist/11418140/favorit-playlist-cover.bb48fdb9b9f4/300x300"
embed = Embed( embed = Embed(
title="Мне нравится", title="Мне нравится",
description="Треки, которые вам понравились.", description="Треки, которые вам понравились.",
color=0xce3a26, color=0xce3a26
) )
embed.set_thumbnail(url=cover_url) embed.set_thumbnail(url=cover_url)
@@ -56,203 +62,143 @@ def _generate_likes_embed(tracks: list[Track]) -> Embed:
if track.duration_ms: if track.duration_ms:
duration += track.duration_ms duration += track.duration_ms
duration_m = duration // 60000 embed.add_field(name="Длительность", value=_format_duration(duration))
duration_s = ceil(duration / 1000) - duration_m * 60 embed.add_field(name="Треки", value=str(len(tracks)))
if duration_s == 60:
duration_m += 1
duration_s = 0
embed.add_field(name="Длительность", value=f"{duration_m}:{duration_s:02}")
if track_count is not None:
embed.add_field(name="Треки", value=str(track_count))
return embed return embed
async def _generate_track_embed(track: Track) -> Embed: async def _generate_track_embed(track: Track) -> Embed:
title = cast(str, track.title) title = track.title
avail = cast(bool, track.available)
artists = track.artists_name()
albums = [cast(str, album.title) for album in track.albums] albums = [cast(str, album.title) for album in track.albums]
lyrics = cast(bool, track.lyrics_available)
duration = cast(int, track.duration_ms)
explicit = track.explicit or track.content_warning explicit = track.explicit or track.content_warning
bg_video = track.background_video_uri year = track.albums[0].year if track.albums else None
metadata = track.meta_data artist = track.artists[0] if track.artists else None
year = track.albums[0].year
artist = track.artists[0]
cover_url = track.get_cover_url('400x400') if track.cover_uri:
color = await _get_average_color_from_url(cover_url) cover_url = track.get_cover_url('400x400')
color = await _get_average_color_from_url(cover_url)
else:
cover_url = None
color = 0x000
if explicit: if explicit and title:
explicit_eid = getenv('EXPLICIT_EID')
if not explicit_eid:
raise ValueError('You must specify explicit emoji id in your enviroment (EXPLICIT_EID).')
title += ' <:explicit:' + explicit_eid + '>' title += ' <:explicit:' + explicit_eid + '>'
duration_m = duration // 60000 if artist:
duration_s = ceil(duration / 1000) - duration_m * 60 artist_url = f"https://music.yandex.ru/artist/{artist.id}"
if duration_s == 60: artist_cover = artist.cover
duration_m += 1
duration_s = 0
artist_url = f"https://music.yandex.ru/artist/{artist.id}" if not artist_cover and artist.op_image:
artist_cover = artist.cover artist_cover_url = artist.get_op_image_url()
elif artist_cover:
if not artist_cover and artist.op_image: artist_cover_url = artist_cover.get_url()
artist_cover_url = artist.get_op_image_url() else:
elif artist_cover: artist_cover_url = None
artist_cover_url = artist_cover.get_url()
else: else:
artist_url = None
artist_cover_url = None artist_cover_url = None
embed = Embed( embed = Embed(
title=title, title=title,
description=", ".join(albums), description=", ".join(albums),
color=color, color=color
) )
embed.set_thumbnail(url=cover_url) embed.set_thumbnail(url=cover_url)
embed.set_author(name=", ".join(artists), url=artist_url, icon_url=artist_cover_url) embed.set_author(name=", ".join(track.artists_name()), url=artist_url, icon_url=artist_cover_url)
embed.add_field(name="Текст песни", value="Есть" if lyrics else "Нет") embed.add_field(name="Текст песни", value="Есть" if track.lyrics_available else "Нет")
embed.add_field(name="Длительность", value=f"{duration_m}:{duration_s:02}")
if isinstance(track.duration_ms, int):
embed.add_field(name="Длительность", value=_format_duration(track.duration_ms))
if year: if year:
embed.add_field(name="Год выпуска", value=str(year)) embed.add_field(name="Год выпуска", value=str(year))
if metadata: if track.background_video_uri:
if metadata.year: embed.add_field(name="Видеофон", value=f"[Ссылка]({track.background_video_uri})")
embed.add_field(name="Год выхода", value=str(metadata.year))
if metadata.number: if not (track.available or track.available_for_premium_users):
embed.add_field(name="Позиция", value=str(metadata.number))
if metadata.composer:
embed.add_field(name="Композитор", value=metadata.composer)
if metadata.version:
embed.add_field(name="Версия", value=metadata.version)
if bg_video:
embed.add_field(name="Видеофон", value=f"[Ссылка]({bg_video})")
if not avail:
embed.set_footer(text=f"Трек в данный момент недоступен.") embed.set_footer(text=f"Трек в данный момент недоступен.")
return embed return embed
async def _generate_album_embed(album: Album) -> Embed: async def _generate_album_embed(album: Album) -> Embed:
title = cast(str, album.title) title = album.title
track_count = album.track_count
artists = album.artists_name()
avail = cast(bool, album.available)
description = album.short_description
year = album.year
version = album.version
bests = album.bests
duration = album.duration_ms
explicit = album.explicit or album.content_warning explicit = album.explicit or album.content_warning
likes_count = album.likes_count
artist = album.artists[0] artist = album.artists[0]
cover_url = album.get_cover_url('400x400') cover_url = album.get_cover_url('400x400')
color = await _get_average_color_from_url(cover_url)
if isinstance(album.labels[0], Label): if isinstance(album.labels[0], Label):
labels = [cast(Label, label).name for label in album.labels] labels = [cast(Label, label).name for label in album.labels]
else: else:
labels = [cast(str, label) for label in album.labels] labels = [cast(str, label) for label in album.labels]
if version: if album.version and title:
title += f' *{version}*' title += f' *{album.version}*'
if explicit: if explicit and title:
explicit_eid = getenv('EXPLICIT_EID')
if not explicit_eid:
raise ValueError('You must specify explicit emoji id in your enviroment.')
title += ' <:explicit:' + explicit_eid + '>' title += ' <:explicit:' + explicit_eid + '>'
artist_url = f"https://music.yandex.ru/artist/{artist.id}" artist_url = f"https://music.yandex.ru/artist/{artist.id}"
artist_cover = artist.cover artist_cover = artist.cover
if not artist_cover and artist.op_image: if not artist_cover and artist.op_image:
artist_cover_url = artist.get_op_image_url() artist_cover_url = artist.get_op_image_url('400x400')
elif artist_cover: elif artist_cover:
artist_cover_url = artist_cover.get_url() artist_cover_url = artist_cover.get_url(size='400x400')
else: else:
artist_cover_url = None artist_cover_url = None
embed = Embed( embed = Embed(
title=title, title=title,
description=description, description=album.short_description,
color=color, color=await _get_average_color_from_url(cover_url)
) )
embed.set_thumbnail(url=cover_url) embed.set_thumbnail(url=cover_url)
embed.set_author(name=", ".join(artists), url=artist_url, icon_url=artist_cover_url) embed.set_author(name=", ".join(album.artists_name()), url=artist_url, icon_url=artist_cover_url)
if year: if album.year:
embed.add_field(name="Год выпуска", value=str(year)) embed.add_field(name="Год выпуска", value=str(album.year))
if duration: if isinstance(album.duration_ms, int):
duration_m = duration // 60000 embed.add_field(name="Длительность", value=_format_duration(album.duration_ms))
duration_s = ceil(duration / 1000) - duration_m * 60
if duration_s == 60:
duration_m += 1
duration_s = 0
embed.add_field(name="Длительность", value=f"{duration_m}:{duration_s:02}")
if track_count is not None: if album.track_count is not None:
if track_count > 1: embed.add_field(name="Треки", value=str(album.track_count) if album.track_count > 1 else "Сингл")
embed.add_field(name="Треки", value=str(track_count))
else:
embed.add_field(name="Треки", value="Сингл")
if likes_count: if album.likes_count is not None:
embed.add_field(name="Лайки", value=str(likes_count)) embed.add_field(name="Лайки", value=str(album.likes_count))
if len(labels) > 1: embed.add_field(name="Лейблы" if len(labels) > 1 else "Лейбл", value=", ".join(labels))
embed.add_field(name="Лейблы", value=", ".join(labels))
else:
embed.add_field(name="Лейбл", value=", ".join(labels))
if not avail: if not (album.available or album.available_for_premium_users):
embed.set_footer(text=f"Альбом в данный момент недоступен.") embed.set_footer(text=f"Альбом в данный момент недоступен.")
return embed return embed
async def _generate_artist_embed(artist: Artist) -> Embed: async def _generate_artist_embed(artist: Artist) -> Embed:
name = cast(str, artist.name)
likes_count = artist.likes_count
avail = cast(bool, artist.available)
counts = artist.counts
description = artist.description
ratings = artist.ratings
popular_tracks = artist.popular_tracks
if not artist.cover: if not artist.cover:
cover_url = artist.get_op_image_url('400x400') cover_url = artist.get_op_image_url('400x400')
else: else:
cover_url = artist.cover.get_url(size='400x400') cover_url = artist.cover.get_url(size='400x400')
color = await _get_average_color_from_url(cover_url)
embed = Embed( embed = Embed(
title=name, title=artist.name,
description=description.text if description else None, description=artist.description.text if artist.description else None,
color=color, color=await _get_average_color_from_url(cover_url)
) )
embed.set_thumbnail(url=cover_url) embed.set_thumbnail(url=cover_url)
if likes_count: if artist.likes_count:
embed.add_field(name="Лайки", value=str(likes_count)) embed.add_field(name="Лайки", value=str(artist.likes_count))
# if ratings: # if ratings:
# embed.add_field(name="Слушателей за месяц", value=str(ratings.month)) # Wrong numbers? # embed.add_field(name="Слушателей за месяц", value=str(ratings.month)) # Wrong numbers
if counts: if artist.counts:
embed.add_field(name="Треки", value=str(counts.tracks)) embed.add_field(name="Треки", value=str(artist.counts.tracks))
embed.add_field(name="Альбомы", value=str(counts.direct_albums)) embed.add_field(name="Альбомы", value=str(artist.counts.direct_albums))
if artist.genres: if artist.genres:
genres = [genre.capitalize() for genre in artist.genres] genres = [genre.capitalize() for genre in artist.genres]
@@ -261,23 +207,12 @@ async def _generate_artist_embed(artist: Artist) -> Embed:
else: else:
embed.add_field(name="Жанр", value=", ".join(genres)) embed.add_field(name="Жанр", value=", ".join(genres))
if not avail: if not artist.available or artist.reason:
embed.set_footer(text=f"Артист в данный момент недоступен.") embed.set_footer(text=f"Артист в данный момент недоступен.")
return embed return embed
async def _generate_playlist_embed(playlist: Playlist) -> Embed: async def _generate_playlist_embed(playlist: Playlist) -> Embed:
title = cast(str, playlist.title)
track_count = playlist.track_count
avail = cast(bool, playlist.available)
description = playlist.description
year = playlist.created
modified = playlist.modified
duration = playlist.duration_ms
likes_count = playlist.likes_count
cover_url = None
if playlist.cover and playlist.cover.uri: if playlist.cover and playlist.cover.uri:
cover_url = f"https://{playlist.cover.uri.replace('%%', '400x400')}" cover_url = f"https://{playlist.cover.uri.replace('%%', '400x400')}"
else: else:
@@ -287,6 +222,8 @@ async def _generate_playlist_embed(playlist: Playlist) -> Embed:
if track and track.albums and track.albums[0].cover_uri: if track and track.albums and track.albums[0].cover_uri:
cover_url = f"https://{track.albums[0].cover_uri.replace('%%', '400x400')}" cover_url = f"https://{track.albums[0].cover_uri.replace('%%', '400x400')}"
break break
else:
cover_url = None
if cover_url: if cover_url:
color = await _get_average_color_from_url(cover_url) color = await _get_average_color_from_url(cover_url)
@@ -294,37 +231,33 @@ async def _generate_playlist_embed(playlist: Playlist) -> Embed:
color = 0x000 color = 0x000
embed = Embed( embed = Embed(
title=title, title=playlist.title,
description=description, description=playlist.description,
color=color, color=color
) )
embed.set_thumbnail(url=cover_url) embed.set_thumbnail(url=cover_url)
if year: if playlist.created:
embed.add_field(name="Год создания", value=str(year).split('-')[0]) embed.add_field(name="Год создания", value=str(playlist.created).split('-')[0])
if modified: if playlist.modified:
embed.add_field(name="Изменён", value=str(modified).split('-')[0]) embed.add_field(name="Изменён", value=str(playlist.modified).split('-')[0])
if duration: if playlist.duration_ms:
duration_m = duration // 60000 embed.add_field(name="Длительность", value=_format_duration(playlist.duration_ms))
duration_s = ceil(duration / 1000) - duration_m * 60
if duration_s == 60:
duration_m += 1
duration_s = 0
embed.add_field(name="Длительность", value=f"{duration_m}:{duration_s:02}")
if track_count is not None: if playlist.track_count is not None:
embed.add_field(name="Треки", value=str(track_count)) embed.add_field(name="Треки", value=str(playlist.track_count))
if likes_count: if playlist.likes_count:
embed.add_field(name="Лайки", value=str(likes_count)) embed.add_field(name="Лайки", value=str(playlist.likes_count))
if not avail: if not playlist.available:
embed.set_footer(text=f"Плейлист в данный момент недоступен.") embed.set_footer(text=f"Плейлист в данный момент недоступен.")
return embed return embed
@lru_cache()
async def _get_average_color_from_url(url: str) -> int: async def _get_average_color_from_url(url: str) -> int:
"""Get image from url and calculate its average color to use in embeds. """Get image from url and calculate its average color to use in embeds.
@@ -358,5 +291,13 @@ async def _get_average_color_from_url(url: str) -> int:
b = b_total // count b = b_total // count
return (r << 16) + (g << 8) + b return (r << 16) + (g << 8) + b
except Exception: except (aiohttp.ClientError, IOError, ValueError):
return 0x000 return 0x000
def _format_duration(duration_ms: int) -> str:
duration_m = duration_ms // 60000
duration_s = ceil(duration_ms / 1000) - duration_m * 60
if duration_s == 60:
duration_m += 1
duration_s = 0
return f"{duration_m}:{duration_s:02}"

12
MusicBot/cogs/utils/voice_extension.py
View File

@@ -163,6 +163,7 @@ class VoiceExtension:
guild = await self.db.get_guild(gid, projection={'vibing': 1, 'current_menu': 1, 'current_track': 1}) guild = await self.db.get_guild(gid, projection={'vibing': 1, 'current_menu': 1, 'current_track': 1})
if not guild['current_menu']: if not guild['current_menu']:
logging.debug("[VC_EXT] No current menu found")
return False return False
menu_message = await self.get_menu_message(ctx, guild['current_menu']) if not menu_message else menu_message menu_message = await self.get_menu_message(ctx, guild['current_menu']) if not menu_message else menu_message
@@ -281,7 +282,7 @@ class VoiceExtension:
uid = viber_id if viber_id else ctx.user_id if isinstance(ctx, discord.RawReactionActionEvent) else ctx.user.id if ctx.user else None uid = viber_id if viber_id else ctx.user_id if isinstance(ctx, discord.RawReactionActionEvent) else ctx.user.id if ctx.user else None
if not uid or not gid: if not uid or not gid:
logging.warning("[VC_EXT] Guild ID or User ID not found in context inside 'vibe_update'") logging.warning("[VC_EXT] Guild ID or User ID not found in context")
return False return False
user = await self.users_db.get_user(uid, projection={'ym_token': 1, 'vibe_settings': 1}) user = await self.users_db.get_user(uid, projection={'ym_token': 1, 'vibe_settings': 1})
@@ -516,6 +517,7 @@ class VoiceExtension:
vc: discord.VoiceClient | None = None, vc: discord.VoiceClient | None = None,
*, *,
after: bool = False, after: bool = False,
client: YMClient | None = None,
menu_message: discord.Message | None = None, menu_message: discord.Message | None = None,
button_callback: bool = False button_callback: bool = False
) -> str | None: ) -> str | None:
@@ -526,6 +528,7 @@ class VoiceExtension:
ctx (ApplicationContext | Interaction | RawReactionActionEvent): Context ctx (ApplicationContext | Interaction | RawReactionActionEvent): Context
vc (discord.VoiceClient, optional): Voice client. vc (discord.VoiceClient, optional): Voice client.
after (bool, optional): Whether the function is being called by the after callback. Defaults to False. after (bool, optional): Whether the function is being called by the after callback. Defaults to False.
client (YMClient | None, optional): Yandex Music client. Defaults to None.
menu_message (discord.Message | None): Menu message. If None, fetches menu from channel using message id from database. Defaults to None. menu_message (discord.Message | None): Menu message. If None, fetches menu from channel using message id from database. Defaults to None.
button_callback (bool, optional): Should be True if the function is being called from button callback. Defaults to False. button_callback (bool, optional): Should be True if the function is being called from button callback. Defaults to False.
@@ -548,13 +551,6 @@ class VoiceExtension:
logging.debug("[VC_EXT] Playback is stopped, skipping after callback.") logging.debug("[VC_EXT] Playback is stopped, skipping after callback.")
return None return None
if not (client := await self.init_ym_client(ctx, user['ym_token'])):
return None
if not (vc := await self.get_voice_client(ctx) if not vc else vc):
logging.debug("[VC_EXT] Voice client not found in 'next_track'")
return None
if guild['current_track'] and guild['current_menu'] and not guild['repeat']: if guild['current_track'] and guild['current_menu'] and not guild['repeat']:
logging.debug("[VC_EXT] Adding current track to history") logging.debug("[VC_EXT] Adding current track to history")
await self.db.modify_track(gid, guild['current_track'], 'previous', 'insert') await self.db.modify_track(gid, guild['current_track'], 'previous', 'insert')

25
MusicBot/cogs/voice.py
View File

@@ -204,8 +204,8 @@ class Voice(Cog, VoiceExtension):
@voice.command(name="menu", description="Создать или обновить меню проигрывателя.") @voice.command(name="menu", description="Создать или обновить меню проигрывателя.")
async def menu(self, ctx: discord.ApplicationContext) -> None: async def menu(self, ctx: discord.ApplicationContext) -> None:
logging.info(f"[VOICE] Menu command invoked by user {ctx.author.id} in guild {ctx.guild.id}") logging.info(f"[VOICE] Menu command invoked by user {ctx.author.id} in guild {ctx.guild.id}")
if await self.voice_check(ctx): if await self.voice_check(ctx) and not await self.send_menu_message(ctx):
await self.send_menu_message(ctx) await ctx.respond("❌ Не удалось создать меню.", ephemeral=True)
@voice.command(name="join", description="Подключиться к голосовому каналу, в котором вы сейчас находитесь.") @voice.command(name="join", description="Подключиться к голосовому каналу, в котором вы сейчас находитесь.")
async def join(self, ctx: discord.ApplicationContext) -> None: async def join(self, ctx: discord.ApplicationContext) -> None:
@@ -213,17 +213,17 @@ class Voice(Cog, VoiceExtension):
member = cast(discord.Member, ctx.author) member = cast(discord.Member, ctx.author)
guild = await self.db.get_guild(ctx.guild.id, projection={'allow_change_connect': 1}) guild = await self.db.get_guild(ctx.guild.id, projection={'allow_change_connect': 1})
vc = await self.get_voice_client(ctx)
await ctx.defer(ephemeral=True)
if not member.guild_permissions.manage_channels and not guild['allow_change_connect']: if not member.guild_permissions.manage_channels and not guild['allow_change_connect']:
response_message = "❌ У вас нет прав для выполнения этой команды." response_message = "❌ У вас нет прав для выполнения этой команды."
elif vc and vc.is_connected():
response_message = "❌ Бот уже находится в голосовом канале. Выключите его с помощью команды /voice leave."
elif isinstance(ctx.channel, discord.VoiceChannel): elif isinstance(ctx.channel, discord.VoiceChannel):
try: try:
await ctx.channel.connect() await ctx.channel.connect()
except TimeoutError: except TimeoutError:
response_message = "❌ Не удалось подключиться к голосовому каналу." response_message = "❌ Не удалось подключиться к голосовому каналу."
except discord.ClientException:
response_message = "❌ Бот уже находится в голосовом канале. Выключите его с помощью команды /voice leave."
else: else:
response_message = "✅ Подключение успешно!" response_message = "✅ Подключение успешно!"
else: else:
@@ -302,6 +302,10 @@ class Voice(Cog, VoiceExtension):
await self.users_db.update(ctx.user.id, {'queue_page': 0}) await self.users_db.update(ctx.user.id, {'queue_page': 0})
tracks = await self.db.get_tracks_list(ctx.guild.id, 'next') tracks = await self.db.get_tracks_list(ctx.guild.id, 'next')
if len(tracks) == 0:
await ctx.respond("❌ Очередь пуста.", ephemeral=True)
return
embed = generate_queue_embed(0, tracks) embed = generate_queue_embed(0, tracks)
await ctx.respond(embed=embed, view=await QueueView(ctx).init(), ephemeral=True) await ctx.respond(embed=embed, view=await QueueView(ctx).init(), ephemeral=True)
@@ -340,6 +344,7 @@ class Voice(Cog, VoiceExtension):
) )
return return
await ctx.defer(ephemeral=True)
res = await self.stop_playing(ctx, full=True) res = await self.stop_playing(ctx, full=True)
if res: if res:
await ctx.respond("✅ Воспроизведение остановлено.", delete_after=15, ephemeral=True) await ctx.respond("✅ Воспроизведение остановлено.", delete_after=15, ephemeral=True)
@@ -435,18 +440,16 @@ class Voice(Cog, VoiceExtension):
} }
) )
return return
feedback = await self.update_vibe(ctx, _type, _id)
if not feedback: if not await self.update_vibe(ctx, _type, _id):
await ctx.respond("❌ Операция не удалась. Возможно, у вес нет подписки на Яндекс Музыку.", delete_after=15, ephemeral=True) await ctx.respond("❌ Операция не удалась. Возможно, у вес нет подписки на Яндекс Музыку.", delete_after=15, ephemeral=True)
return return
if guild['current_menu']: if guild['current_menu']:
await ctx.respond("✅ Моя Волна включена.", delete_after=15, ephemeral=True) await ctx.respond("✅ Моя Волна включена.", delete_after=15, ephemeral=True)
else: elif not await self.send_menu_message(ctx, disable=True):
await self.send_menu_message(ctx, disable=True) await ctx.respond("❌ Не удалось отправить меню. Попробуйте позже.", delete_after=15, ephemeral=True)
next_track = await self.db.get_track(ctx.guild_id, 'next') next_track = await self.db.get_track(ctx.guild_id, 'next')
if next_track: if next_track:
await self._play_track(ctx, next_track) await self.play_track(ctx, next_track)

1
MusicBot/main.py
View File

@@ -5,7 +5,6 @@ import discord
from discord.ext.commands import Bot from discord.ext.commands import Bot
intents = discord.Intents.default() intents = discord.Intents.default()
intents.message_content = True
bot = Bot(intents=intents) bot = Bot(intents=intents)
cogs_list = [ cogs_list = [

39
MusicBot/ui/find.py
View File

@@ -19,11 +19,11 @@ class PlayButton(Button, VoiceExtension):
logging.debug(f"[FIND] Callback triggered for type: '{type(self.item).__name__}'") logging.debug(f"[FIND] Callback triggered for type: '{type(self.item).__name__}'")
if not interaction.guild: if not interaction.guild:
logging.warning("[FIND] No guild found in PlayButton callback") logging.info("[FIND] No guild found in PlayButton callback")
await interaction.respond("❌ Эта команда доступна только на серверах.", ephemeral=True, delete_after=15)
return return
if not await self.voice_check(interaction): if not await self.voice_check(interaction):
logging.debug("[FIND] Voice check failed in PlayButton callback")
return return
gid = interaction.guild.id gid = interaction.guild.id
@@ -41,7 +41,7 @@ class PlayButton(Button, VoiceExtension):
album = await self.item.with_tracks_async() album = await self.item.with_tracks_async()
if not album or not album.volumes: if not album or not album.volumes:
logging.debug("[FIND] Failed to fetch album tracks in PlayButton callback") logging.debug("[FIND] Failed to fetch album tracks in PlayButton callback")
await interaction.respond("❌ Не удалось получить треки альбома.", ephemeral=True) await interaction.respond("❌ Не удалось получить треки альбома.", ephemeral=True, delete_after=15)
return return
tracks = [track for volume in album.volumes for track in volume] tracks = [track for volume in album.volumes for track in volume]
@@ -53,7 +53,7 @@ class PlayButton(Button, VoiceExtension):
artist_tracks = await self.item.get_tracks_async() artist_tracks = await self.item.get_tracks_async()
if not artist_tracks: if not artist_tracks:
logging.debug("[FIND] Failed to fetch artist tracks in PlayButton callback") logging.debug("[FIND] Failed to fetch artist tracks in PlayButton callback")
await interaction.respond("❌ Не удалось получить треки артиста.", ephemeral=True) await interaction.respond("❌ Не удалось получить треки артиста.", ephemeral=True, delete_after=15)
return return
tracks = artist_tracks.tracks.copy() tracks = artist_tracks.tracks.copy()
@@ -65,7 +65,7 @@ class PlayButton(Button, VoiceExtension):
short_tracks = await self.item.fetch_tracks_async() short_tracks = await self.item.fetch_tracks_async()
if not short_tracks: if not short_tracks:
logging.debug("[FIND] Failed to fetch playlist tracks in PlayButton callback") logging.debug("[FIND] Failed to fetch playlist tracks in PlayButton callback")
await interaction.respond("❌ Не удалось получить треки из плейлиста.", delete_after=15) await interaction.respond("❌ Не удалось получить треки из плейлиста.", ephemeral=True, delete_after=15)
return return
tracks = [cast(Track, short_track.track) for short_track in short_tracks] tracks = [cast(Track, short_track.track) for short_track in short_tracks]
@@ -77,7 +77,7 @@ class PlayButton(Button, VoiceExtension):
tracks = self.item.copy() tracks = self.item.copy()
if not tracks: if not tracks:
logging.debug("[FIND] Empty tracks list in PlayButton callback") logging.debug("[FIND] Empty tracks list in PlayButton callback")
await interaction.respond("❌ Не удалось получить треки.", delete_after=15) await interaction.respond("❌ Не удалось получить треки.", ephemeral=True, delete_after=15)
return return
action = 'add_playlist' action = 'add_playlist'
@@ -109,21 +109,20 @@ class PlayButton(Button, VoiceExtension):
) )
return return
logging.debug(f"[FIND] Skipping vote for '{action}'")
if guild['current_menu']: if guild['current_menu']:
await interaction.respond(response_message, delete_after=15) await interaction.respond(response_message, delete_after=15)
else: elif not await self.send_menu_message(interaction, disable=True):
await self.send_menu_message(interaction, disable=True) await interaction.respond('❌ Не удалось отправить сообщение.', ephemeral=True, delete_after=15)
if guild['current_track'] is not None: if guild['current_track']:
logging.debug(f"[FIND] Adding tracks to queue") logging.debug(f"[FIND] Adding tracks to queue")
await self.db.modify_track(gid, tracks, 'next', 'extend') await self.db.modify_track(gid, tracks, 'next', 'extend')
else: else:
logging.debug(f"[FIND] Playing track") logging.debug(f"[FIND] Playing track")
track = tracks.pop(0) track = tracks.pop(0)
await self.db.modify_track(gid, tracks, 'next', 'extend') await self.db.modify_track(gid, tracks, 'next', 'extend')
await self.play_track(interaction, track) if not await self.play_track(interaction, track):
await interaction.respond('❌ Не удалось воспроизвести трек.', ephemeral=True, delete_after=15)
if interaction.message: if interaction.message:
await interaction.message.delete() await interaction.message.delete()
@@ -146,6 +145,7 @@ class MyVibeButton(Button, VoiceExtension):
logging.warning(f"[VIBE] Guild ID is None in button callback") logging.warning(f"[VIBE] Guild ID is None in button callback")
return return
track_type_map = { track_type_map = {
Track: 'track', Album: 'album', Artist: 'artist', Playlist: 'playlist', list: 'user' Track: 'track', Album: 'album', Artist: 'artist', Playlist: 'playlist', list: 'user'
} }
@@ -153,7 +153,7 @@ class MyVibeButton(Button, VoiceExtension):
if isinstance(self.item, Playlist): if isinstance(self.item, Playlist):
if not self.item.owner: if not self.item.owner:
logging.warning(f"[VIBE] Playlist owner is None") logging.warning(f"[VIBE] Playlist owner is None")
await interaction.respond("❌ Не удалось получить информацию о плейлисте.", ephemeral=True) await interaction.respond("❌ Не удалось получить информацию о плейлисте. Отсутствует владелец.", ephemeral=True, delete_after=15)
return return
_id = self.item.owner.login + '_' + str(self.item.kind) _id = self.item.owner.login + '_' + str(self.item.kind)
@@ -162,12 +162,11 @@ class MyVibeButton(Button, VoiceExtension):
else: else:
_id = 'onyourwave' _id = 'onyourwave'
await self.send_menu_message(interaction, disable=True) guild = await self.db.get_guild(gid, projection={'current_menu': 1})
await self.update_vibe( if not guild['current_menu'] and not await self.send_menu_message(interaction, disable=True):
interaction, await interaction.respond('❌ Не удалось отправить сообщение.', ephemeral=True, delete_after=15)
track_type_map[type(self.item)],
_id await self.update_vibe(interaction, track_type_map[type(self.item)], _id)
)
next_track = await self.db.get_track(gid, 'next') next_track = await self.db.get_track(gid, 'next')
if next_track: if next_track:
@@ -208,6 +207,6 @@ class ListenView(View):
async def on_timeout(self) -> None: async def on_timeout(self) -> None:
try: try:
return await super().on_timeout() return await super().on_timeout()
except discord.NotFound: except discord.HTTPException:
pass pass
self.stop() self.stop()

13
MusicBot/ui/menu.py
View File

@@ -2,7 +2,7 @@ import logging
from typing import Self, cast from typing import Self, cast
from discord.ui import View, Button, Item, Select from discord.ui import View, Button, Item, Select
from discord import VoiceChannel, ButtonStyle, Interaction, ApplicationContext, RawReactionActionEvent, Embed, ComponentType, SelectOption, Member from discord import VoiceChannel, ButtonStyle, Interaction, ApplicationContext, RawReactionActionEvent, Embed, ComponentType, SelectOption, Member, HTTPException
import yandex_music.exceptions import yandex_music.exceptions
from yandex_music import TrackLyrics, Playlist, ClientAsync as YMClient from yandex_music import TrackLyrics, Playlist, ClientAsync as YMClient
@@ -348,7 +348,7 @@ class MyVibeSelect(Select, VoiceExtension):
await interaction.edit(view=view) await interaction.edit(view=view)
class MyVibeSettingsView(View, VoiceExtension): class MyVibeSettingsView(View, VoiceExtension):
def __init__(self, interaction: Interaction, *items: Item, timeout: float | None = None, disable_on_timeout: bool = True): def __init__(self, interaction: Interaction, *items: Item, timeout: float | None = 360, disable_on_timeout: bool = True):
View.__init__(self, *items, timeout=timeout, disable_on_timeout=disable_on_timeout) View.__init__(self, *items, timeout=timeout, disable_on_timeout=disable_on_timeout)
VoiceExtension.__init__(self, None) VoiceExtension.__init__(self, None)
self.interaction = interaction self.interaction = interaction
@@ -410,6 +410,13 @@ class MyVibeSettingsView(View, VoiceExtension):
self.add_item(select) self.add_item(select)
return self return self
async def on_timeout(self) -> None:
try:
return await super().on_timeout()
except HTTPException:
pass
self.stop()
class MyVibeSettingsButton(Button, VoiceExtension): class MyVibeSettingsButton(Button, VoiceExtension):
def __init__(self, **kwargs): def __init__(self, **kwargs):
@@ -530,7 +537,7 @@ class MenuView(View, VoiceExtension):
if not self.ctx.guild_id: if not self.ctx.guild_id:
return self return self
self.guild = await self.db.get_guild(self.ctx.guild_id) self.guild = await self.db.get_guild(self.ctx.guild_id, projection={'repeat': 1, 'shuffle': 1, 'current_track': 1, 'vibing': 1})
if self.guild['repeat']: if self.guild['repeat']:
self.repeat_button.style = ButtonStyle.success self.repeat_button.style = ButtonStyle.success

28
MusicBot/ui/other.py
View File

@@ -2,7 +2,7 @@ from math import ceil
from typing import Self, Any from typing import Self, Any
from discord.ui import View, Button, Item from discord.ui import View, Button, Item
from discord import ApplicationContext, ButtonStyle, Interaction, Embed from discord import ApplicationContext, ButtonStyle, Interaction, Embed, HTTPException
from MusicBot.cogs.utils.voice_extension import VoiceExtension from MusicBot.cogs.utils.voice_extension import VoiceExtension
@@ -27,10 +27,11 @@ class QueueNextButton(Button, VoiceExtension):
def __init__(self, **kwargs): def __init__(self, **kwargs):
Button.__init__(self, **kwargs) Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None) VoiceExtension.__init__(self, None)
async def callback(self, interaction: Interaction) -> None: async def callback(self, interaction: Interaction) -> None:
if not interaction.user or not interaction.guild: if not interaction.user or not interaction.guild:
return return
user = await self.users_db.get_user(interaction.user.id) user = await self.users_db.get_user(interaction.user.id)
page = user['queue_page'] + 1 page = user['queue_page'] + 1
await self.users_db.update(interaction.user.id, {'queue_page': page}) await self.users_db.update(interaction.user.id, {'queue_page': page})
@@ -42,10 +43,11 @@ class QueuePrevButton(Button, VoiceExtension):
def __init__(self, **kwargs): def __init__(self, **kwargs):
Button.__init__(self, **kwargs) Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None) VoiceExtension.__init__(self, None)
async def callback(self, interaction: Interaction) -> None: async def callback(self, interaction: Interaction) -> None:
if not interaction.user or not interaction.guild: if not interaction.user or not interaction.guild:
return return
user = await self.users_db.get_user(interaction.user.id) user = await self.users_db.get_user(interaction.user.id)
page = user['queue_page'] - 1 page = user['queue_page'] - 1
await self.users_db.update(interaction.user.id, {'queue_page': page}) await self.users_db.update(interaction.user.id, {'queue_page': page})
@@ -54,30 +56,36 @@ class QueuePrevButton(Button, VoiceExtension):
await interaction.edit(embed=embed, view=await QueueView(interaction).init()) await interaction.edit(embed=embed, view=await QueueView(interaction).init())
class QueueView(View, VoiceExtension): class QueueView(View, VoiceExtension):
def __init__(self, ctx: ApplicationContext | Interaction, *items: Item, timeout: float | None = 360, disable_on_timeout: bool = True): def __init__(self, ctx: ApplicationContext | Interaction, *items: Item, timeout: float | None = 360, disable_on_timeout: bool = False):
View.__init__(self, *items, timeout=timeout, disable_on_timeout=disable_on_timeout) View.__init__(self, *items, timeout=timeout, disable_on_timeout=disable_on_timeout)
VoiceExtension.__init__(self, None) VoiceExtension.__init__(self, None)
self.ctx = ctx self.ctx = ctx
self.next_button = QueueNextButton(style=ButtonStyle.primary, emoji='▶️') self.next_button = QueueNextButton(style=ButtonStyle.primary, emoji='▶️')
self.prev_button = QueuePrevButton(style=ButtonStyle.primary, emoji='◀️') self.prev_button = QueuePrevButton(style=ButtonStyle.primary, emoji='◀️')
async def init(self) -> Self: async def init(self) -> Self:
if not self.ctx.user or not self.ctx.guild: if not self.ctx.user or not self.ctx.guild:
return self return self
tracks = await self.db.get_tracks_list(self.ctx.guild.id, 'next') tracks = await self.db.get_tracks_list(self.ctx.guild.id, 'next')
user = await self.users_db.get_user(self.ctx.user.id) user = await self.users_db.get_user(self.ctx.user.id)
count = 15 * user['queue_page'] count = 15 * user['queue_page']
if not tracks[count + 15:]: if not tracks[count + 15:]:
self.next_button.disabled = True self.next_button.disabled = True
if not tracks[:count]: if not tracks[:count]:
self.prev_button.disabled = True self.prev_button.disabled = True
self.add_item(self.prev_button) self.add_item(self.prev_button)
self.add_item(self.next_button) self.add_item(self.next_button)
return self return self
async def on_timeout(self) -> None:
try:
await super().on_timeout()
except HTTPException:
pass
self.stop()