music-assistant-server

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, 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 (
19    BCAlbum,
20    BCTrack,
21    CollectionItem,
22    CollectionSummary,
23    CollectionType,
24    FanItem,
25    FeedResponse,
26    FollowingItem,
27)
28from mashumaro.exceptions import UnserializableDataError
29from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
30from music_assistant_models.enums import (
31    ConfigEntryType,
32    ImageType,
33    MediaType,
34    StreamType,
35)
36from music_assistant_models.errors import (
37    InvalidDataError,
38    LoginFailed,
39    MediaNotFoundError,
40    RateLimited,
41)
42from music_assistant_models.media_items import (
43    Album,
44    Artist,
45    AudioFormat,
46    BrowseFolder,
47    ItemMapping,
48    MediaItemImage,
49    MediaItemType,
50    RecommendationFolder,
51    SearchResults,
52    Track,
53    UniqueList,
54)
55from music_assistant_models.streamdetails import StreamDetails
56
57from music_assistant.constants import CONF_ENTRY_UNOFFICIAL_PROVIDER
58from music_assistant.controllers.cache import use_cache
59from music_assistant.helpers.throttle_retry import ThrottlerManager, throttle_with_retries
60from music_assistant.mass import MusicAssistant
61from music_assistant.models import ProviderInstanceType
62from music_assistant.models.music_provider import MusicProvider
63
64from .constants import (
65    BROWSE_FANS,
66    BROWSE_FEED,
67    BROWSE_FOLLOWERS,
68    BROWSE_FOLLOWING,
69    BROWSE_WISHLIST,
70    CACHE_EMPTY_RESULTS,
71    CACHE_METADATA,
72    CACHE_USER_LISTS,
73    CONF_IDENTITY,
74    CONF_TOP_TRACKS_LIMIT,
75    DEFAULT_TOP_TRACKS_LIMIT,
76    PERSON_SUB_FOLDERS,
77    PERSON_SUB_ROUTES,
78    SUPPORTED_FEATURES,
79)
80from .converters import BandcampConverters
81
82if TYPE_CHECKING:
83    from music_assistant_models.provider import ProviderManifest
84
85
86async def setup(
87    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
88) -> ProviderInstanceType:
89    """Initialize provider(instance) with given configuration."""
90    return BandcampProvider(mass, manifest, config, SUPPORTED_FEATURES)
91
92
93def split_id(id_: str) -> tuple[int, int, int]:
94    """
95    Return (artist_id, album_id, track_id). Missing parts are returned as 0.
96
97    :param id_: Compound ID string, e.g. "123-456-789".
98    :raises InvalidDataError: If the ID contains non-numeric parts.
99    """
100    try:
101        parts = id_.split("-")
102        part_0 = int(parts[0])
103        part_1 = int(parts[1]) if len(parts) > 1 else 0
104        part_2 = int(parts[2]) if len(parts) > 2 else 0
105    except (ValueError, IndexError) as error:
106        raise InvalidDataError(f"Malformed Bandcamp ID: {id_}") from error
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    _slug_to_fan_id: dict[str, int]  # unbounded; eviction would break back-navigation
116    throttler: ThrottlerManager = ThrottlerManager(
117        rate_limit=50,  # requests per period seconds
118        period=10,
119        initial_backoff=3,  # Bandcamp responds with Retry-After 3
120        retry_attempts=5,
121    )
122    top_tracks_limit: int
123
124    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
125        """Return Config entries to configure this provider."""
126        return (
127            CONF_ENTRY_UNOFFICIAL_PROVIDER,
128            ConfigEntry(
129                key=CONF_TOP_TRACKS_LIMIT,
130                type=ConfigEntryType.INTEGER,
131                required=False,
132                default_value=DEFAULT_TOP_TRACKS_LIMIT,
133                advanced=True,
134            ),
135        )
136
137    async def handle_async_init(self) -> None:
138        """Handle async init of the Bandcamp provider."""
139        identity = self.get_setup_value(CONF_IDENTITY)
140        self.top_tracks_limit = cast(
141            "int", self.config.get_value(CONF_TOP_TRACKS_LIMIT, DEFAULT_TOP_TRACKS_LIMIT)
142        )
143        self._client = BandcampAPIClient(
144            session=self.mass.http_session,
145            identity_token=identity,
146            default_retry_after=3,  # Bandcamp responds with Retry-After 3
147        )
148        self._converters = BandcampConverters(self.domain, self.instance_id)
149        self._slug_to_fan_id = {}
150
151        # The provider can function without login (search and streaming),
152        # but if credentials were explicitly configured, validate them now.
153        # A bad login fails hard so the user can fix it immediately;
154        # transient errors (rate limits, network) are logged and the provider
155        # continues since the login may still be valid.
156        if identity:
157            try:
158                await self._client.get_collection_summary()
159            except BandcampMustBeLoggedInError as error:
160                raise LoginFailed("Bandcamp login is invalid or expired.") from error
161            except BandcampAPIError as error:
162                self.logger.warning("Could not validate Bandcamp login: %s", error)
163
164    @property
165    def is_streaming_provider(self) -> bool:
166        """Return True if the provider is a streaming provider."""
167        return True
168
169    @throttle_with_retries
170    async def search(
171        self, search_query: str, media_types: list[MediaType], limit: int = 50
172    ) -> SearchResults:
173        """Perform search on music provider."""
174        results = SearchResults()
175        if not media_types:
176            return results
177
178        try:
179            search_results = await self._client.search(search_query)
180        except BandcampNotFoundError as error:
181            raise MediaNotFoundError("No results for Bandcamp search") from error
182        except BandcampRateLimitError as error:
183            raise RateLimited(
184                "Bandcamp rate limit reached", backoff_time=error.retry_after
185            ) from error
186        except BandcampAPIError as error:
187            raise InvalidDataError("Unexpected error during Bandcamp search") from error
188
189        for item in search_results[:limit]:
190            try:
191                if isinstance(item, SearchResultTrack) and MediaType.TRACK in media_types:
192                    results.tracks = [*results.tracks, self._converters.track_from_search(item)]
193                elif isinstance(item, SearchResultAlbum) and MediaType.ALBUM in media_types:
194                    results.albums = [*results.albums, self._converters.album_from_search(item)]
195                elif isinstance(item, SearchResultArtist) and MediaType.ARTIST in media_types:
196                    results.artists = [*results.artists, self._converters.artist_from_search(item)]
197            except BandcampAPIError as error:
198                self.logger.warning("Failed to convert search result item: %s", error)
199                continue
200
201        return results
202
203    async def get_recommendations(self) -> list[RecommendationFolder]:
204        """Get this provider's available recommendation rows, without items."""
205        if not self._client.identity:
206            return []
207        return [
208            RecommendationFolder(
209                item_id="feed",
210                provider=self.instance_id,
211                name="Bandcamp Feed",
212                translation_key="feed",
213                icon="mdi-rss",
214                is_playable=True,
215            ),
216            RecommendationFolder(
217                item_id="wishlist",
218                provider=self.instance_id,
219                name="Wishlist",
220                translation_key="wishlist",
221                icon="mdi-heart",
222                is_playable=True,
223            ),
224        ]
225
226    async def get_recommendation_items(
227        self, item_id: str
228    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
229        """
230        Get the items for a single recommendation row.
231
232        :param item_id: The item_id of the row, as returned by get_recommendations.
233        """
234        if not self._client.identity:
235            return UniqueList()
236        if item_id == "feed":
237            return UniqueList(await self._get_feed_tracks())
238        if item_id == "wishlist":
239            return UniqueList(await self._browse_person_content(None, CollectionType.WISHLIST))
240        return UniqueList()
241
242    @throttle_with_retries
243    async def _fetch_collection_page(
244        self,
245        collection_type: CollectionType,
246        older_than_token: str | None,
247        fan_id: int | None,
248    ) -> CollectionSummary:
249        """
250        Fetch a single page of collection items with throttling and retry.
251
252        :param collection_type: The type of collection to fetch.
253        :param older_than_token: Pagination cursor from the previous page.
254        :param fan_id: Fan ID to query. None = authenticated user.
255        """
256        try:
257            return await self._client.get_collection_items(
258                collection_type,
259                older_than_token=older_than_token,
260                fan_id=fan_id,
261            )
262        except BandcampRateLimitError as error:
263            raise RateLimited(
264                "Bandcamp rate limit reached", backoff_time=error.retry_after
265            ) from error
266
267    async def _get_all_collection_items(
268        self,
269        collection_type: CollectionType,
270        fan_id: int | None = None,
271    ) -> list[CollectionItem | FollowingItem | FanItem]:
272        """
273        Fetch all pages of a collection endpoint.
274
275        :param collection_type: The type of collection to fetch.
276        :param fan_id: Fan ID to query. None = authenticated user.
277        """
278        all_items: list[CollectionItem | FollowingItem | FanItem] = []
279        older_than_token: str | None = None
280        seen_tokens: set[str] = set()
281        while True:
282            page = await self._fetch_collection_page(collection_type, older_than_token, fan_id)
283            all_items.extend(page.items)
284            self.logger.debug(
285                "Fetched %d items for %s (has_more=%s, last_token=%s, total=%d)",
286                len(page.items),
287                collection_type.value,
288                page.has_more,
289                page.last_token,
290                len(all_items),
291            )
292            if not page.has_more or not page.last_token:
293                break
294            if page.last_token in seen_tokens:
295                self.logger.warning(
296                    "Pagination loop detected for %s: token %s already seen, stopping",
297                    collection_type.value,
298                    page.last_token,
299                )
300                break
301            seen_tokens.add(page.last_token)
302            older_than_token = page.last_token
303        return all_items
304
305    async def get_library_artists(self) -> AsyncGenerator[Artist]:
306        """Retrieve library artists from Bandcamp."""
307        if not self._client.identity:  # library requires identity
308            return
309
310        try:
311            items = await self._get_all_collection_items(CollectionType.COLLECTION)
312            band_ids = set()
313            for item in items:
314                if item.item_type == "band":
315                    band_ids.add(item.item_id)
316                elif item.item_type == "album":
317                    band_ids.add(item.band_id)
318
319            for band_id in band_ids:
320                yield await self.get_artist(str(band_id))
321                await asyncio.sleep(0)  # Yield control to avoid blocking
322
323        except BandcampMustBeLoggedInError as error:
324            self.logger.error("Error getting Bandcamp library artists: Wrong identity token.")
325            raise LoginFailed("Wrong Bandcamp identity token.") from error
326        except BandcampNotFoundError as error:
327            raise MediaNotFoundError("Bandcamp library artists returned no results") from error
328        except BandcampRateLimitError as error:
329            raise RateLimited(
330                "Bandcamp rate limit reached", backoff_time=error.retry_after
331            ) from error
332        except BandcampAPIError as error:
333            raise MediaNotFoundError("Failed to get library artists") from error
334
335    async def get_library_albums(self) -> AsyncGenerator[Album]:
336        """Retrieve library albums from Bandcamp."""
337        if not self._client.identity:  # library requires identity
338            return
339
340        try:
341            items = await self._get_all_collection_items(CollectionType.COLLECTION)
342            for item in items:
343                if item.item_type == "album":
344                    yield await self.get_album(f"{item.band_id}-{item.item_id}")
345                    await asyncio.sleep(0)  # Yield control to avoid blocking
346        except BandcampMustBeLoggedInError as error:
347            self.logger.error("Error getting Bandcamp library albums: Wrong identity token.")
348            raise LoginFailed("Wrong Bandcamp identity token.") from error
349        except BandcampNotFoundError as error:
350            raise MediaNotFoundError("Bandcamp library albums returned no results") from error
351        except BandcampRateLimitError as error:
352            raise RateLimited(
353                "Bandcamp rate limit reached", backoff_time=error.retry_after
354            ) from error
355        except BandcampAPIError as error:
356            raise MediaNotFoundError("Failed to get library albums") from error
357
358    async def get_library_tracks(self) -> AsyncGenerator[Track]:
359        """Retrieve library tracks from Bandcamp."""
360        if not self._client.identity:  # library requires identity
361            return
362
363        async for album in self.get_library_albums():
364            tracks = await self.get_album_tracks(album.item_id)
365            for track in tracks:
366                yield track
367                await asyncio.sleep(0)  # Yield control to avoid blocking
368
369    @use_cache(CACHE_METADATA)
370    @throttle_with_retries
371    async def get_artist(self, prov_artist_id: str) -> Artist:
372        """Get full artist details by id."""
373        try:
374            api_artist = await self._client.get_artist(prov_artist_id)
375            return self._converters.artist_from_api(api_artist)
376        except BandcampNotFoundError as error:
377            raise MediaNotFoundError(f"Artist {prov_artist_id} not found on Bandcamp") from error
378        except BandcampRateLimitError as error:
379            raise RateLimited(
380                "Bandcamp rate limit reached", backoff_time=error.retry_after
381            ) from error
382        except BandcampAPIError as error:
383            raise MediaNotFoundError(f"Failed to get artist {prov_artist_id}") from error
384
385    @use_cache(CACHE_METADATA)
386    @throttle_with_retries
387    async def get_album(self, prov_album_id: str) -> Album:
388        """Get full album details by id."""
389        artist_id, album_id, _ = split_id(prov_album_id)
390        try:
391            api_album = await self._client.get_album(artist_id, album_id)
392            return self._converters.album_from_api(api_album)
393        except BandcampNotFoundError as error:
394            raise MediaNotFoundError(f"Album {prov_album_id} not found on Bandcamp") from error
395        except BandcampRateLimitError as error:
396            raise RateLimited(
397                "Bandcamp rate limit reached", backoff_time=error.retry_after
398            ) from error
399        except BandcampAPIError as error:
400            raise MediaNotFoundError(f"Failed to get album {prov_album_id}") from error
401
402    @throttle_with_retries
403    async def _fetch_api_track(self, item_id: str) -> tuple[BCTrack, BCAlbum | None]:
404        """
405        Fetch a raw API track and its parent album by compound item ID.
406
407        Uses get_album when album_id is present (most tracks), falling back
408        to get_track for standalone tracks (album_id=0).
409
410        :param item_id: Compound track ID in the form artist_id-album_id-track_id.
411        """
412        artist_id, album_id, track_id = split_id(item_id)
413        if not track_id:
414            album_id, track_id = 0, album_id
415
416        try:
417            if album_id:
418                api_album = await self._client.get_album(artist_id, album_id)
419                api_track = next((t for t in api_album.tracks if t.id == track_id), None)
420                if not api_track:
421                    raise MediaNotFoundError(f"Track {item_id} not found in album on Bandcamp")
422                return api_track, api_album
423            return await self._client.get_track(artist_id, track_id), None
424        except BandcampMustBeLoggedInError as error:
425            raise LoginFailed("Bandcamp login is invalid or expired.") from error
426        except BandcampNotFoundError as error:
427            raise MediaNotFoundError(f"Track {item_id} not found on Bandcamp") from error
428        except BandcampRateLimitError as error:
429            raise RateLimited(
430                "Bandcamp rate limit reached", backoff_time=error.retry_after
431            ) from error
432        except BandcampAPIError as error:
433            raise MediaNotFoundError(f"Failed to get track {item_id}") from error
434
435    @use_cache(CACHE_METADATA)
436    async def get_track(self, prov_track_id: str) -> Track:
437        """Get full track details by id."""
438        api_track, api_album = await self._fetch_api_track(prov_track_id)
439        if api_album:
440            return self._converters.track_from_api(
441                track=api_track,
442                album_id=api_album.id,
443                album_name=api_album.title,
444                album_image_url=api_album.art_url,
445            )
446        return self._converters.track_from_api(
447            track=api_track,
448            album_id=api_track.album.id if api_track.album else None,
449            album_name=api_track.album.title if api_track.album else "",
450            album_image_url=api_track.album.art_url if api_track.album else "",
451        )
452
453    @use_cache(CACHE_METADATA)
454    @throttle_with_retries
455    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
456        """Get all tracks in an album."""
457        artist_id, album_id, _ = split_id(prov_album_id)
458        try:
459            api_album = await self._client.get_album(artist_id, album_id)
460            if api_album.tracks:
461                return [
462                    self._converters.track_from_api(
463                        track=track,
464                        album_id=album_id,
465                        album_name=api_album.title,
466                        album_image_url=api_album.art_url,
467                    )
468                    for track in api_album.tracks
469                    if track.streaming_url  # Only include tracks with streaming URLs
470                ]
471
472            return []
473
474        except BandcampNotFoundError as error:
475            raise MediaNotFoundError(
476                f"Album tracks for {prov_album_id} not found on Bandcamp"
477            ) from error
478        except BandcampRateLimitError as error:
479            raise RateLimited(
480                "Bandcamp rate limit reached", backoff_time=error.retry_after
481            ) from error
482        except BandcampAPIError as error:
483            raise MediaNotFoundError(f"Failed to get albums tracks for {prov_album_id}") from error
484
485    @use_cache(CACHE_METADATA)
486    @throttle_with_retries
487    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
488        """Get albums by an artist."""
489        try:
490            api_discography = await self._client.get_artist_discography(prov_artist_id)
491            return [
492                self._converters.album_from_discography_item(item)
493                for item in api_discography
494                if item.get("item_type") == "album" and item.get("item_id")
495            ]
496        except BandcampNotFoundError as error:
497            raise MediaNotFoundError(
498                f"Artist {prov_artist_id} albums not found on Bandcamp"
499            ) from error
500        except BandcampRateLimitError as error:
501            raise RateLimited(
502                "Bandcamp rate limit reached", backoff_time=error.retry_after
503            ) from error
504        except BandcampAPIError as error:
505            raise MediaNotFoundError(f"Failed to get albums for artist {prov_artist_id}") from error
506
507    @use_cache(CACHE_METADATA)
508    @throttle_with_retries
509    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
510        """Get top tracks of an artist."""
511        tracks: list[Track] = []
512        # get_artist_albums and get_album_tracks already handle exceptions and rate limiting
513        albums = await self.get_artist_albums(prov_artist_id)
514        albums.sort(key=lambda album: (album.year is None, album.year or 0), reverse=True)
515        for album in albums:
516            tracks.extend(await self.get_album_tracks(album.item_id))
517            if len(tracks) >= self.top_tracks_limit:
518                break
519
520        return tracks[: self.top_tracks_limit]
521
522    @throttle_with_retries
523    async def _fetch_feed(self) -> FeedResponse:
524        """Fetch the authenticated user's feed with throttling and retry."""
525        try:
526            return await self._client.get_feed()
527        except BandcampRateLimitError as error:
528            raise RateLimited(
529                "Bandcamp rate limit reached", backoff_time=error.retry_after
530            ) from error
531
532    async def _get_feed_tracks(self) -> list[Track]:
533        """Fetch and convert the streamable tracks from the user's feed."""
534        cache_key = "_feed_tracks"
535        cached = await self.mass.cache.get(cache_key, provider=self.instance_id, base_class=Track)
536        if cached is not None:
537            return cached  # type: ignore[no-any-return]
538        tracks: list[Track] = []
539        async with self._map_api_errors("Failed to get Bandcamp feed"):
540            feed = await self._fetch_feed()
541            tracks = [
542                self._converters.track_from_feed(track)
543                for track in feed.track_list
544                if track.streaming_url
545            ]
546        await self.mass.cache.set(
547            cache_key,
548            [t.to_dict() for t in tracks],
549            expiration=CACHE_USER_LISTS if tracks else CACHE_EMPTY_RESULTS,
550            provider=self.instance_id,
551        )
552        return tracks
553
554    async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
555        """
556        Browse this provider's items.
557
558        :param path: The path to browse, (e.g. provider_id://artists).
559        """
560        subpath = path.split("://")[1] if "://" in path else ""
561        # Filter empty segments from double-slashes or trailing slashes
562        path_parts = [p for p in subpath.split("/") if p]
563        base = f"{self.instance_id}://"
564
565        # Route fan/follower paths (supports arbitrary nesting depth)
566        if path_parts and path_parts[0] in (BROWSE_FANS, BROWSE_FOLLOWERS):
567            return await self._browse_person(path_parts, base)
568
569        # The feed/wishlist recommendation folders resolve to their tracks here when played;
570        # the folder's explicit path is dropped on deserialization, so play arrives as the
571        # bare item_id slug (e.g. ".../feed") rather than ".../recommendations/feed".
572        if path_parts == [BROWSE_FEED]:
573            return await self._get_feed_tracks()
574        if path_parts == [BROWSE_WISHLIST]:
575            return await self._browse_person_content(None, CollectionType.WISHLIST)
576        if path_parts == [BROWSE_FOLLOWING]:
577            return await self._browse_person_following(None)
578
579        # Delegate standard library paths and root listing to the base class
580        result = list(await super().browse(path))
581
582        # At root level, append custom browse folders when authenticated.
583        # These top-level folders query the authenticated user (person_id=None);
584        # person-specific paths (e.g. fans/42/wishlist) work without authentication
585        # since the Bandcamp API only requires identity for the "me" shortcut.
586        if not path_parts and self._client.identity:
587            # Collection is excluded — the user's own collection is the standard library.
588            for folder_id, folder_name in (
589                (BROWSE_WISHLIST, "Wishlist"),
590                (BROWSE_FOLLOWING, "Following"),
591                (BROWSE_FANS, "Fans"),
592                (BROWSE_FOLLOWERS, "Followers"),
593            ):
594                result.append(
595                    BrowseFolder(
596                        item_id=folder_id,
597                        provider=self.instance_id,
598                        path=base + folder_id,
599                        name=folder_name,
600                        translation_key=folder_id,
601                    )
602                )
603
604        return result
605
606    async def _browse_person(
607        self,
608        path_parts: list[str],
609        base: str,
610    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
611        """
612        Route person browse paths: fans/followers and their sub-categories.
613
614        Pattern: (fans|followers)[/{id}[/(collection|wishlist|following|fans|followers)]*]
615        """
616        # Top-level: authenticated user's fans or followers
617        if len(path_parts) == 1:
618            collection_type = (
619                CollectionType.FOLLOWING_FANS
620                if path_parts[0] == BROWSE_FANS
621                else CollectionType.FOLLOWERS
622            )
623            return await self._browse_person_people(collection_type, f"{base}{path_parts[0]}")
624
625        tail = path_parts[-1]
626
627        # Path ends with a person identifier (numeric ID or slug) → show their 5 sub-folders
628        person_id = await self._resolve_person_segment(tail)
629        if person_id is not None:
630            if person_id <= 0:
631                raise InvalidDataError(f"Invalid person ID in browse path: {tail}")
632            return self._browse_person_root(person_id, f"{base}{'/'.join(path_parts)}")
633
634        # Path ends with a sub-category → person identifier is second-to-last
635        if len(path_parts) < 2:
636            raise InvalidDataError(f"Invalid browse path: {base}{'/'.join(path_parts)}")
637        person_id = await self._resolve_person_segment(path_parts[-2])
638        if person_id is None:
639            raise InvalidDataError(f"Invalid browse path: {base}{'/'.join(path_parts)}")
640        if person_id <= 0:
641            raise InvalidDataError(f"Invalid person ID in browse path: {path_parts[-2]}")
642
643        route = PERSON_SUB_ROUTES.get(tail)
644        if route is None:
645            raise InvalidDataError(f"Unknown browse sub-category: {tail}")
646
647        method_kind, collection_type = route
648        if method_kind == "content":
649            return await self._browse_person_content(person_id, collection_type)
650        if method_kind == "following":
651            return await self._browse_person_following(person_id)
652        if method_kind != "people":
653            raise InvalidDataError(f"Unknown route kind: {method_kind}")
654        canon = "/".join(path_parts)
655        return await self._browse_person_people(collection_type, f"{base}{canon}", person_id)
656
657    # --- Person browse helpers (fans, followers, and social graph traversal) ---
658
659    async def _resolve_person_segment(self, segment: str) -> int | None:
660        """
661        Resolve a path segment to a fan_id.
662
663        Checks the slug→fan_id cache first, then tries numeric parse.
664        For unknown slugs, rebuilds the cache from fan/follower lists and retries.
665        Returns None if the segment is neither a known slug nor a valid int,
666        or if it is a known sub-route name (e.g. "collection", "wishlist").
667        """
668        if segment in self._slug_to_fan_id:
669            return self._slug_to_fan_id[segment]
670        try:
671            return int(segment)
672        except ValueError:
673            pass
674        # Known sub-route names are structural, not user slugs
675        if segment in PERSON_SUB_ROUTES:
676            return None
677        # Slug not in cache and not numeric — rebuild from parent lists and retry
678        await self._rebuild_slug_cache()
679        return self._slug_to_fan_id.get(segment)
680
681    async def _rebuild_slug_cache(self) -> None:
682        """Re-fetch fan/follower lists to rebuild the slug→fan_id map."""
683        base = f"{self.instance_id}://"
684        for collection_type, folder_id in (
685            (CollectionType.FOLLOWING_FANS, BROWSE_FANS),
686            (CollectionType.FOLLOWERS, BROWSE_FOLLOWERS),
687        ):
688            with suppress(Exception):
689                await self._browse_person_people(collection_type, f"{base}{folder_id}")
690
691    @staticmethod
692    def _fan_slug(person: FanItem) -> str | None:
693        """
694        Extract the URL slug from a FanItem's url.
695
696        e.g. "https://bandcamp.com/teancom" → "teancom"
697        """
698        if person.url:
699            slug: str = person.url.rstrip("/").rsplit("/", 1)[-1]
700            if slug:
701                return slug
702        return None
703
704    def _people_to_folders(self, items: list[FanItem], base_path: str) -> list[BrowseFolder]:
705        """Convert a list of people to BrowseFolder items with thumbnails."""
706        folders: list[BrowseFolder] = []
707        for person in items:
708            slug = self._fan_slug(person)
709            if slug:
710                self._slug_to_fan_id[slug] = person.fan_id
711            path_segment = slug or str(person.fan_id)
712            folder = BrowseFolder(
713                item_id=f"person_{person.fan_id}",
714                provider=self.instance_id,
715                path=f"{base_path}/{path_segment}",
716                name=person.name or f"User {person.fan_id}",
717            )
718            if person.image_url:
719                folder.image = MediaItemImage(
720                    type=ImageType.THUMB,
721                    path=person.image_url,
722                    provider=self.instance_id,
723                    remotely_accessible=True,
724                )
725            folders.append(folder)
726        return folders
727
728    def _browse_person_root(self, person_id: int, base_path: str) -> list[BrowseFolder]:
729        """Return the 5 sub-folders for a person's profile."""
730        return [
731            BrowseFolder(
732                item_id=f"person_{person_id}_{sub_id}",
733                provider=self.instance_id,
734                path=f"{base_path}/{sub_id}",
735                name=name,
736                translation_key=sub_id,
737            )
738            for sub_id, name in PERSON_SUB_FOLDERS
739        ]
740
741    @asynccontextmanager
742    async def _map_api_errors(self, context: str) -> AsyncIterator[None]:
743        """Map Bandcamp API exceptions to MusicAssistant exceptions."""
744        try:
745            yield
746        except BandcampMustBeLoggedInError as error:
747            raise LoginFailed("Wrong Bandcamp identity token.") from error
748        except BandcampRateLimitError as error:
749            raise RateLimited(
750                "Bandcamp rate limit reached", backoff_time=error.retry_after
751            ) from error
752        except BandcampAPIError as error:
753            raise MediaNotFoundError(context) from error
754
755    @staticmethod
756    def _deserialize_content_item(item: dict[str, object]) -> Album | Track:
757        """Deserialize a cached content item back to its model type."""
758        media_type = item.get("media_type")
759        if media_type == MediaType.ALBUM:
760            return Album.from_dict(item)
761        if media_type == MediaType.TRACK:
762            return Track.from_dict(item)
763        msg = f"Unexpected media_type in cached content item: {media_type}"
764        raise ValueError(msg)
765
766    @throttle_with_retries
767    async def _browse_person_content(
768        self, person_id: int | None, collection_type: CollectionType
769    ) -> list[Album | Track]:
770        """
771        Fetch a person's collection or wishlist items.
772
773        :param person_id: Person to query. None = authenticated user.
774        """
775        cache_key = f"_browse_person_content_{person_id}_{collection_type.value}"
776        cached = await self.mass.cache.get(cache_key, provider=self.instance_id)
777        if cached is not None:
778            try:
779                return [self._deserialize_content_item(item) for item in cached]
780            except LookupError, ValueError, UnserializableDataError, InvalidDataError:
781                self.logger.warning("Stale cache for %s, fetching fresh", cache_key)
782        results: list[Album | Track] = []
783        context = f"Failed to get {collection_type.value} for person {person_id}"
784        async with self._map_api_errors(context):
785            items = await self._get_all_collection_items(collection_type, fan_id=person_id)
786            for item in items:
787                with suppress(MediaNotFoundError):
788                    if item.item_type == "album":
789                        results.append(await self.get_album(f"{item.band_id}-{item.item_id}"))
790                    elif item.item_type == "track":
791                        results.append(await self.get_track(f"{item.band_id}-0-{item.item_id}"))
792        await self.mass.cache.set(
793            cache_key,
794            [item.to_dict() for item in results],
795            expiration=CACHE_USER_LISTS if results else CACHE_EMPTY_RESULTS,
796            provider=self.instance_id,
797        )
798        return results
799
800    @throttle_with_retries
801    async def _browse_person_following(self, person_id: int | None) -> list[Artist]:
802        """
803        Fetch a person's followed artists.
804
805        :param person_id: Person to query. None = authenticated user.
806        """
807        cache_key = f"_browse_person_following_{person_id}"
808        cached = await self.mass.cache.get(cache_key, provider=self.instance_id, base_class=Artist)
809        if cached is not None:
810            return cached  # type: ignore[no-any-return]
811        artists: list[Artist] = []
812        async with self._map_api_errors(f"Failed to get following for person {person_id}"):
813            collection = await self._get_all_collection_items(
814                CollectionType.FOLLOWING, fan_id=person_id
815            )
816            for item in collection:
817                try:
818                    artists.append(await self.get_artist(item.band_id))
819                except MediaNotFoundError:
820                    self.logger.warning(
821                        "Artist not found for band_id %s (%s)", item.band_id, item.name
822                    )
823        await self.mass.cache.set(
824            cache_key,
825            [a.to_dict() for a in artists],
826            expiration=CACHE_USER_LISTS if artists else CACHE_EMPTY_RESULTS,
827            provider=self.instance_id,
828        )
829        return artists
830
831    @throttle_with_retries
832    async def _browse_person_people(
833        self,
834        collection_type: CollectionType,
835        base_path: str,
836        person_id: int | None = None,
837    ) -> list[BrowseFolder]:
838        """
839        Fetch a person's fans or followers as browsable folders.
840
841        :param collection_type: FOLLOWING_FANS or FOLLOWERS.
842        :param base_path: Browse path prefix for the resulting folder links.
843        :param person_id: Person to query. None = authenticated user.
844        """
845        # base_path included intentionally: folder links differ per navigation path.
846        cache_key = f"_browse_person_people_{person_id}_{collection_type.value}_{base_path}"
847        cached = await self.mass.cache.get(
848            cache_key, provider=self.instance_id, base_class=BrowseFolder
849        )
850        if cached is not None:
851            for folder in cached:
852                segment = folder.path.rstrip("/").rsplit("/", 1)[-1]
853                fan_id_str = folder.item_id.removeprefix("person_")
854                with suppress(ValueError):
855                    self._slug_to_fan_id[segment] = int(fan_id_str)
856            return cached  # type: ignore[no-any-return]
857        context = f"Failed to get {collection_type.value} for person {person_id}"
858        async with self._map_api_errors(context):
859            collection = await self._get_all_collection_items(collection_type, fan_id=person_id)
860            folders = self._people_to_folders(collection, base_path)
861        await self.mass.cache.set(
862            cache_key,
863            [f.to_dict() for f in folders],
864            expiration=CACHE_USER_LISTS if folders else CACHE_EMPTY_RESULTS,
865            provider=self.instance_id,
866        )
867        return folders
868
869    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
870        """
871        Return the content details for the given track.
872
873        Fetches fresh from the Bandcamp API since streaming URLs may expire.
874        """
875        api_track, _ = await self._fetch_api_track(item_id)
876
877        streaming_url, bitrate, content_type = self._converters.streaming_url_from_api(
878            api_track.streaming_url or {}
879        )
880        if not streaming_url:
881            raise MediaNotFoundError(f"No streaming URL found for track {item_id}")
882
883        return StreamDetails(
884            item_id=item_id,
885            provider=self.instance_id,
886            audio_format=AudioFormat(
887                content_type=content_type,
888                bit_rate=bitrate,
889            ),
890            stream_type=StreamType.HTTP,
891            media_type=media_type,
892            path=streaming_url,
893            can_seek=True,
894            allow_seek=True,
895        )
896