mirror of
https://github.com/Matthww/TuneBot.git
synced 2026-09-21 23:37:47 +00:00
85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
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)
|