/
/
/
1"""Bandcamp music provider support for MusicAssistant."""
2
3import asyncio
4from collections.abc import AsyncGenerator, AsyncIterator, Sequence
5from contextlib import asynccontextmanager, suppress
6from typing import TYPE_CHECKING, Any, cast
7
8from bandcamp_async_api import (
9 BandcampAPIClient,
10 BandcampAPIError,
11 BandcampMustBeLoggedInError,
12 BandcampNotFoundError,
13 BandcampRateLimitError,
14 SearchResultAlbum,
15 SearchResultArtist,
16 SearchResultItem,
17 SearchResultTrack,
18)
19from bandcamp_async_api.models import (
20 BCAlbum,
21 BCTrack,
22 CollectionItem,
23 CollectionSummary,
24 CollectionType,
25 FanItem,
26 FeedResponse,
27 FollowingItem,
28)
29from mashumaro.exceptions import UnserializableDataError
30from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
31from music_assistant_models.enums import (
32 ConfigEntryType,
33 ImageType,
34 MediaType,
35 StreamType,
36)
37from music_assistant_models.errors import (
38 InvalidDataError,
39 LoginFailed,
40 MediaNotFoundError,
41 RateLimited,
42 ResourceTemporarilyUnavailable,
43 RetriesExhausted,
44)
45from music_assistant_models.media_items import (
46 Album,
47 Artist,
48 AudioFormat,
49 BrowseFolder,
50 ItemMapping,
51 MediaItemImage,
52 MediaItemType,
53 RecommendationFolder,
54 SearchResults,
55 Track,
56 UniqueList,
57)
58from music_assistant_models.streamdetails import StreamDetails
59
60from music_assistant.constants import CONF_ENTRY_UNOFFICIAL_PROVIDER
61from music_assistant.controllers.cache import use_cache
62from music_assistant.helpers.throttle_retry import ThrottlerManager, throttle_with_retries
63from music_assistant.mass import MusicAssistant
64from music_assistant.models import ProviderInstanceType
65from music_assistant.models.music_provider import MusicProvider
66
67from ._ids import make_artist_id, parse_artist_id, slugify_performer
68from .constants import (
69 BROWSE_FANS,
70 BROWSE_FEED,
71 BROWSE_FOLLOWERS,
72 BROWSE_FOLLOWING,
73 BROWSE_WISHLIST,
74 CACHE_EMPTY_RESULTS,
75 CACHE_METADATA,
76 CACHE_USER_LISTS,
77 CONF_IDENTITY,
78 CONF_TOP_TRACKS_LIMIT,
79 DEFAULT_TOP_TRACKS_LIMIT,
80 PERSON_SUB_FOLDERS,
81 PERSON_SUB_ROUTES,
82 SUPPORTED_FEATURES,
83)
84from .converters import BandcampConverters, DiscographyItem
85
86if TYPE_CHECKING:
87 from music_assistant_models.provider import ProviderManifest
88
89
90async def setup(
91 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
92) -> ProviderInstanceType:
93 """Initialize provider(instance) with given configuration."""
94 return BandcampProvider(mass, manifest, config, SUPPORTED_FEATURES)
95
96
97def split_id(id_: str) -> tuple[int, int, int]:
98 """
99 Return (artist_id, album_id, track_id). Missing parts are returned as 0.
100
101 :param id_: Compound ID string, e.g. "123-456-789".
102 :raises InvalidDataError: If the ID contains non-numeric parts.
103 """
104 try:
105 parts = id_.split("-")
106 part_0 = int(parts[0])
107 part_1 = int(parts[1]) if len(parts) > 1 else 0
108 part_2 = int(parts[2]) if len(parts) > 2 else 0
109 except (ValueError, IndexError) as error:
110 raise InvalidDataError(f"Malformed Bandcamp ID: {id_}") from error
111 return part_0, part_1, part_2
112
113
114class BandcampProvider(MusicProvider):
115 """Bandcamp provider support."""
116
117 _client: BandcampAPIClient
118 _converters: BandcampConverters
119 _slug_to_fan_id: dict[str, int] # unbounded; eviction would break back-navigation
120 throttler: ThrottlerManager = ThrottlerManager(
121 rate_limit=50, # requests per period seconds
122 period=10,
123 initial_backoff=3, # Bandcamp responds with Retry-After 3
124 retry_attempts=5,
125 )
126 top_tracks_limit: int
127
128 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
129 """Return Config entries to configure this provider."""
130 return (
131 CONF_ENTRY_UNOFFICIAL_PROVIDER,
132 ConfigEntry(
133 key=CONF_TOP_TRACKS_LIMIT,
134 type=ConfigEntryType.INTEGER,
135 required=False,
136 default_value=DEFAULT_TOP_TRACKS_LIMIT,
137 advanced=True,
138 ),
139 )
140
141 async def handle_async_init(self) -> None:
142 """Handle async init of the Bandcamp provider."""
143 identity = self.get_setup_value(CONF_IDENTITY)
144 self.top_tracks_limit = cast(
145 "int", self.config.get_value(CONF_TOP_TRACKS_LIMIT, DEFAULT_TOP_TRACKS_LIMIT)
146 )
147 self._client = BandcampAPIClient(
148 session=self.mass.http_session,
149 identity_token=identity,
150 default_retry_after=3, # Bandcamp responds with Retry-After 3
151 )
152 self._converters = BandcampConverters(self.domain, self.instance_id)
153 self._slug_to_fan_id = {}
154
155 # The provider can function without login (search and streaming),
156 # but if credentials were explicitly configured, validate them now.
157 # A bad login fails hard so the user can fix it immediately;
158 # transient errors (rate limits, network) are logged and the provider
159 # continues since the login may still be valid.
160 if identity:
161 try:
162 await self._client.get_collection_summary()
163 except BandcampMustBeLoggedInError as error:
164 raise LoginFailed("Bandcamp login is invalid or expired.") from error
165 except BandcampAPIError as error:
166 self.logger.warning("Could not validate Bandcamp login: %s", error)
167
168 @property
169 def is_streaming_provider(self) -> bool:
170 """Return True if the provider is a streaming provider."""
171 return True
172
173 @throttle_with_retries
174 async def search(
175 self, search_query: str, media_types: list[MediaType], limit: int = 50
176 ) -> SearchResults:
177 """
178 Search Bandcamp for matching media.
179
180 :param search_query: Text to search for.
181 :param media_types: Media types to include in the results.
182 :param limit: Maximum number of results to return.
183 :returns: Matching Bandcamp media.
184 """
185 results = SearchResults()
186 if not media_types:
187 return results
188
189 try:
190 search_results = await self._client.search(search_query)
191 except BandcampNotFoundError as error:
192 raise MediaNotFoundError("No results for Bandcamp search") from error
193 except BandcampRateLimitError as error:
194 raise RateLimited(
195 "Bandcamp rate limit reached", backoff_time=error.retry_after
196 ) from error
197 except BandcampAPIError as error:
198 raise InvalidDataError("Unexpected error during Bandcamp search") from error
199
200 capped = search_results[:limit]
201 # Map band_id -> SearchResultArtist for cross-result dedup. When an
202 # album/track's `band_name` slug matches the band's own slug, the
203 # album is by the band itself and we use the plain `{band_id}` ID;
204 # otherwise we synthesize `{band_id}:{slug}`.
205 bands_by_id: dict[int, SearchResultArtist] = {
206 item.id: item for item in capped if isinstance(item, SearchResultArtist)
207 }
208 artist_id_by_item: dict[int, str] = await self._resolve_search_artist_ids(
209 capped, bands_by_id
210 )
211 artist_ids_seen: set[str] = set()
212 synthetic_artists: list[Artist] = []
213
214 for item in capped:
215 try:
216 if isinstance(item, SearchResultTrack) and MediaType.TRACK in media_types:
217 results.tracks = [
218 *results.tracks,
219 self._converters.track_from_search(
220 item, artist_item_id=artist_id_by_item[id(item)]
221 ),
222 ]
223 elif isinstance(item, SearchResultAlbum) and MediaType.ALBUM in media_types:
224 results.albums = [
225 *results.albums,
226 self._converters.album_from_search(
227 item, artist_item_id=artist_id_by_item[id(item)]
228 ),
229 ]
230 elif isinstance(item, SearchResultArtist) and MediaType.ARTIST in media_types:
231 artist_ids_seen.add(str(item.id))
232 results.artists = [*results.artists, self._converters.artist_from_search(item)]
233 except BandcampAPIError as error:
234 self.logger.warning("Failed to convert search result item: %s", error)
235 continue
236
237 if MediaType.ARTIST in media_types:
238 for item in capped:
239 if not isinstance(item, (SearchResultAlbum, SearchResultTrack)):
240 continue
241 if not item.artist_name:
242 continue
243 artist_item_id = artist_id_by_item[id(item)]
244 if artist_item_id in artist_ids_seen:
245 continue
246 artist_ids_seen.add(artist_item_id)
247 if ":" in artist_item_id:
248 synthetic_artists.append(
249 self._converters.synthetic_artist(
250 band_id=item.artist_id,
251 performer_name=item.artist_name,
252 url=item.artist_url or None,
253 image_url=item.image_url,
254 )
255 )
256 continue
257 if int(artist_item_id) == item.artist_id:
258 # Same band as the row's claimed page â its `b` row just
259 # didn't make the cap; re-introducing it here would surface
260 # a band the user wasn't searching for.
261 continue
262 with suppress(MediaNotFoundError, ResourceTemporarilyUnavailable, RetriesExhausted):
263 results.artists = [*results.artists, await self.get_artist(artist_item_id)]
264
265 if synthetic_artists:
266 results.artists = [*results.artists, *synthetic_artists][:limit]
267
268 return results
269
270 async def get_recommendations(self) -> list[RecommendationFolder]:
271 """Get this provider's available recommendation rows, without items."""
272 if not self._client.identity:
273 return []
274 return [
275 RecommendationFolder(
276 item_id="feed",
277 provider=self.instance_id,
278 name="Bandcamp Feed",
279 translation_key="feed",
280 icon="mdi-rss",
281 is_playable=True,
282 ),
283 RecommendationFolder(
284 item_id="wishlist",
285 provider=self.instance_id,
286 name="Wishlist",
287 translation_key="wishlist",
288 icon="mdi-heart",
289 is_playable=True,
290 ),
291 ]
292
293 async def get_recommendation_items(
294 self, item_id: str
295 ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
296 """
297 Get the items for a single recommendation row.
298
299 :param item_id: The item_id of the row, as returned by get_recommendations.
300 """
301 if not self._client.identity:
302 return UniqueList()
303 if item_id == "feed":
304 return UniqueList(await self._get_feed_tracks())
305 if item_id == "wishlist":
306 return UniqueList(await self._browse_person_content(None, CollectionType.WISHLIST))
307 return UniqueList()
308
309 async def _resolve_search_artist_ids(
310 self,
311 capped: Sequence[SearchResultItem],
312 bands_by_id: dict[int, SearchResultArtist],
313 ) -> dict[int, str]:
314 """
315 Resolve artist item IDs for album and track search results.
316
317 :param capped: Search results to resolve.
318 :param bands_by_id: Band results keyed by Bandcamp ID.
319 :returns: Artist item IDs keyed by search-result object identity.
320 """
321 rows: list[SearchResultAlbum | SearchResultTrack] = [
322 row for row in capped if isinstance(row, (SearchResultAlbum, SearchResultTrack))
323 ]
324 slug_to_name: dict[str, str] = {}
325 for row in rows:
326 performer = row.artist_name or ""
327 band = bands_by_id.get(row.artist_id)
328 if band and slugify_performer(band.name) == slugify_performer(performer):
329 continue
330 slug = slugify_performer(performer)
331 if slug:
332 slug_to_name.setdefault(slug, performer)
333
334 slug_to_real_id = await self._lookup_performer_band_ids_parallel(slug_to_name)
335
336 resolved: dict[int, str] = {}
337 for row in rows:
338 performer = row.artist_name or ""
339 band = bands_by_id.get(row.artist_id)
340 if band and slugify_performer(band.name) == slugify_performer(performer):
341 resolved[id(row)] = str(row.artist_id)
342 continue
343 real_id = slug_to_real_id.get(slugify_performer(performer))
344 if real_id is not None:
345 resolved[id(row)] = str(real_id)
346 continue
347 resolved[id(row)] = make_artist_id(row.artist_id, performer)
348 return resolved
349
350 async def _lookup_performer_band_ids_parallel(
351 self, names_by_slug: dict[str, str]
352 ) -> dict[str, int | None]:
353 """
354 Resolve performer names to Bandcamp artist IDs.
355
356 :param names_by_slug: Performer names keyed by normalized slug.
357 :returns: Resolved artist IDs, or ``None`` for unresolved performers.
358 """
359 if not names_by_slug:
360 return {}
361 slugs = list(names_by_slug)
362 raw_results = await asyncio.gather(
363 *(self._lookup_performer_band_id(names_by_slug[slug]) for slug in slugs),
364 return_exceptions=True,
365 )
366 out: dict[str, int | None] = {}
367 for slug, result in zip(slugs, raw_results, strict=True):
368 if isinstance(result, asyncio.CancelledError):
369 raise result
370 if isinstance(result, Exception):
371 self.logger.warning(
372 "performer band lookup failed for %r: %s",
373 names_by_slug[slug],
374 result,
375 )
376 out[slug] = None
377 elif isinstance(result, BaseException):
378 raise result
379 else:
380 out[slug] = result
381 return out
382
383 async def _lookup_performer_band_id(self, performer_name: str) -> int | None:
384 """Find the band_id for a performer who has their own Bandcamp page."""
385 target_slug = slugify_performer(performer_name)
386 if not target_slug:
387 return None
388 cache_key = f"performer_band_id.{target_slug}"
389 cached = await self.mass.cache.get(cache_key, provider=self.instance_id)
390 if cached is not None:
391 try:
392 cached_int = int(cached)
393 except ValueError, TypeError:
394 self.logger.warning(
395 "Discarding corrupt performer_band_id cache for %r: %r",
396 target_slug,
397 cached,
398 )
399 else:
400 # Negative results are persisted as 0 since the cache layer
401 # treats None as a miss.
402 return cached_int or None
403 band_id = await self._fetch_performer_band_id(performer_name, target_slug)
404 await self.mass.cache.set(
405 cache_key,
406 band_id or 0,
407 expiration=CACHE_METADATA,
408 provider=self.instance_id,
409 )
410 return band_id
411
412 @throttle_with_retries
413 async def _fetch_performer_band_id(self, performer_name: str, target_slug: str) -> int | None:
414 """Autocomplete-search for ``performer_name``; return the first non-label band match."""
415 try:
416 results = await self._client.search(performer_name)
417 except BandcampRateLimitError as error:
418 raise RateLimited(
419 "Bandcamp rate limit reached", backoff_time=error.retry_after
420 ) from error
421 except BandcampAPIError as error:
422 self.logger.warning(
423 "Bandcamp autocomplete failed for performer %r: %s", performer_name, error
424 )
425 raise
426 for item in results:
427 # Skip labels so a same-named label doesn't masquerade as the band page.
428 if (
429 isinstance(item, SearchResultArtist)
430 and not item.is_label
431 and slugify_performer(item.name) == target_slug
432 ):
433 return int(item.id)
434 return None
435
436 @throttle_with_retries
437 async def _fetch_collection_page(
438 self,
439 collection_type: CollectionType,
440 older_than_token: str | None,
441 fan_id: int | None,
442 ) -> CollectionSummary:
443 """
444 Fetch a single page of collection items with throttling and retry.
445
446 :param collection_type: The type of collection to fetch.
447 :param older_than_token: Pagination cursor from the previous page.
448 :param fan_id: Fan ID to query. None = authenticated user.
449 """
450 try:
451 return await self._client.get_collection_items(
452 collection_type,
453 older_than_token=older_than_token,
454 fan_id=fan_id,
455 )
456 except BandcampRateLimitError as error:
457 raise RateLimited(
458 "Bandcamp rate limit reached", backoff_time=error.retry_after
459 ) from error
460
461 async def _get_all_collection_items(
462 self,
463 collection_type: CollectionType,
464 fan_id: int | None = None,
465 ) -> list[CollectionItem | FollowingItem | FanItem]:
466 """
467 Fetch all pages of a collection endpoint.
468
469 :param collection_type: The type of collection to fetch.
470 :param fan_id: Fan ID to query. None = authenticated user.
471 """
472 all_items: list[CollectionItem | FollowingItem | FanItem] = []
473 older_than_token: str | None = None
474 seen_tokens: set[str] = set()
475 while True:
476 page = await self._fetch_collection_page(collection_type, older_than_token, fan_id)
477 all_items.extend(page.items)
478 self.logger.debug(
479 "Fetched %d items for %s (has_more=%s, last_token=%s, total=%d)",
480 len(page.items),
481 collection_type.value,
482 page.has_more,
483 page.last_token,
484 len(all_items),
485 )
486 if not page.has_more or not page.last_token:
487 break
488 if page.last_token in seen_tokens:
489 self.logger.warning(
490 "Pagination loop detected for %s: token %s already seen, stopping",
491 collection_type.value,
492 page.last_token,
493 )
494 break
495 seen_tokens.add(page.last_token)
496 older_than_token = page.last_token
497 return all_items
498
499 async def get_library_artists(self) -> AsyncGenerator[Artist]:
500 """Retrieve library artists from Bandcamp."""
501 if not self._client.identity: # library requires identity
502 return
503
504 try:
505 items = await self._get_all_collection_items(CollectionType.COLLECTION)
506 band_ids = set()
507 for item in items:
508 if item.item_type == "band":
509 band_ids.add(item.item_id)
510 elif item.item_type == "album":
511 band_ids.add(item.band_id)
512
513 for band_id in band_ids:
514 yield await self.get_artist(str(band_id))
515 await asyncio.sleep(0) # Yield control to avoid blocking
516
517 except BandcampMustBeLoggedInError as error:
518 self.logger.error("Error getting Bandcamp library artists: Wrong identity token.")
519 raise LoginFailed("Wrong Bandcamp identity token.") from error
520 except BandcampNotFoundError as error:
521 raise MediaNotFoundError("Bandcamp library artists returned no results") from error
522 except BandcampRateLimitError as error:
523 raise RateLimited(
524 "Bandcamp rate limit reached", backoff_time=error.retry_after
525 ) from error
526 except BandcampAPIError as error:
527 raise MediaNotFoundError("Failed to get library artists") from error
528
529 async def get_library_albums(self) -> AsyncGenerator[Album]:
530 """Retrieve library albums from Bandcamp."""
531 if not self._client.identity: # library requires identity
532 return
533
534 try:
535 items = await self._get_all_collection_items(CollectionType.COLLECTION)
536 for item in items:
537 if item.item_type == "album":
538 yield await self.get_album(f"{item.band_id}-{item.item_id}")
539 await asyncio.sleep(0) # Yield control to avoid blocking
540 except BandcampMustBeLoggedInError as error:
541 self.logger.error("Error getting Bandcamp library albums: Wrong identity token.")
542 raise LoginFailed("Wrong Bandcamp identity token.") from error
543 except BandcampNotFoundError as error:
544 raise MediaNotFoundError("Bandcamp library albums returned no results") from error
545 except BandcampRateLimitError as error:
546 raise RateLimited(
547 "Bandcamp rate limit reached", backoff_time=error.retry_after
548 ) from error
549 except BandcampAPIError as error:
550 raise MediaNotFoundError("Failed to get library albums") from error
551
552 async def get_library_tracks(self) -> AsyncGenerator[Track]:
553 """Retrieve library tracks from Bandcamp."""
554 if not self._client.identity: # library requires identity
555 return
556
557 async for album in self.get_library_albums():
558 tracks = await self.get_album_tracks(album.item_id)
559 for track in tracks:
560 yield track
561 await asyncio.sleep(0) # Yield control to avoid blocking
562
563 @use_cache(CACHE_METADATA)
564 @throttle_with_retries
565 async def get_artist(self, prov_artist_id: str) -> Artist:
566 """
567 Get full artist details by ID.
568
569 :param prov_artist_id: Bandcamp artist or synthetic artist ID.
570 :returns: The resolved Music Assistant artist.
571 :raises InvalidDataError: If the artist ID is malformed.
572 :raises MediaNotFoundError: If the artist cannot be resolved.
573 """
574 try:
575 band_id, performer_slug = parse_artist_id(prov_artist_id)
576 except ValueError as error:
577 raise InvalidDataError(f"Malformed Bandcamp artist ID: {prov_artist_id}") from error
578
579 if performer_slug is None:
580 try:
581 api_artist = await self._client.get_artist(band_id)
582 return self._converters.artist_from_api(api_artist)
583 except BandcampNotFoundError as error:
584 raise MediaNotFoundError(
585 f"Artist {prov_artist_id} not found on Bandcamp"
586 ) from error
587 except BandcampRateLimitError as error:
588 raise RateLimited(
589 "Bandcamp rate limit reached", backoff_time=error.retry_after
590 ) from error
591 except BandcampAPIError as error:
592 raise MediaNotFoundError(f"Failed to get artist {prov_artist_id}") from error
593
594 # Synthetic: locate matching items in the band's discography and
595 # build an artist scoped to that performer. Falls back to the real
596 # band when the slug actually matches the band's own name (e.g.
597 # cached IDs constructed before disambiguation was reliable).
598 return await self._get_synthetic_artist(prov_artist_id, band_id, performer_slug)
599
600 async def _get_synthetic_artist(
601 self, prov_artist_id: str, band_id: int, performer_slug: str
602 ) -> Artist:
603 """Resolve a synthetic artist ID to a Music Assistant artist."""
604 try:
605 api_artist = await self._client.get_artist(band_id)
606 except BandcampNotFoundError as error:
607 raise MediaNotFoundError(f"Artist {prov_artist_id} not found on Bandcamp") from error
608 except BandcampRateLimitError as error:
609 raise RateLimited(
610 "Bandcamp rate limit reached", backoff_time=error.retry_after
611 ) from error
612 except BandcampAPIError as error:
613 raise MediaNotFoundError(f"Failed to get artist {prov_artist_id}") from error
614
615 # Resolve the hosting artist first so legacy owner-slug synthetic IDs
616 # collapse to the real artist before discography filtering.
617 if slugify_performer(api_artist.name) == performer_slug:
618 return self._converters.artist_from_api(api_artist)
619
620 # A synthetic performer is valid only when its explicit credit appears
621 # in the hosting page's discography.
622 try:
623 api_discography = await self._fetch_discography(band_id)
624 except BandcampNotFoundError as error:
625 raise MediaNotFoundError(f"Artist {prov_artist_id} not found on Bandcamp") from error
626 except BandcampRateLimitError as error:
627 raise RateLimited(
628 "Bandcamp rate limit reached", backoff_time=error.retry_after
629 ) from error
630 except BandcampAPIError as error:
631 raise MediaNotFoundError(f"Failed to get artist {prov_artist_id}") from error
632
633 matching = self._filter_discography_by_performer(api_discography, performer_slug)
634 if not matching:
635 raise MediaNotFoundError(f"Artist {prov_artist_id} not found on Bandcamp")
636
637 first = matching[0]
638 performer_name = str(first.get("artist_name") or "")
639 if not performer_name or slugify_performer(performer_name) != performer_slug:
640 raise MediaNotFoundError(f"Artist {prov_artist_id} not found on Bandcamp")
641 art_id = first.get("art_id")
642 image_url = f"https://f4.bcbits.com/img/a{art_id}_0.jpg" if art_id else None
643 # The performer doesn't have their own Bandcamp page; surface the
644 # hosting band's URL so the artist tile links somewhere meaningful
645 # (matching what the search-emission path passes through).
646 return self._converters.synthetic_artist(
647 band_id=band_id,
648 performer_name=performer_name,
649 url=api_artist.url,
650 image_url=image_url,
651 )
652
653 @use_cache(CACHE_METADATA)
654 @throttle_with_retries
655 async def _fetch_discography(self, band_id: int) -> list[dict[str, Any]]:
656 """
657 Fetch a band's discography.
658
659 :param band_id: Bandcamp ID of the page owner.
660 :returns: Raw discography entries.
661 """
662 # Return type is `list[dict[str, Any]]` rather than
663 # `list[DiscographyItem]`: the cache controller's deserializer
664 # uses `isinstance` checks which TypedDict does not support.
665 # Callers cast at the converter boundary.
666 result: list[dict[str, Any]] = await self._client.get_artist_discography(band_id)
667 return result
668
669 @staticmethod
670 def _filter_discography_by_performer(
671 items: list[dict[str, Any]], performer_slug: str
672 ) -> list[dict[str, Any]]:
673 """Filter discography rows down to those credited to a given performer slug."""
674 return [
675 item
676 for item in items
677 if slugify_performer(str(item.get("artist_name") or "")) == performer_slug
678 ]
679
680 async def _resolve_artist_item_id(
681 self, *, band_id: int, performer: str | None, band_name: str
682 ) -> str:
683 """Resolve a single album/track's artist item_id (no batch context)."""
684 if not performer or slugify_performer(performer) == slugify_performer(band_name):
685 return str(band_id)
686 real_band_id = await self._lookup_performer_band_id(performer)
687 if real_band_id is not None:
688 return str(real_band_id)
689 return make_artist_id(band_id, performer)
690
691 @use_cache(CACHE_METADATA)
692 @throttle_with_retries
693 async def get_album(self, prov_album_id: str) -> Album:
694 """Get full album details by id."""
695 artist_id, album_id, _ = split_id(prov_album_id)
696 try:
697 api_album = await self._client.get_album(artist_id, album_id)
698 except BandcampNotFoundError as error:
699 raise MediaNotFoundError(f"Album {prov_album_id} not found on Bandcamp") from error
700 except BandcampRateLimitError as error:
701 raise RateLimited(
702 "Bandcamp rate limit reached", backoff_time=error.retry_after
703 ) from error
704 except BandcampAPIError as error:
705 raise MediaNotFoundError(f"Failed to get album {prov_album_id}") from error
706 artist_item_id = await self._resolve_artist_item_id(
707 band_id=api_album.artist.id,
708 performer=api_album.tralbum_artist,
709 band_name=api_album.artist.name,
710 )
711 return self._converters.album_from_api(api_album, artist_item_id=artist_item_id)
712
713 @throttle_with_retries
714 async def _fetch_api_track(self, item_id: str) -> tuple[BCTrack, BCAlbum | None]:
715 """
716 Fetch a raw API track and its parent album by compound item ID.
717
718 Uses get_album when album_id is present (most tracks), falling back
719 to get_track for standalone tracks (album_id=0).
720
721 :param item_id: Compound track ID in the form artist_id-album_id-track_id.
722 """
723 artist_id, album_id, track_id = split_id(item_id)
724 if not track_id:
725 album_id, track_id = 0, album_id
726
727 try:
728 if album_id:
729 api_album = await self._client.get_album(artist_id, album_id)
730 api_track = next((t for t in api_album.tracks if t.id == track_id), None)
731 if not api_track:
732 raise MediaNotFoundError(f"Track {item_id} not found in album on Bandcamp")
733 return api_track, api_album
734 return await self._client.get_track(artist_id, track_id), None
735 except BandcampMustBeLoggedInError as error:
736 raise LoginFailed("Bandcamp login is invalid or expired.") from error
737 except BandcampNotFoundError as error:
738 raise MediaNotFoundError(f"Track {item_id} not found on Bandcamp") from error
739 except BandcampRateLimitError as error:
740 raise RateLimited(
741 "Bandcamp rate limit reached", backoff_time=error.retry_after
742 ) from error
743 except BandcampAPIError as error:
744 raise MediaNotFoundError(f"Failed to get track {item_id}") from error
745
746 @use_cache(CACHE_METADATA)
747 async def get_track(self, prov_track_id: str) -> Track:
748 """Get full track details by id."""
749 api_track, api_album = await self._fetch_api_track(prov_track_id)
750 if api_album:
751 artist_item_id = await self._resolve_artist_item_id(
752 band_id=api_album.artist.id,
753 performer=api_album.tralbum_artist,
754 band_name=api_album.artist.name,
755 )
756 return self._converters.track_from_api(
757 track=api_track,
758 album_id=api_album.id,
759 album_name=api_album.title,
760 album_image_url=api_album.art_url or "",
761 tralbum_artist=api_album.tralbum_artist,
762 artist_item_id=artist_item_id,
763 )
764 # Standalone tracks (album_id=0) carry the performer credit on
765 # the track itself when fetched directly from tralbum_details.
766 artist_item_id = await self._resolve_artist_item_id(
767 band_id=api_track.artist.id,
768 performer=api_track.tralbum_artist,
769 band_name=api_track.artist.name,
770 )
771 return self._converters.track_from_api(
772 track=api_track,
773 album_id=api_track.album.id if api_track.album else None,
774 album_name=api_track.album.title if api_track.album else "",
775 album_image_url=(api_track.album.art_url if api_track.album else "") or "",
776 tralbum_artist=api_track.tralbum_artist,
777 artist_item_id=artist_item_id,
778 )
779
780 @use_cache(CACHE_METADATA)
781 @throttle_with_retries
782 async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
783 """Get all tracks in an album."""
784 artist_id, album_id, _ = split_id(prov_album_id)
785 try:
786 api_album = await self._client.get_album(artist_id, album_id)
787 except BandcampNotFoundError as error:
788 raise MediaNotFoundError(
789 f"Album tracks for {prov_album_id} not found on Bandcamp"
790 ) from error
791 except BandcampRateLimitError as error:
792 raise RateLimited(
793 "Bandcamp rate limit reached", backoff_time=error.retry_after
794 ) from error
795 except BandcampAPIError as error:
796 raise MediaNotFoundError(f"Failed to get albums tracks for {prov_album_id}") from error
797 if not api_album.tracks:
798 return []
799 artist_item_id = await self._resolve_artist_item_id(
800 band_id=api_album.artist.id,
801 performer=api_album.tralbum_artist,
802 band_name=api_album.artist.name,
803 )
804 return [
805 self._converters.track_from_api(
806 track=track,
807 album_id=album_id,
808 album_name=api_album.title,
809 album_image_url=api_album.art_url or "",
810 tralbum_artist=api_album.tralbum_artist,
811 artist_item_id=artist_item_id,
812 )
813 for track in api_album.tracks
814 if track.streaming_url # Only include tracks with streaming URLs
815 ]
816
817 @use_cache(CACHE_METADATA)
818 @throttle_with_retries
819 async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
820 """
821 Get albums by an artist.
822
823 For real artist IDs this returns the band's full discography (the
824 original behavior). For synthetic IDs (``{band_id}:{slug}``) this
825 filters the band's discography to only the items where the
826 performer matches.
827 """
828 try:
829 band_id, performer_slug = parse_artist_id(prov_artist_id)
830 except ValueError as error:
831 raise InvalidDataError(f"Malformed Bandcamp artist ID: {prov_artist_id}") from error
832
833 if performer_slug is not None:
834 try:
835 api_artist = await self._client.get_artist(band_id)
836 except BandcampNotFoundError as error:
837 raise MediaNotFoundError(
838 f"Artist {prov_artist_id} albums not found on Bandcamp"
839 ) from error
840 except BandcampRateLimitError as error:
841 raise RateLimited(
842 "Bandcamp rate limit reached", backoff_time=error.retry_after
843 ) from error
844 except BandcampAPIError as error:
845 raise MediaNotFoundError(
846 f"Failed to get albums for artist {prov_artist_id}"
847 ) from error
848 if slugify_performer(api_artist.name) == performer_slug:
849 performer_slug = None
850
851 try:
852 api_discography = await self._fetch_discography(band_id)
853 except BandcampNotFoundError as error:
854 raise MediaNotFoundError(
855 f"Artist {prov_artist_id} albums not found on Bandcamp"
856 ) from error
857 except BandcampRateLimitError as error:
858 raise RateLimited(
859 "Bandcamp rate limit reached", backoff_time=error.retry_after
860 ) from error
861 except BandcampAPIError as error:
862 raise MediaNotFoundError(f"Failed to get albums for artist {prov_artist_id}") from error
863
864 items = [
865 item
866 for item in api_discography
867 if item.get("item_type") == "album" and item.get("item_id")
868 ]
869 if performer_slug is not None:
870 items = self._filter_discography_by_performer(items, performer_slug)
871
872 # Pre-resolve so this listing's artist links match what `get_album`
873 # produces on click; otherwise list and detail views diverge for the
874 # same performer.
875 names_by_slug: dict[str, str] = {}
876 for item in items:
877 performer = str(item.get("artist_name") or "")
878 band_name = str(item.get("band_name") or "")
879 if not performer:
880 continue
881 slug = slugify_performer(performer)
882 if not slug or slug == slugify_performer(band_name):
883 continue
884 names_by_slug.setdefault(slug, performer)
885 slug_to_real_id = await self._lookup_performer_band_ids_parallel(names_by_slug)
886
887 return [
888 self._converters.album_from_discography_item(
889 cast("DiscographyItem", item),
890 artist_item_id=self._discography_artist_item_id(item, slug_to_real_id),
891 )
892 for item in items
893 ]
894
895 @staticmethod
896 def _discography_artist_item_id(
897 item: dict[str, Any], slug_to_real_id: dict[str, int | None]
898 ) -> str:
899 """Sync counterpart of ``_resolve_artist_item_id`` for a discography row."""
900 band_id = int(item.get("band_id") or 0)
901 performer = str(item.get("artist_name") or "")
902 band_name = str(item.get("band_name") or "")
903 if not performer or slugify_performer(performer) == slugify_performer(band_name):
904 return str(band_id)
905 real_id = slug_to_real_id.get(slugify_performer(performer))
906 if real_id is not None:
907 return str(real_id)
908 return make_artist_id(band_id, performer)
909
910 @use_cache(CACHE_METADATA)
911 @throttle_with_retries
912 async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
913 """Get top tracks of an artist."""
914 tracks: list[Track] = []
915 # get_artist_albums and get_album_tracks already handle exceptions and rate limiting
916 albums = await self.get_artist_albums(prov_artist_id)
917 albums.sort(key=lambda album: (album.year is None, album.year or 0), reverse=True)
918 for album in albums:
919 tracks.extend(await self.get_album_tracks(album.item_id))
920 if len(tracks) >= self.top_tracks_limit:
921 break
922
923 return tracks[: self.top_tracks_limit]
924
925 @throttle_with_retries
926 async def _fetch_feed(self) -> FeedResponse:
927 """Fetch the authenticated user's feed with throttling and retry."""
928 try:
929 return await self._client.get_feed()
930 except BandcampRateLimitError as error:
931 raise RateLimited(
932 "Bandcamp rate limit reached", backoff_time=error.retry_after
933 ) from error
934
935 async def _get_feed_tracks(self) -> list[Track]:
936 """Fetch and convert the streamable tracks from the user's feed."""
937 cache_key = "_feed_tracks"
938 cached = await self.mass.cache.get(cache_key, provider=self.instance_id, base_class=Track)
939 if cached is not None:
940 return cached # type: ignore[no-any-return]
941 tracks: list[Track] = []
942 async with self._map_api_errors("Failed to get Bandcamp feed"):
943 feed = await self._fetch_feed()
944 tracks = [
945 self._converters.track_from_feed(track)
946 for track in feed.track_list
947 if track.streaming_url
948 ]
949 await self.mass.cache.set(
950 cache_key,
951 [t.to_dict() for t in tracks],
952 expiration=CACHE_USER_LISTS if tracks else CACHE_EMPTY_RESULTS,
953 provider=self.instance_id,
954 )
955 return tracks
956
957 async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
958 """
959 Browse this provider's items.
960
961 :param path: The path to browse, (e.g. provider_id://artists).
962 """
963 subpath = path.split("://")[1] if "://" in path else ""
964 # Filter empty segments from double-slashes or trailing slashes
965 path_parts = [p for p in subpath.split("/") if p]
966 base = f"{self.instance_id}://"
967
968 # Route fan/follower paths (supports arbitrary nesting depth)
969 if path_parts and path_parts[0] in (BROWSE_FANS, BROWSE_FOLLOWERS):
970 return await self._browse_person(path_parts, base)
971
972 # The feed/wishlist recommendation folders resolve to their tracks here when played;
973 # the folder's explicit path is dropped on deserialization, so play arrives as the
974 # bare item_id slug (e.g. ".../feed") rather than ".../recommendations/feed".
975 if path_parts == [BROWSE_FEED]:
976 return await self._get_feed_tracks()
977 if path_parts == [BROWSE_WISHLIST]:
978 return await self._browse_person_content(None, CollectionType.WISHLIST)
979 if path_parts == [BROWSE_FOLLOWING]:
980 return await self._browse_person_following(None)
981
982 # Delegate standard library paths and root listing to the base class
983 result = list(await super().browse(path))
984
985 # At root level, append custom browse folders when authenticated.
986 # These top-level folders query the authenticated user (person_id=None);
987 # person-specific paths (e.g. fans/42/wishlist) work without authentication
988 # since the Bandcamp API only requires identity for the "me" shortcut.
989 if not path_parts and self._client.identity:
990 # Collection is excluded â the user's own collection is the standard library.
991 for folder_id, folder_name in (
992 (BROWSE_WISHLIST, "Wishlist"),
993 (BROWSE_FOLLOWING, "Following"),
994 (BROWSE_FANS, "Fans"),
995 (BROWSE_FOLLOWERS, "Followers"),
996 ):
997 result.append(
998 BrowseFolder(
999 item_id=folder_id,
1000 provider=self.instance_id,
1001 path=base + folder_id,
1002 name=folder_name,
1003 translation_key=folder_id,
1004 )
1005 )
1006
1007 return result
1008
1009 async def _browse_person(
1010 self,
1011 path_parts: list[str],
1012 base: str,
1013 ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1014 """
1015 Route person browse paths: fans/followers and their sub-categories.
1016
1017 Pattern: (fans|followers)[/{id}[/(collection|wishlist|following|fans|followers)]*]
1018 """
1019 # Top-level: authenticated user's fans or followers
1020 if len(path_parts) == 1:
1021 collection_type = (
1022 CollectionType.FOLLOWING_FANS
1023 if path_parts[0] == BROWSE_FANS
1024 else CollectionType.FOLLOWERS
1025 )
1026 return await self._browse_person_people(collection_type, f"{base}{path_parts[0]}")
1027
1028 tail = path_parts[-1]
1029
1030 # Path ends with a person identifier (numeric ID or slug) â show their 5 sub-folders
1031 person_id = await self._resolve_person_segment(tail)
1032 if person_id is not None:
1033 if person_id <= 0:
1034 raise InvalidDataError(f"Invalid person ID in browse path: {tail}")
1035 return self._browse_person_root(person_id, f"{base}{'/'.join(path_parts)}")
1036
1037 # Path ends with a sub-category â person identifier is second-to-last
1038 if len(path_parts) < 2:
1039 raise InvalidDataError(f"Invalid browse path: {base}{'/'.join(path_parts)}")
1040 person_id = await self._resolve_person_segment(path_parts[-2])
1041 if person_id is None:
1042 raise InvalidDataError(f"Invalid browse path: {base}{'/'.join(path_parts)}")
1043 if person_id <= 0:
1044 raise InvalidDataError(f"Invalid person ID in browse path: {path_parts[-2]}")
1045
1046 route = PERSON_SUB_ROUTES.get(tail)
1047 if route is None:
1048 raise InvalidDataError(f"Unknown browse sub-category: {tail}")
1049
1050 method_kind, collection_type = route
1051 if method_kind == "content":
1052 return await self._browse_person_content(person_id, collection_type)
1053 if method_kind == "following":
1054 return await self._browse_person_following(person_id)
1055 if method_kind != "people":
1056 raise InvalidDataError(f"Unknown route kind: {method_kind}")
1057 canon = "/".join(path_parts)
1058 return await self._browse_person_people(collection_type, f"{base}{canon}", person_id)
1059
1060 # --- Person browse helpers (fans, followers, and social graph traversal) ---
1061
1062 async def _resolve_person_segment(self, segment: str) -> int | None:
1063 """
1064 Resolve a path segment to a fan_id.
1065
1066 Returns None if the segment is neither a known slug nor a valid
1067 int, or if it is a known sub-route name (e.g. "collection",
1068 "wishlist").
1069 """
1070 if segment in self._slug_to_fan_id:
1071 return self._slug_to_fan_id[segment]
1072 try:
1073 return int(segment)
1074 except ValueError:
1075 pass
1076 # Known sub-route names are structural, not user slugs
1077 if segment in PERSON_SUB_ROUTES:
1078 return None
1079 # Slug not in cache and not numeric â rebuild from parent lists and retry
1080 await self._rebuild_slug_cache()
1081 return self._slug_to_fan_id.get(segment)
1082
1083 async def _rebuild_slug_cache(self) -> None:
1084 """Re-fetch fan/follower lists to rebuild the slugâfan_id map."""
1085 base = f"{self.instance_id}://"
1086 for collection_type, folder_id in (
1087 (CollectionType.FOLLOWING_FANS, BROWSE_FANS),
1088 (CollectionType.FOLLOWERS, BROWSE_FOLLOWERS),
1089 ):
1090 with suppress(LoginFailed, RateLimited, MediaNotFoundError, RetriesExhausted):
1091 await self._browse_person_people(collection_type, f"{base}{folder_id}")
1092
1093 @staticmethod
1094 def _fan_slug(person: FanItem) -> str | None:
1095 """
1096 Extract the URL slug from a FanItem's url.
1097
1098 e.g. "https://bandcamp.com/teancom" â "teancom"
1099 """
1100 if person.url:
1101 slug: str = person.url.rstrip("/").rsplit("/", 1)[-1]
1102 if slug:
1103 return slug
1104 return None
1105
1106 def _people_to_folders(self, items: list[FanItem], base_path: str) -> list[BrowseFolder]:
1107 """Convert a list of people to BrowseFolder items with thumbnails."""
1108 folders: list[BrowseFolder] = []
1109 for person in items:
1110 slug = self._fan_slug(person)
1111 if slug:
1112 self._slug_to_fan_id[slug] = person.fan_id
1113 path_segment = slug or str(person.fan_id)
1114 folder = BrowseFolder(
1115 item_id=f"person_{person.fan_id}",
1116 provider=self.instance_id,
1117 path=f"{base_path}/{path_segment}",
1118 name=person.name or f"User {person.fan_id}",
1119 )
1120 if person.image_url:
1121 folder.image = MediaItemImage(
1122 type=ImageType.THUMB,
1123 path=person.image_url,
1124 provider=self.instance_id,
1125 remotely_accessible=True,
1126 )
1127 folders.append(folder)
1128 return folders
1129
1130 def _browse_person_root(self, person_id: int, base_path: str) -> list[BrowseFolder]:
1131 """Return the 5 sub-folders for a person's profile."""
1132 return [
1133 BrowseFolder(
1134 item_id=f"person_{person_id}_{sub_id}",
1135 provider=self.instance_id,
1136 path=f"{base_path}/{sub_id}",
1137 name=name,
1138 translation_key=sub_id,
1139 )
1140 for sub_id, name in PERSON_SUB_FOLDERS
1141 ]
1142
1143 @asynccontextmanager
1144 async def _map_api_errors(self, context: str) -> AsyncIterator[None]:
1145 """Map Bandcamp API exceptions to MusicAssistant exceptions."""
1146 try:
1147 yield
1148 except BandcampMustBeLoggedInError as error:
1149 raise LoginFailed("Wrong Bandcamp identity token.") from error
1150 except BandcampRateLimitError as error:
1151 raise RateLimited(
1152 "Bandcamp rate limit reached", backoff_time=error.retry_after
1153 ) from error
1154 except BandcampAPIError as error:
1155 raise MediaNotFoundError(context) from error
1156
1157 @staticmethod
1158 def _deserialize_content_item(item: dict[str, object]) -> Album | Track:
1159 """Deserialize a cached content item back to its model type."""
1160 media_type = item.get("media_type")
1161 if media_type == MediaType.ALBUM:
1162 return Album.from_dict(item)
1163 if media_type == MediaType.TRACK:
1164 return Track.from_dict(item)
1165 msg = f"Unexpected media_type in cached content item: {media_type}"
1166 raise ValueError(msg)
1167
1168 @throttle_with_retries
1169 async def _browse_person_content(
1170 self, person_id: int | None, collection_type: CollectionType
1171 ) -> list[Album | Track]:
1172 """
1173 Fetch a person's collection or wishlist items.
1174
1175 :param person_id: Person to query. None = authenticated user.
1176 """
1177 cache_key = f"_browse_person_content_{person_id}_{collection_type.value}"
1178 cached = await self.mass.cache.get(cache_key, provider=self.instance_id)
1179 if cached is not None:
1180 try:
1181 return [self._deserialize_content_item(item) for item in cached]
1182 except LookupError, ValueError, UnserializableDataError, InvalidDataError:
1183 self.logger.warning("Stale cache for %s, fetching fresh", cache_key)
1184 results: list[Album | Track] = []
1185 context = f"Failed to get {collection_type.value} for person {person_id}"
1186 async with self._map_api_errors(context):
1187 items = await self._get_all_collection_items(collection_type, fan_id=person_id)
1188 for item in items:
1189 with suppress(MediaNotFoundError):
1190 if item.item_type == "album":
1191 results.append(await self.get_album(f"{item.band_id}-{item.item_id}"))
1192 elif item.item_type == "track":
1193 results.append(await self.get_track(f"{item.band_id}-0-{item.item_id}"))
1194 await self.mass.cache.set(
1195 cache_key,
1196 [item.to_dict() for item in results],
1197 expiration=CACHE_USER_LISTS if results else CACHE_EMPTY_RESULTS,
1198 provider=self.instance_id,
1199 )
1200 return results
1201
1202 @throttle_with_retries
1203 async def _browse_person_following(self, person_id: int | None) -> list[Artist]:
1204 """
1205 Fetch a person's followed artists.
1206
1207 :param person_id: Person to query. None = authenticated user.
1208 """
1209 cache_key = f"_browse_person_following_{person_id}"
1210 cached = await self.mass.cache.get(cache_key, provider=self.instance_id, base_class=Artist)
1211 if cached is not None:
1212 return cached # type: ignore[no-any-return]
1213 artists: list[Artist] = []
1214 async with self._map_api_errors(f"Failed to get following for person {person_id}"):
1215 collection = await self._get_all_collection_items(
1216 CollectionType.FOLLOWING, fan_id=person_id
1217 )
1218 for item in collection:
1219 try:
1220 artists.append(await self.get_artist(str(item.band_id)))
1221 except MediaNotFoundError:
1222 self.logger.warning(
1223 "Artist not found for band_id %s (%s)", item.band_id, item.name
1224 )
1225 await self.mass.cache.set(
1226 cache_key,
1227 [a.to_dict() for a in artists],
1228 expiration=CACHE_USER_LISTS if artists else CACHE_EMPTY_RESULTS,
1229 provider=self.instance_id,
1230 )
1231 return artists
1232
1233 @throttle_with_retries
1234 async def _browse_person_people(
1235 self,
1236 collection_type: CollectionType,
1237 base_path: str,
1238 person_id: int | None = None,
1239 ) -> list[BrowseFolder]:
1240 """
1241 Fetch a person's fans or followers as browsable folders.
1242
1243 :param collection_type: FOLLOWING_FANS or FOLLOWERS.
1244 :param base_path: Browse path prefix for the resulting folder links.
1245 :param person_id: Person to query. None = authenticated user.
1246 """
1247 # base_path included intentionally: folder links differ per navigation path.
1248 cache_key = f"_browse_person_people_{person_id}_{collection_type.value}_{base_path}"
1249 cached = await self.mass.cache.get(
1250 cache_key, provider=self.instance_id, base_class=BrowseFolder
1251 )
1252 if cached is not None:
1253 for folder in cached:
1254 segment = folder.path.rstrip("/").rsplit("/", 1)[-1]
1255 fan_id_str = folder.item_id.removeprefix("person_")
1256 with suppress(ValueError):
1257 self._slug_to_fan_id[segment] = int(fan_id_str)
1258 return cached # type: ignore[no-any-return]
1259 context = f"Failed to get {collection_type.value} for person {person_id}"
1260 async with self._map_api_errors(context):
1261 collection = await self._get_all_collection_items(collection_type, fan_id=person_id)
1262 folders = self._people_to_folders(collection, base_path)
1263 await self.mass.cache.set(
1264 cache_key,
1265 [f.to_dict() for f in folders],
1266 expiration=CACHE_USER_LISTS if folders else CACHE_EMPTY_RESULTS,
1267 provider=self.instance_id,
1268 )
1269 return folders
1270
1271 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
1272 """
1273 Return the content details for the given track.
1274
1275 Fetches fresh from the Bandcamp API since streaming URLs may expire.
1276 """
1277 api_track, _ = await self._fetch_api_track(item_id)
1278
1279 streaming_url, bitrate, content_type = self._converters.streaming_url_from_api(
1280 api_track.streaming_url or {}
1281 )
1282 if not streaming_url:
1283 raise MediaNotFoundError(f"No streaming URL found for track {item_id}")
1284
1285 return StreamDetails(
1286 item_id=item_id,
1287 provider=self.instance_id,
1288 audio_format=AudioFormat(
1289 content_type=content_type,
1290 bit_rate=bitrate,
1291 ),
1292 stream_type=StreamType.HTTP,
1293 media_type=media_type,
1294 path=streaming_url,
1295 can_seek=True,
1296 allow_seek=True,
1297 )
1298