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