/
/
/
1"""Bandcamp music provider support for MusicAssistant."""
2
3import asyncio
4from collections.abc import AsyncGenerator
5from contextlib import suppress
6from typing import cast
7
8from bandcamp_async_api import (
9 BandcampAPIClient,
10 BandcampAPIError,
11 BandcampMustBeLoggedInError,
12 BandcampNotFoundError,
13 SearchResultAlbum,
14 SearchResultArtist,
15 SearchResultTrack,
16)
17from bandcamp_async_api.models import CollectionType
18from music_assistant_models.config_entries import ConfigEntry, ConfigValueType, ProviderConfig
19from music_assistant_models.enums import (
20 ConfigEntryType,
21 MediaType,
22 ProviderFeature,
23 StreamType,
24)
25from music_assistant_models.errors import (
26 InvalidDataError,
27 LoginFailed,
28 MediaNotFoundError,
29)
30from music_assistant_models.media_items import (
31 Album,
32 Artist,
33 AudioFormat,
34 SearchResults,
35 Track,
36)
37from music_assistant_models.provider import ProviderManifest
38from music_assistant_models.streamdetails import StreamDetails
39
40from music_assistant.controllers.cache import use_cache
41from music_assistant.helpers.throttle_retry import ThrottlerManager, throttle_with_retries
42from music_assistant.mass import MusicAssistant
43from music_assistant.models import ProviderInstanceType
44from music_assistant.models.music_provider import MusicProvider
45
46from .converters import BandcampConverters
47
48SUPPORTED_FEATURES = {
49 ProviderFeature.LIBRARY_ARTISTS,
50 ProviderFeature.LIBRARY_ALBUMS,
51 ProviderFeature.LIBRARY_TRACKS,
52 ProviderFeature.SEARCH,
53 ProviderFeature.ARTIST_ALBUMS,
54 ProviderFeature.ARTIST_TOPTRACKS,
55}
56
57CONF_IDENTITY = "identity"
58CONF_TOP_TRACKS_LIMIT = "top_tracks_limit"
59DEFAULT_TOP_TRACKS_LIMIT = 50
60CACHE = 3600 * 24 * 30 # Cache for 30 days
61
62
63async def setup(
64 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
65) -> ProviderInstanceType:
66 """Initialize provider(instance) with given configuration."""
67 return BandcampProvider(mass, manifest, config, SUPPORTED_FEATURES)
68
69
70# noinspection PyTypeHints,PyUnusedLocal
71async def get_config_entries(
72 mass: MusicAssistant, # noqa: ARG001
73 instance_id: str | None = None, # noqa: ARG001
74 action: str | None = None, # noqa: ARG001
75 values: dict[str, ConfigValueType] | None = None,
76) -> tuple[ConfigEntry, ...]:
77 """Return Config entries to setup this provider."""
78 return (
79 ConfigEntry(
80 key=CONF_IDENTITY,
81 type=ConfigEntryType.SECURE_STRING,
82 label="Identity token",
83 required=False,
84 description="Identity token from Bandcamp cookies for account collection access."
85 " Log in https://bandcamp.com and extract browser cookie named 'identity'.",
86 value=values.get(CONF_IDENTITY) if values else None,
87 ),
88 ConfigEntry(
89 key=CONF_TOP_TRACKS_LIMIT,
90 type=ConfigEntryType.INTEGER,
91 label="Artist Top Tracks search limit",
92 required=False,
93 description="Search limit while getting artist top tracks.",
94 value=values.get(CONF_TOP_TRACKS_LIMIT) if values else DEFAULT_TOP_TRACKS_LIMIT,
95 default_value=DEFAULT_TOP_TRACKS_LIMIT,
96 category="advanced",
97 ),
98 )
99
100
101def split_id(id_: str) -> tuple[int, int | None, int | None]:
102 """Return (artist_id, album_id, track_id). Missing parts are returned as 0."""
103 parts = id_.split("-")
104 part_0 = int(parts[0])
105 part_1 = int(parts[1]) if len(parts) > 1 else 0
106 part_2 = int(parts[2]) if len(parts) > 2 else 0
107 return part_0, part_1, part_2
108
109
110class BandcampProvider(MusicProvider):
111 """Bandcamp provider support."""
112
113 _client: BandcampAPIClient
114 _converters: BandcampConverters
115 throttler: ThrottlerManager
116 top_tracks_limit: int
117
118 async def handle_async_init(self) -> None:
119 """Handle async init of the Bandcamp provider."""
120 identity = self.config.get_value(CONF_IDENTITY)
121 self.top_tracks_limit = cast(
122 "int", self.config.get_value(CONF_TOP_TRACKS_LIMIT, DEFAULT_TOP_TRACKS_LIMIT)
123 )
124
125 # Initialize the new async API client
126 self._client = BandcampAPIClient(session=self.mass.http_session, identity_token=identity)
127
128 self.throttler = ThrottlerManager(rate_limit=1, period=2)
129 self._converters = BandcampConverters(self.domain, self.instance_id)
130
131 @property
132 def is_streaming_provider(self) -> bool:
133 """Return True if the provider is a streaming provider."""
134 return True
135
136 @use_cache(CACHE)
137 @throttle_with_retries
138 async def search(
139 self, search_query: str, media_types: list[MediaType], limit: int = 50
140 ) -> SearchResults:
141 """Perform search on music provider."""
142 results = SearchResults()
143 if not self._client.identity:
144 return results
145
146 if not media_types:
147 return results
148
149 try:
150 search_results = await self._client.search(search_query)
151 except BandcampNotFoundError as error:
152 raise MediaNotFoundError("No results for Bandcamp search") from error
153 except BandcampAPIError as error:
154 raise InvalidDataError("Unexpected error during Bandcamp search") from error
155
156 for item in search_results[:limit]:
157 try:
158 if isinstance(item, SearchResultTrack) and MediaType.TRACK in media_types:
159 results.tracks = [*results.tracks, self._converters.track_from_search(item)]
160 elif isinstance(item, SearchResultAlbum) and MediaType.ALBUM in media_types:
161 results.albums = [*results.albums, self._converters.album_from_search(item)]
162 elif isinstance(item, SearchResultArtist) and MediaType.ARTIST in media_types:
163 results.artists = [*results.artists, self._converters.artist_from_search(item)]
164 except BandcampAPIError as error:
165 self.logger.warning("Failed to convert search result item: %s", error)
166 continue
167
168 return results
169
170 async def get_library_artists(self) -> AsyncGenerator[Artist, None]:
171 """Retrieve library artists from Bandcamp."""
172 if not self._client.identity: # library requires identity
173 return
174
175 try:
176 collection = await self._client.get_collection_items(CollectionType.COLLECTION)
177 band_ids = set()
178 for item in collection.items:
179 if item.item_type == "band":
180 band_ids.add(item.item_id)
181 elif item.item_type == "album":
182 band_ids.add(item.band_id)
183
184 for band_id in band_ids:
185 yield await self.get_artist(band_id)
186 await asyncio.sleep(0) # Yield control to avoid blocking
187
188 except BandcampMustBeLoggedInError as error:
189 self.logger.error("Error getting Bandcamp library artists: Wrong identity token.")
190 raise LoginFailed("Wrong Bandcamp identity token.") from error
191 except BandcampNotFoundError as error:
192 raise MediaNotFoundError("Bandcamp library artists returned no results") from error
193 except BandcampAPIError as error:
194 raise MediaNotFoundError("Failed to get library artists") from error
195
196 async def get_library_albums(self) -> AsyncGenerator[Album, None]:
197 """Retrieve library albums from Bandcamp."""
198 if not self._client.identity: # library requires identity
199 return
200
201 try:
202 api_collection = await self._client.get_collection_items(CollectionType.COLLECTION)
203 for item in api_collection.items:
204 if item.item_type == "album":
205 yield await self.get_album(f"{item.band_id}-{item.item_id}")
206 await asyncio.sleep(0) # Yield control to avoid blocking
207 except BandcampMustBeLoggedInError as error:
208 self.logger.error("Error getting Bandcamp library albums: Wrong identity token.")
209 raise LoginFailed("Wrong Bandcamp identity token.") from error
210 except BandcampNotFoundError as error:
211 raise MediaNotFoundError("Bandcamp library albums returned no results") from error
212 except BandcampAPIError as error:
213 raise MediaNotFoundError("Failed to get library albums") from error
214
215 async def get_library_tracks(self) -> AsyncGenerator[Track, None]:
216 """Retrieve library tracks from Bandcamp."""
217 if not self._client.identity: # library requires identity
218 return
219
220 try:
221 async for album in self.get_library_albums():
222 tracks = await self.get_album_tracks(album.item_id)
223 for track in tracks:
224 yield track
225 await asyncio.sleep(0) # Yield control to avoid blocking
226 except BandcampMustBeLoggedInError as error:
227 self.logger.error("Error getting Bandcamp library tracks: Wrong identity token.")
228 raise LoginFailed("Wrong Bandcamp identity token.") from error
229 except BandcampNotFoundError as error:
230 raise MediaNotFoundError("Bandcamp library tracks returned no results") from error
231 except BandcampAPIError as error:
232 raise MediaNotFoundError("Failed to get library tracks") from error
233
234 @use_cache(CACHE)
235 async def get_artist(self, prov_artist_id: str | int) -> Artist:
236 """Get full artist details by id."""
237 try:
238 api_artist = await self._client.get_artist(prov_artist_id)
239 return self._converters.artist_from_api(api_artist)
240 except BandcampNotFoundError as error:
241 raise MediaNotFoundError(
242 f"Bandcamp artist {prov_artist_id} search returned no results"
243 ) from error
244 except BandcampAPIError as error:
245 raise MediaNotFoundError(f"Failed to get artist {prov_artist_id}") from error
246
247 @use_cache(CACHE)
248 async def get_album(self, prov_album_id: str) -> Album:
249 """Get full album details by id."""
250 artist_id, album_id, _ = split_id(prov_album_id)
251 try:
252 api_album = await self._client.get_album(artist_id, album_id)
253 return self._converters.album_from_api(api_album)
254 except BandcampNotFoundError as error:
255 raise MediaNotFoundError(
256 f"Bandcamp album {prov_album_id} search returned no results"
257 ) from error
258 except BandcampAPIError as error:
259 raise MediaNotFoundError(f"Failed to get album {prov_album_id}") from error
260
261 @use_cache(CACHE)
262 async def get_track(self, prov_track_id: str) -> Track:
263 """Get full track details by id."""
264 artist_id, album_id, track_id = split_id(prov_track_id)
265 if track_id is None: # artist_id-track_id
266 album_id, track_id = None, album_id
267
268 try:
269 if all((artist_id, album_id, track_id)):
270 api_album = await self._client.get_album(artist_id, album_id)
271 api_track = next((_ for _ in api_album.tracks if _.id == track_id), None)
272 return self._converters.track_from_api(
273 track=api_track,
274 album_id=api_album.id,
275 album_name=api_album.title,
276 album_image_url=api_album.art_url,
277 )
278 if not album_id:
279 api_track = await self._client.get_track(artist_id, track_id)
280 return self._converters.track_from_api(
281 track=api_track,
282 album_id=api_track.album.id if api_track.album else None,
283 album_name=api_track.album.title if api_track.album else "",
284 album_image_url=api_track.album.art_url if api_track.album else "",
285 )
286 raise MediaNotFoundError(f"Track {prov_track_id} not found on Bandcamp")
287 except BandcampNotFoundError as error:
288 raise MediaNotFoundError(
289 f"Bandcamp track {prov_track_id} search returned no results"
290 ) from error
291 except BandcampAPIError as error:
292 raise MediaNotFoundError(f"Failed to get track {prov_track_id}") from error
293
294 @use_cache(CACHE)
295 async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
296 """Get all tracks in an album."""
297 artist_id, album_id, _ = split_id(prov_album_id)
298 try:
299 api_album = await self._client.get_album(artist_id, album_id)
300 if api_album.tracks:
301 return [
302 self._converters.track_from_api(
303 track=track,
304 album_id=album_id,
305 album_name=api_album.title,
306 album_image_url=api_album.art_url,
307 )
308 for track in api_album.tracks
309 if track.streaming_url # Only include tracks with streaming URLs
310 ]
311
312 return []
313
314 except BandcampNotFoundError as error:
315 raise MediaNotFoundError(
316 f"Bandcamp album {prov_album_id} tracks search returned no results"
317 ) from error
318 except BandcampAPIError as error:
319 raise MediaNotFoundError(f"Failed to get albums tracks for {prov_album_id}") from error
320
321 @use_cache(CACHE)
322 async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
323 """Get albums by an artist."""
324 albums = []
325 try:
326 api_discography = await self._client.get_artist_discography(prov_artist_id)
327 for item in api_discography:
328 if item.get("item_type") == "album" and item.get("item_id"):
329 album = None
330
331 with suppress(MediaNotFoundError):
332 album = await self.get_album(f"{item['band_id']}-{item['item_id']}")
333
334 with suppress(MediaNotFoundError):
335 album = album or await self.get_album(f"{prov_artist_id}-{item['item_id']}")
336
337 if album:
338 albums.append(album)
339
340 except BandcampNotFoundError as error:
341 raise MediaNotFoundError(
342 f"Bandcamp artist {prov_artist_id} albums search returned no results"
343 ) from error
344 except BandcampAPIError as error:
345 raise MediaNotFoundError(f"Failed to get albums for artist {prov_artist_id}") from error
346
347 return albums
348
349 @use_cache(CACHE)
350 async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
351 """Get top tracks of an artist."""
352 tracks: list[Track] = []
353 try:
354 albums = await self.get_artist_albums(prov_artist_id)
355 albums.sort(key=lambda album: (album.year is None, album.year or 0), reverse=True)
356 for album in albums:
357 tracks.extend(await self.get_album_tracks(album.item_id))
358 if len(tracks) >= self.top_tracks_limit:
359 break
360
361 except BandcampNotFoundError as error:
362 raise MediaNotFoundError(
363 f"Bandcamp artist {prov_artist_id} top tracks search returned no results"
364 ) from error
365
366 except BandcampAPIError as error:
367 raise MediaNotFoundError(
368 f"Failed to get toptracks for artist {prov_artist_id}"
369 ) from error
370
371 return tracks[: self.top_tracks_limit]
372
373 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
374 """Return the content details for the given track."""
375 try:
376 # consider _client to avoid caching if the track urls become dynamic
377 track_ma = await self.get_track(item_id)
378 if not track_ma.metadata.links:
379 raise MediaNotFoundError(
380 f"No streaming links found for track {item_id}. Please report this"
381 )
382
383 link = next(iter(track_ma.metadata.links))
384 if not link:
385 raise MediaNotFoundError(
386 f"No streaming URL found for track {item_id}. Please report this"
387 )
388
389 streaming_url = link.url
390 if not streaming_url:
391 raise MediaNotFoundError(
392 f"No streaming URL found for track {item_id}: {streaming_url}"
393 )
394
395 return StreamDetails(
396 item_id=item_id,
397 provider=self.instance_id,
398 audio_format=AudioFormat(),
399 stream_type=StreamType.HTTP,
400 media_type=media_type,
401 path=streaming_url,
402 can_seek=True,
403 allow_seek=True,
404 )
405
406 except BandcampNotFoundError as error:
407 raise MediaNotFoundError(
408 f"Bandcamp stream details search for {media_type} {item_id} returned no results"
409 ) from error
410 except BandcampAPIError as error:
411 raise MediaNotFoundError(
412 f"Stream details not available for {media_type} {item_id}"
413 ) from error
414