music-assistant-server

1"""Nugs.net musicprovider support for MusicAssistant."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator
6from datetime import UTC, datetime
7from time import time
8from typing import TYPE_CHECKING, Any
9
10from aiohttp import ClientTimeout
11from music_assistant_models.enums import (
12    ContentType,
13    ImageType,
14    MediaType,
15    ProviderFeature,
16    StreamType,
17)
18from music_assistant_models.errors import (
19    InvalidDataError,
20    LoginFailed,
21    MediaNotFoundError,
22    ResourceTemporarilyUnavailable,
23)
24from music_assistant_models.media_items import (
25    Album,
26    Artist,
27    AudioFormat,
28    BrowseFolder,
29    ItemMapping,
30    MediaItemImage,
31    MediaItemMetadata,
32    MediaItemType,
33    Playlist,
34    ProviderMapping,
35    RecommendationFolder,
36    Track,
37    UniqueList,
38)
39from music_assistant_models.streamdetails import StreamDetails
40
41from music_assistant.constants import (
42    CONF_ENTRY_UNOFFICIAL_PROVIDER,
43    CONF_PASSWORD,
44    CONF_USERNAME,
45)
46from music_assistant.controllers.cache import use_cache
47from music_assistant.helpers.json import json_loads
48from music_assistant.helpers.util import infer_album_type, parse_title_and_version
49from music_assistant.models.music_provider import MusicProvider
50
51if TYPE_CHECKING:
52    from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
53    from music_assistant_models.provider import ProviderManifest
54
55    from music_assistant.mass import MusicAssistant
56    from music_assistant.models import ProviderInstanceType
57
58SUPPORTED_FEATURES = {
59    ProviderFeature.BROWSE,
60    ProviderFeature.LIBRARY_ARTISTS,
61    ProviderFeature.LIBRARY_ALBUMS,
62    ProviderFeature.LIBRARY_PLAYLISTS,
63    ProviderFeature.ARTIST_ALBUMS,
64    ProviderFeature.RECOMMENDATIONS,
65}
66
67
68async def setup(
69    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
70) -> ProviderInstanceType:
71    """Initialize provider(instance) with given configuration."""
72    return NugsProvider(mass, manifest, config, SUPPORTED_FEATURES)
73
74
75class NugsProvider(MusicProvider):
76    """Provider implementation for Nugs.net."""
77
78    _auth_token: str | None = None
79    _token_expiry: float = 0
80
81    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
82        """Return Config entries to configure this provider."""
83        return (CONF_ENTRY_UNOFFICIAL_PROVIDER,)
84
85    async def handle_async_init(self) -> None:
86        """Handle async initialization of the provider."""
87        await self.login()
88
89    async def get_library_artists(self) -> AsyncGenerator[Artist]:
90        """Retrieve library artists from nugs.net."""
91        artist_data = await self._get_all_items("stash", "artists/favorite/")
92        for item in artist_data:
93            if item and item["id"]:
94                yield self._parse_artist(item)
95
96    async def get_library_albums(self) -> AsyncGenerator[Album]:
97        """Retrieve library albums from the provider."""
98        album_data = await self._get_all_items("stash", "releases/favorite")
99        for item in album_data:
100            if item and item["id"]:
101                yield self._parse_album(item)
102
103    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
104        """Retrieve playlists from the provider."""
105        playlist_data = await self._get_all_items("stash", "playlists/")
106        for item in playlist_data:
107            if item and item["id"]:
108                yield self._parse_playlist(item)
109
110    @use_cache(3600 * 24 * 14)  # Cache for 14 days
111    async def get_artist(self, prov_artist_id: str) -> Artist:
112        """Get artist details by id."""
113        endpoint = f"/releases/recent?limit=1&artistIds={prov_artist_id}"
114        artist_response = await self._get_data("catalog", endpoint)
115        artist_data = artist_response["items"][0]["artist"]
116        return self._parse_artist(artist_data)
117
118    @use_cache(3600 * 24 * 14, allow_expired_cache=True)  # Cache for 14 days
119    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
120        """Get a list of all albums for the given artist."""
121        params = {
122            "artistIds": prov_artist_id,
123            "contentType": "any",
124        }
125        return [
126            self._parse_album(item)
127            for item in await self._get_all_items("catalog", "releases/recent", **params)
128            if (item and item["id"])
129        ]
130
131    @use_cache(3600 * 24 * 14)  # Cache for 14 days
132    async def get_album(self, prov_album_id: str) -> Album:
133        """Get album details by id."""
134        endpoint = f"shows/{prov_album_id}"
135        response = await self._get_data("catalog", endpoint)
136        return self._parse_album(response["Response"])
137
138    @use_cache(3600 * 24 * 14)  # Cache for 14 days
139    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
140        """Get full playlist details by id."""
141        endpoint = f"playlists/{prov_playlist_id}"
142        response = await self._get_data("stash", endpoint)
143        return self._parse_playlist(response["items"])
144
145    async def get_track(self, prov_track_id: str) -> Track:
146        """Get full track details by id."""
147        cache_key = f"nugs_track_{prov_track_id}"
148        cached: Track | None = await self.mass.cache.get(
149            cache_key, provider=self.instance_id, base_class=Track
150        )
151        if cached:
152            return cached
153        raise MediaNotFoundError(f"Track {prov_track_id} not found")
154
155    @use_cache(3600 * 24 * 14, allow_expired_cache=True)  # Cache for 14 days
156    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
157        """Get all album tracks for given album id."""
158        endpoint = f"shows/{prov_album_id}"
159        response = await self._get_data("catalog", endpoint)
160        album_data = response["Response"]
161        artist = await self.get_artist(album_data["artistID"])
162        album = self._get_item_mapping(
163            MediaType.ALBUM, album_data["containerID"], album_data["containerInfo"]
164        )
165        image = f"https://api.livedownloads.com{album_data['img']['url']}"
166        tracks = [
167            self._parse_track(item, artist=artist, album=album, image_url=image)
168            for item in album_data["tracks"]
169            if item["trackID"]
170        ]
171        await self._cache_tracks(tracks)
172        return tracks
173
174    @use_cache(3600, allow_expired_cache=True)  # Cache for 1 hour
175    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
176        """Get playlist tracks."""
177        result: list[Track] = []
178        if page > 0:
179            # paging not yet supported
180            return []
181        endpoint = f"/playlists/{prov_playlist_id}/playlist-tracks/all"
182        nugs_result = await self._get_data("stash", endpoint)
183        for index, item in enumerate(nugs_result["items"], 1):
184            track = self._parse_track(item)
185            track.position = index
186            result.append(track)
187        await self._cache_tracks(result)
188        return result
189
190    async def get_recommendations(self) -> list[RecommendationFolder]:
191        """Get this provider's available recommendation rows, without items."""
192        return [
193            RecommendationFolder(
194                name="Most Popular",
195                translation_key="nugs_popular_shows",
196                item_id="nugs_popular_shows",
197                provider=self.instance_id,
198            ),
199            RecommendationFolder(
200                name="Recommended Shows",
201                translation_key="nugs_recommended_shows",
202                item_id="nugs_recommended_shows",
203                provider=self.instance_id,
204            ),
205            RecommendationFolder(
206                name="Recent Shows",
207                translation_key="nugs_recent_shows",
208                item_id="nugs_recent_shows",
209                provider=self.instance_id,
210            ),
211        ]
212
213    async def get_recommendation_items(
214        self, item_id: str
215    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
216        """
217        Get the items for a single recommendation row.
218
219        :param item_id: The item_id of the row, as returned by get_recommendations.
220        """
221        # caching lives on the folder-level helper: a RecommendationFolder is reconstructed
222        # from cached json via base_class, unlike this method's parameterized union type
223        folder = await self._get_recommendation_folder(item_id)
224        if folder is None:
225            return UniqueList()
226        return folder.items
227
228    async def _cache_tracks(self, tracks: list[Track]) -> None:
229        """Cache individual tracks persistently for later lookup by get_track."""
230        for track in tracks:
231            await self.mass.cache.set(
232                f"nugs_track_{track.item_id}",
233                track.to_dict(),
234                expiration=3600 * 24 * 14,
235                provider=self.instance_id,
236                persistent=True,
237            )
238
239    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
240        """Return the content details for the given track when it will be streamed."""
241        stream_url = await self._get_stream_url(item_id)
242        return StreamDetails(
243            item_id=item_id,
244            provider=self.instance_id,
245            audio_format=AudioFormat(
246                content_type=ContentType.UNKNOWN,
247            ),
248            stream_type=StreamType.HTTP,
249            path=stream_url,
250        )
251
252    def _parse_artist(self, artist_obj: dict[str, Any]) -> Artist:
253        """Parse nugs artist object to generic layout."""
254        artist_id = artist_obj.get("artistID") or artist_obj.get("id")
255        artist_name = artist_obj.get("artistName") or artist_obj.get("name")
256        artist = Artist(
257            item_id=str(artist_id),
258            provider=self.instance_id,
259            name=str(artist_name),
260            provider_mappings={
261                ProviderMapping(
262                    item_id=str(artist_id),
263                    provider_domain=self.domain,
264                    provider_instance=self.instance_id,
265                    url=f"https://catalog.nugs.net/api/v1/artists?ids={artist_id}",
266                )
267            },
268        )
269        if artist_obj.get("avatarImage"):
270            artist.metadata.add_image(
271                MediaItemImage(
272                    type=ImageType.THUMB,
273                    path=artist_obj["avatarImage"]["url"],
274                    provider=self.instance_id,
275                    remotely_accessible=True,
276                )
277            )
278        return artist
279
280    def _parse_album(self, album_obj: dict[str, Any]) -> Album:
281        """Parse nugs release/show/album object to generic album layout."""
282        item_id = album_obj.get("releaseId") or album_obj.get("id") or album_obj.get("containerID")
283        title = album_obj.get("title") or album_obj.get("containerInfo")
284        name, version = parse_title_and_version(str(title))
285        album = Album(
286            item_id=str(item_id),
287            provider=self.instance_id,
288            name=name,
289            version=version,
290            provider_mappings={
291                ProviderMapping(
292                    item_id=str(item_id),
293                    provider_domain=self.domain,
294                    provider_instance=self.instance_id,
295                )
296            },
297        )
298
299        artist_obj = album_obj.get("artist", False) or {
300            "id": album_obj["artistID"],
301            "name": album_obj["artistName"],
302        }
303        if artist_obj.get("name") and artist_obj.get("id"):
304            album.artists.append(self._parse_artist(artist_obj))
305
306        path: str | None = None
307        if album_obj.get("image"):
308            path = album_obj["image"]["url"]
309        if album_obj.get("img"):
310            path = f"https://api.livedownloads.com{album_obj['img']['url']}"
311        if path:
312            album.metadata.add_image(
313                MediaItemImage(
314                    type=ImageType.THUMB,
315                    path=path,
316                    provider=self.instance_id,
317                    remotely_accessible=True,
318                )
319            )
320        year = album_obj.get("performanceDateYear", False)
321        if not year:
322            date = album_obj.get("performanceDate", False) or album_obj.get(
323                "albumreleaseDate", False
324            )
325            if date:
326                year = date.split("-")[0]
327        if year:
328            album.year = int(year)
329
330        # No album type info in this provider so try and infer it
331        album.album_type = infer_album_type(album.name, album.version)
332
333        return album
334
335    def _parse_playlist(self, playlist_obj: dict[str, Any]) -> Playlist:
336        """Parse nugs playlist object to generic layout."""
337        return Playlist(
338            item_id=playlist_obj["id"],
339            provider=self.instance_id,
340            name=playlist_obj["name"],
341            provider_mappings={
342                ProviderMapping(
343                    item_id=playlist_obj["id"],
344                    provider_domain=self.domain,
345                    provider_instance=self.instance_id,
346                )
347            },
348            metadata=MediaItemMetadata(
349                images=UniqueList(
350                    [
351                        MediaItemImage(
352                            type=ImageType.THUMB,
353                            path=playlist_obj["imageUrl"],
354                            provider=self.instance_id,
355                            remotely_accessible=True,
356                        )
357                    ]
358                ),
359            ),
360            is_editable=False,
361        )
362
363    def _parse_track(
364        self,
365        track_obj: dict[str, Any],
366        artist: Artist | None = None,
367        album: Album | ItemMapping | None = None,
368        image_url: str | None = None,
369    ) -> Track:
370        """Parse response from inconsistent nugs.net APIs to a Track model object."""
371        track_id = (
372            track_obj.get("trackId") or track_obj.get("trackID") or track_obj.get("trackLabel")
373        )
374        track_name = track_obj.get("name") or track_obj.get("songTitle")
375        name, version = parse_title_and_version(str(track_name))
376
377        track = Track(
378            item_id=str(track_id),
379            provider=self.instance_id,
380            name=name,
381            version=version,
382            provider_mappings={
383                ProviderMapping(
384                    item_id=str(track_id),
385                    provider_domain=self.domain,
386                    provider_instance=self.instance_id,
387                    available=True,
388                )
389            },
390        )
391
392        if artist:
393            track.artists.append(artist)
394        if (
395            track_obj.get("artist")
396            and isinstance(track_obj.get("artist"), dict)
397            and track_obj["artist"].get("id")
398        ):
399            track.artists.append(
400                self._get_item_mapping(
401                    MediaType.ARTIST, track_obj["artist"]["id"], track_obj["artist"]["name"]
402                )
403            )
404        if not track.artists:
405            msg = "Track is missing artists"
406            raise InvalidDataError(msg)
407
408        if album:
409            track.album = album
410        if image_url is None and track_obj.get("image"):
411            image_url = track_obj["image"]["url"]
412        if image_url:
413            track.metadata.add_image(
414                MediaItemImage(
415                    type=ImageType.THUMB,
416                    path=image_url,
417                    provider=self.instance_id,
418                    remotely_accessible=True,
419                )
420            )
421        duration = track_obj.get("durationSeconds") or track_obj.get("totalRunningTime")
422        if duration:
423            track.duration = int(duration)
424        return track
425
426    async def _get_stream_url(self, item_id: str) -> Any:
427        subscription_info = await self._get_data("subscription", "")
428        dt_start = datetime.strptime(subscription_info["startedAt"], "%m/%d/%Y %H:%M:%S").replace(
429            tzinfo=UTC
430        )
431        dt_end = datetime.strptime(subscription_info["endsAt"], "%m/%d/%Y %H:%M:%S").replace(
432            tzinfo=UTC
433        )
434        user_info = await self._get_data("user", "")
435        url = "https://streamapi.nugs.net/bigriver/subplayer.aspx"
436        timeout = ClientTimeout(total=120)
437        params = {
438            "platformID": -1,
439            "app": 1,
440            "HLS": 1,
441            "orgn": "websdk",
442            "method": "subPlayer",
443            "trackId": item_id,
444            "subCostplanIDAccessList": subscription_info["plan"]["id"],
445            "startDateStamp": int(dt_start.timestamp()),
446            "endDateStamp": int(dt_end.timestamp()),
447            "nn_userID": user_info["userId"],
448            "subscriptionID": subscription_info["legacySubscriptionId"],
449        }
450        async with (
451            self.mass.http_session.get(url, params=params, ssl=True, timeout=timeout) as response,
452        ):
453            response.raise_for_status()
454            content = await response.text()
455            stream = json_loads(content)
456            if not stream.get("streamLink"):
457                raise MediaNotFoundError("No stream found for song %s.", item_id)
458            return stream["streamLink"]
459
460    def _get_item_mapping(self, media_type: MediaType, key: str, name: str) -> ItemMapping:
461        return ItemMapping(
462            media_type=media_type,
463            item_id=key,
464            provider=self.instance_id,
465            name=name,
466        )
467
468    async def login(self) -> Any:
469        """Login to nugs.net and return the token."""
470        if self._auth_token and (self._token_expiry > time()):
471            return self._auth_token
472        if not self.get_setup_value(CONF_USERNAME) or not self.get_setup_value(CONF_PASSWORD):
473            msg = "Invalid login credentials"
474            raise LoginFailed(msg)
475        login_data = {
476            "username": self.get_setup_value(CONF_USERNAME),
477            "password": self.get_setup_value(CONF_PASSWORD),
478            "scope": "offline_access nugsnet:api nugsnet:legacyapi openid profile email",
479            "grant_type": "password",
480            "client_id": "Eg7HuH873H65r5rt325UytR5429",
481        }
482        token = None
483        url = "https://id.nugs.net/connect/token"
484        timeout = ClientTimeout(total=120)
485        async with (
486            self.mass.http_session.post(
487                url, data=login_data, ssl=True, timeout=timeout
488            ) as response,
489        ):
490            # Handle errors
491            if response.status == 401:
492                raise LoginFailed("Invalid Nugs.net username or password")
493            # handle temporary server error
494            if response.status in (502, 503):
495                raise ResourceTemporarilyUnavailable(backoff_time=30)
496            response.raise_for_status()
497            token = await response.json()
498            self._auth_token = token["access_token"]
499            self._token_expiry = time() + token["expires_in"]
500        return token["access_token"]
501
502    async def _get_data(self, nugs_api: str, endpoint: str, **kwargs: Any) -> Any:
503        """Return the requested data from one of various nugs.net API."""
504        headers = {}
505        url: str | None = None
506        timeout = ClientTimeout(total=120)
507        tokeninfo = kwargs.pop("tokeninfo", None)
508        if tokeninfo is None:
509            tokeninfo = await self.login()
510        headers = {"Authorization": f"Bearer {tokeninfo}"}
511        if nugs_api == "catalog":
512            url = f"https://catalog.nugs.net/api/v1/{endpoint}"
513        if nugs_api == "stash":
514            url = f"https://stash.nugs.net/api/v1/me/{endpoint}"
515        if nugs_api == "subscription":
516            url = "https://subscriptions.nugs.net/api/v1/me/subscriptions"
517        if nugs_api == "user":
518            url = "https://stash.nugs.net/api/v1/stash"
519        if not url:
520            raise MediaNotFoundError(f"{nugs_api} not found")
521        async with (
522            self.mass.http_session.get(
523                url, headers=headers, params=kwargs, ssl=True, timeout=timeout
524            ) as response,
525        ):
526            if response.status == 404:
527                raise MediaNotFoundError(f"{url} not found")
528            response.raise_for_status()
529            return await response.json()
530
531    async def _get_all_items(
532        self, nugs_api: str, endpoint: str, **kwargs: Any
533    ) -> list[dict[str, Any]]:
534        limit = 100
535        offset = 0
536        total = 0
537        all_items = []
538        while True:
539            kwargs["limit"] = limit
540            kwargs["offset"] = offset
541            result = await self._get_data(nugs_api, endpoint, **kwargs)
542            total = result["total"]
543            all_items += result["items"]
544            if total <= offset + limit:
545                break
546            offset += limit
547        return all_items
548
549    @use_cache(3600 * 4, base_class=RecommendationFolder)  # Cache for 4 hours
550    async def _get_recommendation_folder(self, item_id: str) -> RecommendationFolder | None:
551        """
552        Fetch a single recommendation row, including its items.
553
554        :param item_id: The item_id of the row (unknown ids yield None).
555        """
556        folder = next(
557            (row for row in await self.get_recommendations() if row.item_id == item_id), None
558        )
559        if folder is None:
560            return None
561        albums: list[Album]
562        if item_id == "nugs_popular_shows":
563            popular_data = await self._get_data("catalog", "releases/popular", limit=20)
564            albums = []
565            for item in popular_data["items"]:
566                response = await self._get_data("catalog", f"shows/{item['id']}")
567                albums.append(self._parse_album(response["Response"]))
568        elif item_id == "nugs_recommended_shows":
569            recommended_data = await self._get_data("catalog", "me/releases/recommendations")
570            albums = [self._parse_album(item) for item in recommended_data["items"]]
571        else:  # nugs_recent_shows
572            recent_data = await self._get_data("catalog", "releases/recent", limit=50)
573            albums = [self._parse_album(item) for item in recent_data["items"]]
574        folder.items = UniqueList(albums)
575        return folder
576