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