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