music-assistant-server

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