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.logger.debug("Parse artist failed: %s", artist, exc_info=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:
179                self.logger.debug(
180                    "Unexpected Soundcloud API response when parsing playlists: %s",
181                    item,
182                )
183                continue
184
185            try:
186                playlist = await self._get_playlist_object(
187                    prov_playlist_id=raw_playlist["id"],
188                )
189
190                yield await self._parse_playlist(playlist)
191            except (KeyError, TypeError, InvalidDataError, IndexError) as error:
192                self.logger.debug(
193                    "Failed to obtain Soundcloud playlist details: %s",
194                    raw_playlist,
195                    exc_info=error,
196                )
197                continue
198
199        self.logger.debug(
200            "Processing Soundcloud library playlists took %s seconds",
201            round(time.time() - time_start, 2),
202        )
203
204    async def get_library_tracks(self) -> AsyncGenerator[Track]:
205        """Retrieve library tracks from Soundcloud."""
206        time_start = time.time()
207        drm_protected = 0
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 DrmProtectedTrackError:
212                drm_protected += 1
213                continue
214            except (KeyError, TypeError, InvalidDataError, IndexError) as error:
215                # somehow certain track id's don't exist (anymore)
216                self.logger.debug(
217                    "%s: Parse track with id %s failed: %s",
218                    type(error).__name__,
219                    track["id"],
220                    track,
221                )
222                continue
223
224        if drm_protected:
225            self.logger.info(
226                "Skipped %s DRM protected track(s) while syncing the library: "
227                "Soundcloud only allows those to be played in its own apps",
228                drm_protected,
229            )
230
231        self.logger.debug(
232            "Processing Soundcloud library tracks took %s seconds",
233            round(time.time() - time_start, 2),
234        )
235
236    async def get_recommendations(self) -> list[RecommendationFolder]:
237        """Get this provider's available recommendation rows, without items."""
238        rows = await self._recommendation_rows_from_payload()
239        rows.append(
240            RecommendationFolder(
241                name="SoundCloud Feed",
242                translation_key="soundcloud_feed",
243                item_id=f"{self.instance_id}_sc_subscribed_feed",
244                provider=self.instance_id,
245                icon="mdi-rss",
246            )
247        )
248        return rows
249
250    async def get_recommendation_items(
251        self, item_id: str
252    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
253        """
254        Get the items for a single recommendation row.
255
256        :param item_id: The item_id of the row, as returned by get_recommendations.
257        """
258        if item_id == f"{self.instance_id}_sc_subscribed_feed":
259            return UniqueList(await self._get_subscribed_feed_tracks())
260        return await self._recommendation_items_from_payload(item_id)
261
262    async def _fetch_recommendation_payload(self) -> list[RecommendationFolder]:
263        """Fetch and parse the mixed-selection collections as folders with items."""
264        folders: list[RecommendationFolder] = []
265        recommendations = await self._soundcloud.get_mixed_selection(40)
266        for collection in recommendations.get("collection", []):
267            folder = RecommendationFolder(
268                name=collection["title"],
269                item_id=f"{self.instance_id}_{collection['id']}",
270                provider=self.instance_id,
271                icon="mdi-playlist-music",
272            )
273            for playlist in (collection.get("items") or {}).get("collection", []):
274                # Each items can be a track, playlist, album or artist but seems playlists only
275                if playlist.get("kind") == "system-playlist":
276                    folder.items.append(await self._parse_playlist(playlist))
277                else:
278                    self.logger.debug(
279                        "Unknown item type in collection for SoundCloud: %s", playlist.get("kind")
280                    )
281                    continue
282            folders.append(folder)
283        return folders
284
285    @use_cache(3600 * 3)  # Cache for 3 hours
286    async def _get_subscribed_feed_tracks(self) -> list[Track]:
287        """Fetch and parse the tracks of the subscribed feed."""
288        tracks: list[Track] = []
289        feed = await self._soundcloud.get_subscribe_feed(40)
290        if not feed or "collection" not in feed:
291            return tracks
292        for item in feed["collection"]:
293            if item.get("type") == "track" or item.get("type") == "track-repost":
294                try:
295                    tracks.append(await self._parse_track(item.get("track")))
296                except (KeyError, TypeError, InvalidDataError, IndexError) as error:
297                    # a single unplayable track (e.g. DRM protected) must not empty the feed
298                    self.logger.debug("Parse track failed: %s", item, exc_info=error)
299                    continue
300            else:
301                self.logger.debug(
302                    "Unknown type in subscribed feed for SoundCloud: %s", item.get("type")
303                )
304                continue
305        return tracks
306
307    @use_cache(3600 * 24 * 14)  # Cache for 14 days
308    async def get_artist(self, prov_artist_id: str) -> Artist:
309        """Get full artist details by id."""
310        artist_obj = await self._soundcloud.get_user_details(prov_artist_id)
311        try:
312            if artist_obj:
313                artist = await self._parse_artist(artist_obj)
314        except (KeyError, TypeError, InvalidDataError, IndexError) as error:
315            self.logger.debug("Parse artist failed: %s", artist_obj, exc_info=error)
316        return artist
317
318    @use_cache(3600 * 24 * 14)  # Cache for 14 days
319    async def get_track(self, prov_track_id: str) -> Track:
320        """Get full track details by id."""
321        track_obj = await self._soundcloud.get_track_details(prov_track_id)
322        try:
323            return await self._parse_track(track_obj[0])
324        except (KeyError, TypeError, InvalidDataError, IndexError) as error:
325            self.logger.debug("Parse track failed: %s", track_obj, exc_info=error)
326            msg = f"Soundcloud track {prov_track_id} is not available"
327            raise MediaNotFoundError(msg) from error
328
329    @use_cache(3600 * 24 * 14)  # Cache for 14 days
330    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
331        """Get full playlist details by id."""
332        playlist_obj = await self._get_playlist_object(prov_playlist_id)
333        try:
334            playlist = await self._parse_playlist(playlist_obj)
335        except (KeyError, TypeError, InvalidDataError, IndexError) as error:
336            self.logger.debug("Parse playlist failed: %s", playlist_obj, exc_info=error)
337        return playlist
338
339    async def _get_playlist_object(self, prov_playlist_id: str) -> dict[str, Any]:
340        """Get playlist object from Soundcloud API based on playlist ID type."""
341        # Handle playlist id's which are actually numbers
342        prov_playlist_id = str(prov_playlist_id)
343        if prov_playlist_id.startswith("soundcloud:system-playlists"):
344            # Handle system playlists
345            result = await self._soundcloud.get_system_playlist_details(prov_playlist_id)
346            return cast("dict[str, Any]", result)
347        # Handle regular playlists
348        result = await self._soundcloud.get_playlist_details(prov_playlist_id)
349        return cast("dict[str, Any]", result)
350
351    @use_cache(3600 * 3, allow_expired_cache=True)  # Cache for 3 hours
352    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
353        """Get playlist tracks."""
354        result: list[Track] = []
355        if page > 0:
356            # TODO: soundcloud doesn't seem to support paging for playlist tracks ?!
357            return result
358        playlist_obj = await self._get_playlist_object(prov_playlist_id)
359        if "tracks" not in playlist_obj:
360            return result
361        drm_protected = 0
362        for index, item in enumerate(playlist_obj["tracks"], 1):
363            try:
364                # Skip some ugly "tracks" entries, example:
365                # {'id': 123, 'kind': 'track', 'monetization_model': 'NOT_APPLICABLE'}
366                if "title" in item:
367                    if track := await self._parse_track(item, index):
368                        result.append(track)
369                # But also try to get the track details if the track is not in the playlist
370                else:
371                    track_details = await self._soundcloud.get_track_details(item["id"])
372                    if track := await self._parse_track(track_details[0], index):
373                        result.append(track)
374            except DrmProtectedTrackError:
375                drm_protected += 1
376                continue
377            except (KeyError, TypeError, InvalidDataError, IndexError) as error:
378                self.logger.debug("Parse track failed: %s", item, exc_info=error)
379                continue
380        if drm_protected:
381            self.logger.debug(
382                "Skipped %s DRM protected track(s) in Soundcloud playlist %s",
383                drm_protected,
384                prov_playlist_id,
385            )
386        return result
387
388    @use_cache(3600 * 24 * 14, allow_expired_cache=True)  # Cache for 14 days
389    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
390        """Get a list of (max 100, API doesn't allow a higher limit) tracks for the given artist."""
391        tracks_obj = await self._soundcloud.get_tracks_from_user(prov_artist_id, 100)
392
393        tracks: list[Track] = []
394
395        # Try multiple fallback mechanisms to get tracks collection
396        collection = self._extract_collection(tracks_obj)
397
398        # If still no collection, try getting popular tracks
399        if not collection:
400            try:
401                popular_tracks_obj = await self._soundcloud.get_popular_tracks_user(
402                    prov_artist_id, 100
403                )
404                collection = self._extract_collection(popular_tracks_obj)
405            except ClientError as error:
406                self.logger.debug("Failed to get popular tracks: %s", error)
407
408        # If no collection found, log warning and return empty list
409        if not collection:
410            self.logger.warning(
411                "No tracks found for artist %s (tried collection, items, and popular tracks)",
412                prov_artist_id,
413            )
414            return tracks
415
416        for item in collection:
417            song = await self._soundcloud.get_track_details(item["id"])
418            try:
419                track = await self._parse_track(song[0])
420                tracks.append(track)
421            except (KeyError, TypeError, InvalidDataError, IndexError) as error:
422                self.logger.debug("Parse track failed: %s", song, exc_info=error)
423                continue
424        return tracks
425
426    def _extract_collection(
427        self, api_response: dict[str, Any] | None
428    ) -> list[dict[str, Any]] | None:
429        """Extract collection or items from SoundCloud API response."""
430        if not api_response:
431            return None
432        return api_response.get("collection") or api_response.get("items")
433
434    @use_cache(3600 * 24 * 14, allow_expired_cache=True)  # Cache for 14 days
435    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
436        """Retrieve a dynamic list of tracks based on the provided item."""
437        tracks_obj = await self._soundcloud.get_recommended(prov_track_id, limit)
438        tracks: list[Track] = []
439
440        # Check if we have a valid response with tracks collection
441        collection = self._extract_collection(tracks_obj)
442
443        if not collection:
444            self.logger.warning("No similar tracks found for track %s", prov_track_id)
445            return tracks
446
447        for item in collection:
448            song = await self._soundcloud.get_track_details(item["id"])
449            try:
450                track = await self._parse_track(song[0])
451                tracks.append(track)
452            except (KeyError, TypeError, InvalidDataError, IndexError) as error:
453                self.logger.debug("Parse track failed: %s", song, exc_info=error)
454                continue
455
456        return tracks
457
458    async def _get_stream_url(self, track_info: dict[str, Any]) -> str | None:
459        """
460        Get stream URL, preferring progressive (HTTP) over HLS.
461
462        SoundCloud HLS playlists can have limited content windows (~10 min) which
463        cause seeking failures mid-track. Progressive HTTP URLs support full
464        range-based seeking across the entire track duration.
465
466        :param track_info: Raw track object as returned by the Soundcloud API.
467        """
468        track_auth = track_info.get("track_authorization")
469        if not track_auth:
470            return None
471        transcodings = track_info.get("media", {}).get("transcodings", [])
472        # Two passes: prefer progressive mp3, fall back to any mp3 (which may be HLS)
473        for preferred_protocol in ("progressive", None):
474            for transcoding in transcodings:
475                preset = transcoding.get("preset", "")
476                protocol = transcoding.get("format", {}).get("protocol", "")
477                if not preset.startswith("mp3"):
478                    continue
479                if preferred_protocol is not None and protocol != preferred_protocol:
480                    continue
481                stream_url = (
482                    f"{transcoding['url']}?client_id={self._soundcloud.client_id}"
483                    f"&track_authorization={track_auth}"
484                )
485                req = await self._soundcloud.get(stream_url, headers=self._soundcloud.headers)
486                if isinstance(req, dict) and "url" in req:
487                    return str(req["url"])
488        return None
489
490    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
491        """Return the content details for the given track when it will be streamed."""
492        full_json = await self._soundcloud.get_track_details(item_id)
493        track_info = full_json[0] if full_json and isinstance(full_json, list) else None
494        if track_info and _is_drm_protected(track_info):
495            # this track should never have been imported, but it may predate that check
496            msg = (
497                f"Soundcloud track {item_id} is DRM protected, "
498                "which Soundcloud only allows to be played in its own apps"
499            )
500            raise MediaNotFoundError(msg)
501        url = await self._get_stream_url(track_info) if track_info else None
502        if not url:
503            msg = f"No stream URL available for Soundcloud track {item_id}"
504            raise MediaNotFoundError(msg)
505        # Parse CDN URL expiry to avoid seeking with an expired URL.
506        # SoundCloud CDN URLs are short-lived; seeking starts a new FFmpeg process
507        # that makes a fresh HTTP request to the stored URL, which may have expired.
508        expiration = 30  # conservative default if expiry cannot be determined
509        if parsed_qs := parse_qs(urlparse(url).query):
510            for param in ("Expires", "expire"):
511                if expire_ts := parsed_qs.get(param, [None])[0]:
512                    expiration = max(30, int(expire_ts) - int(time.time()) - 10)
513                    break
514        return StreamDetails(
515            provider=self.instance_id,
516            item_id=item_id,
517            # let ffmpeg work out the details itself as
518            # soundcloud uses a mix of different content types and streaming methods
519            audio_format=AudioFormat(
520                content_type=ContentType.UNKNOWN,
521            ),
522            stream_type=StreamType.HLS
523            if url.startswith("https://cf-hls-media.sndcdn.com")
524            else StreamType.HTTP,
525            path=url,
526            can_seek=True,
527            allow_seek=True,
528            expiration=expiration,
529        )
530
531    async def _parse_artist(self, artist_obj: dict[str, Any]) -> Artist:
532        """Parse a Soundcloud user response to Artist model object."""
533        artist_id = None
534        permalink = artist_obj["permalink"]
535        if artist_obj.get("id"):
536            artist_id = artist_obj["id"]
537        if not artist_id:
538            msg = "Artist does not have a valid ID"
539            raise InvalidDataError(msg)
540        artist_id = str(artist_id)
541        artist = Artist(
542            item_id=artist_id,
543            name=artist_obj["username"],
544            provider=self.domain,
545            provider_mappings={
546                ProviderMapping(
547                    item_id=str(artist_id),
548                    provider_domain=self.domain,
549                    provider_instance=self.instance_id,
550                    url=f"https://soundcloud.com/{permalink}",
551                )
552            },
553        )
554        if artist_obj.get("description"):
555            artist.metadata.description = artist_obj["description"]
556        # skip default_avatar placeholder; it has no high-res variant and 404s after transform
557        if (avatar_url := artist_obj.get("avatar_url")) and "default_avatar" not in avatar_url:
558            img_url = self._transform_artwork_url(avatar_url)
559            artist.metadata.images = UniqueList(
560                [
561                    MediaItemImage(
562                        type=ImageType.THUMB,
563                        path=img_url,
564                        provider=self.instance_id,
565                        remotely_accessible=True,
566                    )
567                ]
568            )
569        return artist
570
571    async def _parse_playlist(self, playlist_obj: dict[str, Any]) -> Playlist:
572        """Parse a Soundcloud Playlist response to a Playlist object."""
573        playlist_id = str(playlist_obj["id"])
574        # Remove the "Related tracks" prefix from the playlist name
575        playlist_obj["title"] = playlist_obj["title"].removeprefix("Related tracks: ")
576
577        playlist = Playlist(
578            item_id=playlist_id,
579            provider=self.domain,
580            name=playlist_obj["title"],
581            provider_mappings={
582                ProviderMapping(
583                    item_id=playlist_id,
584                    provider_domain=self.domain,
585                    provider_instance=self.instance_id,
586                )
587            },
588        )
589        playlist.is_editable = False
590        if playlist_obj.get("description"):
591            playlist.metadata.description = playlist_obj["description"]
592        artwork_url = playlist_obj.get("artwork_url") or playlist_obj.get("calculated_artwork_url")
593        if artwork_url:
594            playlist.metadata.images = UniqueList(
595                [
596                    MediaItemImage(
597                        type=ImageType.THUMB,
598                        path=self._transform_artwork_url(artwork_url),
599                        provider=self.instance_id,
600                        remotely_accessible=True,
601                    )
602                ]
603            )
604        if not artwork_url:
605            # fall back to the artwork of the first track that has one
606            for track_obj in playlist_obj.get("tracks", []):
607                if track_obj.get("artwork_url"):
608                    artwork_url = track_obj["artwork_url"]
609                    break
610        if playlist_obj.get("genre"):
611            playlist.metadata.genres = {playlist_obj["genre"]}
612        if playlist_obj.get("tag_list"):
613            playlist.metadata.style = playlist_obj["tag_list"]
614        return playlist
615
616    async def _parse_track(self, track_obj: dict[str, Any], playlist_position: int = 0) -> Track:
617        """Parse a Soundcloud Track response to a Track model object."""
618        if _is_drm_protected(track_obj):
619            msg = f"Track {track_obj.get('id')} is DRM protected and can not be streamed"
620            raise DrmProtectedTrackError(msg)
621        name, version = parse_title_and_version(track_obj["title"])
622        track_id = str(track_obj["id"])
623        track = Track(
624            item_id=track_id,
625            provider=self.domain,
626            name=name,
627            version=version,
628            duration=int(track_obj["duration"] / 1000),
629            provider_mappings={
630                ProviderMapping(
631                    item_id=track_id,
632                    provider_domain=self.domain,
633                    provider_instance=self.instance_id,
634                    audio_format=AudioFormat(
635                        content_type=ContentType.MP3,
636                    ),
637                    url=track_obj["permalink_url"],
638                )
639            },
640            position=playlist_position,
641        )
642        user_id = track_obj["user"]["id"]
643        user = await self._soundcloud.get_user_details(user_id)
644        artist = await self._parse_artist(user)
645        if artist and artist.item_id not in {x.item_id for x in track.artists}:
646            track.artists.append(artist)
647
648        if track_obj.get("artwork_url"):
649            track.metadata.images = UniqueList(
650                [
651                    MediaItemImage(
652                        type=ImageType.THUMB,
653                        path=self._transform_artwork_url(track_obj["artwork_url"]),
654                        provider=self.instance_id,
655                        remotely_accessible=True,
656                    )
657                ]
658            )
659
660        if track_obj.get("description"):
661            track.metadata.description = track_obj["description"]
662        if track_obj.get("genre"):
663            track.metadata.genres = {track_obj["genre"]}
664        if track_obj.get("tag_list"):
665            track.metadata.style = track_obj["tag_list"]
666        return track
667
668    def _transform_artwork_url(self, artwork_url: str) -> str:
669        """Patch artwork URL to a high quality thumbnail."""
670        # This is undocumented in their API docs, but was previously
671        return artwork_url.replace("large", "t500x500")
672
673
674def _is_drm_protected(track_obj: dict[str, Any]) -> bool:
675    """
676    Return if the given Soundcloud track object is DRM protected.
677
678    :param track_obj: Raw track object as returned by the Soundcloud API. A partial object
679        without media details is not considered DRM protected.
680    """
681    transcodings = track_obj.get("media", {}).get("transcodings", [])
682    return any(
683        DRM_PROTOCOL_MARKER in transcoding.get("format", {}).get("protocol", "")
684        for transcoding in transcodings
685    )
686