import hashlib import time # https://github.com/jlieth/legacy-scrobbler from .funkwhale_startup import PLUGIN class ScrobblerException(Exception): pass def handshake_v1(session, url, username, password): timestamp = str(int(time.time())).encode("utf-8") password_hash = hashlib.md5(password.encode("utf-8")).hexdigest() auth = hashlib.md5(password_hash.encode("utf-8") + timestamp).hexdigest() params = { "hs": "true", "p": "1.2", "c": PLUGIN["name"], "v": PLUGIN["version"], "u": username, "t": timestamp, "a": auth, } PLUGIN["logger"].debug( "Performing scrobbler handshake for username %s at %s", username, url ) handshake_response = session.get(url, params=params) # process response result = handshake_response.text.split("\n") if len(result) >= 4 and result[0] == "OK": session_key = result[1] nowplaying_url = result[2] scrobble_url = result[3] elif result[0] == "BANNED": raise ScrobblerException("BANNED") elif result[0] == "BADAUTH": raise ScrobblerException("BADAUTH") elif result[0] == "BADTIME": raise ScrobblerException("BADTIME") else: raise ScrobblerException(handshake_response.text) PLUGIN["logger"].debug("Handshake successful, scrobble url: %s", scrobble_url) return session_key, nowplaying_url, scrobble_url def submit_scrobble_v1(session, scrobble_time, track, session_key, scrobble_url): payload = get_scrobble_payload(track, scrobble_time) PLUGIN["logger"].debug("Sending scrobble with payload %s", payload) payload["s"] = session_key response = session.post(scrobble_url, payload) response.raise_for_status() if response.text.startswith("OK"): return elif response.text.startswith("BADSESSION"): raise ScrobblerException("Remote server says the session is invalid") else: raise ScrobblerException(response.text) PLUGIN["logger"].debug("Scrobble successfull!") def submit_now_playing_v1(session, track, session_key, now_playing_url): payload = get_scrobble_payload(track, date=None, suffix="") PLUGIN["logger"].debug("Sending now playing with payload %s", payload) payload["s"] = session_key response = session.post(now_playing_url, payload) response.raise_for_status() if response.text.startswith("OK"): return elif response.text.startswith("BADSESSION"): raise ScrobblerException("Remote server says the session is invalid") else: raise ScrobblerException(response.text) PLUGIN["logger"].debug("Now playing successfull!") def get_scrobble_payload(track, date, suffix="[0]"): """ Documentation available at https://web.archive.org/web/20190531021725/https://www.last.fm/api/submissions """ upload = track.uploads.filter(duration__gte=0).first() data = { "a{}".format(suffix): track.artist.name, "t{}".format(suffix): track.title, "l{}".format(suffix): upload.duration if upload else 0, "b{}".format(suffix): (track.album.title if track.album else "") or "", "n{}".format(suffix): track.position or "", "m{}".format(suffix): str(track.mbid) or "", "o{}".format(suffix): "P", # Source: P = chosen by user } if date: data["i{}".format(suffix)] = int(date.timestamp()) return data