Varuous fixes and enhancements for the Soundcloud provider

music_assistant/providers/soundcloud/__init__.py

@@ -2,9 +2,8 @@
 
 from __future__ import annotations
 
-import asyncio
 import time
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from music_assistant_models.config_entries import ConfigEntry, ConfigValueType
 from music_assistant_models.enums import (
@@ -46,7 +45,7 @@
 
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncGenerator, Callable
+    from collections.abc import AsyncGenerator
 
     from music_assistant_models.config_entries import ProviderConfig
     from music_assistant_models.provider import ProviderManifest
@@ -98,14 +97,9 @@ async def get_config_entries(
 class SoundcloudMusicProvider(MusicProvider):
     """Provider for Soundcloud."""
 
-    _headers = None
-    _context = None
-    _cookies = None
-    _signature_timestamp = 0
-    _cipher = None
-    _user_id = None
-    _soundcloud = None
-    _me = None
+    _user_id: str = ""
+    _soundcloud: SoundcloudAsyncAPI = None
+    _me: dict[str, Any] = {}
 
     async def handle_async_init(self) -> None:
         """Set up the Soundcloud provider."""
@@ -121,12 +115,8 @@ def supported_features(self) -> set[ProviderFeature]:
         """Return the features supported by this Provider."""
         return SUPPORTED_FEATURES
 
-    @classmethod
-    async def _run_async(cls, call: Callable, *args, **kwargs):  # noqa: ANN206
-        return await asyncio.to_thread(call, *args, **kwargs)
-
     async def search(
-        self, search_query: str, media_types=list[MediaType], limit: int = 10
+        self, search_query: str, media_types: list[MediaType], limit: int = 10
     ) -> SearchResults:
         """Perform search on musicprovider.
 
@@ -215,42 +205,45 @@ async def get_library_playlists(self) -> AsyncGenerator[Playlist, None]:
     async def get_library_tracks(self) -> AsyncGenerator[Track, None]:
         """Retrieve library tracks from Soundcloud."""
         time_start = time.time()
-        async for item in self._soundcloud.get_tracks_liked():
-            track = await self._soundcloud.get_track_details(item)
+        async for track in self._soundcloud.get_track_details_liked(self._user_id):
             try:
-                yield await self._parse_track(track[0])
-            except IndexError:
-                continue
-            except (KeyError, TypeError, InvalidDataError) as error:
-                self.logger.debug("Parse track failed: %s", track, exc_info=error)
+                yield await self._parse_track(track)
+            except (KeyError, TypeError, InvalidDataError, IndexError) as error:
+                # somehow certain track id's don't exist (anymore)
+                self.logger.debug(
+                    "%s: Parse track with id %s failed: %s",
+                    error.__name__,
+                    track["id"],
+                    track,
+                )
                 continue
 
         self.logger.debug(
             "Processing Soundcloud library tracks took %s seconds",
             round(time.time() - time_start, 2),
         )
 
-    async def get_artist(self, prov_artist_id) -> Artist:
+    async def get_artist(self, prov_artist_id: str) -> Artist:
         """Get full artist details by id."""
-        artist_obj = await self._soundcloud.get_user_details(user_id=prov_artist_id)
+        artist_obj = await self._soundcloud.get_user_details(prov_artist_id)
         try:
-            artist = await self._parse_artist(artist_obj=artist_obj) if artist_obj else None
+            artist = await self._parse_artist(artist_obj) if artist_obj else None
         except (KeyError, TypeError, InvalidDataError, IndexError) as error:
             self.logger.debug("Parse artist failed: %s", artist_obj, exc_info=error)
         return artist
 
-    async def get_track(self, prov_track_id) -> Track:
+    async def get_track(self, prov_track_id: str) -> Track:
         """Get full track details by id."""
-        track_obj = await self._soundcloud.get_track_details(track_id=prov_track_id)
+        track_obj = await self._soundcloud.get_track_details(prov_track_id)
         try:
             track = await self._parse_track(track_obj[0])
         except (KeyError, TypeError, InvalidDataError, IndexError) as error:
             self.logger.debug("Parse track failed: %s", track_obj, exc_info=error)
         return track
 
-    async def get_playlist(self, prov_playlist_id) -> Playlist:
+    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
         """Get full playlist details by id."""
-        playlist_obj = await self._soundcloud.get_playlist_details(playlist_id=prov_playlist_id)
+        playlist_obj = await self._soundcloud.get_playlist_details(prov_playlist_id)
         try:
             playlist = await self._parse_playlist(playlist_obj)
         except (KeyError, TypeError, InvalidDataError, IndexError) as error:
@@ -263,25 +256,25 @@ async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> lis
         if page > 0:
             # TODO: soundcloud doesn't seem to support paging for playlist tracks ?!
             return result
-        playlist_obj = await self._soundcloud.get_playlist_details(playlist_id=prov_playlist_id)
+        playlist_obj = await self._soundcloud.get_playlist_details(prov_playlist_id)
         if "tracks" not in playlist_obj:
             return result
         for index, item in enumerate(playlist_obj["tracks"], 1):
-            # TODO: is it really needed to grab the entire track with an api call ?
-            song = await self._soundcloud.get_track_details(item["id"])
             try:
-                if track := await self._parse_track(song[0], index):
-                    result.append(track)
+                # Skip some ugly "tracks" entries, example:
+                # {'id': 123, 'kind': 'track', 'monetization_model': 'NOT_APPLICABLE',
+                # 'policy': 'ALLOW'}
+                if "title" in item:
+                    if track := await self._parse_track(item, index):
+                        result.append(track)
             except (KeyError, TypeError, InvalidDataError, IndexError) as error:
-                self.logger.debug("Parse track failed: %s", song, exc_info=error)
+                self.logger.debug("Parse track failed: %s", item, exc_info=error)
                 continue
         return result
 
-    async def get_artist_toptracks(self, prov_artist_id) -> list[Track]:
+    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
         """Get a list of 25 most popular tracks for the given artist."""
-        tracks_obj = await self._soundcloud.get_popular_tracks_user(
-            user_id=prov_artist_id, limit=25
-        )
+        tracks_obj = await self._soundcloud.get_popular_tracks_user(prov_artist_id, 25)
         tracks = []
         for item in tracks_obj["collection"]:
             song = await self._soundcloud.get_track_details(item["id"])
@@ -293,9 +286,9 @@ async def get_artist_toptracks(self, prov_artist_id) -> list[Track]:
                 continue
         return tracks
 
-    async def get_similar_tracks(self, prov_track_id, limit=25) -> list[Track]:
+    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
         """Retrieve a dynamic list of tracks based on the provided item."""
-        tracks_obj = await self._soundcloud.get_recommended(track_id=prov_track_id, limit=limit)
+        tracks_obj = await self._soundcloud.get_recommended(prov_track_id, limit)
         tracks = []
         for item in tracks_obj["collection"]:
             song = await self._soundcloud.get_track_details(item["id"])
@@ -353,7 +346,7 @@ async def _parse_artist(self, artist_obj: dict) -> Artist:
         if artist_obj.get("description"):
             artist.metadata.description = artist_obj["description"]
         if artist_obj.get("avatar_url"):
-            img_url = artist_obj["avatar_url"]
+            img_url = self._transform_artwork_url(artist_obj["avatar_url"])
             artist.metadata.images = [
                 MediaItemImage(
                     type=ImageType.THUMB,

music_assistant/providers/soundcloud/manifest.json

@@ -4,7 +4,7 @@
   "name": "Soundcloud",
   "description": "Support for the Soundcloud streaming provider in Music Assistant.",
   "codeowners": ["@domanchi", "@gieljnssns"],
-  "requirements": ["soundcloudpy==0.1.0"],
+  "requirements": ["soundcloudpy==0.1.2"],
   "documentation": "https://music-assistant.io/music-providers/soundcloud/",
   "multi_instance": true
 }

mypy.ini

@@ -21,4 +21,4 @@ disallow_untyped_decorators = true
 disallow_untyped_defs = true
 warn_return_any = true
 warn_unreachable = true
-packages=tests,music_assistant.providers.builtin,music_assistant.providers.filesystem_local,music_assistant.providers.filesystem_smb,music_assistant.providers.fully_kiosk,music_assistant.providers.jellyfin,music_assistant.providers.plex,music_assistant.providers.radiobrowser,music_assistant.providers.test,music_assistant.providers.theaudiodb,music_assistant.providers.tidal
+packages=tests,music_assistant.providers.builtin,music_assistant.providers.filesystem_local,music_assistant.providers.filesystem_smb,music_assistant.providers.fully_kiosk,music_assistant.providers.jellyfin,music_assistant.providers.plex,music_assistant.providers.radiobrowser,music_assistant.providers.test,music_assistant.providers.theaudiodb,music_assistant.providers.tidal,music_assistant.providers.soundcloud

requirements_all.txt

@@ -41,7 +41,7 @@ radios==0.3.2
 shortuuid==1.0.13
 snapcast==2.3.6
 soco==0.30.6
-soundcloudpy==0.1.0
+soundcloudpy==0.1.2
 sxm==0.2.8
 tidalapi==0.8.3
 unidecode==1.3.8