Added a ListenBrainz plugin

Allows users to submit their listenings to ListenBrainz.org.

api/config/settings/common.py

@@ -94,6 +94,7 @@ Path to a directory containing Funkwhale plugins. These will be imported at runt
 sys.path.append(FUNKWHALE_PLUGINS_PATH)
 CORE_PLUGINS = [
     "funkwhale_api.contrib.scrobbler",
+    "funkwhale_api.contrib.listenbrainz",
 ]
 
 LOAD_CORE_PLUGINS = env.bool("FUNKWHALE_LOAD_CORE_PLUGINS", default=True)

api/funkwhale_api/contrib/listenbrainz/client.py

+# Copyright (c) 2018 Philipp Wolfer <ph.wolfer@gmail.com>
+#
+# Permission is hereby granted, free of charge, to any person obtaining
+# a copy of this software and associated documentation files (the
+# "Software"), to deal in the Software without restriction, including
+# without limitation the rights to use, copy, modify, merge, publish,
+# distribute, sublicense, and/or sell copies of the Software, and to
+# permit persons to whom the Software is furnished to do so, subject to
+# the following conditions:
+#
+# The above copyright notice and this permission notice shall be
+# included in all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+import json
+import logging
+import ssl
+import time
+from http.client import HTTPSConnection
+
+HOST_NAME = "api.listenbrainz.org"
+PATH_SUBMIT = "/1/submit-listens"
+SSL_CONTEXT = ssl.create_default_context()
+
+
+class Track:
+    """
+    Represents a single track to submit.
+
+    See https://listenbrainz.readthedocs.io/en/latest/dev/json.html
+    """
+
+    def __init__(self, artist_name, track_name, release_name=None, additional_info={}):
+        """
+        Create a new Track instance
+        @param artist_name as str
+        @param track_name as str
+        @param release_name as str
+        @param additional_info as dict
+        """
+        self.artist_name = artist_name
+        self.track_name = track_name
+        self.release_name = release_name
+        self.additional_info = additional_info
+
+    @staticmethod
+    def from_dict(data):
+        return Track(
+            data["artist_name"],
+            data["track_name"],
+            data.get("release_name", None),
+            data.get("additional_info", {}),
+        )
+
+    def to_dict(self):
+        return {
+            "artist_name": self.artist_name,
+            "track_name": self.track_name,
+            "release_name": self.release_name,
+            "additional_info": self.additional_info,
+        }
+
+    def __repr__(self):
+        return "Track(%s, %s)" % (self.artist_name, self.track_name)
+
+
+class ListenBrainzClient:
+    """
+    Submit listens to ListenBrainz.org.
+
+    See https://listenbrainz.readthedocs.io/en/latest/dev/api.html
+    """
+
+    def __init__(self, user_token, logger=logging.getLogger(__name__)):
+        self.__next_request_time = 0
+        self.user_token = user_token
+        self.logger = logger
+
+    def listen(self, listened_at, track):
+        """
+        Submit a listen for a track
+        @param listened_at as int
+        @param entry as Track
+        """
+        payload = _get_payload(track, listened_at)
+        return self._submit("single", [payload])
+
+    def playing_now(self, track):
+        """
+        Submit a playing now notification for a track
+        @param track as Track
+        """
+        payload = _get_payload(track)
+        return self._submit("playing_now", [payload])
+
+    def import_tracks(self, tracks):
+        """
+        Import a list of tracks as (listened_at, Track) pairs
+        @param track as [(int, Track)]
+        """
+        payload = _get_payload_many(tracks)
+        return self._submit("import", payload)
+
+    def _submit(self, listen_type, payload, retry=0):
+        self._wait_for_ratelimit()
+        self.logger.debug("ListenBrainz %s: %r", listen_type, payload)
+        data = {"listen_type": listen_type, "payload": payload}
+        headers = {
+            "Authorization": "Token %s" % self.user_token,
+            "Content-Type": "application/json",
+        }
+        body = json.dumps(data)
+        conn = HTTPSConnection(HOST_NAME, context=SSL_CONTEXT)
+        conn.request("POST", PATH_SUBMIT, body, headers)
+        response = conn.getresponse()
+        response_text = response.read()
+        try:
+            response_data = json.loads(response_text)
+        except json.decoder.JSONDecodeError:
+            response_data = response_text
+
+        self._handle_ratelimit(response)
+        log_msg = "Response %s: %r" % (response.status, response_data)
+        if response.status == 429 and retry < 5:  # Too Many Requests
+            self.logger.warning(log_msg)
+            return self._submit(listen_type, payload, retry + 1)
+        elif response.status == 200:
+            self.logger.debug(log_msg)
+        else:
+            self.logger.error(log_msg)
+        return response
+
+    def _wait_for_ratelimit(self):
+        now = time.time()
+        if self.__next_request_time > now:
+            delay = self.__next_request_time - now
+            self.logger.debug("Rate limit applies, delay %d", delay)
+            time.sleep(delay)
+
+    def _handle_ratelimit(self, response):
+        remaining = int(response.getheader("X-RateLimit-Remaining", 0))
+        reset_in = int(response.getheader("X-RateLimit-Reset-In", 0))
+        self.logger.debug("X-RateLimit-Remaining: %i", remaining)
+        self.logger.debug("X-RateLimit-Reset-In: %i", reset_in)
+        if remaining == 0:
+            self.__next_request_time = time.time() + reset_in
+
+
+def _get_payload_many(tracks):
+    payload = []
+    for (listened_at, track) in tracks:
+        data = _get_payload(track, listened_at)
+        payload.append(data)
+    return payload
+
+
+def _get_payload(track, listened_at=None):
+    data = {"track_metadata": track.to_dict()}
+    if listened_at is not None:
+        data["listened_at"] = listened_at
+    return data

api/funkwhale_api/contrib/listenbrainz/funkwhale_ready.py

+from config import plugins
+from .funkwhale_startup import PLUGIN
+from .client import ListenBrainzClient, Track
+
+
+@plugins.register_hook(plugins.LISTENING_CREATED, PLUGIN)
+def submit_listen(listening, conf, **kwargs):
+    user_token = conf["user_token"]
+    if not user_token:
+        return
+
+    logger = PLUGIN["logger"]
+    logger.info("Submitting listen to ListenBrainz")
+    client = ListenBrainzClient(user_token=user_token, logger=logger)
+    track = get_track(listening.track)
+    client.listen(int(listening.creation_date.timestamp()), track)
+
+
+def get_track(track):
+    artist = track.artist.name
+    title = track.title
+    album = None
+    additional_info = {
+        "listening_from": "Funkwhale",
+        "recording_mbid": str(track.mbid),
+        "tracknumber": track.position,
+        "discnumber": track.disc_number,
+    }
+
+    if track.album:
+        if track.album.title:
+            album = track.album.title
+        if track.album.mbid:
+            additional_info["release_mbid"] = str(track.album.mbid)
+
+    if track.artist.mbid:
+        additional_info["artist_mbids"] = [str(track.artist.mbid)]
+
+    return Track(artist, title, album, additional_info)

api/funkwhale_api/contrib/listenbrainz/funkwhale_startup.py

+from config import plugins
+
+
+PLUGIN = plugins.get_plugin_config(
+    name="listenbrainz",
+    label="ListenBrainz",
+    description="A plugin that allows you to submit your listens to ListenBrainz.",
+    version="0.1",
+    user=True,
+    conf=[
+        {
+            "name": "user_token",
+            "type": "text",
+            "label": "Your ListenBrainz user token",
+            "help": "You can find your user token in your ListenBrainz profile at https://listenbrainz.org/profile/",
+        }
+    ],
+)

changes/changelog.d/listenbrainz.enhancement

+Added a ListenBrainz plugin to submit listenings
\ No newline at end of file