refactor(youtubedl): clean up formatting and use dataclasses

Also ignore messages whose embeds have changed.

commands/utils.py

@@ -11,6 +11,7 @@ class Command(enum.Enum):
     LEAVE = "leave"
     PAUSE = "pause"
     PLAY = "play"
+    PLAYING = "playing"
     PURGE = "purge"
     QUEUE = "queue"
     RELOAD = "reload"
@@ -24,6 +25,14 @@ def match_token(token: str) -> list[Command]:
     if token.lower() == "r":
         return [Command.RELOAD]
 
+    if exact_match := list(
+        filter(
+            lambda command: command.value == token.lower(),
+            Command.__members__.values(),
+        )
+    ):
+        return exact_match
+
     return list(
         filter(
             lambda command: command.value.startswith(token.lower()),

commands/voice.py

@@ -1,7 +1,11 @@
-import math
+import itertools
+
+import disnake
+import disnake_paginator
 
 import arguments
 import commands
+import constants
 import utils
 import youtubedl
 from state import client, players
@@ -50,6 +54,11 @@ async def queue_or_play(message):
         action="store_true",
         help="play the specified song immediately",
     )
+    parser.add_argument(
+        "--next",
+        action="store_true",
+        help="play the specified song next",
+    )
     parser.add_argument(
         "-t",
         "--remove-title",
@@ -61,37 +70,10 @@ async def queue_or_play(message):
         type=int,
         help="remove queued songs by queuer",
     )
-    parser.add_argument(
-        "-d",
-        "--duration",
-        action="store_true",
-        help="print duration of queued songs",
-    )
-    parser.add_argument(
-        "-p",
-        "--page",
-        type=int,
-        default=1,
-        help="print the specified page of the queue",
-    )
     if not (args := await parser.parse_args(message, tokens)):
         return
 
-    if args.duration:
+    if args.clear:
-        queued_songs = players[message.guild.id].queue
-        formatted_duration = utils.format_duration(
-            sum(
-                [
-                    queued.player.duration if queued.player.duration else 0
-                    for queued in queued_songs
-                ]
-            )
-        )
-        await utils.reply(
-            message,
-            f"queue is **{formatted_duration or '0 seconds'}** long (**{len(queued_songs)}** queued)",
-        )
-    elif args.clear:
         players[message.guild.id].queue.clear()
         await utils.add_check_reaction(message)
         return
@@ -111,7 +93,7 @@ async def queue_or_play(message):
                     targets.append(queued)
                     continue
             if q := args.remove_queuer:
-                if q == queued.queuer:
+                if q == queued.trigger_message.author.id:
                     targets.append(queued)
         if not args.remove_multiple:
             targets = targets[:1]
@@ -123,6 +105,25 @@ async def queue_or_play(message):
             f"removed **{len(targets)}** queued {'song' if len(targets) == 1 else 'songs'}",
         )
     elif query := args.query:
+        if (
+            not message.channel.permissions_for(message.author).manage_channels
+            and len(
+                list(
+                    filter(
+                        lambda queued: queued.trigger_message.author.id
+                        == message.author.id,
+                        players[message.guild.id].queue,
+                    )
+                )
+            )
+            >= 5
+        ):
+            await utils.reply(
+                message,
+                "you can only queue **5 songs** without the manage channels permission!",
+            )
+            return
+
         try:
             async with message.channel.typing():
                 player = await youtubedl.YTDLSource.from_url(
@@ -131,14 +132,13 @@ async def queue_or_play(message):
                 player.volume = float(args.volume) / 100.0
         except Exception as e:
             await utils.reply(
-                message,
+                message, f"**failed to queue:** `{e}`", suppress_embeds=True
-                f"**unable to queue {query}:** {e}",
             )
             return
 
-        queued = youtubedl.QueuedSong(player, message.author.id)
+        queued = youtubedl.QueuedSong(player, message)
 
-        if args.now:
+        if args.now or args.next:
             players[message.guild.id].queue_add_front(queued)
         else:
             players[message.guild.id].queue_add(queued)
@@ -157,45 +157,68 @@ async def queue_or_play(message):
             )
     else:
         if tokens[0].lower() == "play":
-            message.guild.voice_client.resume()
+            await resume(message)
-            await utils.reply(
-                message,
-                "resumed!",
-            )
         else:
-            args.page = max(
+            if players[message.guild.id].queue:
-                min(args.page, math.ceil(len(players[message.guild.id].queue) / 10)), 1
+                formatted_duration = utils.format_duration(
-            )
+                    sum(
-            queue_list = lambda: "\n".join(
                         [
-                    f"**{i + 1}.** {queued.format(with_queuer=True, hide_preview=True)}"
+                            queued.player.duration if queued.player.duration else 0
-                    for i, queued in list(enumerate(players[message.guild.id].queue))[
+                            for queued in players[message.guild.id].queue
-                        (args.page - 1) * 10 : args.page * 10
-                    ]
                         ]
                     )
-            currently_playing = (
-                lambda: f"**0.** {'(paused) ' if message.guild.voice_client.is_paused() else ''} {players[message.guild.id].current.format(with_queuer=True)}"
                 )
-            if (
+
-                not players[message.guild.id].queue
+                def embed(description):
-                and not message.guild.voice_client.source
+                    e = disnake.Embed(
-            ):
+                        title="Queued",
+                        description=description,
+                        color=constants.EMBED_COLOR,
+                    )
+                    if formatted_duration:
+                        e.set_footer(text=f"{formatted_duration} long")
+                    return e
+
+                await disnake_paginator.ButtonPaginator(
+                    invalid_user_function=utils.invalid_user_handler,
+                    color=constants.EMBED_COLOR,
+                    segments=list(
+                        map(
+                            embed,
+                            [
+                                "\n\n".join(
+                                    [
+                                        f"**{i + 1}.** {queued.format(show_queuer=True, hide_preview=True, multiline=True)}"
+                                        for i, queued in batch
+                                    ]
+                                )
+                                for batch in itertools.batched(
+                                    enumerate(players[message.guild.id].queue), 10
+                                )
+                            ],
+                        )
+                    ),
+                ).start(disnake_paginator.wrappers.MessageInteractionWrapper(message))
+            else:
                 await utils.reply(
                     message,
-                    "nothing is playing or queued!",
+                    "nothing is queued!",
                 )
-            elif not players[message.guild.id].queue:
+
-                await utils.reply(message, currently_playing())
+
-            elif not message.guild.voice_client.source:
+async def playing(message):
+    if not command_allowed(message):
+        return
+
+    if message.guild.voice_client.source:
         await utils.reply(
             message,
-                    queue_list(),
+            f"{'(paused) ' if message.guild.voice_client.is_paused() else ''} {players[message.guild.id].current.format(show_queuer=True)}",
         )
     else:
         await utils.reply(
             message,
-                    currently_playing() + "\n" + queue_list(),
+            "nothing is playing!",
         )
 
 
@@ -316,7 +339,9 @@ def play_next(message, once=False):
                 message.channel.send(f"error while trying to play: `{e}`")
             )
             return
-        client.loop.create_task(message.channel.send(f"**0.** {queued.format()}"))
+        client.loop.create_task(
+            message.channel.send(f"**0.** {queued.format(show_queuer=True)}")
+        )
 
 
 async def ensure_joined(message):

constants.py

@@ -19,6 +19,7 @@ RELOADABLE_MODULES = [
 ]
 
 YTDL_OPTIONS = {
+    "color": "never",
     "default_search": "auto",
     "format": "bestaudio/best",
     "ignoreerrors": False,

core.py

@@ -12,7 +12,7 @@ import commands
 import constants
 import core
 import utils
-from state import client, command_locks
+from state import client, command_locks, executed_messages
 
 
 async def on_message(message):
@@ -80,20 +80,23 @@ async def on_message(message):
 
                 if len(output) > 2000:
                     output = output.replace("`", "\\`")
-                    pager = disnake_paginator.ButtonPaginator(
+                    await disnake_paginator.ButtonPaginator(
                         prefix=f"```\n",
                         suffix="```",
+                        invalid_user_function=utils.invalid_user_handler,
                         color=constants.EMBED_COLOR,
                         segments=disnake_paginator.split(output),
-                        invalid_user_function=utils.invalid_user_handler,
+                    ).start(
-                    )
-                    await pager.start(
                         disnake_paginator.wrappers.MessageInteractionWrapper(message)
                     )
                 elif len(output.strip()) == 0:
                     await utils.add_check_reaction(message)
                 else:
-                    await message.channel.send(output)
+                    if message.id in executed_messages:
+                        await executed_messages[message.id].edit(output)
+                    else:
+                        response = await message.channel.send(output)
+                        executed_messages[message.id] = response
             case C.CLEAR | C.PURGE if message.author.id in constants.OWNERS:
                 await commands.tools.clear(message)
             case C.JOIN:
@@ -116,6 +119,8 @@ async def on_message(message):
                 await commands.bot.help(message)
             case C.UPTIME:
                 await commands.bot.uptime(message)
+            case C.PLAYING:
+                await commands.voice.playing(message)
     except Exception as e:
         await utils.reply(
             message,

events.py

@@ -16,24 +16,32 @@ async def trigger_dynamic_handlers(event_type: str, *data):
                 )
 
 
-@client.event
-async def on_message_edit(before, after):
-    await events.trigger_dynamic_handlers("on_message_edit", before, after)
-
-    await core.on_message(after)
-
-
-@client.event
 async def on_message(message):
     await events.trigger_dynamic_handlers("on_message", message)
 
     await core.on_message(message)
 
 
-@client.event
+async def on_message_edit(before, after):
+    await events.trigger_dynamic_handlers("on_message_edit", before, after)
+
+    if before.content == after.content:
+        return
+
+    await core.on_message(after)
+
+
 async def on_voice_state_update(member, before, after):
     await events.trigger_dynamic_handlers(
         "on_voice_state_update", member, before, after
     )
 
     await core.on_voice_state_update(member, before, after)
+
+
+for k, v in client.get_listeners().items():
+    for f in v:
+        client.remove_listener(f, k)
+client.add_listener(on_message, "on_message")
+client.add_listener(on_message_edit, "on_message_edit")
+client.add_listener(on_voice_state_update, "on_voice_state_update")

state.py

@@ -2,8 +2,9 @@ import time
 
 import disnake
 
-players = {}
 command_locks = {}
+executed_messages = {}
+players = {}
 
 intents = disnake.Intents.default()
 intents.message_content = True

utils.py

@@ -33,8 +33,10 @@ async def add_check_reaction(message):
     await message.add_reaction("✅")
 
 
-async def reply(message, *args):
+async def reply(message, *args, **kwargs):
-    await message.reply(*args, allowed_mentions=disnake.AllowedMentions.none())
+    await message.reply(
+        *args, **kwargs, allowed_mentions=disnake.AllowedMentions.none()
+    )
 
 
 async def invalid_user_handler(interaction):

youtubedl.py

@@ -1,5 +1,6 @@
 import asyncio
 import collections
+from dataclasses import dataclass
 from typing import Any, Optional
 
 import disnake
@@ -50,10 +51,43 @@ class YTDLSource(disnake.PCMVolumeTransformer):
         return self.__repr__()
 
 
+@dataclass
+class QueuedSong:
+    player: YTDLSource
+    trigger_message: disnake.Message
+
+    def format(self, show_queuer=False, hide_preview=False, multiline=False) -> str:
+        if multiline:
+            return (
+                f"[`{self.player.title}`]({'<' if hide_preview else ''}{self.player.original_url}{'>' if hide_preview else ''})\n**duration:** {self.format_duration(self.player.duration) if self.player.duration else '[live]'}"
+                + (
+                    f", **queuer:** <@{self.trigger_message.author.id}>"
+                    if show_queuer
+                    else ""
+                )
+            )
+        else:
+            return (
+                f"[`{self.player.title}`]({'<' if hide_preview else ''}{self.player.original_url}{'>' if hide_preview else ''}) [{self.format_duration(self.player.duration) if self.player.duration else 'live'}]"
+                + (f" (<@{self.trigger_message.author.id}>)" if show_queuer else "")
+            )
+
+    def format_duration(self, duration: int) -> str:
+        hours, duration = divmod(duration, 3600)
+        minutes, duration = divmod(duration, 60)
+        segments = [hours, minutes, duration]
+        if len(segments) == 3 and segments[0] == 0:
+            del segments[0]
+        return f"{':'.join(f'{s:0>2}' for s in segments)}"
+
+    def __str__(self):
+        return self.__repr__()
+
+
+@dataclass
 class QueuedPlayer:
-    def __init__(self):
+    queue = collections.deque()
-        self.queue = collections.deque()
+    current: Optional[QueuedSong] = None
-        self.current = None
 
     def queue_pop(self):
         popped = self.queue.popleft()
@@ -66,35 +100,6 @@ class QueuedPlayer:
     def queue_add_front(self, item):
         self.queue.appendleft(item)
 
-    def __repr__(self):
-        return f"<QueuedPlayer current={self.current} queue={self.queue}>"
-
-    def __str__(self):
-        return self.__repr__()
-
-
-class QueuedSong:
-    def __init__(self, player, queuer):
-        self.player = player
-        self.queuer = queuer
-
-    def format(self, with_queuer=False, hide_preview=False) -> str:
-        return (
-            f"[`{self.player.title}`]({'<' if hide_preview else ''}{self.player.original_url}{'>' if hide_preview else ''}) [{self.format_duration(self.player.duration) if self.player.duration else 'live'}]"
-            + (f" (<@{self.queuer}>)" if with_queuer else "")
-        )
-
-    def format_duration(self, duration: int) -> str:
-        hours, duration = divmod(duration, 3600)
-        minutes, duration = divmod(duration, 60)
-        segments = [hours, minutes, duration]
-        if len(segments) == 3 and segments[0] == 0:
-            del segments[0]
-        return f"{':'.join(f'{s:0>2}' for s in segments)}"
-
-    def __repr__(self):
-        return f"<QueuedSong player={self.player} queuer={self.queuer}>"
-
     def __str__(self):
         return self.__repr__()