/
/
/
1"""iBroadcast support for MusicAssistant."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any
6from urllib.parse import urlparse, urlunparse
7
8from ibroadcastaio import IBroadcastClient
9from music_assistant_models.enums import (
10 ContentType,
11 ImageType,
12 MediaType,
13 ProviderFeature,
14 StreamType,
15)
16from music_assistant_models.errors import InvalidDataError, LoginFailed, MediaNotFoundError
17from music_assistant_models.media_items import (
18 Album,
19 Artist,
20 AudioFormat,
21 ItemMapping,
22 MediaItemImage,
23 Playlist,
24 ProviderMapping,
25 Track,
26 UniqueList,
27)
28from music_assistant_models.streamdetails import StreamDetails
29
30from music_assistant.constants import (
31 CONF_PASSWORD,
32 CONF_USERNAME,
33 UNKNOWN_ARTIST,
34 VARIOUS_ARTISTS_MBID,
35 VARIOUS_ARTISTS_NAME,
36)
37from music_assistant.controllers.cache import use_cache
38from music_assistant.helpers.util import infer_album_type, parse_title_and_version
39from music_assistant.models.music_provider import (
40 DEFAULT_MAX_CONCURRENT_STREAMS,
41 MusicProvider,
42)
43
44SUPPORTED_FEATURES = {
45 ProviderFeature.LIBRARY_ARTISTS,
46 ProviderFeature.LIBRARY_TRACKS,
47 ProviderFeature.LIBRARY_ALBUMS,
48 ProviderFeature.LIBRARY_PLAYLISTS,
49 ProviderFeature.BROWSE,
50 ProviderFeature.ARTIST_ALBUMS,
51}
52
53
54if TYPE_CHECKING:
55 from collections.abc import AsyncGenerator
56
57 from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
58 from music_assistant_models.provider import ProviderManifest
59
60 from music_assistant.mass import MusicAssistant
61 from music_assistant.models import ProviderInstanceType
62
63
64async def setup(
65 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
66) -> ProviderInstanceType:
67 """Initialize provider(instance) with given configuration."""
68 return IBroadcastProvider(mass, manifest, config, SUPPORTED_FEATURES)
69
70
71class IBroadcastProvider(MusicProvider):
72 """Provider for iBroadcast."""
73
74 _user_id: str
75 _client: IBroadcastClient
76
77 @property
78 def is_streaming_provider(self) -> bool:
79 """Return False: the catalog is the account's own uploaded collection."""
80 return False
81
82 @property
83 def max_concurrent_streams(self) -> int:
84 """Keep the conservative streaming default, as this is a hosted service."""
85 return DEFAULT_MAX_CONCURRENT_STREAMS
86
87 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
88 """Return Config entries to setup this provider."""
89 return ()
90
91 async def handle_async_init(self) -> None:
92 """Set up the iBroadcast provider."""
93 username = self.get_setup_value(CONF_USERNAME)
94 password = self.get_setup_value(CONF_PASSWORD)
95 if not username or not password:
96 msg = "Invalid login credentials"
97 raise LoginFailed(msg)
98 self._client = IBroadcastClient(self.mass.http_session)
99 status = await self._client.login(username, password)
100 self._user_id = status["user"]["id"]
101
102 # temporary call to refresh library until ibroadcast provides a detailed api
103 await self._client.refresh_library()
104
105 async def get_library_albums(self) -> AsyncGenerator[Album]:
106 """Retrieve library albums from ibroadcast."""
107 for album in (await self._client.get_albums()).values():
108 try:
109 yield await self._parse_album(album)
110 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
111 self._report_skipped_item(MediaType.ALBUM, album, "album_id", error)
112 continue
113
114 @use_cache(3600 * 24 * 7) # Cache for 7 days
115 async def get_album(self, prov_album_id: str) -> Album:
116 """Get full album details by id."""
117 album_obj = await self._client.get_album(int(prov_album_id))
118 return await self._parse_album(album_obj)
119
120 async def get_library_artists(self) -> AsyncGenerator[Artist]:
121 """Retrieve all library artists from iBroadcast."""
122 for artist in (await self._client.get_artists()).values():
123 try:
124 yield await self._parse_artist(artist)
125 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
126 self._report_skipped_item(MediaType.ARTIST, artist, "artist_id", error)
127 continue
128
129 @use_cache(3600 * 24 * 7, allow_expired_cache=True) # Cache for 7 days
130 async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
131 """Get a list of albums for the given artist."""
132 albums_objs = [
133 album
134 for album in (await self._client.get_albums()).values()
135 if album["artist_id"] == int(prov_artist_id)
136 ]
137 albums = []
138 for album in albums_objs:
139 try:
140 albums.append(await self._parse_album(album))
141 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
142 self.logger.debug("Parse album failed: %s", album, exc_info=error)
143 continue
144 return albums
145
146 @use_cache(3600 * 24 * 7, allow_expired_cache=True) # Cache for 7 days
147 async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
148 """Get album tracks for given album id."""
149 album = await self._client.get_album(int(prov_album_id))
150 return await self._get_tracks(album["tracks"])
151
152 @use_cache(3600 * 24 * 7) # Cache for 7 days
153 async def get_track(self, prov_track_id: str) -> Track:
154 """Get full track details by id."""
155 track_obj = await self._client.get_track(int(prov_track_id))
156 return await self._parse_track(track_obj)
157
158 @use_cache(3600 * 24 * 7) # Cache for 7 days
159 async def get_artist(self, prov_artist_id: str) -> Artist:
160 """Get full artist details by id."""
161 artist_obj = await self._client.get_artist(int(prov_artist_id))
162 return await self._parse_artist(artist_obj)
163
164 async def get_library_tracks(self) -> AsyncGenerator[Track]:
165 """Retrieve library tracks from iBroadcast."""
166 for track in (await self._client.get_tracks()).values():
167 try:
168 yield await self._parse_track(track)
169 except IndexError as error:
170 self._report_skipped_item(MediaType.TRACK, track, "track_id", error)
171 continue
172 except (KeyError, TypeError, InvalidDataError) as error:
173 self._report_skipped_item(MediaType.TRACK, track, "track_id", error)
174 continue
175
176 def _get_artist_item_mapping(self, artist_id: str, artist_obj: dict[str, Any]) -> ItemMapping:
177 if (not artist_id and artist_obj["name"] == "Various Artists") or artist_id == "0":
178 artist_id = VARIOUS_ARTISTS_MBID
179 return self._get_item_mapping(MediaType.ARTIST, artist_id, str(artist_obj.get("name")))
180
181 def _get_item_mapping(self, media_type: MediaType, key: str, name: str) -> ItemMapping:
182 return ItemMapping(
183 media_type=media_type,
184 item_id=key,
185 provider=self.instance_id,
186 name=name,
187 )
188
189 async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
190 """Retrieve playlists from iBroadcast."""
191 for playlist in (await self._client.get_playlists()).values():
192 # Skip the auto generated playlist
193 if playlist["type"] != "recently-played" and playlist["type"] != "thumbsup":
194 yield await self._parse_playlist(playlist)
195
196 @use_cache(3600 * 24 * 7) # Cache for 7 days
197 async def get_playlist(self, prov_playlist_id: str) -> Playlist:
198 """Get full playlist details by id."""
199 playlist_obj = await self._client.get_playlist(int(prov_playlist_id))
200 try:
201 playlist = await self._parse_playlist(playlist_obj)
202 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
203 self.logger.debug("Parse playlist failed: %s", playlist_obj, exc_info=error)
204 raise MediaNotFoundError(f"Playlist {prov_playlist_id} could not be parsed") from error
205 return playlist
206
207 @use_cache(3600, allow_expired_cache=True) # Cache for 1 hour
208 async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
209 """Get playlist tracks."""
210 tracks: list[Track] = []
211 if page > 0:
212 return tracks
213 playlist_obj = await self._client.get_playlist(int(prov_playlist_id))
214 if "tracks" not in playlist_obj:
215 return tracks
216 return await self._get_tracks(playlist_obj["tracks"], True)
217
218 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
219 """Return the content details for the given track when it will be streamed."""
220 # How to buildup a stream url:
221 # [streaming_server]/[url]?Expires=[now]&Signature=[user token]&file_id=[file ID]
222 # &user_id=[user ID]&platform=[your app name]&version=[your app version]
223 # See https://devguide.ibroadcast.com/?p=streaming-server
224 url = await self._client.get_full_stream_url(int(item_id), "music-assistant")
225
226 # Replace the bitrate path segment with "orig" to request the original upload format.
227 # By default the bitrate is always 128kbps.
228 parsed = urlparse(url)
229 path_parts = parsed.path.split("/")
230 if len(path_parts) > 1 and path_parts[1].isdigit():
231 path_parts[1] = "orig"
232 url = urlunparse(parsed._replace(path="/".join(path_parts)))
233
234 return StreamDetails(
235 provider=self.instance_id,
236 item_id=item_id,
237 audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
238 stream_type=StreamType.HTTP,
239 path=url,
240 can_seek=True,
241 allow_seek=True,
242 )
243
244 async def _get_tracks(self, track_ids: list[int], is_playlist: bool = False) -> list[Track]:
245 """Retrieve a list of tracks based on provided track IDs."""
246 tracks = []
247 for index, track_id in enumerate(track_ids, 1):
248 track_obj = await self._client.get_track(track_id)
249 if track_obj is not None:
250 track = await self._parse_track(track_obj)
251 if is_playlist:
252 track.position = index
253 tracks.append(track)
254 return tracks
255
256 async def _parse_artist(self, artist_obj: dict[str, Any]) -> Artist:
257 """Parse a iBroadcast user response to Artist model object."""
258 artist_id = artist_obj["artist_id"]
259 artist = Artist(
260 item_id=artist_id,
261 name=artist_obj["name"],
262 provider=self.instance_id,
263 provider_mappings={
264 ProviderMapping(
265 item_id=artist_id,
266 provider_domain=self.domain,
267 provider_instance=self.instance_id,
268 url=f"https://media.ibroadcast.com/?view=container&container_id={artist_id}&type=artists",
269 )
270 },
271 )
272 # Artwork
273 if "artwork_id" in artist_obj:
274 artist.metadata.images = UniqueList(
275 [
276 MediaItemImage(
277 type=ImageType.THUMB,
278 path=await self._client.get_artist_artwork_url(artist_id),
279 provider=self.instance_id,
280 remotely_accessible=True,
281 )
282 ]
283 )
284 return artist
285
286 async def _parse_album(self, album_obj: dict[str, Any]) -> Album:
287 """Parse ibroadcast album object to generic layout."""
288 album_id = album_obj["album_id"]
289 name, version = parse_title_and_version(album_obj["name"])
290 album = Album(
291 item_id=album_id,
292 provider=self.instance_id,
293 name=name,
294 year=album_obj["year"],
295 version=version,
296 provider_mappings={
297 ProviderMapping(
298 item_id=album_id,
299 provider_domain=self.domain,
300 provider_instance=self.instance_id,
301 audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
302 url=f"https://media.ibroadcast.com/?view=container&container_id={album_id}&type=albums",
303 )
304 },
305 )
306 if album_obj["artist_id"] == 0:
307 artist = Artist(
308 item_id=VARIOUS_ARTISTS_MBID,
309 name=VARIOUS_ARTISTS_NAME,
310 provider=self.instance_id,
311 provider_mappings={
312 ProviderMapping(
313 item_id=VARIOUS_ARTISTS_MBID,
314 provider_domain=self.domain,
315 provider_instance=self.instance_id,
316 )
317 },
318 )
319 album.artists.append(artist)
320 else:
321 artist_mapping = self._get_item_mapping(
322 MediaType.ARTIST,
323 album_obj["artist_id"],
324 (await self._client.get_artist(album_obj["artist_id"]))["name"]
325 if await self._client.get_artist(album_obj["artist_id"])
326 else UNKNOWN_ARTIST,
327 )
328 album.artists.append(artist_mapping)
329
330 if "rating" in album_obj and album_obj["rating"] == 5:
331 album.favorite = True
332 # iBroadcast doesn't seem to know album type - try inference
333 album.album_type = infer_album_type(name, version)
334
335 # There is only an artwork in the tracks, lets get the first track one
336 artwork_url = await self._client.get_album_artwork_url(album_id)
337 if artwork_url:
338 album.metadata.images = UniqueList([self._get_artwork_object(artwork_url)])
339 return album
340
341 def _get_artwork_object(self, url: str) -> MediaItemImage:
342 return MediaItemImage(
343 type=ImageType.THUMB,
344 path=url,
345 provider=self.instance_id,
346 remotely_accessible=True,
347 )
348
349 async def _parse_track(self, track_obj: dict[str, Any]) -> Track:
350 """Parse an iBroadcast track object to a Track model object."""
351 track = Track(
352 item_id=track_obj["track_id"],
353 provider=self.instance_id,
354 name=track_obj["title"],
355 provider_mappings={
356 ProviderMapping(
357 item_id=track_obj["track_id"],
358 provider_domain=self.domain,
359 provider_instance=self.instance_id,
360 available=not track_obj["trashed"],
361 audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
362 )
363 },
364 )
365 if track_obj["album_id"]:
366 album = await self._client.get_album(track_obj["album_id"])
367
368 if "rating" in track_obj and track_obj["rating"] == 5:
369 track.favorite = True
370 if "length" in track_obj and str(track_obj["length"]).isdigit():
371 track.duration = track_obj["length"]
372 # use the disc number if available
373 if album and album["disc"] > 0:
374 track.disc_number = album["disc"]
375 track.track_number = int(track_obj["track"])
376 # otherwise, track number might look like 201, meaning, disc 2, track 1
377 elif track_obj["track"] > 99:
378 track.disc_number = int(str(track_obj["track"])[:1])
379 track.track_number = int(str(track_obj["track"])[1:])
380 # or just the track number and no disc number
381 else:
382 track.track_number = int(track_obj["track"])
383 # Track artists
384 if "artist_id" in track_obj:
385 artist_id = track_obj["artist_id"]
386 track.artists = UniqueList(
387 [self._get_artist_item_mapping(artist_id, await self._client.get_artist(artist_id))]
388 )
389 # additional artists structure: 'artists_additional': [[artist id, phrase, type]]
390 track.artists.extend(
391 [
392 self._get_artist_item_mapping(
393 additional_artist[0],
394 await self._client.get_artist(additional_artist[0]),
395 )
396 for additional_artist in track_obj["artists_additional"]
397 if additional_artist[0]
398 ]
399 )
400 # guard that track has valid artists
401 if not track.artists:
402 msg = "Track is missing artists"
403 raise InvalidDataError(msg)
404
405 # Artwork
406 track.metadata.images = UniqueList(
407 [
408 self._get_artwork_object(
409 await self._client.get_track_artwork_url(track_obj["track_id"])
410 )
411 ]
412 )
413 # Genre
414 genres: set[str] = set()
415 if track_obj["genre"]:
416 genres.add(track_obj["genre"])
417 if track_obj["genres_additional"]:
418 genres.add(track_obj["genres_additional"])
419 track.metadata.genres = genres
420 # album info
421 if album:
422 track.album = self._get_item_mapping(
423 MediaType.ALBUM, track_obj["album_id"], album["name"]
424 )
425 return track
426
427 async def _parse_playlist(self, playlist_obj: dict[str, Any]) -> Playlist:
428 """Parse an iBroadcast Playlist response to a Playlist object."""
429 playlist_id = str(playlist_obj["playlist_id"])
430 playlist = Playlist(
431 item_id=playlist_id,
432 provider=self.instance_id,
433 name=playlist_obj["name"],
434 provider_mappings={
435 ProviderMapping(
436 item_id=playlist_id,
437 provider_domain=self.domain,
438 provider_instance=self.instance_id,
439 )
440 },
441 )
442 # Can be supported in future, the API has options available
443 playlist.is_editable = False
444 playlist.metadata.images = UniqueList(
445 [
446 self._get_artwork_object(
447 await self._client.get_playlist_artwork_url(int(playlist_id))
448 )
449 ]
450 )
451 if "description" in playlist_obj:
452 playlist.metadata.description = playlist_obj["description"]
453 return playlist
454
455 def _report_skipped_item(
456 self, media_type: MediaType, item_obj: dict[str, Any], id_key: str, err: Exception
457 ) -> None:
458 """
459 Report a library item that was dropped while listing the library.
460
461 :param media_type: Media type of the skipped item.
462 :param item_obj: Raw api object of the skipped item.
463 :param id_key: Key under which the raw object holds the item id, which iBroadcast
464 returns as a number while the library stores it as text.
465 :param err: The error that made the item unusable.
466 """
467 item_id = item_obj.get(id_key)
468 self.report_skipped_sync_item(
469 media_type, str(item_id) if item_id is not None else None, err
470 )
471