scrobbler.py 3.42 KB
Newer Older
Eliot Berriot's avatar
Eliot Berriot committed
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
import hashlib
import time

import time

from funkwhale_api import plugins

from . import scrobbler

# https://github.com/jlieth/legacy-scrobbler
plugin = plugins.get_plugin("fw_scrobbler")


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,
    }

    session = plugin.get_requests_session()
    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]
41
        nowplaying_url = result[2]
Eliot Berriot's avatar
Eliot Berriot committed
42
43
44
45
46
47
48
49
50
51
52
        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)
53
    return session_key, nowplaying_url, scrobble_url
Eliot Berriot's avatar
Eliot Berriot committed
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71


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!")


72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
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
    """
Eliot Berriot's avatar
Eliot Berriot committed
92
    upload = track.uploads.filter(duration__gte=0).first()
93
94
95
96
97
98
99
100
    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 or "",
        "n{}".format(suffix): track.position or "",
        "m{}".format(suffix): str(track.mbid) or "",
        "o{}".format(suffix): "P",  # Source: P = chosen by user
Eliot Berriot's avatar
Eliot Berriot committed
101
    }
102
103
104
    if date:
        data["i{}".format(suffix)] = int(date.timestamp())
    return data