impr: Rework QueueView.

This commit is contained in:
Lemon4ksan
2025-03-16 18:18:22 +03:00
parent 1228516800
commit bb0135f1d6
4 changed files with 51 additions and 62 deletions

View File

@@ -333,15 +333,13 @@ class Voice(Cog, VoiceExtension):
if not await self.voice_check(ctx): if not await self.voice_check(ctx):
return return
await self.users_db.update(ctx.user.id, {'queue_page': 0})
tracks = await self.db.get_tracks_list(ctx.guild_id, 'next') tracks = await self.db.get_tracks_list(ctx.guild_id, 'next')
if len(tracks) == 0: if len(tracks) == 0:
await self.respond(ctx, "error", "Очередь прослушивания пуста.", delete_after=15, ephemeral=True) await self.respond(ctx, "error", "Очередь прослушивания пуста.", delete_after=15, ephemeral=True)
return return
embed = generate_queue_embed(0, tracks) await ctx.respond(embed=generate_queue_embed(0, tracks), view=QueueView(ctx, tracks), ephemeral=True)
await ctx.respond(embed=embed, view=await QueueView(ctx).init(), ephemeral=True)
logging.info(f"[VOICE] Queue embed sent to user {ctx.author.id} in guild {ctx.guild_id}") logging.info(f"[VOICE] Queue embed sent to user {ctx.author.id} in guild {ctx.guild_id}")

View File

@@ -20,9 +20,6 @@ guilds: AsyncCollection[ExplicitGuild] = db.guilds
class BaseUsersDatabase: class BaseUsersDatabase:
DEFAULT_USER = User( DEFAULT_USER = User(
ym_token=None, ym_token=None,
playlists=[],
playlists_page=0,
queue_page=0,
vibe_batch_id=None, vibe_batch_id=None,
vibe_type=None, vibe_type=None,
vibe_id=None, vibe_id=None,

View File

@@ -8,9 +8,6 @@ VibeSettingsOptions: TypeAlias = Literal[
class User(TypedDict, total=False): # Don't forget to change base.py if you add a new field class User(TypedDict, total=False): # Don't forget to change base.py if you add a new field
ym_token: str | None ym_token: str | None
playlists: list[tuple[str, int]]
playlists_page: int
queue_page: int
vibe_batch_id: str | None vibe_batch_id: str | None
vibe_type: Literal['track', 'album', 'artist', 'playlist', 'user'] | None vibe_type: Literal['track', 'album', 'artist', 'playlist', 'user'] | None
vibe_id: str | int | None vibe_id: str | int | None
@@ -19,9 +16,6 @@ class User(TypedDict, total=False): # Don't forget to change base.py if you add
class ExplicitUser(TypedDict): class ExplicitUser(TypedDict):
_id: int _id: int
ym_token: str | None ym_token: str | None
playlists: list[tuple[str, int]] # name / tracks count
playlists_page: int
queue_page: int
vibe_batch_id: str | None vibe_batch_id: str | None
vibe_type: Literal['track', 'album', 'artist', 'playlist', 'user'] | None vibe_type: Literal['track', 'album', 'artist', 'playlist', 'user'] | None
vibe_id: str | int | None vibe_id: str | int | None

View File

@@ -1,91 +1,91 @@
from math import ceil from math import ceil
from typing import Self, Any from typing import Any
from discord.ui import View, Button, Item from discord.ui import View, Button, Item
from discord import ApplicationContext, ButtonStyle, Interaction, Embed, HTTPException from discord import ApplicationContext, ButtonStyle, Interaction, Embed
from MusicBot.cogs.utils.voice_extension import VoiceExtension from MusicBot.cogs.utils.voice_extension import VoiceExtension
def generate_queue_embed(page: int, tracks_list: list[dict[str, Any]]) -> Embed: def generate_queue_embed(page: int, tracks_list: list[dict[str, Any]]) -> Embed:
count = 15 * page count = 15 * page
length = len(tracks_list) length = len(tracks_list)
embed = Embed( embed = Embed(
title=f"Всего: {length}", title=f"Всего: {length}",
color=0xfed42b, color=0xfed42b,
) )
embed.set_author(name="Очередь треков") embed.set_author(name="Очередь треков")
embed.set_footer(text=f"Страница {page + 1} из {ceil(length / 15)}") embed.set_footer(text=f"Страница {page + 1} из {ceil(length / 15)}")
for i, track in enumerate(tracks_list[count:count + 15], start=1 + count): for i, track in enumerate(tracks_list[count:count + 15], start=1 + count):
duration = track['duration_ms'] if track['duration_ms']:
if duration: duration_m = track['duration_ms'] // 60000
duration_m = duration // 60000 duration_s = ceil(track['duration_ms'] / 1000) - duration_m * 60
duration_s = ceil(duration / 1000) - duration_m * 60
embed.add_field(name=f"{i} - {track['title']} - {duration_m}:{duration_s:02d}", value="", inline=False) embed.add_field(name=f"{i} - {track['title']} - {duration_m}:{duration_s:02d}", value="", inline=False)
return embed return embed
class QueueNextButton(Button, VoiceExtension): class QueueNextButton(Button):
def __init__(self, **kwargs): def __init__(self, root:' QueueView', **kwargs):
Button.__init__(self, **kwargs) Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None) self.root = root
async def callback(self, interaction: Interaction) -> None: async def callback(self, interaction: Interaction) -> None:
if not interaction.user or not interaction.guild: self.root.page += 1
return self.root.update()
embed = generate_queue_embed(self.root.page, self.root.tracks)
await interaction.edit(embed=embed, view=self.root)
user = await self.users_db.get_user(interaction.user.id) class QueuePrevButton(Button):
page = user['queue_page'] + 1 def __init__(self, root: 'QueueView', **kwargs):
await self.users_db.update(interaction.user.id, {'queue_page': page})
tracks = await self.db.get_tracks_list(interaction.guild.id, 'next')
embed = generate_queue_embed(page, tracks)
await interaction.edit(embed=embed, view=await QueueView(interaction).init())
class QueuePrevButton(Button, VoiceExtension):
def __init__(self, **kwargs):
Button.__init__(self, **kwargs) Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None) self.root = root
async def callback(self, interaction: Interaction) -> None: async def callback(self, interaction: Interaction) -> None:
if not interaction.user or not interaction.guild: self.root.page -= 1
return self.root.update()
embed = generate_queue_embed(self.root.page, self.root.tracks)
user = await self.users_db.get_user(interaction.user.id) await interaction.edit(embed=embed, view=self.root)
page = user['queue_page'] - 1
await self.users_db.update(interaction.user.id, {'queue_page': page})
tracks = await self.db.get_tracks_list(interaction.guild.id, 'next')
embed = generate_queue_embed(page, tracks)
await interaction.edit(embed=embed, view=await QueueView(interaction).init())
class QueueView(View, VoiceExtension): class QueueView(View, VoiceExtension):
def __init__(self, ctx: ApplicationContext | Interaction, *items: Item, timeout: float | None = 360, disable_on_timeout: bool = False): def __init__(
self,
ctx: ApplicationContext | Interaction,
tracks: list[dict[str, Any]],
*items: Item,
timeout: float | None = 360,
disable_on_timeout: bool = False
):
View.__init__(self, *items, timeout=timeout, disable_on_timeout=disable_on_timeout) View.__init__(self, *items, timeout=timeout, disable_on_timeout=disable_on_timeout)
VoiceExtension.__init__(self, None) VoiceExtension.__init__(self, None)
self.ctx = ctx self.ctx = ctx
self.next_button = QueueNextButton(style=ButtonStyle.primary, emoji='▶️') self.tracks = tracks
self.prev_button = QueuePrevButton(style=ButtonStyle.primary, emoji='◀️') self.page = 0
async def init(self) -> Self: self.next_button = QueueNextButton(self, style=ButtonStyle.primary, emoji='▶️')
if not self.ctx.user or not self.ctx.guild: self.prev_button = QueuePrevButton(self, style=ButtonStyle.primary, emoji='◀️', disabled=True)
return self
if not self.tracks[15:]:
tracks = await self.db.get_tracks_list(self.ctx.guild.id, 'next')
user = await self.users_db.get_user(self.ctx.user.id)
count = 15 * user['queue_page']
if not tracks[count + 15:]:
self.next_button.disabled = True self.next_button.disabled = True
if not tracks[:count]:
self.prev_button.disabled = True self.prev_button.disabled = True
self.add_item(self.prev_button) self.add_item(self.prev_button)
self.add_item(self.next_button) self.add_item(self.next_button)
return self def update(self):
count = 15 * self.page
if self.tracks[15:]:
self.next_button.disabled = False
else:
self.next_button.disabled = True
if self.tracks[:count]:
self.prev_button.disabled = False
else:
self.prev_button.disabled = True
async def on_timeout(self) -> None: async def on_timeout(self) -> None:
try:
await super().on_timeout()
except HTTPException:
pass
self.stop() self.stop()