feat: add player progress tracking

commands/voice.py

@@ -153,10 +153,7 @@ async def queue_or_play(message, edited=False):
         else:
             players[message.guild.id].queue_add(queued)
 
-        if (
+        if not message.guild.voice_client.source:
-            not message.guild.voice_client.is_playing()
-            and not message.guild.voice_client.is_paused()
-        ):
             play_next(message)
         elif args.now:
             message.guild.voice_client.stop()
@@ -219,10 +216,29 @@ async def playing(message):
     if not command_allowed(message):
         return
 
-    if message.guild.voice_client.source:
+    if source := message.guild.voice_client.source:
+        bar_length = 35
+        progress = source.original.progress / source.duration
+
+        embed = disnake.Embed(
+            color=constants.EMBED_COLOR,
+            title=source.title,
+            description=f"{'⏸️ ' if message.guild.voice_client.is_paused() else ''}"
+            f"`[{'#'*int(progress * bar_length)}{'-'*int((1 - progress) * bar_length)}]`"
+            f"{youtubedl.format_duration(int(source.original.progress))} / {youtubedl.format_duration(source.duration)}",
+            url=source.original_url,
+        )
+        embed.add_field(name="Volume", value=f"{int(source.volume*100)}%")
+        embed.add_field(name="Views", value=f"{source.view_count:,}")
+        embed.add_field(
+            name="Queuer",
+            value=players[message.guild.id].current.trigger_message.author.mention,
+        )
+        embed.set_image(source.thumbnail_url)
+
         await utils.reply(
             message,
-            f"{'(paused) ' if message.guild.voice_client.is_paused() else ''} {players[message.guild.id].current.format(show_queuer=True)}",
+            embed=embed,
         )
     else:
         await utils.reply(
@@ -244,10 +260,7 @@ async def skip(message):
     else:
         message.guild.voice_client.stop()
         await utils.add_check_reaction(message)
-        if (
+        if not message.guild.voice_client.source:
-            not message.guild.voice_client.is_playing()
-            and not message.guild.voice_client.is_paused()
-        ):
             play_next(message)
 
 

youtubedl.py

@@ -11,14 +11,33 @@ import constants
 ytdl = yt_dlp.YoutubeDL(constants.YTDL_OPTIONS)
 
 
+class TrackedAudioSource(disnake.AudioSource):
+    def __init__(self, source):
+        self._source = source
+        self.count = 0
+
+    def read(self) -> bytes:
+        data = self._source.read()
+        if data:
+            self.count += 1
+        return data
+
+    @property
+    def progress(self) -> float:
+        return self.count * 0.02
+
+
 class YTDLSource(disnake.PCMVolumeTransformer):
     def __init__(
-        self, source: disnake.AudioSource, *, data: dict[str, Any], volume: float = 0.5
+        self, source: TrackedAudioSource, *, data: dict[str, Any], volume: float = 0.5
     ):
         super().__init__(source, volume)
-        self.title = data.get("title")
+
-        self.original_url = data.get("original_url")
         self.duration = data.get("duration")
+        self.original_url = data.get("original_url")
+        self.thumbnail_url = data.get("thumbnail")
+        self.title = data.get("title")
+        self.view_count = data.get("view_count")
 
     @classmethod
     async def from_url(
@@ -37,9 +56,11 @@ class YTDLSource(disnake.PCMVolumeTransformer):
             data = data["entries"][0]
 
         return cls(
+            TrackedAudioSource(
                 disnake.FFmpegPCMAudio(
                     data["url"] if stream else ytdl.prepare_filename(data),
                     before_options="-vn -reconnect 1",
+                )
             ),
             data=data,
         )
@@ -59,7 +80,7 @@ class QueuedSong:
     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"[`{self.player.title}`]({'<' if hide_preview else ''}{self.player.original_url}{'>' if hide_preview else ''})\n**duration:** {format_duration(self.player.duration) if self.player.duration else '[live]'}"
                 + (
                     f", **queuer:** <@{self.trigger_message.author.id}>"
                     if show_queuer
@@ -68,18 +89,10 @@ class QueuedSong:
             )
         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.player.title}`]({'<' if hide_preview else ''}{self.player.original_url}{'>' if hide_preview else ''}) **[{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__()
 
@@ -104,6 +117,15 @@ class QueuedPlayer:
         return self.__repr__()
 
 
+def format_duration(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 __reload_module__():
     global ytdl
     ytdl = yt_dlp.YoutubeDL(constants.YTDL_OPTIONS)