Add images to tracks and playlists

mopidy_soundcloud/actor.py

@@ -14,7 +14,7 @@ def __init__(self, config, audio):
         super().__init__()
         self.config = config
         self.remote = SoundCloudClient(config)
-        self.library = SoundCloudLibraryProvider(backend=self)
+        self.library = SoundCloudLibraryProvider(self.remote, backend=self)
         self.playback = SoundCloudPlaybackProvider(audio=audio, backend=self)
 
         self.uri_schemes = ["soundcloud", "sc"]
@@ -28,7 +28,7 @@ def on_start(self):
 class SoundCloudPlaybackProvider(backend.PlaybackProvider):
     def translate_uri(self, uri):
         track_id = self.backend.remote.parse_track_uri(uri)
-        track = self.backend.remote.get_track(track_id, True)
+        track = self.backend.remote.get_parsed_track(track_id, True)
         if track is None:
             return None
         return track.uri

mopidy_soundcloud/images.py

@@ -0,0 +1,119 @@
+import itertools
+import logging
+import operator
+import urllib.parse
+
+from mopidy import models
+from mopidy_soundcloud.soundcloud import SoundCloudClient
+
+# NOTE: current file adapted from https://github.com/mopidy/mopidy-spotify
+#   - /mopidy-spotify/images.py
+
+logger = logging.getLogger(__name__)
+
+
+class SoundCloudImageProvider:
+    _API_MAX_IDS_PER_REQUEST = 50
+
+    _cache = {}  # (type, id) -> [Image(), ...]
+
+    # For reference
+    _ARTWORK_MAP = {
+        "mini": 16,
+        "tiny": 20,
+        "small": 32,
+        "badge": 47,
+        "t67x67": 67,
+        "large": 100,
+        "t300x300": 300,
+        "crop": 400,
+        "t500x500": 500,
+        "original": 0,
+    }
+
+    def __init__(self, web_client: SoundCloudClient):
+        self.web_client = web_client
+
+    def get_images(self, uris):
+        result = {}
+        uri_type_getter = operator.itemgetter("type")
+        uris = sorted((self._parse_uri(u) for u in uris), key=uri_type_getter)
+        for uri_type, group in itertools.groupby(uris, uri_type_getter):
+            batch = []
+            for uri in group:
+                if uri["key"] in self._cache:
+                    result[uri["uri"]] = self._cache[uri["key"]]
+                elif uri_type == "playlist":
+                    result.update(self._process_set(uri))
+                else:
+                    batch.append(uri)
+                    if len(batch) >= self._API_MAX_IDS_PER_REQUEST:
+                        result.update(self._process_uris(batch))
+                        batch = []
+            result.update(self._process_uris(batch))
+        return result
+
+    def _parse_uri(self, uri):
+        parsed_uri = urllib.parse.urlparse(uri)
+        uri_type, uri_id = None, None
+
+        if parsed_uri.scheme == "soundcloud":
+            uri_type, uri_id = parsed_uri.path.split("/")[:2]
+        elif parsed_uri.scheme in ("http", "https"):
+            if "soundcloud.com" in parsed_uri.netloc:
+                uri_type, uri_id = parsed_uri.path.split("/")[1:3]
+
+        supported_types = ("song", "album", "artist", "playlist")
+        if uri_type and uri_type in supported_types and uri_id:
+            return {
+                "uri": uri,
+                "type": uri_type,
+                "id": self.web_client.parse_track_uri(uri_id),
+                "key": (uri_type, uri_id),
+            }
+
+        raise ValueError(f"Could not parse {repr(uri)} as a SoundCloud URI")
+
+    def _process_set(self, uri):
+        tracks = self.web_client.get_set(uri["id"])
+        set_images = tuple()
+        for track in tracks:
+            set_images += (*self._process_track(track),)
+
+        self._cache[uri["key"]] = set_images
+
+        return {uri["uri"]: set_images}
+
+    def _process_uris(self, uris):
+        result = {}
+        for uri in uris:
+            if uri["key"] not in self._cache:
+                track = self.web_client.get_track(uri["id"])
+                self._cache[uri["key"]] = self._process_track(track)
+
+            result.update({uri["uri"]: self._cache[uri["key"]]})
+        return result
+
+    @staticmethod
+    def _process_track(track):
+        images = []
+        if track:
+            image_sources = [
+                track.get("artwork_url"),
+                track.get("calculated_artwork_url"),
+            ]
+
+            # Only include avatar images if no other images found
+            if image_sources.count(None) == len(image_sources):
+                image_sources = [
+                    track.get("user", {}).get("avatar_url"),
+                    track.get("avatar_url"),
+                ]
+
+            for image_url in image_sources:
+                if image_url is not None:
+                    image_url = image_url.replace("large", "t500x500")
+                    image = models.Image(uri=image_url, height=500, width=500)
+                    images.append(image)
+
+        return tuple(images)

mopidy_soundcloud/library.py

@@ -6,6 +6,8 @@
 from mopidy import backend, models
 from mopidy.models import SearchResult, Track
 
+from mopidy_soundcloud.images import SoundCloudImageProvider
+
 logger = logging.getLogger(__name__)
 
 
@@ -38,14 +40,19 @@ class SoundCloudLibraryProvider(backend.LibraryProvider):
         uri="soundcloud:directory", name="SoundCloud"
     )
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, web_client, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.vfs = {"soundcloud:directory": {}}
         self.add_to_vfs(new_folder("Following", ["following"]))
         self.add_to_vfs(new_folder("Liked", ["liked"]))
         self.add_to_vfs(new_folder("Sets", ["sets"]))
         self.add_to_vfs(new_folder("Stream", ["stream"]))
 
+        self.image_provider = SoundCloudImageProvider(web_client)
+
+    def get_images(self, uris):
+        return self.image_provider.get_images(uris)
+
     def add_to_vfs(self, _model):
         self.vfs["soundcloud:directory"][_model.uri] = _model
 
@@ -144,9 +151,24 @@ def lookup(self, uri):
             uri = uri.replace("sc:", "")
             return self.backend.remote.resolve_url(uri)
 
+        if "directory" in uri:
+            return
+
+        if "stream" in uri:
+            return list(self.backend.remote.get_user_stream())
+
+        if "liked" in uri:
+            return list(self.backend.remote.get_likes())
+
+        if "sets" in uri:
+            return list(self.backend.remote.get_sets())
+
+        if "following" in uri:
+            return list(self.backend.remote.get_followings())
+
         try:
             track_id = self.backend.remote.parse_track_uri(uri)
-            track = self.backend.remote.get_track(track_id)
+            track = self.backend.remote.get_parsed_track(track_id)
             if track is None:
                 logger.info(
                     f"Failed to lookup {uri}: SoundCloud track not found"

mopidy_soundcloud/soundcloud.py

@@ -235,13 +235,18 @@ def get_tracks(self, user_id=None):
 
     # Public
     @cache()
-    def get_track(self, track_id, streamable=False):
+    def get_track(self, track_id):
         logger.debug(f"Getting info for track with ID {track_id}")
         try:
-            return self.parse_track(self._get(f"tracks/{track_id}"), streamable)
+            return self._get(f"tracks/{track_id}")
         except Exception:
             return None
 
+    @cache()
+    def get_parsed_track(self, track_id, streamable=False):
+        track = self.get_track(track_id)
+        return self.parse_track(track, streamable)
+
     @staticmethod
     def parse_track_uri(track):
         logger.debug(f"Parsing track {track}")

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = Mopidy-SoundCloud
-version = 3.0.1
+version = 3.1.0
 url = https://github.com/mopidy/mopidy-soundcloud
 author = Janez Troha
 author_email = [email protected]