/
/
/
1"""Soundcloud support for MusicAssistant."""
2
3from __future__ import annotations
4
5import time
6from typing import TYPE_CHECKING, Any, cast
7
8from music_assistant_models.config_entries import ConfigEntry, ConfigValueType
9from music_assistant_models.enums import (
10 ConfigEntryType,
11 ContentType,
12 ImageType,
13 MediaType,
14 ProviderFeature,
15 StreamType,
16)
17from music_assistant_models.errors import InvalidDataError, LoginFailed
18from music_assistant_models.media_items import (
19 Artist,
20 AudioFormat,
21 MediaItemImage,
22 Playlist,
23 ProviderMapping,
24 RecommendationFolder,
25 SearchResults,
26 Track,
27 UniqueList,
28)
29from music_assistant_models.streamdetails import StreamDetails
30from soundcloudpy import SoundcloudAsyncAPI
31
32from music_assistant.controllers.cache import use_cache
33from music_assistant.helpers.util import parse_title_and_version
34from music_assistant.models.music_provider import MusicProvider
35
36CONF_CLIENT_ID = "client_id"
37CONF_AUTHORIZATION = "authorization"
38
39SUPPORTED_FEATURES = {
40 ProviderFeature.LIBRARY_ARTISTS,
41 ProviderFeature.LIBRARY_TRACKS,
42 ProviderFeature.LIBRARY_PLAYLISTS,
43 ProviderFeature.BROWSE,
44 ProviderFeature.SEARCH,
45 ProviderFeature.ARTIST_TOPTRACKS,
46 ProviderFeature.SIMILAR_TRACKS,
47 ProviderFeature.RECOMMENDATIONS,
48}
49
50
51if TYPE_CHECKING:
52 from collections.abc import AsyncGenerator
53
54 from music_assistant_models.config_entries import ProviderConfig
55 from music_assistant_models.provider import ProviderManifest
56
57 from music_assistant.mass import MusicAssistant
58 from music_assistant.models import ProviderInstanceType
59
60
61async def setup(
62 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
63) -> ProviderInstanceType:
64 """Initialize provider(instance) with given configuration."""
65 if not config.get_value(CONF_CLIENT_ID) or not config.get_value(CONF_AUTHORIZATION):
66 msg = "Invalid login credentials"
67 raise LoginFailed(msg)
68 return SoundcloudMusicProvider(mass, manifest, config, SUPPORTED_FEATURES)
69
70
71async def get_config_entries(
72 mass: MusicAssistant,
73 instance_id: str | None = None,
74 action: str | None = None,
75 values: dict[str, ConfigValueType] | None = None,
76) -> tuple[ConfigEntry, ...]:
77 """
78 Return Config entries to setup this provider.
79
80 instance_id: id of an existing provider instance (None if new instance setup).
81 action: [optional] action key called from config entries UI.
82 values: the (intermediate) raw values for config entries sent with the action.
83 """
84 # ruff: noqa: ARG001
85 return (
86 ConfigEntry(
87 key=CONF_CLIENT_ID,
88 type=ConfigEntryType.SECURE_STRING,
89 label="Client ID",
90 required=True,
91 ),
92 ConfigEntry(
93 key=CONF_AUTHORIZATION,
94 type=ConfigEntryType.SECURE_STRING,
95 label="Authorization",
96 required=True,
97 ),
98 )
99
100
101class SoundcloudMusicProvider(MusicProvider):
102 """Provider for Soundcloud."""
103
104 _user_id: str = ""
105 _soundcloud: SoundcloudAsyncAPI = None
106 _me: dict[str, Any] = {}
107
108 async def handle_async_init(self) -> None:
109 """Set up the Soundcloud provider."""
110 client_id = self.config.get_value(CONF_CLIENT_ID)
111 auth_token = self.config.get_value(CONF_AUTHORIZATION)
112 self._soundcloud = SoundcloudAsyncAPI(auth_token, client_id, self.mass.http_session)
113 await self._soundcloud.login()
114 self._me = await self._soundcloud.get_account_details()
115 self._user_id = self._me["id"]
116
117 @use_cache(3600 * 48) # Cache for 48 hours
118 async def search(
119 self, search_query: str, media_types: list[MediaType], limit: int = 10
120 ) -> SearchResults:
121 """Perform search on musicprovider.
122
123 :param search_query: Search query.
124 :param media_types: A list of media_types to include.
125 :param limit: Number of items to return in the search (per type).
126 """
127 result = SearchResults()
128 searchtypes = []
129 if MediaType.ARTIST in media_types:
130 searchtypes.append("artist")
131 if MediaType.TRACK in media_types:
132 searchtypes.append("track")
133 if MediaType.PLAYLIST in media_types:
134 searchtypes.append("playlist")
135
136 media_types = [
137 x for x in media_types if x in (MediaType.ARTIST, MediaType.TRACK, MediaType.PLAYLIST)
138 ]
139 if not media_types:
140 return result
141
142 searchresult = await self._soundcloud.search(search_query, limit)
143
144 for item in searchresult["collection"]:
145 media_type = item["kind"]
146 if media_type == "user" and MediaType.ARTIST in media_types:
147 result.artists = [*result.artists, await self._parse_artist(item)]
148 elif media_type == "track" and MediaType.TRACK in media_types:
149 if item.get("duration") == item.get("full_duration"):
150 # skip if it's a preview track (e.g. in case of free accounts)
151 result.tracks = [*result.tracks, await self._parse_track(item)]
152 elif media_type == "playlist" and MediaType.PLAYLIST in media_types:
153 result.playlists = [*result.playlists, await self._parse_playlist(item)]
154
155 return result
156
157 async def get_library_artists(self) -> AsyncGenerator[Artist, None]:
158 """Retrieve all library artists from Soundcloud."""
159 time_start = time.time()
160
161 following = await self._soundcloud.get_following(self._user_id)
162 self.logger.debug(
163 "Processing Soundcloud library artists took %s seconds",
164 round(time.time() - time_start, 2),
165 )
166 for artist in following["collection"]:
167 try:
168 yield await self._parse_artist(artist)
169 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
170 self.logger.debug("Parse artist failed: %s", artist, exc_info=error)
171 continue
172
173 async def get_library_playlists(self) -> AsyncGenerator[Playlist, None]:
174 """Retrieve all library playlists from Soundcloud."""
175 time_start = time.time()
176 async for item in self._soundcloud.get_account_playlists():
177 try:
178 raw_playlist = item["playlist"]
179 except KeyError:
180 self.logger.debug(
181 "Unexpected Soundcloud API response when parsing playlists: %s",
182 item,
183 )
184 continue
185
186 try:
187 playlist = await self._get_playlist_object(
188 prov_playlist_id=raw_playlist["id"],
189 )
190
191 yield await self._parse_playlist(playlist)
192 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
193 self.logger.debug(
194 "Failed to obtain Soundcloud playlist details: %s",
195 raw_playlist,
196 exc_info=error,
197 )
198 continue
199
200 self.logger.debug(
201 "Processing Soundcloud library playlists took %s seconds",
202 round(time.time() - time_start, 2),
203 )
204
205 async def get_library_tracks(self) -> AsyncGenerator[Track, None]:
206 """Retrieve library tracks from Soundcloud."""
207 time_start = time.time()
208 async for track in self._soundcloud.get_track_details_liked(self._user_id):
209 try:
210 yield await self._parse_track(track)
211 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
212 # somehow certain track id's don't exist (anymore)
213 self.logger.debug(
214 "%s: Parse track with id %s failed: %s",
215 type(error).__name__,
216 track["id"],
217 track,
218 )
219 continue
220
221 self.logger.debug(
222 "Processing Soundcloud library tracks took %s seconds",
223 round(time.time() - time_start, 2),
224 )
225
226 @use_cache(3600 * 3) # Cache for 3 hours
227 async def recommendations(self) -> list[RecommendationFolder]:
228 """Get available recommendations."""
229 # Part 1, the mixed selections
230 recommendations = await self._soundcloud.get_mixed_selection(20)
231 folders = []
232 for collection in recommendations.get("collection", []):
233 folder = RecommendationFolder(
234 name=collection["title"],
235 item_id=f"{self.instance_id}_{collection['id']}",
236 provider=self.instance_id,
237 icon="mdi-playlist-music",
238 )
239 for playlist in collection.get("items").get("collection", []):
240 # Each items can be a track, playlist, album or artist but seems playlists only
241 if playlist.get("kind") == "system-playlist":
242 folder.items.append(await self._parse_playlist(playlist))
243 else:
244 self.logger.debug(
245 "Unknown item type in collection for SoundCloud: %s", playlist.get("kind")
246 )
247 continue
248 folders.append(folder)
249 # Part 2, the subscribed feed
250 feed = await self._soundcloud.get_subscribe_feed(20)
251 if feed and "collection" in feed:
252 folder = RecommendationFolder(
253 name="SoundCloud Feed",
254 item_id=f"{self.instance_id}_sc_subscribed_feed",
255 provider=self.instance_id,
256 icon="mdi-rss",
257 )
258 for item in feed["collection"]:
259 if item.get("type") == "track" or item.get("type") == "track-repost":
260 folder.items.append(await self._parse_track(item.get("track")))
261 else:
262 self.logger.debug(
263 "Unknown type in subscribed feed for SoundCloud: %s", item.get("type")
264 )
265 continue
266 folders.append(folder)
267 return folders
268
269 @use_cache(3600 * 24 * 14) # Cache for 14 days
270 async def get_artist(self, prov_artist_id: str) -> Artist:
271 """Get full artist details by id."""
272 artist_obj = await self._soundcloud.get_user_details(prov_artist_id)
273 try:
274 if artist_obj:
275 artist = await self._parse_artist(artist_obj)
276 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
277 self.logger.debug("Parse artist failed: %s", artist_obj, exc_info=error)
278 return artist
279
280 @use_cache(3600 * 24 * 14) # Cache for 14 days
281 async def get_track(self, prov_track_id: str) -> Track:
282 """Get full track details by id."""
283 track_obj = await self._soundcloud.get_track_details(prov_track_id)
284 try:
285 track = await self._parse_track(track_obj[0])
286 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
287 self.logger.debug("Parse track failed: %s", track_obj, exc_info=error)
288 return track
289
290 @use_cache(3600 * 24 * 14) # Cache for 14 days
291 async def get_playlist(self, prov_playlist_id: str) -> Playlist:
292 """Get full playlist details by id."""
293 playlist_obj = await self._get_playlist_object(prov_playlist_id)
294 try:
295 playlist = await self._parse_playlist(playlist_obj)
296 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
297 self.logger.debug("Parse playlist failed: %s", playlist_obj, exc_info=error)
298 return playlist
299
300 async def _get_playlist_object(self, prov_playlist_id: str) -> dict[str, Any]:
301 """Get playlist object from Soundcloud API based on playlist ID type."""
302 # Handle playlist id's which are actually numbers
303 prov_playlist_id = str(prov_playlist_id)
304 if prov_playlist_id.startswith("soundcloud:system-playlists"):
305 # Handle system playlists
306 result = await self._soundcloud.get_system_playlist_details(prov_playlist_id)
307 return cast("dict[str, Any]", result)
308 # Handle regular playlists
309 result = await self._soundcloud.get_playlist_details(prov_playlist_id)
310 return cast("dict[str, Any]", result)
311
312 @use_cache(3600 * 3) # Cache for 3 hours
313 async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
314 """Get playlist tracks."""
315 result: list[Track] = []
316 if page > 0:
317 # TODO: soundcloud doesn't seem to support paging for playlist tracks ?!
318 return result
319 playlist_obj = await self._get_playlist_object(prov_playlist_id)
320 if "tracks" not in playlist_obj:
321 return result
322 for index, item in enumerate(playlist_obj["tracks"], 1):
323 try:
324 # Skip some ugly "tracks" entries, example:
325 # {'id': 123, 'kind': 'track', 'monetization_model': 'NOT_APPLICABLE'}
326 if "title" in item:
327 if track := await self._parse_track(item, index):
328 result.append(track)
329 # But also try to get the track details if the track is not in the playlist
330 else:
331 track_details = await self._soundcloud.get_track_details(item["id"])
332 if track := await self._parse_track(track_details[0], index):
333 result.append(track)
334 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
335 self.logger.debug("Parse track failed: %s", item, exc_info=error)
336 continue
337 return result
338
339 @use_cache(3600 * 24 * 14) # Cache for 14 days
340 async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
341 """Get a list of (max 500) tracks for the given artist."""
342 tracks_obj = await self._soundcloud.get_tracks_from_user(prov_artist_id, 500)
343
344 tracks = []
345 for item in tracks_obj["collection"]:
346 song = await self._soundcloud.get_track_details(item["id"])
347 try:
348 track = await self._parse_track(song[0])
349 tracks.append(track)
350 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
351 self.logger.debug("Parse track failed: %s", song, exc_info=error)
352 continue
353 return tracks
354
355 @use_cache(3600 * 24 * 14) # Cache for 14 days
356 async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
357 """Retrieve a dynamic list of tracks based on the provided item."""
358 tracks_obj = await self._soundcloud.get_recommended(prov_track_id, limit)
359 tracks = []
360 for item in tracks_obj["collection"]:
361 song = await self._soundcloud.get_track_details(item["id"])
362 try:
363 track = await self._parse_track(song[0])
364 tracks.append(track)
365 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
366 self.logger.debug("Parse track failed: %s", song, exc_info=error)
367 continue
368
369 return tracks
370
371 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
372 """Return the content details for the given track when it will be streamed."""
373 url: str = await self._soundcloud.get_stream_url(track_id=item_id, presets=["mp3"])
374 return StreamDetails(
375 provider=self.instance_id,
376 item_id=item_id,
377 # let ffmpeg work out the details itself as
378 # soundcloud uses a mix of different content types and streaming methods
379 audio_format=AudioFormat(
380 content_type=ContentType.UNKNOWN,
381 ),
382 stream_type=StreamType.HLS
383 if url.startswith("https://cf-hls-media.sndcdn.com")
384 else StreamType.HTTP,
385 path=url,
386 can_seek=True,
387 allow_seek=True,
388 )
389
390 async def _parse_artist(self, artist_obj: dict[str, Any]) -> Artist:
391 """Parse a Soundcloud user response to Artist model object."""
392 artist_id = None
393 permalink = artist_obj["permalink"]
394 if artist_obj.get("id"):
395 artist_id = artist_obj["id"]
396 if not artist_id:
397 msg = "Artist does not have a valid ID"
398 raise InvalidDataError(msg)
399 artist_id = str(artist_id)
400 artist = Artist(
401 item_id=artist_id,
402 name=artist_obj["username"],
403 provider=self.domain,
404 provider_mappings={
405 ProviderMapping(
406 item_id=str(artist_id),
407 provider_domain=self.domain,
408 provider_instance=self.instance_id,
409 url=f"https://soundcloud.com/{permalink}",
410 )
411 },
412 )
413 if artist_obj.get("description"):
414 artist.metadata.description = artist_obj["description"]
415 if artist_obj.get("avatar_url"):
416 img_url = self._transform_artwork_url(artist_obj["avatar_url"])
417 artist.metadata.images = UniqueList(
418 [
419 MediaItemImage(
420 type=ImageType.THUMB,
421 path=img_url,
422 provider=self.instance_id,
423 remotely_accessible=True,
424 )
425 ]
426 )
427 return artist
428
429 async def _parse_playlist(self, playlist_obj: dict[str, Any]) -> Playlist:
430 """Parse a Soundcloud Playlist response to a Playlist object."""
431 playlist_id = str(playlist_obj["id"])
432 # Remove the "Related tracks" prefix from the playlist name
433 playlist_obj["title"] = playlist_obj["title"].removeprefix("Related tracks: ")
434
435 playlist = Playlist(
436 item_id=playlist_id,
437 provider=self.domain,
438 name=playlist_obj["title"],
439 provider_mappings={
440 ProviderMapping(
441 item_id=playlist_id,
442 provider_domain=self.domain,
443 provider_instance=self.instance_id,
444 )
445 },
446 )
447 playlist.is_editable = False
448 if playlist_obj.get("description"):
449 playlist.metadata.description = playlist_obj["description"]
450 if playlist_obj.get("artwork_url"):
451 playlist.metadata.images = UniqueList(
452 [
453 MediaItemImage(
454 type=ImageType.THUMB,
455 path=self._transform_artwork_url(playlist_obj["artwork_url"]),
456 provider=self.instance_id,
457 remotely_accessible=True,
458 )
459 ]
460 )
461 if playlist_obj.get("genre"):
462 playlist.metadata.genres = playlist_obj["genre"]
463 if playlist_obj.get("tag_list"):
464 playlist.metadata.style = playlist_obj["tag_list"]
465 return playlist
466
467 async def _parse_track(self, track_obj: dict[str, Any], playlist_position: int = 0) -> Track:
468 """Parse a Soundcloud Track response to a Track model object."""
469 name, version = parse_title_and_version(track_obj["title"])
470 track_id = str(track_obj["id"])
471 track = Track(
472 item_id=track_id,
473 provider=self.domain,
474 name=name,
475 version=version,
476 duration=track_obj["duration"] / 1000,
477 provider_mappings={
478 ProviderMapping(
479 item_id=track_id,
480 provider_domain=self.domain,
481 provider_instance=self.instance_id,
482 audio_format=AudioFormat(
483 content_type=ContentType.MP3,
484 ),
485 url=track_obj["permalink_url"],
486 )
487 },
488 position=playlist_position,
489 )
490 user_id = track_obj["user"]["id"]
491 user = await self._soundcloud.get_user_details(user_id)
492 artist = await self._parse_artist(user)
493 if artist and artist.item_id not in {x.item_id for x in track.artists}:
494 track.artists.append(artist)
495
496 if track_obj.get("artwork_url"):
497 track.metadata.images = UniqueList(
498 [
499 MediaItemImage(
500 type=ImageType.THUMB,
501 path=self._transform_artwork_url(track_obj["artwork_url"]),
502 provider=self.instance_id,
503 remotely_accessible=True,
504 )
505 ]
506 )
507
508 if track_obj.get("description"):
509 track.metadata.description = track_obj["description"]
510 if track_obj.get("genre"):
511 track.metadata.genres = {track_obj["genre"]}
512 if track_obj.get("tag_list"):
513 track.metadata.style = track_obj["tag_list"]
514 return track
515
516 def _transform_artwork_url(self, artwork_url: str) -> str:
517 """Patch artwork URL to a high quality thumbnail."""
518 # This is undocumented in their API docs, but was previously
519 return artwork_url.replace("large", "t500x500")
520