/
/
/
1"""Soundcloud support for MusicAssistant."""
2
3from __future__ import annotations
4
5import time
6from typing import TYPE_CHECKING, Any, cast
7from urllib.parse import parse_qs, quote, urlparse
8
9from aiohttp import ClientError
10from music_assistant_models.enums import (
11 ContentType,
12 ImageType,
13 MediaType,
14 ProviderFeature,
15 StreamType,
16)
17from music_assistant_models.errors import InvalidDataError, LoginFailed, MediaNotFoundError
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.constants import CONF_ENTRY_UNOFFICIAL_PROVIDER
33from music_assistant.controllers.cache import use_cache
34from music_assistant.helpers.util import parse_title_and_version
35from music_assistant.models.music_provider import MusicProvider, describe_sync_error
36from music_assistant.models.recommendation_payload import RecommendationPayloadMixin
37
38CONF_CLIENT_ID = "client_id"
39CONF_AUTHORIZATION = "authorization"
40
41SUPPORTED_FEATURES = {
42 ProviderFeature.LIBRARY_ARTISTS,
43 ProviderFeature.LIBRARY_TRACKS,
44 ProviderFeature.LIBRARY_PLAYLISTS,
45 ProviderFeature.BROWSE,
46 ProviderFeature.SEARCH,
47 ProviderFeature.ARTIST_TOPTRACKS,
48 ProviderFeature.SIMILAR_TRACKS,
49 ProviderFeature.RECOMMENDATIONS,
50}
51
52# When searching, the duration is compared with the full duration to check if it's a preview track etc.
53# Sometimes, for non preview tracks, the duration is off by a bit compared to the full duration so any differences below
54# this tolerance are acceptable
55SEARCH_DURATION_COMPARISON_TOLERANCE = 1000
56
57# Soundcloud serves DRM protected (encrypted HLS) audio for part of its catalog, mostly major
58# label releases. Playing those requires a Widevine/FairPlay CDM, which we do not have.
59# Such a track does still advertise plain mp3 transcodings, but requesting one of those returns
60# a 404, so the presence of an encrypted protocol is what tells us the track is unplayable.
61DRM_PROTOCOL_MARKER = "encrypted"
62
63
64class DrmProtectedTrackError(MediaNotFoundError):
65 """Error raised when a Soundcloud track can only be played in Soundcloud's own apps."""
66
67 def __init__(self, item_id: str | int | None) -> None:
68 """Initialize with the id of the track that can not be streamed."""
69 super().__init__(
70 f"Soundcloud track {item_id} is DRM protected, "
71 "which Soundcloud only allows to be played in its own apps"
72 )
73
74
75if TYPE_CHECKING:
76 from collections.abc import AsyncGenerator
77
78 from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
79 from music_assistant_models.media_items import BrowseFolder, ItemMapping, MediaItemType
80 from music_assistant_models.provider import ProviderManifest
81
82 from music_assistant.mass import MusicAssistant
83 from music_assistant.models import ProviderInstanceType
84
85
86async def setup(
87 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
88) -> ProviderInstanceType:
89 """Initialize provider(instance) with given configuration."""
90 return SoundcloudMusicProvider(mass, manifest, config, SUPPORTED_FEATURES)
91
92
93class SoundcloudMusicProvider(RecommendationPayloadMixin, MusicProvider):
94 """Provider for Soundcloud."""
95
96 # keep the pre-refactor 3h refresh interval for the mixed-selections payload
97 recommendation_payload_ttl = 3600 * 3
98
99 _user_id: str = ""
100 _soundcloud: SoundcloudAsyncAPI = None
101 _me: dict[str, Any]
102
103 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
104 """Return Config entries to configure this provider."""
105 return (CONF_ENTRY_UNOFFICIAL_PROVIDER,)
106
107 async def handle_async_init(self) -> None:
108 """Set up the Soundcloud provider."""
109 client_id = self.get_setup_value(CONF_CLIENT_ID)
110 auth_token = self.get_setup_value(CONF_AUTHORIZATION)
111 if not client_id or not auth_token:
112 msg = "Invalid login credentials"
113 raise LoginFailed(msg)
114 self._soundcloud = SoundcloudAsyncAPI(auth_token, client_id, self.mass.http_session)
115 await self._soundcloud.login()
116 self._me = await self._soundcloud.get_account_details()
117 # the API returns the id as a number, while it is only ever used to build request urls
118 self._user_id = str(self._me["id"])
119
120 @use_cache(3600 * 48) # Cache for 48 hours
121 async def search(
122 self, search_query: str, media_types: list[MediaType], limit: int = 10
123 ) -> SearchResults:
124 """
125 Perform search on musicprovider.
126
127 :param search_query: Search query.
128 :param media_types: A list of media_types to include.
129 :param limit: Number of items to return in the search (per type).
130 """
131 result = SearchResults()
132
133 media_types = [
134 x for x in media_types if x in (MediaType.ARTIST, MediaType.TRACK, MediaType.PLAYLIST)
135 ]
136 if not media_types:
137 return result
138
139 searchresult = await self._soundcloud.search(quote(search_query), limit)
140
141 drm_protected = 0
142 for item in searchresult["collection"]:
143 try:
144 media_type = item["kind"]
145 if media_type == "user" and MediaType.ARTIST in media_types:
146 result.artists = [*result.artists, await self._parse_artist(item)]
147 elif media_type == "track" and MediaType.TRACK in media_types:
148 duration = item.get("duration", 0)
149 full_duration = item.get("full_duration", 0)
150 if abs(duration - full_duration) < SEARCH_DURATION_COMPARISON_TOLERANCE:
151 # skip preview/snippet tracks (e.g. in case of free accounts)
152 # where duration is significantly shorter than full_duration
153 result.tracks = [*result.tracks, await self._parse_track(item)]
154 elif media_type == "playlist" and MediaType.PLAYLIST in media_types:
155 result.playlists = [*result.playlists, await self._parse_playlist(item)]
156 except DrmProtectedTrackError:
157 drm_protected += 1
158 continue
159 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
160 # a single unusable result must not discard the rest of the search results
161 self.logger.debug(
162 "Skipping search result %s: %s", _item_id(item), describe_sync_error(error)
163 )
164 continue
165
166 if drm_protected:
167 self.logger.debug(
168 "Skipped %s DRM protected track(s) in Soundcloud search results", drm_protected
169 )
170 return result
171
172 async def get_library_artists(self) -> AsyncGenerator[Artist]:
173 """Retrieve all library artists from Soundcloud."""
174 time_start = time.time()
175
176 following = await self._soundcloud.get_following(self._user_id)
177 self.logger.debug(
178 "Processing Soundcloud library artists took %s seconds",
179 round(time.time() - time_start, 2),
180 )
181 for artist in following["collection"]:
182 try:
183 yield await self._parse_artist(artist)
184 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
185 self._report_skipped_item(MediaType.ARTIST, artist, error)
186 continue
187
188 async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
189 """Retrieve all library playlists from Soundcloud."""
190 time_start = time.time()
191 async for item in self._soundcloud.get_account_playlists():
192 try:
193 raw_playlist = item["playlist"]
194 except KeyError as error:
195 # the entry holds no playlist payload at all, so there is no id to name
196 self.report_skipped_sync_item(MediaType.PLAYLIST, None, error)
197 continue
198
199 try:
200 playlist = await self._get_playlist_object(
201 prov_playlist_id=raw_playlist["id"],
202 )
203
204 yield await self._parse_playlist(playlist)
205 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
206 self._report_skipped_item(MediaType.PLAYLIST, raw_playlist, error)
207 continue
208
209 self.logger.debug(
210 "Processing Soundcloud library playlists took %s seconds",
211 round(time.time() - time_start, 2),
212 )
213
214 async def get_library_tracks(self) -> AsyncGenerator[Track]:
215 """Retrieve library tracks from Soundcloud."""
216 time_start = time.time()
217 drm_protected = 0
218 async for track in self._soundcloud.get_track_details_liked(self._user_id):
219 try:
220 yield await self._parse_track(track)
221 except DrmProtectedTrackError:
222 # a permanent restriction on the track rather than a failure to read it, so
223 # it is counted and logged once below instead of reported as a sync failure
224 drm_protected += 1
225 continue
226 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
227 # somehow certain track id's don't exist (anymore)
228 self._report_skipped_item(MediaType.TRACK, track, error)
229 continue
230
231 if drm_protected:
232 self.logger.info(
233 "Skipped %s DRM protected track(s) while syncing the library: "
234 "Soundcloud only allows those to be played in its own apps",
235 drm_protected,
236 )
237
238 self.logger.debug(
239 "Processing Soundcloud library tracks took %s seconds",
240 round(time.time() - time_start, 2),
241 )
242
243 async def get_recommendations(self) -> list[RecommendationFolder]:
244 """Get this provider's available recommendation rows, without items."""
245 rows = await self._recommendation_rows_from_payload()
246 rows.append(
247 RecommendationFolder(
248 name="SoundCloud Feed",
249 translation_key="soundcloud_feed",
250 item_id=f"{self.instance_id}_sc_subscribed_feed",
251 provider=self.instance_id,
252 icon="mdi-rss",
253 )
254 )
255 return rows
256
257 async def get_recommendation_items(
258 self, item_id: str
259 ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
260 """
261 Get the items for a single recommendation row.
262
263 :param item_id: The item_id of the row, as returned by get_recommendations.
264 """
265 if item_id == f"{self.instance_id}_sc_subscribed_feed":
266 return UniqueList(await self._get_subscribed_feed_tracks())
267 return await self._recommendation_items_from_payload(item_id)
268
269 async def _fetch_recommendation_payload(self) -> list[RecommendationFolder]:
270 """Fetch and parse the mixed-selection collections as folders with items."""
271 folders: list[RecommendationFolder] = []
272 recommendations = await self._soundcloud.get_mixed_selection(40)
273 for collection in recommendations.get("collection", []):
274 folder = RecommendationFolder(
275 name=collection["title"],
276 item_id=f"{self.instance_id}_{collection['id']}",
277 provider=self.instance_id,
278 icon="mdi-playlist-music",
279 )
280 for playlist in (collection.get("items") or {}).get("collection", []):
281 # Each items can be a track, playlist, album or artist but seems playlists only
282 if playlist.get("kind") == "system-playlist":
283 folder.items.append(await self._parse_playlist(playlist))
284 else:
285 self.logger.debug(
286 "Unknown item type in collection for SoundCloud: %s", playlist.get("kind")
287 )
288 continue
289 folders.append(folder)
290 return folders
291
292 @use_cache(3600 * 3) # Cache for 3 hours
293 async def _get_subscribed_feed_tracks(self) -> list[Track]:
294 """Fetch and parse the tracks of the subscribed feed."""
295 tracks: list[Track] = []
296 feed = await self._soundcloud.get_subscribe_feed(40)
297 if not feed or "collection" not in feed:
298 return tracks
299 drm_protected = 0
300 for item in feed["collection"]:
301 if item.get("type") == "track" or item.get("type") == "track-repost":
302 try:
303 tracks.append(await self._parse_track(item.get("track")))
304 except DrmProtectedTrackError:
305 drm_protected += 1
306 continue
307 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
308 # a single unusable track must not empty the feed
309 self.logger.debug(
310 "Skipping feed track %s: %s",
311 _item_id(item.get("track")),
312 describe_sync_error(error),
313 )
314 continue
315 else:
316 self.logger.debug(
317 "Unknown type in subscribed feed for SoundCloud: %s", item.get("type")
318 )
319 continue
320 if drm_protected:
321 self.logger.debug(
322 "Skipped %s DRM protected track(s) in the Soundcloud subscribed feed",
323 drm_protected,
324 )
325 return tracks
326
327 @use_cache(3600 * 24 * 14) # Cache for 14 days
328 async def get_artist(self, prov_artist_id: str) -> Artist:
329 """Get full artist details by id."""
330 artist_obj = await self._soundcloud.get_user_details(prov_artist_id)
331 try:
332 if artist_obj:
333 artist = await self._parse_artist(artist_obj)
334 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
335 self.logger.debug("Skipping artist %s: %s", prov_artist_id, describe_sync_error(error))
336 return artist
337
338 @use_cache(3600 * 24 * 14) # Cache for 14 days
339 async def get_track(self, prov_track_id: str) -> Track:
340 """Get full track details by id."""
341 track_obj = await self._soundcloud.get_track_details(prov_track_id)
342 try:
343 return await self._parse_track(track_obj[0])
344 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
345 self.logger.debug("Skipping track %s: %s", prov_track_id, describe_sync_error(error))
346 msg = f"Soundcloud track {prov_track_id} is not available"
347 raise MediaNotFoundError(msg) from error
348
349 @use_cache(3600 * 24 * 14) # Cache for 14 days
350 async def get_playlist(self, prov_playlist_id: str) -> Playlist:
351 """Get full playlist details by id."""
352 playlist_obj = await self._get_playlist_object(prov_playlist_id)
353 try:
354 playlist = await self._parse_playlist(playlist_obj)
355 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
356 self.logger.debug(
357 "Skipping playlist %s: %s", prov_playlist_id, describe_sync_error(error)
358 )
359 return playlist
360
361 async def _get_playlist_object(self, prov_playlist_id: str) -> dict[str, Any]:
362 """Get playlist object from Soundcloud API based on playlist ID type."""
363 # Handle playlist id's which are actually numbers
364 prov_playlist_id = str(prov_playlist_id)
365 if prov_playlist_id.startswith("soundcloud:system-playlists"):
366 # Handle system playlists
367 result = await self._soundcloud.get_system_playlist_details(prov_playlist_id)
368 return cast("dict[str, Any]", result)
369 # Handle regular playlists
370 result = await self._soundcloud.get_playlist_details(prov_playlist_id)
371 return cast("dict[str, Any]", result)
372
373 @use_cache(3600 * 3, allow_expired_cache=True) # Cache for 3 hours
374 async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
375 """Get playlist tracks."""
376 result: list[Track] = []
377 if page > 0:
378 # TODO: soundcloud doesn't seem to support paging for playlist tracks ?!
379 return result
380 playlist_obj = await self._get_playlist_object(prov_playlist_id)
381 if "tracks" not in playlist_obj:
382 return result
383 drm_protected = 0
384 for index, item in enumerate(playlist_obj["tracks"], 1):
385 try:
386 # Skip some ugly "tracks" entries, example:
387 # {'id': 123, 'kind': 'track', 'monetization_model': 'NOT_APPLICABLE'}
388 if "title" in item:
389 if track := await self._parse_track(item, index):
390 result.append(track)
391 # But also try to get the track details if the track is not in the playlist
392 else:
393 track_details = await self._soundcloud.get_track_details(item["id"])
394 if track := await self._parse_track(track_details[0], index):
395 result.append(track)
396 except DrmProtectedTrackError:
397 drm_protected += 1
398 continue
399 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
400 self.logger.debug(
401 "Skipping track %s in playlist %s: %s",
402 _item_id(item),
403 prov_playlist_id,
404 describe_sync_error(error),
405 )
406 continue
407 if drm_protected:
408 self.logger.debug(
409 "Skipped %s DRM protected track(s) in Soundcloud playlist %s",
410 drm_protected,
411 prov_playlist_id,
412 )
413 return result
414
415 @use_cache(3600 * 24 * 14, allow_expired_cache=True) # Cache for 14 days
416 async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
417 """Get a list of (max 100, API doesn't allow a higher limit) tracks for the given artist."""
418 tracks_obj = await self._soundcloud.get_tracks_from_user(prov_artist_id, 100)
419
420 tracks: list[Track] = []
421
422 # Try multiple fallback mechanisms to get tracks collection
423 collection = self._extract_collection(tracks_obj)
424
425 # If still no collection, try getting popular tracks
426 if not collection:
427 try:
428 popular_tracks_obj = await self._soundcloud.get_popular_tracks_user(
429 prov_artist_id, 100
430 )
431 collection = self._extract_collection(popular_tracks_obj)
432 except ClientError as error:
433 self.logger.debug("Failed to get popular tracks: %s", error)
434
435 # If no collection found, log warning and return empty list
436 if not collection:
437 self.logger.warning(
438 "No tracks found for artist %s (tried collection, items, and popular tracks)",
439 prov_artist_id,
440 )
441 return tracks
442
443 drm_protected = 0
444 for item in collection:
445 song = await self._soundcloud.get_track_details(item["id"])
446 try:
447 track = await self._parse_track(song[0])
448 tracks.append(track)
449 except DrmProtectedTrackError:
450 drm_protected += 1
451 continue
452 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
453 self.logger.debug("Skipping track %s: %s", item["id"], describe_sync_error(error))
454 continue
455 if drm_protected:
456 self.logger.debug(
457 "Skipped %s DRM protected track(s) in the top tracks of Soundcloud artist %s",
458 drm_protected,
459 prov_artist_id,
460 )
461 return tracks
462
463 def _extract_collection(
464 self, api_response: dict[str, Any] | None
465 ) -> list[dict[str, Any]] | None:
466 """Extract collection or items from SoundCloud API response."""
467 if not api_response:
468 return None
469 return api_response.get("collection") or api_response.get("items")
470
471 @use_cache(3600 * 24 * 14, allow_expired_cache=True) # Cache for 14 days
472 async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
473 """Retrieve a dynamic list of tracks based on the provided item."""
474 tracks_obj = await self._soundcloud.get_recommended(prov_track_id, limit)
475 tracks: list[Track] = []
476
477 # Check if we have a valid response with tracks collection
478 collection = self._extract_collection(tracks_obj)
479
480 if not collection:
481 self.logger.warning("No similar tracks found for track %s", prov_track_id)
482 return tracks
483
484 drm_protected = 0
485 for item in collection:
486 song = await self._soundcloud.get_track_details(item["id"])
487 try:
488 track = await self._parse_track(song[0])
489 tracks.append(track)
490 except DrmProtectedTrackError:
491 drm_protected += 1
492 continue
493 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
494 self.logger.debug("Skipping track %s: %s", item["id"], describe_sync_error(error))
495 continue
496
497 if drm_protected:
498 self.logger.debug(
499 "Skipped %s DRM protected track(s) in tracks similar to Soundcloud track %s",
500 drm_protected,
501 prov_track_id,
502 )
503 return tracks
504
505 async def _get_stream_url(self, track_info: dict[str, Any]) -> str | None:
506 """
507 Get stream URL, preferring progressive (HTTP) over HLS.
508
509 SoundCloud HLS playlists can have limited content windows (~10 min) which
510 cause seeking failures mid-track. Progressive HTTP URLs support full
511 range-based seeking across the entire track duration.
512
513 :param track_info: Raw track object as returned by the Soundcloud API.
514 """
515 track_auth = track_info.get("track_authorization")
516 if not track_auth:
517 return None
518 transcodings = track_info.get("media", {}).get("transcodings", [])
519 # Two passes: prefer progressive mp3, fall back to any mp3 (which may be HLS)
520 for preferred_protocol in ("progressive", None):
521 for transcoding in transcodings:
522 preset = transcoding.get("preset", "")
523 protocol = transcoding.get("format", {}).get("protocol", "")
524 if not preset.startswith("mp3"):
525 continue
526 if preferred_protocol is not None and protocol != preferred_protocol:
527 continue
528 stream_url = (
529 f"{transcoding['url']}?client_id={self._soundcloud.client_id}"
530 f"&track_authorization={track_auth}"
531 )
532 req = await self._soundcloud.get(stream_url, headers=self._soundcloud.headers)
533 if isinstance(req, dict) and "url" in req:
534 return str(req["url"])
535 return None
536
537 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
538 """Return the content details for the given track when it will be streamed."""
539 full_json = await self._soundcloud.get_track_details(item_id)
540 track_info = full_json[0] if full_json and isinstance(full_json, list) else None
541 if track_info and _is_drm_protected(track_info):
542 # this track should never have been imported, but it may predate that check
543 raise DrmProtectedTrackError(item_id)
544 url = await self._get_stream_url(track_info) if track_info else None
545 if not url:
546 msg = f"No stream URL available for Soundcloud track {item_id}"
547 raise MediaNotFoundError(msg)
548 # Parse CDN URL expiry to avoid seeking with an expired URL.
549 # SoundCloud CDN URLs are short-lived; seeking starts a new FFmpeg process
550 # that makes a fresh HTTP request to the stored URL, which may have expired.
551 expiration = 30 # conservative default if expiry cannot be determined
552 if parsed_qs := parse_qs(urlparse(url).query):
553 for param in ("Expires", "expire"):
554 if expire_ts := parsed_qs.get(param, [None])[0]:
555 expiration = max(30, int(expire_ts) - int(time.time()) - 10)
556 break
557 return StreamDetails(
558 provider=self.instance_id,
559 item_id=item_id,
560 # let ffmpeg work out the details itself as
561 # soundcloud uses a mix of different content types and streaming methods
562 audio_format=AudioFormat(
563 content_type=ContentType.UNKNOWN,
564 ),
565 stream_type=StreamType.HLS
566 if url.startswith("https://cf-hls-media.sndcdn.com")
567 else StreamType.HTTP,
568 path=url,
569 can_seek=True,
570 allow_seek=True,
571 expiration=expiration,
572 )
573
574 async def _parse_artist(self, artist_obj: dict[str, Any]) -> Artist:
575 """Parse a Soundcloud user response to Artist model object."""
576 artist_id = None
577 permalink = artist_obj["permalink"]
578 if artist_obj.get("id"):
579 artist_id = artist_obj["id"]
580 if not artist_id:
581 msg = "Artist does not have a valid ID"
582 raise InvalidDataError(msg)
583 artist_id = str(artist_id)
584 artist = Artist(
585 item_id=artist_id,
586 name=artist_obj["username"],
587 provider=self.domain,
588 provider_mappings={
589 ProviderMapping(
590 item_id=str(artist_id),
591 provider_domain=self.domain,
592 provider_instance=self.instance_id,
593 url=f"https://soundcloud.com/{permalink}",
594 )
595 },
596 )
597 if artist_obj.get("description"):
598 artist.metadata.description = artist_obj["description"]
599 # skip default_avatar placeholder; it has no high-res variant and 404s after transform
600 if (avatar_url := artist_obj.get("avatar_url")) and "default_avatar" not in avatar_url:
601 img_url = self._transform_artwork_url(avatar_url)
602 artist.metadata.images = UniqueList(
603 [
604 MediaItemImage(
605 type=ImageType.THUMB,
606 path=img_url,
607 provider=self.instance_id,
608 remotely_accessible=True,
609 )
610 ]
611 )
612 return artist
613
614 async def _parse_playlist(self, playlist_obj: dict[str, Any]) -> Playlist:
615 """Parse a Soundcloud Playlist response to a Playlist object."""
616 playlist_id = str(playlist_obj["id"])
617 # Remove the "Related tracks" prefix from the playlist name
618 playlist_obj["title"] = playlist_obj["title"].removeprefix("Related tracks: ")
619
620 playlist = Playlist(
621 item_id=playlist_id,
622 provider=self.domain,
623 name=playlist_obj["title"],
624 provider_mappings={
625 ProviderMapping(
626 item_id=playlist_id,
627 provider_domain=self.domain,
628 provider_instance=self.instance_id,
629 )
630 },
631 )
632 playlist.is_editable = False
633 if playlist_obj.get("description"):
634 playlist.metadata.description = playlist_obj["description"]
635 artwork_url = playlist_obj.get("artwork_url") or playlist_obj.get("calculated_artwork_url")
636 if artwork_url:
637 playlist.metadata.images = UniqueList(
638 [
639 MediaItemImage(
640 type=ImageType.THUMB,
641 path=self._transform_artwork_url(artwork_url),
642 provider=self.instance_id,
643 remotely_accessible=True,
644 )
645 ]
646 )
647 if not artwork_url:
648 # fall back to the artwork of the first track that has one
649 for track_obj in playlist_obj.get("tracks", []):
650 if track_obj.get("artwork_url"):
651 artwork_url = track_obj["artwork_url"]
652 break
653 if playlist_obj.get("genre"):
654 playlist.metadata.genres = {playlist_obj["genre"]}
655 if playlist_obj.get("tag_list"):
656 playlist.metadata.style = playlist_obj["tag_list"]
657 return playlist
658
659 async def _parse_track(self, track_obj: dict[str, Any], playlist_position: int = 0) -> Track:
660 """Parse a Soundcloud Track response to a Track model object."""
661 if _is_drm_protected(track_obj):
662 raise DrmProtectedTrackError(track_obj.get("id"))
663 name, version = parse_title_and_version(track_obj["title"])
664 track_id = str(track_obj["id"])
665 track = Track(
666 item_id=track_id,
667 provider=self.domain,
668 name=name,
669 version=version,
670 duration=int(track_obj["duration"] / 1000),
671 provider_mappings={
672 ProviderMapping(
673 item_id=track_id,
674 provider_domain=self.domain,
675 provider_instance=self.instance_id,
676 audio_format=AudioFormat(
677 content_type=ContentType.MP3,
678 ),
679 url=track_obj["permalink_url"],
680 )
681 },
682 position=playlist_position,
683 )
684 user_id = track_obj["user"]["id"]
685 user = await self._soundcloud.get_user_details(user_id)
686 artist = await self._parse_artist(user)
687 if artist and artist.item_id not in {x.item_id for x in track.artists}:
688 track.artists.append(artist)
689
690 if track_obj.get("artwork_url"):
691 track.metadata.images = UniqueList(
692 [
693 MediaItemImage(
694 type=ImageType.THUMB,
695 path=self._transform_artwork_url(track_obj["artwork_url"]),
696 provider=self.instance_id,
697 remotely_accessible=True,
698 )
699 ]
700 )
701
702 if track_obj.get("description"):
703 track.metadata.description = track_obj["description"]
704 if track_obj.get("genre"):
705 track.metadata.genres = {track_obj["genre"]}
706 if track_obj.get("tag_list"):
707 track.metadata.style = track_obj["tag_list"]
708 return track
709
710 def _transform_artwork_url(self, artwork_url: str) -> str:
711 """Patch artwork URL to a high quality thumbnail."""
712 # This is undocumented in their API docs, but was previously
713 return artwork_url.replace("large", "t500x500")
714
715 def _report_skipped_item(
716 self, media_type: MediaType, item_obj: dict[str, Any], err: Exception
717 ) -> None:
718 """
719 Report a library item that was dropped while listing the library.
720
721 :param media_type: Media type of the skipped item.
722 :param item_obj: Raw api object of the skipped item, whose id is its item id.
723 :param err: The error that made the item unusable.
724 """
725 item_id = item_obj.get("id")
726 self.report_skipped_sync_item(media_type, str(item_id) if item_id else None, err)
727
728
729def _item_id(item_obj: Any) -> Any:
730 """Return the id of a raw api object, or None if it does not have a readable one."""
731 # called from the handlers that log an unusable object, so it must never raise itself
732 return item_obj.get("id") if isinstance(item_obj, dict) else None
733
734
735def _is_drm_protected(track_obj: dict[str, Any]) -> bool:
736 """
737 Return if the given Soundcloud track object is DRM protected.
738
739 :param track_obj: Raw track object as returned by the Soundcloud API. A partial object
740 without media details is not considered DRM protected.
741 """
742 transcodings = track_obj.get("media", {}).get("transcodings", [])
743 return any(
744 DRM_PROTOCOL_MARKER in transcoding.get("format", {}).get("protocol", "")
745 for transcoding in transcodings
746 )
747