Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
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