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

feat: Add bot logging for debugging.

Browse Source
This commit is contained in:
Lemon4ksan
2025-01-24 17:36:27 +03:00
parent 859d60de35
commit 3f9698fa7b
8 changed files with 804 additions and 945 deletions
Download Patch File Download Diff File Expand all files Collapse all files

143
MusicBot/cogs/general.py
View File

@@ -1,4 +1,5 @@
from typing import cast
import logging
from typing import Literal, cast
from asyncio import gather
import discord
@@ -10,11 +11,9 @@ from yandex_music import ClientAsync as YMClient
from yandex_music import Track, Album, Artist, Playlist
from MusicBot.database import BaseUsersDatabase, BaseGuildsDatabase
from MusicBot.cogs.utils.find import (
process_album, process_track, process_artist, process_playlist,
ListenAlbum, ListenTrack, ListenArtist, ListenPlaylist, ListenLikesPlaylist
)
from MusicBot.cogs.utils.misc import MyPlaylists, generate_playlist_embed, generate_likes_embed
from MusicBot.cogs.utils.find import ListenView, generate_item_embed
from MusicBot.cogs.utils.misc import generate_playlists_embed, generate_likes_embed
from MusicBot.cogs.utils.views import MyPlaylists
def setup(bot):
bot.add_cog(General(bot))
@@ -36,6 +35,7 @@ class General(Cog):
default='all'
)
async def help(self, ctx: discord.ApplicationContext, command: str) -> None:
logging.debug(f"Help command invoked by {ctx.user.id} for command '{command}'")
response_message = None
embed = discord.Embed(
color=0xfed42b
@@ -109,41 +109,54 @@ class General(Cog):
@account.command(description="Ввести токен от Яндекс Музыки.")
@discord.option("token", type=discord.SlashCommandOptionType.string, description="Токен.")
async def login(self, ctx: discord.ApplicationContext, token: str) -> None:
logging.debug(f"Login command invoked by user {ctx.author.id} in guild {ctx.guild.id}")
try:
client = await YMClient(token).init()
except yandex_music.exceptions.UnauthorizedError:
logging.debug(f"Invalid token provided by user {ctx.author.id}")
await ctx.respond('❌ Недействительный токен.', delete_after=15, ephemeral=True)
return
about = cast(yandex_music.Status, client.me).to_dict()
uid = ctx.author.id
self.users_db.update(uid, {'ym_token': token})
logging.debug(f"Token saved for user {ctx.author.id}")
await ctx.respond(f'Привет, {about['account']['first_name']}!', delete_after=15, ephemeral=True)
@account.command(description="Удалить токен из датабазы бота.")
async def remove(self, ctx: discord.ApplicationContext) -> None:
logging.debug(f"Remove command invoked by user {ctx.author.id} in guild {ctx.guild.id}")
self.users_db.update(ctx.user.id, {'ym_token': None})
await ctx.respond(f'Токен был удалён.', delete_after=15, ephemeral=True)
@account.command(description="Получить плейлист «Мне нравится»")
async def likes(self, ctx: discord.ApplicationContext) -> None:
logging.debug(f"Likes command invoked by user {ctx.author.id} in guild {ctx.guild.id}")
token = self.users_db.get_ym_token(ctx.user.id)
if not token:
logging.debug(f"No token found for user {ctx.user.id}")
await ctx.respond('❌ Необходимо указать свой токен доступа с помощью команды /login.', delete_after=15, ephemeral=True)
return
client = await YMClient(token).init()
if not client.me or not client.me.account or not client.me.account.uid:
logging.warning(f"Failed to fetch user info for user {ctx.user.id}")
await ctx.respond('❌ Что-то пошло не так. Повторите попытку позже.', delete_after=15, ephemeral=True)
return
likes = await client.users_likes_tracks()
if not likes:
if likes is None:
logging.debug(f"Failed to fetch likes for user {ctx.user.id}")
await ctx.respond('❌ Что-то пошло не так. Повторите попытку позже.', delete_after=15, ephemeral=True)
return
elif not likes:
logging.debug(f"Empty likes for user {ctx.user.id}")
await ctx.respond('❌ У вас нет треков в плейлисте «Мне нравится».', delete_after=15, ephemeral=True)
return
real_tracks = await gather(*[track_short.fetch_track_async() for track_short in likes.tracks], return_exceptions=True)
tracks = [track for track in real_tracks if not isinstance(track, BaseException)] # Can't fetch user tracks
embed = generate_likes_embed(tracks)
await ctx.respond(embed=embed, view=ListenLikesPlaylist(tracks))
logging.debug(f"Successfully fetched likes for user {ctx.user.id}")
await ctx.respond(embed=embed, view=ListenView(tracks))
@account.command(description="Получить ваши плейлисты.")
async def playlists(self, ctx: discord.ApplicationContext) -> None:
@@ -160,7 +173,8 @@ class General(Cog):
(playlist.title if playlist.title else 'Без названия', playlist.track_count if playlist.track_count else 0) for playlist in playlists_list
]
self.users_db.update(ctx.user.id, {'playlists': playlists, 'playlists_page': 0})
embed = generate_playlist_embed(0, playlists)
embed = generate_playlists_embed(0, playlists)
logging.debug(f"Successfully fetched playlists for user {ctx.user.id}")
await ctx.respond(embed=embed, view=MyPlaylists(ctx), ephemeral=True)
@discord.slash_command(description="Найти контент и отправить информацию о нём. Возвращается лучшее совпадение.")
@@ -173,99 +187,114 @@ class General(Cog):
"content_type",
description="Тип искомого контента.",
type=discord.SlashCommandOptionType.string,
choices=['Artist', 'Album', 'Track', 'Playlist', 'User Playlist'],
default='Track'
choices=['Трек', 'Альбом', 'Артист', 'Плейлист', 'Свой плейлист'],
default='Трек'
)
async def find(
self,
ctx: discord.ApplicationContext,
name: str,
content_type: str = 'Track'
content_type: Literal['Трек', 'Альбом', 'Артист', 'Плейлист', 'Свой плейлист'] = 'Трек'
) -> None:
if content_type not in ['Artist', 'Album', 'Track', 'Playlist', 'User Playlist']:
await ctx.respond("❌ Недопустимый тип.", delete_after=15, ephemeral=True)
return
logging.debug(f"User {ctx.user.id} invoked find command for '{content_type}' with name '{name}'")
guild = self.db.get_guild(ctx.guild_id)
token = self.users_db.get_ym_token(ctx.user.id)
if not token:
logging.debug(f"No token found for user {ctx.user.id}")
await ctx.respond("❌ Необходимо указать свой токен доступа с помощью команды /login.", delete_after=15, ephemeral=True)
return
try:
client = await YMClient(token).init()
except yandex_music.exceptions.UnauthorizedError:
logging.debug(f"User {ctx.user.id} provided invalid token")
await ctx.respond("❌ Недействительный токен. Если это не так, попробуйте ещё раз.", delete_after=15, ephemeral=True)
return
if content_type == 'User Playlist':
if content_type == 'Свой плейлист':
if not client.me or not client.me.account or not client.me.account.uid:
logging.warning(f"Failed to get user info for user {ctx.user.id}")
await ctx.respond("❌ Не удалось получить информацию о пользователе.", delete_after=15, ephemeral=True)
return
playlists = await client.users_playlists_list(client.me.account.uid)
result = next((playlist for playlist in playlists if playlist.title == name), None)
if not result:
logging.debug(f"User {ctx.user.id} playlist '{name}' not found")
await ctx.respond("❌ Плейлист не найден.", delete_after=15, ephemeral=True)
return
tracks = await result.fetch_tracks_async()
if not tracks:
logging.debug(f"User {ctx.user.id} playlist '{name}' is empty")
await ctx.respond("❌ Плейлист пуст.", delete_after=15, ephemeral=True)
return
for track_short in tracks:
track = cast(Track, track_short.track)
if (track.explicit or track.content_warning) and not guild['allow_explicit']:
logging.debug(f"User {ctx.user.id} playlist '{name}' contains explicit content and is not allowed on this server")
await ctx.respond("❌ Explicit контент запрещён на этом сервере.", delete_after=15, ephemeral=True)
return
embed = await process_playlist(result)
await ctx.respond(embed=embed, view=ListenPlaylist(result))
embed = await generate_item_embed(result)
view = ListenView(result)
else:
result = await client.search(name, True)
if not result:
logging.warning(f"Failed to search for '{name}' for user {ctx.user.id}")
await ctx.respond("❌ Что-то пошло не так. Повторите попытку позже", delete_after=15, ephemeral=True)
return
content_map = {
'Album': (result.albums, process_album, ListenAlbum),
'Track': (result.tracks, process_track, ListenTrack),
'Artist': (result.artists, process_artist, ListenArtist),
'Playlist': (result.playlists, process_playlist, ListenPlaylist)
}
if content_type == 'Трек':
content = result.tracks
elif content_type == 'Альбом':
content = result.albums
elif content_type == 'Артист':
content = result.artists
elif content_type == 'Плейлист':
content = result.playlists
if content_type in content_map:
content: Album | Track | Artist | Playlist = content_map[content_type][0].results[0]
embed: discord.Embed = await content_map[content_type][1](content)
view = content_map[content_type][2](content)
if isinstance(content, (Track, Album)) and (content.explicit or content.content_warning) and not guild['allow_explicit']:
await ctx.respond("❌ Explicit контент запрещён на этом сервере.", delete_after=15, ephemeral=True)
return
elif isinstance(content, Artist):
tracks = await content.get_tracks_async()
if not tracks:
await ctx.respond("❌ Треки от этого исполнителя не найдены.", delete_after=15, ephemeral=True)
return
for track in tracks:
if (track.explicit or track.content_warning) and not guild['allow_explicit']:
view = None
embed.set_footer(text="Воспроизведение недоступно, так как у автора присутствуют Explicit треки")
break
elif isinstance(content, Playlist):
tracks = await content.fetch_tracks_async()
if not tracks:
await ctx.respond("❌ Треки в этом плейлисте не найдены.", delete_after=15, ephemeral=True)
return
for track_short in content.tracks:
track = cast(Track, track_short.track)
if (track.explicit or track.content_warning) and not guild['allow_explicit']:
view = None
embed.set_footer(text="Воспроизведение недоступно, так как у автора присутствуют Explicit треки")
break
await ctx.respond(embed=embed, view=view)
else:
if not content:
logging.debug(f"User {ctx.user.id} search for '{name}' returned no results")
await ctx.respond("❌ По запросу ничего не найдено.", delete_after=15, ephemeral=True)
return
content = content.results[0]
embed = await generate_item_embed(content)
view = ListenView(content)
if isinstance(content, (Track, Album)) and (content.explicit or content.content_warning) and not guild['allow_explicit']:
logging.debug(f"User {ctx.user.id} search for '{name}' returned explicit content and is not allowed on this server")
await ctx.respond("❌ Explicit контент запрещён на этом сервере.", delete_after=15, ephemeral=True)
return
elif isinstance(content, Artist):
tracks = await content.get_tracks_async()
if not tracks:
logging.debug(f"User {ctx.user.id} search for '{name}' returned no tracks")
await ctx.respond("❌ Треки от этого исполнителя не найдены.", delete_after=15, ephemeral=True)
return
for track in tracks:
if (track.explicit or track.content_warning) and not guild['allow_explicit']:
logging.debug(f"User {ctx.user.id} search for '{name}' returned explicit content and is not allowed on this server")
view = None
embed.set_footer(text="Воспроизведение недоступно, так как у автора присутствуют Explicit треки")
break
elif isinstance(content, Playlist):
tracks = await content.fetch_tracks_async()
if not tracks:
logging.debug(f"User {ctx.user.id} search for '{name}' returned no tracks")
await ctx.respond("❌ Пустой плейлист.", delete_after=15, ephemeral=True)
return
for track_short in content.tracks:
track = cast(Track, track_short.track)
if (track.explicit or track.content_warning) and not guild['allow_explicit']:
logging.debug(f"User {ctx.user.id} search for '{name}' returned explicit content and is not allowed on this server")
view = None
embed.set_footer(text="Воспроизведение недоступно, так как у автора присутствуют Explicit треки")
break
logging.debug(f"Successfully generated '{content_type}' message for user {ctx.author.id}")
await ctx.respond(embed=embed, view=view)

663
MusicBot/cogs/utils/find.py
View File

@@ -1,84 +1,93 @@
from os import getenv
from math import ceil
import logging
from typing import cast
import discord
from yandex_music import Track, Album, Artist, Playlist, Label
from yandex_music import Track, Album, Artist, Playlist
from discord.ui import View, Button, Item
from discord import ButtonStyle, Interaction, Embed
from MusicBot.cogs.utils.voice_extension import VoiceExtension, get_average_color_from_url
from MusicBot.cogs.utils.voice_extension import VoiceExtension
from MusicBot.cogs.utils.misc import generate_track_embed, generate_album_embed, generate_artist_embed, generate_playlist_embed
class PlayTrackButton(Button, VoiceExtension):
def __init__(self, track: Track, **kwargs):
class PlayButton(Button, VoiceExtension):
def __init__(self, item: Track | Album | Artist | Playlist | list[Track], **kwargs):
Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None)
self.track = track
self.item = item
async def callback(self, interaction: Interaction) -> None:
if not interaction.guild or not await self.voice_check(interaction):
logging.debug(f"Callback triggered for type: '{type(self.item).__name__}'")
if not interaction.guild:
logging.warning("No guild found in context.")
return
if not await self.voice_check(interaction):
logging.debug("Voice check failed")
return
gid = interaction.guild.id
guild = self.db.get_guild(gid)
channel = cast(discord.VoiceChannel, interaction.channel)
member = cast(discord.Member, interaction.user)
if guild['vote_add_track'] and len(channel.members) > 2 and not member.guild_permissions.manage_channels:
message = cast(discord.Interaction, await interaction.respond(f"{member.mention} хочет добавить трек **{self.track.title}** в очередь.\n\n Голосуйте за добавление.", delete_after=30))
response = await message.original_response()
await response.add_reaction('✅')
await response.add_reaction('❌')
self.db.update_vote(
gid,
response.id,
{
'positive_votes': list(),
'negative_votes': list(),
'total_members': len(channel.members),
'action': 'add_track',
'vote_content': self.track.to_dict()
}
)
if isinstance(self.item, Track):
tracks = [self.item]
action = 'add_track'
vote_message = f"{member.mention} хочет добавить трек **{self.item.title}** в очередь.\n\n Голосуйте за добавление."
response_message = f"Трек **{self.item.title}** был добавлен в очередь."
play_message = f"Сейчас играет: **{self.item.title}**!"
elif isinstance(self.item, Album):
album = await self.item.with_tracks_async()
if not album or not album.volumes:
logging.debug("Failed to fetch album tracks")
await interaction.respond("Не удалось получить треки альбома.", ephemeral=True)
return
tracks = [track for volume in album.volumes for track in volume]
action = 'add_album'
vote_message = f"{member.mention} хочет добавить альбом **{self.item.title}** в очередь.\n\n Голосуйте за добавление."
response_message = f"Альбом **{self.item.title}** был добавлен в очередь."
play_message = f"Сейчас играет: **{self.item.title}**!"
elif isinstance(self.item, Artist):
artist_tracks = await self.item.get_tracks_async()
if not artist_tracks:
logging.debug("Failed to fetch artist tracks")
await interaction.respond("Не удалось получить треки артиста.", ephemeral=True)
return
tracks = artist_tracks.tracks.copy()
action = 'add_artist'
vote_message = f"{member.mention} хочет добавить треки от **{self.item.name}** в очередь.\n\n Голосуйте за добавление."
response_message = f"Песни артиста **{self.item.name}** были добавлены в очередь."
play_message = f"Сейчас играет: **{self.item.name}**!"
elif isinstance(self.item, Playlist):
short_tracks = await self.item.fetch_tracks_async()
if not short_tracks:
logging.debug("Failed to fetch playlist tracks")
await interaction.respond("❌ Не удалось получить треки из плейлиста.", delete_after=15)
return
tracks = [cast(Track, short_track.track) for short_track in short_tracks]
action = 'add_playlist'
vote_message = f"{member.mention} хочет добавить плейлист **{self.item.title}** в очередь.\n\n Голосуйте за добавление."
response_message = f"Плейлист **{self.item.title}** был добавлен в очередь."
play_message = f"Сейчас играет: **{self.item.title}**!"
elif isinstance(self.item, list):
tracks = self.item.copy()
if not tracks:
logging.debug("Empty tracks list")
await interaction.respond("❌ Не удалось получить треки.", delete_after=15)
return
action = 'add_playlist'
vote_message = f"{member.mention} хочет добавить плейлист **** в очередь.\n\n Голосуйте за добавление."
response_message = f"Плейлист **«Мне нравится»** был добавлен в очередь."
play_message = f"Сейчас играет: **{tracks[0].title}**!"
else:
if guild['current_track']:
self.db.modify_track(gid, self.track, 'next', 'append')
response_message = f"Трек **{self.track.title}** был добавлен в очередь."
else:
await self.play_track(interaction, self.track)
response_message = f"Сейчас играет: **{self.track.title}**!"
raise ValueError(f"Unknown item type: '{type(self.item).__name__}'")
if guild['current_player'] is not None and interaction.message:
await interaction.message.delete()
await interaction.respond(response_message, delete_after=15)
class PlayAlbumButton(Button, VoiceExtension):
def __init__(self, album: Album, **kwargs):
Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None)
self.album = album
async def callback(self, interaction: Interaction) -> None:
if not interaction.guild or not await self.voice_check(interaction):
return
album = await self.album.with_tracks_async()
if not album or not album.volumes:
return
gid = interaction.guild.id
guild = self.db.get_guild(gid)
channel = cast(discord.VoiceChannel, interaction.channel)
member = cast(discord.Member, interaction.user)
tracks: list[Track] = [track for volume in album.volumes for track in volume]
if guild['vote_add_album'] and len(channel.members) > 2 and not member.guild_permissions.manage_channels:
message = cast(discord.Interaction, await interaction.respond(f"{member.mention} хочет добавить альбом **{self.album.title}** в очередь.\n\n Голосуйте за добавление.", delete_after=30))
if guild.get(f'vote_{action}') and len(channel.members) > 2 and not member.guild_permissions.manage_channels:
logging.debug(f"Starting vote for '{action}'")
message = cast(discord.Interaction, await interaction.respond(vote_message, delete_after=30))
response = await message.original_response()
await response.add_reaction('✅')
await response.add_reaction('❌')
@@ -89,513 +98,73 @@ class PlayAlbumButton(Button, VoiceExtension):
'positive_votes': list(),
'negative_votes': list(),
'total_members': len(channel.members),
'action': 'add_album',
'action': action,
'vote_content': [track.to_dict() for track in tracks]
}
)
else:
logging.debug(f"Skipping vote for '{action}'")
if guild['current_track'] is not None:
self.db.modify_track(gid, tracks, 'next', 'extend')
response_message = f"Альбом **{album.title}** был добавлен в очередь."
response_message = response_message
else:
track = tracks.pop(0)
self.db.modify_track(gid, tracks, 'next', 'extend')
await self.play_track(interaction, track)
response_message = f"Сейчас играет: **{track.title}**!"
if guild['current_player'] is not None and interaction.message:
await interaction.message.delete()
else:
await interaction.respond(response_message, delete_after=15)
class PlayArtistButton(Button, VoiceExtension):
def __init__(self, artist: Artist, **kwargs):
Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None)
self.artist = artist
async def callback(self, interaction: Interaction) -> None:
if not interaction.guild or not await self.voice_check(interaction):
return
artist_tracks = await self.artist.get_tracks_async(page_size=500)
if not artist_tracks:
return
gid = interaction.guild.id
guild = self.db.get_guild(gid)
channel = cast(discord.VoiceChannel, interaction.channel)
member = cast(discord.Member, interaction.user)
tracks: list[Track] = artist_tracks.tracks.copy()
if guild['vote_add_artist'] and len(channel.members) > 2 and not member.guild_permissions.manage_channels:
message = cast(discord.Interaction, await interaction.respond(f"{member.mention} хочет добавить треки от **{self.artist.name}** в очередь.\n\n Голосуйте за добавление.", delete_after=30))
response = await message.original_response()
await response.add_reaction('✅')
await response.add_reaction('❌')
self.db.update_vote(
gid,
response.id,
{
'positive_votes': list(),
'negative_votes': list(),
'total_members': len(channel.members),
'action': 'add_album',
'vote_content': [track.to_dict() for track in tracks]
}
)
else:
if guild['current_track'] is not None:
self.db.modify_track(gid, tracks, 'next', 'extend')
response_message = f"Песни артиста **{self.artist.name}** были добавлены в очередь."
else:
track = tracks.pop(0)
self.db.modify_track(gid, tracks, 'next', 'extend')
await self.play_track(interaction, track)
response_message = f"Сейчас играет: **{track.title}**!"
if guild['current_player'] is not None and interaction.message:
await interaction.message.delete()
response_message = play_message
if guild['current_player']:
current_player = await self.get_player_message(interaction, guild['current_player'])
if current_player and interaction.message:
logging.debug(f"Deleting interaction message '{interaction.message.id}': current player '{current_player.id}' found")
await interaction.message.delete()
await interaction.respond(response_message, delete_after=15)
class PlayPlaylistButton(Button, VoiceExtension):
def __init__(self, playlist: Playlist, **kwargs):
Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None)
self.playlist = playlist
async def callback(self, interaction: Interaction) -> None:
if not interaction.guild or not await self.voice_check(interaction):
class ListenView(View):
def __init__(self, item: Track | Album | Artist | Playlist | list[Track], *items: Item, timeout: float | None = None, disable_on_timeout: bool = False):
super().__init__(*items, timeout=timeout, disable_on_timeout=disable_on_timeout)
logging.debug(f"Creating view for type: '{type(item).__name__}'")
if isinstance(item, Track):
link_app = f"yandexmusic://album/{item.albums[0].id}/track/{item.id}"
link_web = f"https://music.yandex.ru/album/{item.albums[0].id}/track/{item.id}"
elif isinstance(item, Album):
link_app = f"yandexmusic://album/{item.id}"
link_web = f"https://music.yandex.ru/album/{item.id}"
elif isinstance(item, Artist):
link_app = f"yandexmusic://artist/{item.id}"
link_web = f"https://music.yandex.ru/artist/{item.id}"
elif isinstance(item, Playlist):
link_app = f"yandexmusic://playlist/{item.playlist_uuid}"
link_web = f"https://music.yandex.ru/playlist/{item.playlist_uuid}"
elif isinstance(item, list): # Can't open other person's likes
self.add_item(PlayButton(item, label="Слушать в голосовом канале", style=ButtonStyle.gray))
return
short_tracks = await self.playlist.fetch_tracks_async()
if not short_tracks:
await interaction.respond("Не удалось получить треки из плейлиста.", delete_after=15)
return
gid = interaction.guild.id
guild = self.db.get_guild(gid)
channel = cast(discord.VoiceChannel, interaction.channel)
member = cast(discord.Member, interaction.user)
tracks: list[Track] = [cast(Track, short_track.track) for short_track in short_tracks]
if guild['vote_add_playlist'] and len(channel.members) > 2 and not member.guild_permissions.manage_channels:
message = cast(discord.Interaction, await interaction.respond(f"{member.mention} хочет добавить плейлист **{self.playlist.title}** в очередь.\n\n Голосуйте за добавление.", delete_after=30))
response = await message.original_response()
await response.add_reaction('✅')
await response.add_reaction('❌')
self.db.update_vote(
gid,
response.id,
{
'positive_votes': list(),
'negative_votes': list(),
'total_members': len(channel.members),
'action': 'add_playlist',
'vote_content': [track.to_dict() for track in tracks]
}
)
else:
if guild['current_track'] is not None:
self.db.modify_track(gid, tracks, 'next', 'extend')
response_message = f"Плейлист **{self.playlist.title}** был добавлен в очередь."
else:
track = tracks.pop(0)
self.db.modify_track(gid, tracks, 'next', 'extend')
await self.play_track(interaction, track)
response_message = f"Сейчас играет: **{self.playlist.title}**!"
if guild['current_player'] is not None and interaction.message:
await interaction.message.delete()
await interaction.respond(response_message, delete_after=15)
class PlayLikesButton(Button, VoiceExtension):
def __init__(self, playlist: list[Track], **kwargs):
Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None)
self.playlist = playlist
async def callback(self, interaction: Interaction):
if not interaction.guild or not await self.voice_check(interaction):
return
playlist = self.playlist.copy()
gid = interaction.guild.id
guild = self.db.get_guild(gid)
channel = cast(discord.VoiceChannel, interaction.channel)
member = cast(discord.Member, interaction.user)
if guild['vote_add_playlist'] and len(channel.members) > 2 and not member.guild_permissions.manage_channels:
message = cast(discord.Interaction, await interaction.respond(f"{member.mention} хочет добавить плейлист **«Мне нравится»** в очередь.\n\n Голосуйте за добавление.", delete_after=30))
response = await message.original_response()
await response.add_reaction('✅')
await response.add_reaction('❌')
self.db.update_vote(
gid,
response.id,
{
'positive_votes': list(),
'negative_votes': list(),
'total_members': len(channel.members),
'action': 'add_playlist',
'vote_content': [track.to_dict() for track in playlist]
}
)
else:
if guild['current_track'] is not None:
self.db.modify_track(gid, playlist, 'next', 'extend')
response_message = f"Плейлист **«Мне нравится»** был добавлен в очередь."
else:
track = playlist.pop(0)
self.db.modify_track(gid, playlist, 'next', 'extend')
await self.play_track(interaction, track)
response_message = f"Сейчас играет: **{track.title}**!"
if guild['current_player'] is not None and interaction.message:
await interaction.message.delete()
await interaction.respond(response_message, delete_after=15)
class ListenLikesPlaylist(View):
def __init__(self, playlist: list[Track], *items: Item, timeout: float | None = None, disable_on_timeout: bool = False):
super().__init__(*items, timeout=timeout, disable_on_timeout=disable_on_timeout)
self.add_item(PlayLikesButton(playlist, label="Слушать в голосовом канале", style=ButtonStyle.gray))
class ListenTrack(View):
def __init__(self, track: Track, *items: Item, timeout: float | None = None, disable_on_timeout: bool = False):
super().__init__(*items, timeout=timeout, disable_on_timeout=disable_on_timeout)
link_app = f"yandexmusic://album/{track.albums[0].id}/track/{track.id}"
link_web = f"https://music.yandex.ru/album/{track.albums[0].id}/track/{track.id}"
self.button1: Button = Button(label="Слушать в приложении", style=ButtonStyle.gray, url=link_app)
self.button2: Button = Button(label="Слушать в браузере", style=ButtonStyle.gray, url=link_web)
self.button3: PlayTrackButton = PlayTrackButton(track, label="Слушать в голосовом канале", style=ButtonStyle.gray)
# self.add_item(self.button1) # Discord doesn't allow well formed URLs in buttons for some reason.
if track.available:
self.add_item(self.button2)
self.add_item(self.button3)
class ListenAlbum(View):
def __init__(self, album: Album, *items: Item, timeout: float | None = None, disable_on_timeout: bool = False):
super().__init__(*items, timeout=timeout, disable_on_timeout=disable_on_timeout)
link_app = f"yandexmusic://album/{album.id}"
link_web = f"https://music.yandex.ru/album/{album.id}"
self.button1: Button = Button(label="Слушать в приложении", style=ButtonStyle.gray, url=link_app)
self.button2: Button = Button(label="Слушать в браузере", style=ButtonStyle.gray, url=link_web)
self.button3: PlayAlbumButton = PlayAlbumButton(album, label="Слушать в голосовом канале", style=ButtonStyle.gray)
# self.add_item(self.button1) # Discord doesn't allow well formed URLs in buttons for some reason.
if album.available:
self.button3: PlayButton = PlayButton(item, label="Слушать в голосовом канале", style=ButtonStyle.gray)
if item.available:
# self.add_item(self.button1) # Discord doesn't allow well formed URLs in buttons for some reason.
self.add_item(self.button2)
self.add_item(self.button3)
class ListenArtist(View):
def __init__(self, artist: Artist, *items: Item, timeout: float | None = None, disable_on_timeout: bool = False):
super().__init__(*items, timeout=timeout, disable_on_timeout=disable_on_timeout)
link_app = f"yandexmusic://artist/{artist.id}"
link_web = f"https://music.yandex.ru/artist/{artist.id}"
self.button1: Button = Button(label="Слушать в приложении", style=ButtonStyle.gray, url=link_app)
self.button2: Button = Button(label="Слушать в браузере", style=ButtonStyle.gray, url=link_web)
self.button3: PlayArtistButton = PlayArtistButton(artist, label="Слушать в голосовом канале", style=ButtonStyle.gray)
# self.add_item(self.button1) # Discord doesn't allow well formed URLs in buttons for some reason.
if artist.available:
self.add_item(self.button2)
self.add_item(self.button3)
class ListenPlaylist(View):
def __init__(self, playlist: Playlist, *items: Item, timeout: float | None = None, disable_on_timeout: bool = False):
super().__init__(*items, timeout=timeout, disable_on_timeout=disable_on_timeout)
link_app = f"yandexmusic://playlist/{playlist.playlist_uuid}"
link_web = f"https://music.yandex.ru/playlist/{playlist.playlist_uuid}"
self.button1: Button = Button(label="Слушать в приложении", style=ButtonStyle.gray, url=link_app)
self.button2: Button = Button(label="Слушать в браузере", style=ButtonStyle.gray, url=link_web)
self.button3: PlayPlaylistButton = PlayPlaylistButton(playlist, label="Слушать в голосовом канале", style=ButtonStyle.gray)
# self.add_item(self.button1) # Discord doesn't allow well formed URLs in buttons for some reason.
if playlist.available:
self.add_item(self.button2)
self.add_item(self.button3)
async def process_track(track: Track) -> Embed:
"""Generate track embed.
async def generate_item_embed(item: Track | Album | Artist | Playlist) -> Embed:
"""Generate item embed.
Args:
track (yandex_music.Track): Track to be processed.
item (yandex_music.Track | yandex_music.Album | yandex_music.Artist | yandex_music.Playlist): Item to be processed.
Returns:
discord.Embed: Track embed.
discord.Embed: Item embed.
"""
title = cast(str, track.title) # casted types are always there, blame JS for that
avail = cast(bool, track.available)
artists = track.artists_name()
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
bg_video = track.background_video_uri
metadata = track.meta_data
year = track.albums[0].year
artist = track.artists[0]
cover_url = track.get_cover_url('400x400')
color = await get_average_color_from_url(cover_url)
if explicit:
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 + '>'
duration_m = duration // 60000
duration_s = ceil(duration / 1000) - duration_m * 60
artist_url = f"https://music.yandex.ru/artist/{artist.id}"
artist_cover = artist.cover
if not artist_cover:
artist_cover_url = artist.get_op_image_url()
logging.debug(f"Generating embed for type: '{type(item).__name__}'")
if isinstance(item, Track):
return await generate_track_embed(item)
elif isinstance(item, Album):
return await generate_album_embed(item)
elif isinstance(item, Artist):
return await generate_artist_embed(item)
elif isinstance(item, Playlist):
return await generate_playlist_embed(item)
else:
artist_cover_url = artist_cover.get_url()
embed = discord.Embed(
title=title,
description=", ".join(albums),
color=color,
)
embed.set_thumbnail(url=cover_url)
embed.set_author(name=", ".join(artists), url=artist_url, icon_url=artist_cover_url)
embed.add_field(name="Текст песни", value="Есть" if lyrics else "Нет")
embed.add_field(name="Длительность", value=f"{duration_m}:{duration_s:02}")
if year:
embed.add_field(name="Год выпуска", value=str(year))
if metadata:
if metadata.year:
embed.add_field(name="Год выхода", value=str(metadata.year))
if metadata.number:
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"Трек в данный момент недоступен.")
return embed
async def process_album(album: Album) -> Embed:
"""Generate album embed.
Args:
album (yandex_music.Album): Album to process.
Returns:
discord.Embed: Album embed.
"""
title = cast(str, 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
likes_count = album.likes_count
artist = album.artists[0]
cover_url = album.get_cover_url('400x400')
color = await get_average_color_from_url(cover_url)
if isinstance(album.labels[0], Label):
labels = [cast(Label, label).name for label in album.labels]
else:
labels = [cast(str, label) for label in album.labels]
if version:
title += f' *{version}*'
if explicit:
explicit_eid = getenv('EXPLICIT_EID')
if not explicit_eid:
raise ValueError('You must specify explicit emoji id in your enviroment.')
title += ' <:explicit:' + explicit_eid + '>'
artist_url = f"https://music.yandex.ru/artist/{artist.id}"
artist_cover = artist.cover
if not artist_cover:
artist_cover_url = artist.get_op_image_url()
else:
artist_cover_url = artist_cover.get_url()
embed = discord.Embed(
title=title,
description=description,
color=color,
)
embed.set_thumbnail(url=cover_url)
embed.set_author(name=", ".join(artists), url=artist_url, icon_url=artist_cover_url)
if year:
embed.add_field(name="Год выпуска", value=str(year))
if duration:
duration_m = duration // 60000
duration_s = ceil(duration / 1000) - duration_m * 60
embed.add_field(name="Длительность", value=f"{duration_m}:{duration_s:02}")
if track_count is not None:
if track_count > 1:
embed.add_field(name="Треки", value=str(track_count))
else:
embed.add_field(name="Треки", value="Сингл")
if likes_count:
embed.add_field(name="Лайки", value=str(likes_count))
if len(labels) > 1:
embed.add_field(name="Лейблы", value=", ".join(labels))
else:
embed.add_field(name="Лейбл", value=", ".join(labels))
if not avail:
embed.set_footer(text=f"Альбом в данный момент недоступен.")
return embed
async def process_artist(artist: Artist) -> Embed:
"""Generate artist embed.
Args:
artist (yandex_music.Artist): Artist to process.
Returns:
discord.Embed: 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:
cover_url = artist.get_op_image_url('400x400')
else:
cover_url = artist.cover.get_url(size='400x400')
color = await get_average_color_from_url(cover_url)
embed = discord.Embed(
title=name,
description=description.text if description else None,
color=color,
)
embed.set_thumbnail(url=cover_url)
if likes_count:
embed.add_field(name="Лайки", value=str(likes_count))
# if ratings:
# embed.add_field(name="Слушателей за месяц", value=str(ratings.month)) # Wrong numbers?
if counts:
embed.add_field(name="Треки", value=str(counts.tracks))
embed.add_field(name="Альбомы", value=str(counts.direct_albums))
if artist.genres:
genres = [genre.capitalize() for genre in artist.genres]
if len(genres) > 1:
embed.add_field(name="Жанры", value=", ".join(genres))
else:
embed.add_field(name="Жанр", value=", ".join(genres))
if not avail:
embed.set_footer(text=f"Артист в данный момент недоступен.")
return embed
async def process_playlist(playlist: Playlist) -> Embed:
"""Generate playlist embed.
Args:
playlist (yandex_music.Playlist): Playlist to process.
Returns:
discord.Embed: Playlist embed.
"""
title = cast(str, playlist.title)
track_count = playlist.track_count
avail = cast(bool, playlist.available)
description = playlist.description_formatted
year = playlist.created
modified = playlist.modified
duration = playlist.duration_ms
likes_count = playlist.likes_count
color = 0x000
cover_url = None
if playlist.cover and playlist.cover.uri:
cover_url = f"https://{playlist.cover.uri.replace('%%', '400x400')}"
else:
tracks = await playlist.fetch_tracks_async()
for i in range(len(tracks)):
track = tracks[i].track
try:
cover_url = f"https://{track.albums[0].cover_uri.replace('%%', '400x400')}" # type: ignore # Errors are being caught below.
break
except (TypeError, IndexError):
continue
if cover_url:
color = await get_average_color_from_url(cover_url)
embed = discord.Embed(
title=title,
description=description,
color=color,
)
embed.set_thumbnail(url=cover_url)
if year:
embed.add_field(name="Год создания", value=str(year).split('-')[0])
if modified:
embed.add_field(name="Изменён", value=str(modified).split('-')[0])
if duration:
duration_m = duration // 60000
duration_s = ceil(duration / 1000) - duration_m * 60
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))
if likes_count:
embed.add_field(name="Лайки", value=str(likes_count))
if not avail:
embed.set_footer(text=f"Плейлист в данный момент недоступен.")
return embed
raise ValueError(f"Unknown item type: {type(item).__name__}")

359
MusicBot/cogs/utils/misc.py
View File

@@ -1,11 +1,13 @@
from typing import Any, cast
from math import ceil
from typing import Any
from os import getenv
from yandex_music import Track
from discord.ui import View, Button, Item
from discord import ButtonStyle, Interaction, ApplicationContext, Embed
import aiohttp
from io import BytesIO
from PIL import Image
from MusicBot.cogs.utils.voice_extension import VoiceExtension
from yandex_music import Track, Album, Artist, Playlist, Label
from discord import Embed
def generate_likes_embed(tracks: list[Track]) -> Embed:
track_count = len(tracks)
@@ -32,7 +34,7 @@ def generate_likes_embed(tracks: list[Track]) -> Embed:
return embed
def generate_playlist_embed(page: int, playlists: list[tuple[str, int]]) -> Embed:
def generate_playlists_embed(page: int, playlists: list[tuple[str, int]]) -> Embed:
count = 15 * page
length = len(playlists)
embed = Embed(
@@ -62,102 +64,277 @@ def generate_queue_embed(page: int, tracks_list: list[dict[str, Any]]) -> Embed:
embed.add_field(name=f"{i} - {track['title']} - {duration_m}:{duration_s:02d}", value="", inline=False)
return embed
class MPNextButton(Button, VoiceExtension):
def __init__(self, **kwargs):
Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None)
async def generate_track_embed(track: Track) -> Embed:
title = cast(str, track.title)
avail = cast(bool, track.available)
artists = track.artists_name()
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
bg_video = track.background_video_uri
metadata = track.meta_data
year = track.albums[0].year
artist = track.artists[0]
cover_url = track.get_cover_url('400x400')
color = await get_average_color_from_url(cover_url)
if explicit:
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 + '>'
duration_m = duration // 60000
duration_s = ceil(duration / 1000) - duration_m * 60
artist_url = f"https://music.yandex.ru/artist/{artist.id}"
artist_cover = artist.cover
if not artist_cover:
artist_cover_url = artist.get_op_image_url()
else:
artist_cover_url = artist_cover.get_url()
embed = Embed(
title=title,
description=", ".join(albums),
color=color,
)
embed.set_thumbnail(url=cover_url)
embed.set_author(name=", ".join(artists), url=artist_url, icon_url=artist_cover_url)
embed.add_field(name="Текст песни", value="Есть" if lyrics else "Нет")
embed.add_field(name="Длительность", value=f"{duration_m}:{duration_s:02}")
if year:
embed.add_field(name="Год выпуска", value=str(year))
if metadata:
if metadata.year:
embed.add_field(name="Год выхода", value=str(metadata.year))
if metadata.number:
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"Трек в данный момент недоступен.")
return embed
async def generate_album_embed(album: Album) -> Embed:
title = cast(str, 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
likes_count = album.likes_count
artist = album.artists[0]
cover_url = album.get_cover_url('400x400')
color = await get_average_color_from_url(cover_url)
if isinstance(album.labels[0], Label):
labels = [cast(Label, label).name for label in album.labels]
else:
labels = [cast(str, label) for label in album.labels]
if version:
title += f' *{version}*'
if explicit:
explicit_eid = getenv('EXPLICIT_EID')
if not explicit_eid:
raise ValueError('You must specify explicit emoji id in your enviroment.')
title += ' <:explicit:' + explicit_eid + '>'
artist_url = f"https://music.yandex.ru/artist/{artist.id}"
artist_cover = artist.cover
if not artist_cover:
artist_cover_url = artist.get_op_image_url()
else:
artist_cover_url = artist_cover.get_url()
embed = Embed(
title=title,
description=description,
color=color,
)
embed.set_thumbnail(url=cover_url)
embed.set_author(name=", ".join(artists), url=artist_url, icon_url=artist_cover_url)
if year:
embed.add_field(name="Год выпуска", value=str(year))
if duration:
duration_m = duration // 60000
duration_s = ceil(duration / 1000) - duration_m * 60
embed.add_field(name="Длительность", value=f"{duration_m}:{duration_s:02}")
if track_count is not None:
if track_count > 1:
embed.add_field(name="Треки", value=str(track_count))
else:
embed.add_field(name="Треки", value="Сингл")
if likes_count:
embed.add_field(name="Лайки", value=str(likes_count))
if len(labels) > 1:
embed.add_field(name="Лейблы", value=", ".join(labels))
else:
embed.add_field(name="Лейбл", value=", ".join(labels))
if not avail:
embed.set_footer(text=f"Альбом в данный момент недоступен.")
return 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:
cover_url = artist.get_op_image_url('400x400')
else:
cover_url = artist.cover.get_url(size='400x400')
color = await get_average_color_from_url(cover_url)
embed = Embed(
title=name,
description=description.text if description else None,
color=color,
)
embed.set_thumbnail(url=cover_url)
if likes_count:
embed.add_field(name="Лайки", value=str(likes_count))
# if ratings:
# embed.add_field(name="Слушателей за месяц", value=str(ratings.month)) # Wrong numbers?
if counts:
embed.add_field(name="Треки", value=str(counts.tracks))
async def callback(self, interaction: Interaction) -> None:
if not interaction.user:
return
user = self.users_db.get_user(interaction.user.id)
page = user['playlists_page'] + 1
self.users_db.update(interaction.user.id, {'playlists_page': page})
embed = generate_playlist_embed(page, user['playlists'])
await interaction.edit(embed=embed, view=MyPlaylists(interaction))
embed.add_field(name="Альбомы", value=str(counts.direct_albums))
class MPPrevButton(Button, VoiceExtension):
def __init__(self, **kwargs):
Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None)
if artist.genres:
genres = [genre.capitalize() for genre in artist.genres]
if len(genres) > 1:
embed.add_field(name="Жанры", value=", ".join(genres))
else:
embed.add_field(name="Жанр", value=", ".join(genres))
if not avail:
embed.set_footer(text=f"Артист в данный момент недоступен.")
return embed
async def callback(self, interaction: Interaction) -> None:
if not interaction.user:
return
user = self.users_db.get_user(interaction.user.id)
page = user['playlists_page'] - 1
self.users_db.update(interaction.user.id, {'playlists_page': page})
embed = generate_playlist_embed(page, user['playlists'])
await interaction.edit(embed=embed, view=MyPlaylists(interaction))
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_formatted
year = playlist.created
modified = playlist.modified
duration = playlist.duration_ms
likes_count = playlist.likes_count
class MyPlaylists(View, VoiceExtension):
def __init__(self, ctx: ApplicationContext | Interaction, *items: Item, timeout: float | None = 3600, disable_on_timeout: bool = True):
View.__init__(self, *items, timeout=timeout, disable_on_timeout=disable_on_timeout)
VoiceExtension.__init__(self, None)
if not ctx.user:
return
user = self.users_db.get_user(ctx.user.id)
count = 10 * user['playlists_page']
color = 0x000
cover_url = None
next_button = MPNextButton(style=ButtonStyle.primary, emoji='▶️')
prev_button = MPPrevButton(style=ButtonStyle.primary, emoji='◀️')
if playlist.cover and playlist.cover.uri:
cover_url = f"https://{playlist.cover.uri.replace('%%', '400x400')}"
else:
tracks = await playlist.fetch_tracks_async()
for i in range(len(tracks)):
track = tracks[i].track
if not track or not track.albums or not track.albums[0].cover_uri:
continue
if not user['playlists'][count + 10:]:
next_button.disabled = True
if not user['playlists'][:count]:
prev_button.disabled = True
if cover_url:
color = await get_average_color_from_url(cover_url)
self.add_item(prev_button)
self.add_item(next_button)
embed = Embed(
title=title,
description=description,
color=color,
)
embed.set_thumbnail(url=cover_url)
class QNextButton(Button, VoiceExtension):
def __init__(self, **kwargs):
Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None)
async def callback(self, interaction: Interaction) -> None:
if not interaction.user or not interaction.guild:
return
user = self.users_db.get_user(interaction.user.id)
page = user['queue_page'] + 1
self.users_db.update(interaction.user.id, {'queue_page': page})
tracks = self.db.get_tracks_list(interaction.guild.id, 'next')
embed = generate_queue_embed(page, tracks)
await interaction.edit(embed=embed, view=QueueView(interaction))
if year:
embed.add_field(name="Год создания", value=str(year).split('-')[0])
class QPrevButton(Button, VoiceExtension):
def __init__(self, **kwargs):
Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None)
async def callback(self, interaction: Interaction) -> None:
if not interaction.user or not interaction.guild:
return
user = self.users_db.get_user(interaction.user.id)
page = user['queue_page'] - 1
self.users_db.update(interaction.user.id, {'queue_page': page})
tracks = self.db.get_tracks_list(interaction.guild.id, 'next')
embed = generate_queue_embed(page, tracks)
await interaction.edit(embed=embed, view=QueueView(interaction))
if modified:
embed.add_field(name="Изменён", value=str(modified).split('-')[0])
class QueueView(View, VoiceExtension):
def __init__(self, ctx: ApplicationContext | Interaction, *items: Item, timeout: float | None = 3600, disable_on_timeout: bool = True):
View.__init__(self, *items, timeout=timeout, disable_on_timeout=disable_on_timeout)
VoiceExtension.__init__(self, None)
if not ctx.user or not ctx.guild:
return
if duration:
duration_m = duration // 60000
duration_s = ceil(duration / 1000) - duration_m * 60
embed.add_field(name="Длительность", value=f"{duration_m}:{duration_s:02}")
tracks = self.db.get_tracks_list(ctx.guild.id, 'next')
user = self.users_db.get_user(ctx.user.id)
count = 15 * user['queue_page']
if track_count is not None:
embed.add_field(name="Треки", value=str(track_count))
next_button = QNextButton(style=ButtonStyle.primary, emoji='▶️')
prev_button = QPrevButton(style=ButtonStyle.primary, emoji='◀️')
if likes_count:
embed.add_field(name="Лайки", value=str(likes_count))
if not tracks[count + 15:]:
next_button.disabled = True
if not tracks[:count]:
prev_button.disabled = True
if not avail:
embed.set_footer(text=f"Плейлист в данный момент недоступен.")
self.add_item(prev_button)
self.add_item(next_button)
return embed
async def get_average_color_from_url(url: str) -> int:
"""Get image from url and calculate its average color to use in embeds.
Args:
url (str): Image url.
Returns:
int: RGB Hex code. 0x000 if failed.
"""
try:
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
response.raise_for_status()
result = await response.read()
img_file = Image.open(BytesIO(result))
img = img_file.convert('RGB')
width, height = img.size
r_total, g_total, b_total = 0, 0, 0
for y in range(height):
for x in range(width):
r, g, b = cast(tuple, img.getpixel((x, y)))
r_total += r
g_total += g
b_total += b
count = width * height
r = r_total // count
g = g_total // count
b = b_total // count
return (r << 16) + (g << 8) + b
except Exception:
return 0x000

7
MusicBot/cogs/utils/player.py
View File

@@ -1,3 +1,4 @@
import logging
from discord.ui import View, Button, Item
from discord import ButtonStyle, Interaction, ApplicationContext
@@ -9,6 +10,7 @@ class ToggleRepeatButton(Button, VoiceExtension):
VoiceExtension.__init__(self, None)
async def callback(self, interaction: Interaction) -> None:
logging.debug('Repeat button callback...')
if not interaction.guild:
return
gid = interaction.guild.id
@@ -22,6 +24,7 @@ class ToggleShuffleButton(Button, VoiceExtension):
VoiceExtension.__init__(self, None)
async def callback(self, interaction: Interaction) -> None:
logging.debug('Shuffle button callback...')
if not interaction.guild:
return
gid = interaction.guild.id
@@ -35,6 +38,7 @@ class PlayPauseButton(Button, VoiceExtension):
VoiceExtension.__init__(self, None)
async def callback(self, interaction: Interaction) -> None:
logging.debug('Play/Pause button callback...')
if not await self.voice_check(interaction):
return
@@ -59,6 +63,7 @@ class NextTrackButton(Button, VoiceExtension):
VoiceExtension.__init__(self, None)
async def callback(self, interaction: Interaction) -> None:
logging.debug('Next track button callback...')
if not await self.voice_check(interaction):
return
title = await self.next_track(interaction)
@@ -71,6 +76,7 @@ class PrevTrackButton(Button, VoiceExtension):
VoiceExtension.__init__(self, None)
async def callback(self, interaction: Interaction) -> None:
logging.debug('Previous track button callback...')
if not await self.voice_check(interaction):
return
title = await self.prev_track(interaction)
@@ -83,6 +89,7 @@ class LikeButton(Button, VoiceExtension):
VoiceExtension.__init__(self, None)
async def callback(self, interaction: Interaction) -> None:
logging.debug('Like button callback...')
if await self.voice_check(interaction):
vc = await self.get_voice_client(interaction)
if not vc or not vc.is_playing:

105
MusicBot/cogs/utils/views.py Normal file
View File

@@ -0,0 +1,105 @@
from discord.ui import View, Button, Item
from discord import ButtonStyle, Interaction, ApplicationContext
from MusicBot.cogs.utils.voice_extension import VoiceExtension
from MusicBot.cogs.utils.misc import generate_playlists_embed, generate_queue_embed
class MPNextButton(Button, VoiceExtension):
def __init__(self, **kwargs):
Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None)
async def callback(self, interaction: Interaction) -> None:
if not interaction.user:
return
user = self.users_db.get_user(interaction.user.id)
page = user['playlists_page'] + 1
self.users_db.update(interaction.user.id, {'playlists_page': page})
embed = generate_playlists_embed(page, user['playlists'])
await interaction.edit(embed=embed, view=MyPlaylists(interaction))
class MPPrevButton(Button, VoiceExtension):
def __init__(self, **kwargs):
Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None)
async def callback(self, interaction: Interaction) -> None:
if not interaction.user:
return
user = self.users_db.get_user(interaction.user.id)
page = user['playlists_page'] - 1
self.users_db.update(interaction.user.id, {'playlists_page': page})
embed = generate_playlists_embed(page, user['playlists'])
await interaction.edit(embed=embed, view=MyPlaylists(interaction))
class MyPlaylists(View, VoiceExtension):
def __init__(self, ctx: ApplicationContext | Interaction, *items: Item, timeout: float | None = 3600, disable_on_timeout: bool = True):
View.__init__(self, *items, timeout=timeout, disable_on_timeout=disable_on_timeout)
VoiceExtension.__init__(self, None)
if not ctx.user:
return
user = self.users_db.get_user(ctx.user.id)
count = 10 * user['playlists_page']
next_button = MPNextButton(style=ButtonStyle.primary, emoji='▶️')
prev_button = MPPrevButton(style=ButtonStyle.primary, emoji='◀️')
if not user['playlists'][count + 10:]:
next_button.disabled = True
if not user['playlists'][:count]:
prev_button.disabled = True
self.add_item(prev_button)
self.add_item(next_button)
class QNextButton(Button, VoiceExtension):
def __init__(self, **kwargs):
Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None)
async def callback(self, interaction: Interaction) -> None:
if not interaction.user or not interaction.guild:
return
user = self.users_db.get_user(interaction.user.id)
page = user['queue_page'] + 1
self.users_db.update(interaction.user.id, {'queue_page': page})
tracks = self.db.get_tracks_list(interaction.guild.id, 'next')
embed = generate_queue_embed(page, tracks)
await interaction.edit(embed=embed, view=QueueView(interaction))
class QPrevButton(Button, VoiceExtension):
def __init__(self, **kwargs):
Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None)
async def callback(self, interaction: Interaction) -> None:
if not interaction.user or not interaction.guild:
return
user = self.users_db.get_user(interaction.user.id)
page = user['queue_page'] - 1
self.users_db.update(interaction.user.id, {'queue_page': page})
tracks = self.db.get_tracks_list(interaction.guild.id, 'next')
embed = generate_queue_embed(page, tracks)
await interaction.edit(embed=embed, view=QueueView(interaction))
class QueueView(View, VoiceExtension):
def __init__(self, ctx: ApplicationContext | Interaction, *items: Item, timeout: float | None = 3600, disable_on_timeout: bool = True):
View.__init__(self, *items, timeout=timeout, disable_on_timeout=disable_on_timeout)
VoiceExtension.__init__(self, None)
if not ctx.user or not ctx.guild:
return
tracks = self.db.get_tracks_list(ctx.guild.id, 'next')
user = self.users_db.get_user(ctx.user.id)
count = 15 * user['queue_page']
next_button = QNextButton(style=ButtonStyle.primary, emoji='▶️')
prev_button = QPrevButton(style=ButtonStyle.primary, emoji='◀️')
if not tracks[count + 15:]:
next_button.disabled = True
if not tracks[:count]:
prev_button.disabled = True
self.add_item(prev_button)
self.add_item(next_button)

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

@@ -1,197 +1,116 @@
import aiohttp
import asyncio
from os import getenv
from math import ceil
import logging
from typing import Literal, cast
from io import BytesIO
from PIL import Image
from yandex_music import Track, ClientAsync
import discord
from discord import Interaction, ApplicationContext, RawReactionActionEvent
from MusicBot.cogs.utils.misc import generate_track_embed
from MusicBot.database import VoiceGuildsDatabase, BaseUsersDatabase
# This should be in find.py but recursive import is a thing
async def generate_player_embed(track: Track) -> discord.Embed:
"""Generate track embed for player.
Args:
track (yandex_music.Track): Track to be processed.
Returns:
discord.Embed: Track embed.
"""
title = cast(str, track.title) # casted types are always there, blame JS for that
avail = cast(bool, track.available)
artists = track.artists_name()
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
bg_video = track.background_video_uri
metadata = track.meta_data
year = track.albums[0].year if track.albums else None
artist = track.artists[0] if track.artists else None
if track.cover_uri:
cover_url = f"https://{track.cover_uri.replace('%%', '400x400')}"
else:
cover_url = None
if cover_url:
color = await get_average_color_from_url(cover_url)
else:
color = None
if explicit:
explicit_eid = getenv('EXPLICIT_EID')
if not explicit_eid:
raise ValueError('You must specify explicit emoji id in your enviroment.')
title += ' <:explicit:' + explicit_eid + '>'
duration_m = duration // 60000
duration_s = ceil(duration / 1000) - duration_m * 60
if artist:
artist_url = f"https://music.yandex.ru/artist/{artist.id}"
artist_cover = artist.cover if artist else None
if artist and not artist_cover:
artist_cover_url = artist.get_op_image_url()
elif artist_cover:
artist_cover_url = artist_cover.get_url()
else:
artist_cover_url = None
else:
artist_url = None
artist_cover_url = None
embed = discord.Embed(
title=title,
description=", ".join(albums),
color=color,
)
embed.set_thumbnail(url=cover_url)
embed.set_author(name=", ".join(artists), url=artist_url, icon_url=artist_cover_url)
embed.add_field(name="Текст песни", value="Есть" if lyrics else "Нет")
embed.add_field(name="Длительность", value=f"{duration_m}:{duration_s:02}")
if year:
embed.add_field(name="Год выпуска", value=str(year))
if metadata:
if metadata.year:
embed.add_field(name="Год выхода", value=str(metadata.year))
if metadata.number:
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"Трек в данный момент недоступен.")
return embed
async def get_average_color_from_url(url: str) -> int:
"""Get image from url and calculate its average color to use in embeds.
Args:
url (str): Image url.
Returns:
int: RGB Hex code. 0x000 if failed.
"""
try:
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
response.raise_for_status()
result = await response.read()
img_file = Image.open(BytesIO(result))
img = img_file.convert('RGB')
width, height = img.size
r_total, g_total, b_total = 0, 0, 0
for y in range(height):
for x in range(width):
r, g, b = cast(tuple, img.getpixel((x, y)))
r_total += r
g_total += g
b_total += b
count = width * height
r = r_total // count
g = g_total // count
b = b_total // count
return (r << 16) + (g << 8) + b
except Exception:
return 0x000
class VoiceExtension:
def __init__(self, bot: discord.Bot | None) -> None:
self.bot = bot
self.db = VoiceGuildsDatabase()
self.users_db = BaseUsersDatabase()
async def update_player_embed(self, ctx: ApplicationContext | Interaction | RawReactionActionEvent, player_mid: int) -> None:
"""Update current player message by its id.
async def update_player_embed(self, ctx: ApplicationContext | Interaction | RawReactionActionEvent, player_mid: int) -> bool:
"""Update current player message by its id. Return True if updated, False if not.
Args:
ctx (ApplicationContext | Interaction): Context.
player_mid (int): Id of the player message. There can only be only one player in the guild.
Returns:
bool: True if updated, False if not.
"""
logging.debug(
f"Updating player embed using " +
"interaction context" if isinstance(ctx, Interaction) else
"application context" if isinstance(ctx, ApplicationContext) else
"raw reaction context" + " ..."
)
player = await self.get_player_message(ctx, player_mid)
if not player:
return False
gid = ctx.guild_id if isinstance(ctx, discord.RawReactionActionEvent) else ctx.guild.id if ctx.guild else None
uid = ctx.user_id if isinstance(ctx, discord.RawReactionActionEvent) else ctx.user.id if ctx.user else None
if not gid or not uid:
logging.warning("Guild ID or User ID not found in context")
return False
token = self.users_db.get_ym_token(uid)
if not token:
logging.debug(f"No token found for user {uid}")
return False
current_track = self.db.get_track(gid, 'current')
if not current_track:
logging.debug("No current track found")
return False
track = cast(Track, Track.de_json(
current_track,
client=ClientAsync(token) # type: ignore # Async client can be used here.
))
embed = await generate_track_embed(track)
if isinstance(ctx, Interaction) and ctx.message and ctx.message.id == player_mid:
# If interaction from player buttons
await ctx.edit(embed=embed)
else:
# If interaction from other buttons or commands. They should have their own response.
await player.edit(embed=embed)
return True
async def get_player_message(self, ctx: ApplicationContext | Interaction | RawReactionActionEvent, player_mid: int) -> discord.Message | None:
"""Fetch the player message by its id. Return the message if found, None if not.
Reset `current_player` field in the database if not found.
Args:
ctx (ApplicationContext | Interaction): Context.
player_mid (int): Id of the player message.
Returns:
discord.Message | None: Player message or None.
"""
logging.debug(f"Fetching player message {player_mid}...")
if not ctx.guild_id:
logging.warning("Guild ID not found in context")
return None
try:
if isinstance(ctx, Interaction):
player = ctx.client.get_message(player_mid)
elif isinstance(ctx, RawReactionActionEvent) and self.bot:
elif isinstance(ctx, RawReactionActionEvent):
if not self.bot:
raise ValueError("Bot instance is not set.")
player = self.bot.get_message(player_mid)
elif isinstance(ctx, ApplicationContext):
player = await ctx.fetch_message(player_mid)
else:
player = None
except discord.DiscordException:
return
raise ValueError(f"Invalid context type: '{type(ctx).__name__}'.")
except discord.DiscordException as e:
logging.debug(f"Failed to get player message: {e}")
self.db.update(ctx.guild_id, {'current_player': None})
return None
if not player:
return
gid = ctx.guild_id if isinstance(ctx, discord.RawReactionActionEvent) else ctx.guild.id if ctx.guild else None
uid = ctx.user_id if isinstance(ctx, discord.RawReactionActionEvent) else ctx.user.id if ctx.user else None
if gid and uid:
token = self.users_db.get_ym_token(uid)
current_track = self.db.get_track(gid, 'current')
if not current_track:
return
track = cast(Track, Track.de_json(
current_track,
client=ClientAsync(token) # type: ignore # Async client can be used here.
)
)
embed = await generate_player_embed(track)
if isinstance(ctx, Interaction) and ctx.message and ctx.message.id == player_mid:
# If interaction from player buttons
await ctx.edit(embed=embed)
else:
# If interaction from other buttons. They should have their own response.
await player.edit(embed=embed)
if player:
logging.debug(f"Player message found")
else:
logging.debug("Player message not found. Resetting current_player field.")
self.db.update(ctx.guild_id, {'current_player': None})
return player
async def voice_check(self, ctx: ApplicationContext | Interaction) -> bool:
"""Check if bot can perform voice tasks and respond if failed.
@@ -201,30 +120,36 @@ class VoiceExtension:
Returns:
bool: Check result.
"""
logging.debug("Checking voice requirements...")
if not ctx.user:
logging.warning("User not found in context.")
return False
token = self.users_db.get_ym_token(ctx.user.id)
if not token:
logging.debug(f"No token found for user {ctx.user.id}")
await ctx.respond("❌ Необходимо указать свой токен доступа с помощью команды /login.", delete_after=15, ephemeral=True)
return False
channel = ctx.channel
if not isinstance(channel, discord.VoiceChannel):
logging.debug("User is not in a voice channel")
await ctx.respond("❌ Вы должны отправить команду в голосовом канале.", delete_after=15, ephemeral=True)
return False
if isinstance(ctx, Interaction):
channels = ctx.client.voice_clients
else:
channels = ctx.bot.voice_clients
voice_chat = discord.utils.get(channels, guild=ctx.guild)
if not voice_chat:
logging.debug("Voice client not found")
await ctx.respond("❌ Добавьте бота в голосовой канал при помощи команды /voice join.", delete_after=15, ephemeral=True)
return False
logging.debug("Voice requirements met")
return True
async def get_voice_client(self, ctx: ApplicationContext | Interaction | RawReactionActionEvent) -> discord.VoiceClient | None:
"""Return voice client for the given guild id. Return None if not present.
@@ -234,27 +159,35 @@ class VoiceExtension:
Returns:
discord.VoiceClient | None: Voice client or None.
"""
logging.debug("Getting voice client...")
if isinstance(ctx, Interaction):
voice_chat = discord.utils.get(ctx.client.voice_clients, guild=ctx.guild)
elif isinstance(ctx, RawReactionActionEvent):
if not self.bot:
raise ValueError("Bot is not set.")
raise ValueError("Bot instance is not set.")
if not ctx.guild_id:
return
logging.warning("Guild ID not found in context")
return None
voice_chat = discord.utils.get(self.bot.voice_clients, guild=await self.bot.fetch_guild(ctx.guild_id))
else:
elif isinstance(ctx, ApplicationContext):
voice_chat = discord.utils.get(ctx.bot.voice_clients, guild=ctx.guild)
else:
raise ValueError(f"Invalid context type: '{type(ctx).__name__}'.")
if voice_chat:
logging.debug(f"Voice client found")
else:
logging.debug("Voice client not found")
return cast((discord.VoiceClient | None), voice_chat)
async def play_track(self, ctx: ApplicationContext | Interaction | RawReactionActionEvent, track: Track) -> str | None:
"""Download ``track`` by its id and play it in the voice channel. Return track title on success.
If sound is already playing, add track id to the queue. There's no response to the context.
Args:
ctx (ApplicationContext | Interaction): Context
track (Track): Track class with id and title specified.
track (Track): Track to play.
Returns:
str | None: Song title or None.
@@ -262,39 +195,45 @@ class VoiceExtension:
gid = ctx.guild_id if isinstance(ctx, discord.RawReactionActionEvent) else ctx.guild.id if ctx.guild else None
uid = ctx.user_id if isinstance(ctx, discord.RawReactionActionEvent) else ctx.user.id if ctx.user else None
if not gid or not uid:
logging.warning("Guild ID or User ID not found in context")
return None
vc = await self.get_voice_client(ctx)
if not vc:
return None
if isinstance(ctx, Interaction):
loop = ctx.client.loop
elif isinstance(ctx, ApplicationContext):
loop = ctx.bot.loop
else:
elif isinstance(ctx, RawReactionActionEvent):
if not self.bot:
raise ValueError("Bot is not set.")
loop = self.bot.loop
else:
raise ValueError(f"Invalid context type: '{type(ctx).__name__}'.")
guild = self.db.get_guild(gid)
await track.download_async(f'music/{gid}.mp3')
song = discord.FFmpegPCMAudio(f'music/{gid}.mp3', options='-vn -filter:a "volume=0.15"')
vc.play(song, after=lambda exc: asyncio.run_coroutine_threadsafe(self.next_track(ctx, after=True), loop))
logging.debug(f"Playing track '{track.title}'")
self.db.set_current_track(gid, track)
self.db.update(gid, {'is_stopped': False})
player = guild['current_player']
if player is not None:
await self.update_player_embed(ctx, player)
return track.title
async def stop_playing(self, ctx: ApplicationContext | Interaction | RawReactionActionEvent) -> None:
logging.debug("Stopping playback...")
gid = ctx.guild_id if isinstance(ctx, discord.RawReactionActionEvent) else ctx.guild.id if ctx.guild else None
if not gid:
logging.warning("Guild ID not found in context")
return
vc = await self.get_voice_client(ctx)
@@ -302,45 +241,51 @@ class VoiceExtension:
self.db.update(gid, {'current_track': None, 'is_stopped': True})
vc.stop()
return
async def next_track(self, ctx: ApplicationContext | Interaction | RawReactionActionEvent, *, after: bool = False) -> str | None:
"""Switch to the next track in the queue. Return track title on success.
Doesn't change track if stopped. Stop playing if tracks list is empty.
Args:
ctx (ApplicationContext | Interaction): Context
after (bool, optional): Whether the function was called by the after callback. Defaults to False.
after (bool, optional): Whether the function is being called by the after callback. Defaults to False.
Returns:
str | None: Track title or None.
"""
logging.debug("Switching to the next track")
gid = ctx.guild_id if isinstance(ctx, discord.RawReactionActionEvent) else ctx.guild.id if ctx.guild else None
uid = ctx.user_id if isinstance(ctx, discord.RawReactionActionEvent) else ctx.user.id if ctx.user else None
if not gid or not uid:
logging.warning("Guild ID or User ID not found in context.")
return
guild = self.db.get_guild(gid)
token = self.users_db.get_ym_token(uid)
title = None
if not token:
logging.debug(f"No token found for user {uid}")
return None
if guild['is_stopped']:
logging.debug("Playback is stopped, skipping...")
return None
if not await self.get_voice_client(ctx): # Silently return if bot got kicked
logging.debug("Voice client not found")
return None
current_track = guild['current_track']
ym_track = None
if guild['repeat'] and after:
return await self.repeat_current_track(ctx)
next_track = guild['current_track']
elif guild['shuffle']:
logging.debug("Shuffling tracks")
next_track = self.db.get_random_track(gid)
else:
logging.debug("Getting next track")
next_track = self.db.get_track(gid, 'next')
if current_track and guild['current_player']:
self.db.modify_track(gid, current_track, 'previous', 'insert')
if guild['current_track'] and guild['current_player']:
self.db.modify_track(gid, guild['current_track'], 'previous', 'insert')
if next_track:
ym_track = Track.de_json(
next_track,
@@ -355,7 +300,11 @@ class VoiceExtension:
if after and not guild['current_player'] and not isinstance(ctx, discord.RawReactionActionEvent):
await ctx.respond(f"Сейчас играет: **{title}**!", delete_after=15)
return title
return title
else:
self.db.update(gid, {'is_stopped': True, 'current_track': None})
return None
async def prev_track(self, ctx: ApplicationContext | Interaction) -> str | None:
"""Switch to the previous track in the queue. Repeat curren the song if no previous tracks.
@@ -367,52 +316,33 @@ class VoiceExtension:
Returns:
str | None: Track title or None.
"""
logging.debug("Switching to the previous track")
if not ctx.guild or not ctx.user:
logging.debug("Guild or User not found in context")
return None
gid = ctx.guild.id
token = self.users_db.get_ym_token(ctx.user.id)
current_track = self.db.get_track(gid, 'current')
prev_track = self.db.get_track(gid, 'previous')
title = None
if prev_track:
ym_track = Track.de_json(
prev_track,
client=ClientAsync(token) # type: ignore # Async client can be used here.
)
await self.stop_playing(ctx)
title = await self.play_track(
ctx,
ym_track # type: ignore # de_json should always work here.
)
elif current_track:
title = await self.repeat_current_track(ctx)
return title
async def repeat_current_track(self, ctx: ApplicationContext | Interaction | RawReactionActionEvent) -> str | None:
"""Repeat current track. Return track title on success.
Args:
ctx (ApplicationContext | Interaction): Context
Returns:
str | None: Track title or None.
"""
gid = ctx.guild_id if isinstance(ctx, discord.RawReactionActionEvent) else ctx.guild.id if ctx.guild else None
uid = ctx.user_id if isinstance(ctx, discord.RawReactionActionEvent) else ctx.user.id if ctx.user else None
if not gid or not uid:
if not token:
logging.debug(f"No token found for user {ctx.user.id}")
return
token = self.users_db.get_ym_token(gid)
current_track = self.db.get_track(gid, 'current')
if current_track:
if prev_track:
logging.debug("Previous track found")
track = prev_track
elif current_track:
logging.debug("No previous track found. Repeating current track")
track = self.db.get_track(gid, 'current')
else:
logging.debug("No previous or current track found")
track = None
if track:
ym_track = Track.de_json(
current_track,
track,
client=ClientAsync(token) # type: ignore # Async client can be used here.
)
await self.stop_playing(ctx)
@@ -425,24 +355,27 @@ class VoiceExtension:
async def like_track(self, ctx: ApplicationContext | Interaction) -> str | Literal['TRACK REMOVED'] | None:
"""Like current track. Return track title on success.
Args:
ctx (ApplicationContext | Interaction): Context.
Returns:
str | None: Track title or None.
"""
if not ctx.guild or not ctx.user:
logging.warning("Guild or User not found in context.")
return None
current_track = self.db.get_track(ctx.guild.id, 'current')
token = self.users_db.get_ym_token(ctx.user.id)
if not current_track or not token:
logging.debug("Current track or token not found")
return None
client = await ClientAsync(token).init()
likes = await client.users_likes_tracks()
if not likes:
logging.debug("No likes found")
return None
ym_track = cast(Track, Track.de_json(
@@ -451,10 +384,13 @@ class VoiceExtension:
)
)
if ym_track.id not in [track.id for track in likes.tracks]:
logging.debug("Track not found in likes. Adding...")
await ym_track.like_async()
return ym_track.title
else:
logging.debug("Track found in likes. Removing...")
if not client.me or not client.me.account or not client.me.account.uid:
logging.debug("Client account not found")
return None
await client.users_likes_tracks_remove(ym_track.id, client.me.account.uid)
return 'TRACK REMOVED'
return 'TRACK REMOVED'

44
MusicBot/cogs/voice.py
View File

@@ -1,3 +1,4 @@
import logging
from typing import cast
import discord
@@ -5,9 +6,10 @@ from discord.ext.commands import Cog
from yandex_music import Track, ClientAsync
from MusicBot.cogs.utils.voice_extension import VoiceExtension, generate_player_embed
from MusicBot.cogs.utils.voice_extension import VoiceExtension
from MusicBot.cogs.utils.player import Player
from MusicBot.cogs.utils.misc import QueueView, generate_queue_embed
from MusicBot.cogs.utils.misc import generate_queue_embed, generate_track_embed
from MusicBot.cogs.utils.views import QueueView
def setup(bot: discord.Bot):
bot.add_cog(Voice(bot))
@@ -24,23 +26,27 @@ class Voice(Cog, VoiceExtension):
@Cog.listener()
async def on_voice_state_update(self, member: discord.Member, before: discord.VoiceState, after: discord.VoiceState) -> None:
logging.debug(f"Voice state update for member {member.id} in guild {member.guild.id}")
gid = member.guild.id
guild = self.db.get_guild(gid)
channel = after.channel or before.channel
if not channel:
logging.debug(f"No channel found for member {member.id}")
return
discord_guild = await self.bot.fetch_guild(gid)
vc = cast(discord.VoiceClient | None, discord.utils.get(self.bot.voice_clients, guild=discord_guild))
if len(channel.members) == 1 and vc:
logging.debug(f"Clearing history and stopping playback for guild {gid}")
self.db.clear_history(gid)
self.db.update(gid, {'current_track': None, 'is_stopped': True})
vc.stop()
elif len(channel.members) > 2 and not guild['always_allow_menu']:
current_player = self.db.get_current_player(gid)
if current_player:
logging.debug(f"Disabling current player for guild {gid} due to multiple members")
self.db.update(gid, {'current_player': None, 'repeat': False, 'shuffle': False})
try:
message = await channel.fetch_message(current_player)
@@ -51,6 +57,7 @@ class Voice(Cog, VoiceExtension):
@Cog.listener()
async def on_raw_reaction_add(self, payload: discord.RawReactionActionEvent) -> None:
logging.debug(f"Reaction added by user {payload.user_id} in channel {payload.channel_id}")
if not self.bot.user or not payload.member:
return
@@ -79,23 +86,29 @@ class Voice(Cog, VoiceExtension):
vote_data = votes[str(payload.message_id)]
if payload.emoji.name == '✅':
logging.debug(f"User {payload.user_id} voted positively for message {payload.message_id}")
vote_data['positive_votes'].append(payload.user_id)
elif payload.emoji.name == '❌':
logging.debug(f"User {payload.user_id} voted negatively for message {payload.message_id}")
vote_data['negative_votes'].append(payload.user_id)
total_members = len(channel.members)
required_votes = 2 if total_members <= 5 else 4 if total_members <= 10 else 6 if total_members <= 15 else 9
if len(vote_data['positive_votes']) >= required_votes:
logging.debug(f"Enough positive votes for message {payload.message_id}")
if vote_data['action'] == 'next':
logging.debug(f"Skipping track for message {payload.message_id}")
self.db.update(guild_id, {'is_stopped': False})
title = await self.next_track(payload)
await message.clear_reactions()
await message.edit(content=f"Сейчас играет: **{title}**!", delete_after=15)
del votes[str(payload.message_id)]
elif vote_data['action'] == 'add_track':
logging.debug(f"Adding track for message {payload.message_id}")
await message.clear_reactions()
track = vote_data['vote_content']
if not track:
logging.debug(f"Recieved empty vote context for message {payload.message_id}")
return
self.db.update(guild_id, {'is_stopped': False})
self.db.modify_track(guild_id, track, 'next', 'append')
@@ -106,9 +119,11 @@ class Voice(Cog, VoiceExtension):
await message.edit(content=f"Сейчас играет: **{title}**!", delete_after=15)
del votes[str(payload.message_id)]
elif vote_data['action'] in ('add_album', 'add_artist', 'add_playlist'):
logging.debug(f"Performing '{vote_data['action']}' action for message {payload.message_id}")
tracks = vote_data['vote_content']
await message.clear_reactions()
if not tracks:
logging.debug(f"Recieved empty vote context for message {payload.message_id}")
return
self.db.update(guild_id, {'is_stopped': False})
self.db.modify_track(guild_id, tracks, 'next', 'extend')
@@ -119,6 +134,7 @@ class Voice(Cog, VoiceExtension):
await message.edit(content=f"Сейчас играет: **{title}**!", delete_after=15)
del votes[str(payload.message_id)]
elif len(vote_data['negative_votes']) >= required_votes:
logging.debug(f"Enough negative votes for message {payload.message_id}")
await message.clear_reactions()
await message.edit(content='Запрос был отклонён.', delete_after=15)
del votes[str(payload.message_id)]
@@ -127,6 +143,7 @@ class Voice(Cog, VoiceExtension):
@Cog.listener()
async def on_raw_reaction_remove(self, payload: discord.RawReactionActionEvent) -> None:
logging.debug(f"Reaction removed by user {payload.user_id} in channel {payload.channel_id}")
if not self.bot.user:
return
@@ -146,14 +163,17 @@ class Voice(Cog, VoiceExtension):
vote_data = votes[str(payload.message_id)]
if payload.emoji.name == '✔️':
logging.debug(f"User {payload.user_id} removed positive vote for message {payload.message_id}")
del vote_data['positive_votes'][payload.user_id]
elif payload.emoji.name == '❌':
logging.debug(f"User {payload.user_id} removed negative vote for message {payload.message_id}")
del vote_data['negative_votes'][payload.user_id]
self.db.update(guild_id, {'votes': votes})
@voice.command(name="menu", description="Создать меню проигрывателя. Доступно только если вы единственный в голосовом канале.")
async def menu(self, ctx: discord.ApplicationContext) -> None:
logging.debug(f"Menu command invoked by user {ctx.author.id} in guild {ctx.guild.id}")
if not await self.voice_check(ctx):
return
@@ -162,11 +182,12 @@ class Voice(Cog, VoiceExtension):
embed = None
if len(channel.members) > 2 and not guild['always_allow_menu']:
logging.debug(f"Action declined: other members are present in the voice channel")
await ctx.respond("❌ Вы не единственный в голосовом канале.", ephemeral=True)
return
if guild['current_track']:
embed = await generate_player_embed(
embed = await generate_track_embed(
Track.de_json(
guild['current_track'],
client=ClientAsync() # type: ignore # Async client can be used here.
@@ -179,6 +200,7 @@ class Voice(Cog, VoiceExtension):
embed.remove_footer()
if guild['current_player']:
logging.debug(f"Deleteing old player menu {guild['current_player']} in guild {ctx.guild.id}")
message = await ctx.fetch_message(guild['current_player'])
await message.delete()
@@ -188,11 +210,12 @@ class Voice(Cog, VoiceExtension):
@voice.command(name="join", description="Подключиться к голосовому каналу, в котором вы сейчас находитесь.")
async def join(self, ctx: discord.ApplicationContext) -> None:
logging.debug(f"Join command invoked by user {ctx.author.id} in guild {ctx.guild.id}")
member = cast(discord.Member, ctx.author)
vc = await self.get_voice_client(ctx)
if not member.guild_permissions.manage_channels:
response_message = "❌ У вас нет прав для выполнения этой команды."
elif vc and vc.is_playing():
elif vc and vc.is_connected():
response_message = "❌ Бот уже находится в голосовом канале. Выключите его с помощью команды /voice leave."
elif isinstance(ctx.channel, discord.VoiceChannel):
await ctx.channel.connect(timeout=15)
@@ -204,6 +227,7 @@ class Voice(Cog, VoiceExtension):
@voice.command(description="Заставить бота покинуть голосовой канал.")
async def leave(self, ctx: discord.ApplicationContext) -> None:
logging.debug(f"Leave command invoked by user {ctx.author.id} in guild {ctx.guild.id}")
member = cast(discord.Member, ctx.author)
if not member.guild_permissions.manage_channels:
await ctx.respond("❌ У вас нет прав для выполнения этой команды.", delete_after=15, ephemeral=True)
@@ -211,13 +235,15 @@ class Voice(Cog, VoiceExtension):
vc = await self.get_voice_client(ctx)
if await self.voice_check(ctx) and vc:
await self.stop_playing(ctx)
self.db.update(ctx.guild.id, {'current_track': None, 'is_stopped': True})
self.db.clear_history(ctx.guild.id)
vc.stop()
await vc.disconnect(force=True)
await ctx.respond("Отключение успешно!", delete_after=15, ephemeral=True)
@queue.command(description="Очистить очередь треков и историю прослушивания.")
async def clear(self, ctx: discord.ApplicationContext) -> None:
logging.debug(f"Clear queue command invoked by user {ctx.author.id} in guild {ctx.guild.id}")
member = cast(discord.Member, ctx.author)
channel = cast(discord.VoiceChannel, ctx.channel)
if len(channel.members) > 2 and not member.guild_permissions.manage_channels:
@@ -228,6 +254,7 @@ class Voice(Cog, VoiceExtension):
@queue.command(description="Получить очередь треков.")
async def get(self, ctx: discord.ApplicationContext) -> None:
logging.debug(f"Get queue command invoked by user {ctx.author.id} in guild {ctx.guild.id}")
if not await self.voice_check(ctx):
return
tracks = self.db.get_tracks_list(ctx.guild.id, 'next')
@@ -237,6 +264,7 @@ class Voice(Cog, VoiceExtension):
@track.command(description="Приостановить текущий трек.")
async def pause(self, ctx: discord.ApplicationContext) -> None:
logging.debug(f"Pause command invoked by user {ctx.author.id} in guild {ctx.guild.id}")
member = cast(discord.Member, ctx.author)
channel = cast(discord.VoiceChannel, ctx.channel)
if len(channel.members) > 2 and not member.guild_permissions.manage_channels:
@@ -253,6 +281,7 @@ class Voice(Cog, VoiceExtension):
@track.command(description="Возобновить текущий трек.")
async def resume(self, ctx: discord.ApplicationContext) -> None:
logging.debug(f"Resume command invoked by user {ctx.author.id} in guild {ctx.guild.id}")
member = cast(discord.Member, ctx.author)
channel = cast(discord.VoiceChannel, ctx.channel)
if len(channel.members) > 2 and not member.guild_permissions.manage_channels:
@@ -269,6 +298,7 @@ class Voice(Cog, VoiceExtension):
@track.command(description="Прервать проигрывание, удалить историю, очередь и текущий плеер.")
async def stop(self, ctx: discord.ApplicationContext) -> None:
logging.debug(f"Stop command invoked by user {ctx.author.id} in guild {ctx.guild.id}")
member = cast(discord.Member, ctx.author)
channel = cast(discord.VoiceChannel, ctx.channel)
if len(channel.members) > 2 and not member.guild_permissions.manage_channels:
@@ -288,6 +318,7 @@ class Voice(Cog, VoiceExtension):
@track.command(description="Переключиться на следующую песню в очереди.")
async def next(self, ctx: discord.ApplicationContext) -> None:
logging.debug(f"Next command invoked by user {ctx.author.id} in guild {ctx.guild.id}")
if not await self.voice_check(ctx):
return
gid = ctx.guild.id
@@ -321,10 +352,13 @@ class Voice(Cog, VoiceExtension):
@track.command(description="Добавить трек в избранное или убрать, если он уже там.")
async def like(self, ctx: discord.ApplicationContext) -> None:
logging.debug(f"Like command invoked by user {ctx.author.id} in guild {ctx.guild.id}")
if await self.voice_check(ctx):
vc = await self.get_voice_client(ctx)
if not vc or not vc.is_playing:
logging.debug(f"No current track in {ctx.guild.id}")
await ctx.respond("Нет воспроизводимого трека.", delete_after=15, ephemeral=True)
return
result = await self.like_track(ctx)
if not result:
await ctx.respond("❌ Операция не удалась.", delete_after=15, ephemeral=True)

20
MusicBot/main.py
View File

@@ -4,12 +4,6 @@ import logging
import discord
from discord.ext.commands import Bot
try:
import coloredlogs
coloredlogs.install()
except ImportError:
pass
intents = discord.Intents.default()
intents.message_content = True
bot = Bot(intents=intents)
@@ -28,6 +22,17 @@ if __name__ == '__main__':
from dotenv import load_dotenv
load_dotenv()
try:
import coloredlogs
coloredlogs.install(level=logging.DEBUG)
except ImportError:
pass
logging.getLogger().setLevel(logging.DEBUG)
logging.getLogger('discord').setLevel(logging.INFO)
logging.getLogger('pymongo').setLevel(logging.INFO)
logging.getLogger('yandex_music').setLevel(logging.WARNING)
if not os.path.exists('music'):
os.mkdir('music')
token = os.getenv('TOKEN')
@@ -37,7 +42,4 @@ if __name__ == '__main__':
for cog in cogs_list:
bot.load_extension(f'MusicBot.cogs.{cog}')
logging.basicConfig(format='%(asctime)s:%(levelname)s:%(name)s: %(message)s')
logging.getLogger('discord').setLevel(logging.INFO)
bot.run(token)