import requests from typing import Any from hashlib import md5 from urllib.parse import quote_plus from app.config import UserConfig from app.models.track import Track from app.utils.auth import get_current_userid from app.utils.threading import background from app.plugins import Plugin, plugin_method from app.logger import log class LastFmPlugin(Plugin): def __init__(self): self.config = UserConfig() super().__init__("lastfm", "Last.fm scrobbler") self.set_active( bool( self.config.lastfmApiKey and self.config.lastfmApiSecret and self.config.lastfmSessionKeys.get(str(get_current_userid())) ) ) def get_api_signature(self, data: dict[str, Any]) -> str: params = {k: v for k, v in data.items()} signature = "".join(f"{k}{v}" for k, v in sorted(params.items())) signature += self.config.lastfmApiSecret return md5(signature.encode("utf-8")).hexdigest() def post(self, data: dict[str, Any], useSessionKey: bool = True): url = "http://ws.audioscrobbler.com/2.0/?format=json" data["api_key"] = self.config.lastfmApiKey if useSessionKey: data["sk"] = self.config.lastfmSessionKeys.get(str(get_current_userid())) data["api_sig"] = self.get_api_signature(data) final_url = ( url + "&" + "&".join(f"{k}={quote_plus(str(v))}" for k, v in data.items()) ) return requests.post(final_url) def get_session_key(self, token: str): data = { "method": "auth.getSession", "token": token, } try: res = self.post(data, useSessionKey=False) return res.json()["session"]["key"] except Exception as e: print("get_session_key error", e) return None @plugin_method @background def scrobble(self, track: Track, timestamp: int): log.info(f"Last.fm: logging track: {track.title} - {track.artists[0]['name']}") data = { "method": "track.scrobble", "artist": track.artists[0]["name"], "track": track.title, "timestamp": timestamp, "album": track.album, "albumArtist": track.albumartists[0]["name"], } log.info(f"scrobble data: {data}") try: res = self.post(data) log.info("scrobble response:" + str(res.text)) log.info("scrobble response json:" + str(res.json())) except Exception as e: log.info("scrobble error" + str(e))