Commit hoarding is a terrible practice

plugins/lastfm_scrobbler/cog.py

@@ -0,0 +1,46 @@
+from typing import TYPE_CHECKING
+
+from discord.ext import commands
+
+from context import CustomContext
+from utils.classes import PluginCog
+
+if TYPE_CHECKING:
+    from bot import TuneBot
+
+
+class LastFMScrobblerCog(PluginCog, name="LastFMScrobbler"):
+    def __init__(self, bot: "TuneBot") -> None:
+        super().__init__(bot)
+        self.plugin = self.get_plugin_instance("lastfm-scrobbler")
+
+    @commands.group(name="lfm", invoke_without_command=True)
+    @commands.cooldown(rate=1, per=5, type=commands.BucketType.user)
+    async def lfm(self, ctx: CustomContext):
+        """Enable/Disable LastFM Scrobbling"""
+        embed = ctx.create_embed()
+        embed.title = f"LastFM scrobbler usage:"
+        embed.description = f"```{ctx.prefix}lfm enable\n{ctx.prefix}lfm del```"
+        await ctx.send(embed=embed)
+
+    @lfm.command(name="enable", aliases=["set"])
+    async def lfm_aut(self, ctx: CustomContext):
+        """Enable LastFM Scrobbling"""
+        embed = ctx.create_embed()
+        embed.title = f"Start scrobbling"
+        auth_url = self.plugin.config["auth_url"] + "?user_id=" + str(ctx.author.id)
+        embed.description = f"[Login]({auth_url})"
+        await ctx.send(embed=embed)
+
+    @lfm.command(name="delete", aliases=["del"])
+    async def lfm_del(self, ctx: CustomContext):
+        """Disable LastFM Scrobbling"""
+        db_key = self.plugin.config["session_key_table"]
+        await self.bot.global_utils.raw_table_del_entry(db_key, [ctx.author.id])
+        embed = ctx.create_embed()
+        embed.title = f"Stopped scrobbling"
+        await ctx.send(embed=embed)
+
+
+def setup(bot: "TuneBot"):
+    bot.add_cog(LastFMScrobblerCog(bot))

plugins/lastfm_scrobbler/service.py

@@ -0,0 +1,84 @@
+import hashlib
+import time
+from dataclasses import dataclass
+from typing import Any
+from typing import TYPE_CHECKING
+
+import aiohttp
+from lavalink.models import AudioTrack
+
+from tunebot.plugins import ServiceEvent
+
+if TYPE_CHECKING:
+    from bot import TuneBot
+
+AnyDict = dict[Any, Any]
+
+
+@dataclass
+class Track:
+    name: str
+    artist: str
+
+
+class LastFMScrobbler:
+    api_url = "http://ws.audioscrobbler.com/2.0/"
+
+    def __init__(self, bot: "TuneBot", config: AnyDict) -> None:
+        self.bot = bot
+        self.plug_conf = config
+
+    async def on_dispatch(self, event: "ServiceEvent", payload: AnyDict):
+        if event == ServiceEvent.TRACK_ENDED:
+            _track: AudioTrack = payload.get("last_track", None)
+            if not _track:
+                print("LastFMScrobbler: Could not find required field `last_track`")
+                return
+
+            # TODO: expand parsing options
+            artist, track_name = _track.title.split(" - ", 1)
+            track = Track(track_name, artist)
+
+            in_voice = payload.get("in_voice", [])
+            for session_key in await self.fetch_lastfm_sessions(in_voice):
+                if not session_key:
+                    continue
+
+                await self.scrobble(session_key, track)
+
+    async def fetch_lastfm_sessions(self, user_ids: list[int]) -> list[str]:
+        table_name = self.plug_conf["config"]["session_key_table"]
+        result: list[str] = await self.bot.global_utils.raw_table_lookup(
+            table_name, user_ids
+        )
+        return result
+
+    async def scrobble(self, session_key: str, track: Track):
+        params: AnyDict = {
+            "method": "track.scrobble",
+            "timestamp": str(int(time.time() - 30)),
+            "track": track.name,
+            "artist": track.artist,
+            "sk": session_key,
+        }
+        resp = await self.lastfm_request(params)
+        if resp.status != 200:
+            fmt = f"Failed to scrobble for user {session_key} on track: {track.artist} - {track.name}"
+            print(fmt)
+
+    async def lastfm_request(self, params: AnyDict) -> aiohttp.ClientResponse:
+        params["api_key"] = self.plug_conf["config"]["lastfm_api_key"]
+        params = {key: params[key] for key in sorted(params)}
+
+        secret = self.plug_conf["config"]["lastfm_api_secret"]
+        sig_str = "".join(key + params[key] for key in params.keys()) + secret
+        params["api_sig"] = hashlib.md5(sig_str.encode("utf8")).hexdigest()
+
+        params["format"] = "json"
+        async with aiohttp.ClientSession() as sess:
+            async with sess.post(self.api_url, params=params) as resp:
+                return resp
+
+
+def setup(bot: "TuneBot", config: AnyDict):
+    return LastFMScrobbler(bot, config)