impr: Further code review.

MusicBot/ui/__init__.py

@@ -0,0 +1,12 @@
+from .other import MyPlaylists, QueueView, generate_queue_embed, generate_playlists_embed
+from .menu import MenuView
+from .find import ListenView
+
+__all__ = [
+    'MyPlaylists',
+    'QueueView',
+    'MenuView',
+    'ListenView',
+    'generate_queue_embed',
+    'generate_playlists_embed'
+]

MusicBot/ui/find.py

@@ -0,0 +1,160 @@
+import logging
+from typing import Literal, cast
+
+import discord
+from yandex_music import Track, Album, Artist, Playlist
+
+from discord.ui import View, Button, Item
+from discord import ButtonStyle, Interaction
+
+from MusicBot.cogs.utils.voice_extension import VoiceExtension
+
+class PlayButton(Button, VoiceExtension):
+    def __init__(self, item: Track | Album | Artist | Playlist | list[Track], **kwargs):
+        Button.__init__(self, **kwargs)
+        VoiceExtension.__init__(self, None)
+        self.item = item
+
+    async def callback(self, interaction: Interaction) -> None:
+        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)
+        action: Literal['add_track', 'add_album', 'add_artist', 'add_playlist']
+
+        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}** был добавлен в очередь."
+
+        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}** был добавлен в очередь."
+
+        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}** были добавлены в очередь."
+
+        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}** был добавлен в очередь."
+
+        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"Плейлист **«Мне нравится»** был добавлен в очередь."
+
+        else:
+            raise ValueError(f"Unknown item type: '{type(self.item).__name__}'")
+
+        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('❌')
+
+            self.db.update_vote(
+                gid,
+                response.id,
+                {
+                    'positive_votes': list(),
+                    'negative_votes': list(),
+                    'total_members': len(channel.members),
+                    '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')
+            else:
+                track = tracks.pop(0)
+                self.db.modify_track(gid, tracks, 'next', 'extend')
+                await self.play_track(interaction, track)
+                response_message = f"Сейчас играет: **{track.title}**!"
+            
+            current_player = None
+            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()
+            else:
+                await interaction.respond(response_message, delete_after=15)
+
+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
+
+        self.button1: Button = Button(label="Слушать в приложении", style=ButtonStyle.gray, url=link_app)
+        self.button2: Button = Button(label="Слушать в браузере", style=ButtonStyle.gray, url=link_web)
+        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)

MusicBot/ui/menu.py

@@ -0,0 +1,181 @@
+import logging
+from typing import Self, cast
+
+from discord.ui import View, Button, Item
+from discord import VoiceChannel, ButtonStyle, Interaction, ApplicationContext, RawReactionActionEvent, Embed
+
+from yandex_music import Track, ClientAsync
+from MusicBot.cogs.utils.voice_extension import VoiceExtension
+
+class ToggleRepeatButton(Button, VoiceExtension):
+    def __init__(self, **kwargs):
+        Button.__init__(self, **kwargs)
+        VoiceExtension.__init__(self, None)
+    
+    async def callback(self, interaction: Interaction) -> None:
+        logging.info('Repeat button callback...')
+        if not interaction.guild:
+            return
+        gid = interaction.guild.id
+        guild = self.db.get_guild(gid)
+        self.db.update(gid, {'repeat': not guild['repeat']})
+        await interaction.edit(view=await MenuView(interaction).init())
+
+class ToggleShuffleButton(Button, VoiceExtension):
+    def __init__(self, **kwargs):
+        Button.__init__(self, **kwargs)
+        VoiceExtension.__init__(self, None)
+    
+    async def callback(self, interaction: Interaction) -> None:
+        logging.info('Shuffle button callback...')
+        if not interaction.guild:
+            return
+        gid = interaction.guild.id
+        guild = self.db.get_guild(gid)
+        self.db.update(gid, {'shuffle': not guild['shuffle']})
+        await interaction.edit(view=await MenuView(interaction).init())
+
+class PlayPauseButton(Button, VoiceExtension):
+    def __init__(self, **kwargs):
+        Button.__init__(self, **kwargs)
+        VoiceExtension.__init__(self, None)
+    
+    async def callback(self, interaction: Interaction) -> None:
+        logging.info('Play/Pause button callback...')
+        if not await self.voice_check(interaction):
+            return
+
+        vc = await self.get_voice_client(interaction)
+        if not vc or not interaction.message:
+            return
+
+        embed = interaction.message.embeds[0]
+
+        if vc.is_paused():
+            vc.resume()
+            embed.remove_footer()
+        else:
+            vc.pause()
+            embed.set_footer(text='Приостановлено')
+
+        await interaction.edit(embed=embed)
+
+class NextTrackButton(Button, VoiceExtension):    
+    def __init__(self, **kwargs):
+        Button.__init__(self, **kwargs)
+        VoiceExtension.__init__(self, None)
+    
+    async def callback(self, interaction: Interaction) -> None:
+        logging.info('Next track button callback...')
+        if not await self.voice_check(interaction):
+            return
+        title = await self.next_track(interaction)
+        if not title:
+            await interaction.respond(f"Нет треков в очереди.", delete_after=15, ephemeral=True)
+
+class PrevTrackButton(Button, VoiceExtension):    
+    def __init__(self, **kwargs):
+        Button.__init__(self, **kwargs)
+        VoiceExtension.__init__(self, None)
+    
+    async def callback(self, interaction: Interaction) -> None:
+        logging.info('Previous track button callback...')
+        if not await self.voice_check(interaction):
+            return
+        title = await self.prev_track(interaction)
+        if not title:
+            await interaction.respond(f"Нет треков в истории.", delete_after=15, ephemeral=True)
+
+class LikeButton(Button, VoiceExtension):
+    def __init__(self, **kwargs):
+        Button.__init__(self, **kwargs)
+        VoiceExtension.__init__(self, None)
+
+    async def callback(self, interaction: Interaction) -> None:
+        logging.info('Like button callback...')
+        if not await self.voice_check(interaction):
+            return
+        
+        if not (vc := await self.get_voice_client(interaction)) or not vc.is_playing:
+            await interaction.respond("❌ Нет воспроизводимого трека.", delete_after=15, ephemeral=True)
+
+        await self.like_track(interaction)
+        await interaction.edit(view=await MenuView(interaction).init())
+
+class LyricsButton(Button, VoiceExtension):
+    def __init__(self, **kwargs):
+        Button.__init__(self, **kwargs)
+        VoiceExtension.__init__(self, None)
+        
+    async def callback(self, interaction: Interaction) -> None:
+        logging.info('Lyrics button callback...')
+
+        if not await self.voice_check(interaction) or not interaction.guild_id or not interaction.user:
+            return
+        
+        ym_token = self.users_db.get_ym_token(interaction.user.id)        
+        current_track = self.db.get_track(interaction.guild_id, 'current')
+        if not current_track or not ym_token:
+            return
+
+        track = cast(Track, Track.de_json(
+            current_track,
+            ClientAsync(ym_token),  # type: ignore  # Async client can be used here
+        ))
+
+        lyrics = await track.get_lyrics_async()
+        if not lyrics:
+            return
+
+        embed = Embed(
+            title=track.title,
+            description='**Текст песни**',
+            color=0xfed42b,
+        )
+        text = await lyrics.fetch_lyrics_async()
+        for subtext in text.split('\n\n'):
+            embed.add_field(name='', value=subtext, inline=False)
+        await interaction.respond(embed=embed, ephemeral=True)
+
+
+class MenuView(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.guild_id:
+            return
+        self.ctx = ctx
+        self.guild = self.db.get_guild(ctx.guild_id)
+
+        self.repeat_button = ToggleRepeatButton(style=ButtonStyle.success if self.guild['repeat'] else ButtonStyle.secondary, emoji='🔂', row=0)
+        self.shuffle_button = ToggleShuffleButton(style=ButtonStyle.success if self.guild['shuffle'] else ButtonStyle.secondary, emoji='🔀', row=0)
+        self.play_pause_button = PlayPauseButton(style=ButtonStyle.primary, emoji='⏯', row=0)
+        self.next_button = NextTrackButton(style=ButtonStyle.primary, emoji='⏭', row=0)
+        self.prev_button = PrevTrackButton(style=ButtonStyle.primary, emoji='⏮', row=0)
+        
+        self.like_button = LikeButton(style=ButtonStyle.secondary, emoji='❤️', row=1)
+        self.lyrics_button = LyricsButton(style=ButtonStyle.secondary, emoji='📋', row=1)
+        
+    async def init(self) -> Self:
+        current_track = self.guild['current_track']
+        likes = await self.get_likes(self.ctx)
+
+        self.add_item(self.repeat_button)
+        self.add_item(self.prev_button)
+        self.add_item(self.play_pause_button)
+        self.add_item(self.next_button)
+        self.add_item(self.shuffle_button)
+        
+        if len(cast(VoiceChannel, self.ctx.channel).members) > 2:
+            self.like_button.disabled = True
+        elif likes and current_track and str(current_track['id']) in [str(like.id) for like in likes]:
+            self.like_button.style = ButtonStyle.success
+
+        if not current_track or not current_track['lyrics_available']:
+            self.lyrics_button.disabled = True
+
+        self.add_item(self.like_button)
+        self.add_item(self.lyrics_button)
+
+        return self

MusicBot/ui/other.py

@@ -0,0 +1,143 @@
+from math import ceil
+from typing import Any
+
+from discord.ui import View, Button, Item
+from discord import ButtonStyle, Interaction, Embed
+
+from MusicBot.cogs.utils.voice_extension import VoiceExtension
+
+from discord.ui import View, Button, Item
+from discord import ButtonStyle, Interaction, ApplicationContext
+
+from MusicBot.cogs.utils.voice_extension import VoiceExtension
+
+def generate_playlists_embed(page: int, playlists: list[tuple[str, int]]) -> Embed:
+    count = 15 * page
+    length = len(playlists)
+    embed = Embed(
+        title=f"Всего плейлистов: {length}",
+        color=0xfed42b
+    )
+    embed.set_author(name="Ваши плейлисты")
+    embed.set_footer(text=f"Страница {page + 1} из {ceil(length / 10)}")
+    for playlist in playlists[count:count + 10]:
+        embed.add_field(name=playlist[0], value=f"{playlist[1]} треков", inline=False)
+    return embed
+
+def generate_queue_embed(page: int, tracks_list: list[dict[str, Any]]) -> Embed:
+    count = 15 * page
+    length = len(tracks_list)
+    embed = Embed(
+        title=f"Всего: {length}",
+        color=0xfed42b,
+    )
+    embed.set_author(name="Очередь треков")
+    embed.set_footer(text=f"Страница {page + 1} из {ceil(length / 15)}")
+    for i, track in enumerate(tracks_list[count:count + 15], start=1 + count):
+        duration = track['duration_ms']
+        if duration:
+            duration_m = duration // 60000
+            duration_s = ceil(duration / 1000) - duration_m * 60
+            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 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 QueueNextButton(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 QueuePrevButton(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 = QueueNextButton(style=ButtonStyle.primary, emoji='▶️')
+        prev_button = QueuePrevButton(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)