music-assistant-server

1"""Converters for Bandcamp API models to Music Assistant models."""
2
3from contextlib import suppress
4from datetime import UTC, datetime
5from typing import TYPE_CHECKING, TypedDict
6
7from music_assistant_models.enums import ContentType, ImageType, MediaType
8from music_assistant_models.media_items import Album as MAAlbum
9from music_assistant_models.media_items import Artist as MAArtist
10from music_assistant_models.media_items import (
11    AudioFormat,
12    ItemMapping,
13    MediaItemImage,
14    ProviderMapping,
15    UniqueList,
16)
17from music_assistant_models.media_items import Track as MATrack
18
19if TYPE_CHECKING:
20    from bandcamp_async_api.models import BCAlbum as APIAlbum
21    from bandcamp_async_api.models import BCArtist as APIArtist
22    from bandcamp_async_api.models import BCTrack as APITrack
23    from bandcamp_async_api.models import (
24        FeedTrack,
25        SearchResultAlbum,
26        SearchResultArtist,
27        SearchResultTrack,
28    )
29
30from ._ids import make_artist_id, slugify_performer
31
32
33class DiscographyItem(TypedDict, total=False):
34    """Raw discography item dict from the band_details API."""
35
36    item_id: int
37    item_type: str
38    band_id: int
39    title: str
40    artist_name: str | None
41    band_name: str | None
42    art_id: int | None
43    release_date: str | None
44    is_purchasable: bool
45
46
47class BandcampConverters:
48    """Converters for Bandcamp API models to Music Assistant models."""
49
50    def __init__(self, domain: str, instance_id: str):
51        """Initialize converters with provider information."""
52        self.domain = domain
53        self.instance_id = instance_id
54
55    @staticmethod
56    def streaming_url_from_api(
57        streaming_info: dict[str, str],
58    ) -> tuple[str | None, int | None, ContentType]:
59        """
60        Parse streaming URL info.
61
62        :param streaming_info: Dict of format keys to URLs from the Bandcamp API.
63        """
64        # Extract streaming URL with priority: mp3-v0 > mp3-320 > mp3-128
65        bitrate = None
66        streaming_url = None
67        content_type = ContentType.MP3
68        if "mp3-v0" in streaming_info:
69            streaming_url = streaming_info["mp3-v0"]
70        elif "mp3-320" in streaming_info:
71            streaming_url = streaming_info["mp3-320"]
72            bitrate = 320
73        elif "mp3-128" in streaming_info:
74            streaming_url = streaming_info["mp3-128"]
75            bitrate = 128
76        elif streaming_info:
77            streaming_url = next(iter(streaming_info.values()))
78            content_type = ContentType.UNKNOWN
79        return streaming_url, bitrate, content_type
80
81    def track_from_search(
82        self, item: SearchResultTrack, *, artist_item_id: str | None = None
83    ) -> MATrack:
84        """
85        Convert a Bandcamp search track.
86
87        :param artist_item_id: Optional resolved artist item ID.
88        """
89        track_id = f"{item.artist_id}-{item.album_id or 0}-{item.id}"
90        artist_item_id = artist_item_id or make_artist_id(item.artist_id, item.artist_name)
91        return MATrack(
92            item_id=track_id,
93            provider=self.instance_id,
94            name=item.name,
95            artists=UniqueList(
96                [
97                    ItemMapping(
98                        media_type=MediaType.ARTIST,
99                        item_id=artist_item_id,
100                        provider=self.instance_id,
101                        name=item.artist_name,
102                    )
103                ]
104            ),
105            album=(
106                ItemMapping(
107                    media_type=MediaType.ALBUM,
108                    item_id=f"{item.artist_id}-{item.album_id or 0}",
109                    provider=self.instance_id,
110                    name=item.album_name,
111                )
112                if item.album_id
113                else None
114            ),
115            provider_mappings={
116                ProviderMapping(
117                    item_id=track_id,
118                    provider_domain=self.domain,
119                    provider_instance=self.instance_id,
120                    url=item.url,
121                )
122            },
123        )
124
125    def album_from_search(
126        self, item: SearchResultAlbum, *, artist_item_id: str | None = None
127    ) -> MAAlbum:
128        """
129        Convert a Bandcamp search album.
130
131        :param artist_item_id: Optional resolved artist item ID.
132        """
133        album_id = f"{item.artist_id}-{item.id}"
134        artist_item_id = artist_item_id or make_artist_id(item.artist_id, item.artist_name)
135        output = MAAlbum(
136            item_id=album_id,
137            provider=self.instance_id,
138            name=item.name,
139            uri=item.url,
140            artists=UniqueList(
141                [
142                    ItemMapping(
143                        media_type=MediaType.ARTIST,
144                        item_id=artist_item_id,
145                        provider=self.instance_id,
146                        name=item.artist_name,
147                        uri=item.artist_url,
148                    )
149                ]
150            ),
151            provider_mappings={
152                ProviderMapping(
153                    item_id=album_id,
154                    provider_domain=self.domain,
155                    provider_instance=self.instance_id,
156                    url=item.url,
157                )
158            },
159        )
160        output.metadata.add_image(
161            MediaItemImage(
162                type=ImageType.THUMB,
163                path=item.image_url,
164                provider=self.instance_id,
165                remotely_accessible=True,
166            )
167        )
168        return output
169
170    def artist_from_search(self, item: SearchResultArtist) -> MAArtist:
171        """Create an Artist from new API SearchResultArtist."""
172        output = MAArtist(
173            item_id=str(item.id),
174            provider=self.instance_id,
175            name=item.name,
176            uri=item.url,
177            provider_mappings={
178                ProviderMapping(
179                    item_id=str(item.id),
180                    provider_domain=self.domain,
181                    provider_instance=self.instance_id,
182                    url=item.url,
183                )
184            },
185        )
186        output.metadata.genres = item.tags
187        if item.url:
188            output.metadata.description = item.url
189        output.metadata.add_image(
190            MediaItemImage(
191                type=ImageType.THUMB,
192                path=item.image_url,
193                provider=self.instance_id,
194                remotely_accessible=True,
195            )
196        )
197        return output
198
199    def track_from_api(
200        self,
201        track: APITrack,
202        album_id: str | int | None = None,
203        album_name: str = "",
204        album_image_url: str = "",
205        *,
206        tralbum_artist: str | None = None,
207        artist_item_id: str | None = None,
208    ) -> MATrack:
209        """
210        Convert a Bandcamp API track.
211
212        :param tralbum_artist: Optional per-album performer credit.
213        :param artist_item_id: Optional resolved artist item ID.
214        """
215        album_id = album_id or 0
216        _, bitrate, content_type = self.streaming_url_from_api(track.streaming_url or {})
217        band_name = track.artist.name
218        display_name = tralbum_artist or band_name
219        artist_item_id = artist_item_id or _resolve_artist_id(
220            band_id=track.artist.id, performer=tralbum_artist, band_name=band_name
221        )
222        output = MATrack(
223            item_id=f"{track.artist.id}-{album_id}-{track.id}",
224            provider=self.instance_id,
225            name=track.title,
226            artists=UniqueList(
227                [
228                    ItemMapping(
229                        media_type=MediaType.ARTIST,
230                        item_id=artist_item_id,
231                        provider=self.instance_id,
232                        name=display_name,
233                    )
234                ]
235            ),
236            disc_number=0,
237            duration=track.duration,
238            provider_mappings={
239                ProviderMapping(
240                    item_id=f"{track.artist.id}-{album_id}-{track.id}",
241                    provider_domain=self.domain,
242                    provider_instance=self.instance_id,
243                    url=track.url,
244                    audio_format=AudioFormat(
245                        content_type=content_type,
246                        bit_rate=bitrate,
247                    ),
248                )
249            },
250        )
251        if track.track_number is not None:
252            output.track_number = track.track_number
253
254        if album_id:
255            output.album = ItemMapping(
256                media_type=MediaType.ALBUM,
257                item_id=f"{track.artist.id}-{album_id}",
258                provider=self.instance_id,
259                name=album_name,
260            )
261        elif hasattr(track, "album") and track.album:
262            # If the track has an album attribute, use that information
263            output.album = ItemMapping(
264                media_type=MediaType.ALBUM,
265                item_id=f"{track.artist.id}-{track.album.id}",
266                provider=self.instance_id,
267                name=track.album.title,
268            )
269        output.metadata.lyrics = track.lyrics
270        if album_image_url:
271            output.metadata.add_image(
272                MediaItemImage(
273                    type=ImageType.THUMB,
274                    path=album_image_url,
275                    provider=self.instance_id,
276                    remotely_accessible=True,
277                )
278            )
279        return output
280
281    def track_from_feed(self, track: FeedTrack) -> MATrack:
282        """Convert a feed track_list entry to MA Track format."""
283        album_id = track.album_id or 0
284        item_id = f"{track.band_id}-{album_id}-{track.track_id}"
285        _, bitrate, content_type = self.streaming_url_from_api(track.streaming_url or {})
286        output = MATrack(
287            item_id=item_id,
288            provider=self.instance_id,
289            name=track.title,
290            duration=int(track.duration) if track.duration else 0,
291            artists=UniqueList(
292                [
293                    ItemMapping(
294                        media_type=MediaType.ARTIST,
295                        item_id=str(track.band_id),
296                        provider=self.instance_id,
297                        name=track.band_name,
298                    )
299                ]
300            ),
301            provider_mappings={
302                ProviderMapping(
303                    item_id=item_id,
304                    provider_domain=self.domain,
305                    provider_instance=self.instance_id,
306                    url=track.track_url,
307                    audio_format=AudioFormat(content_type=content_type, bit_rate=bitrate),
308                )
309            },
310        )
311        if track.track_num is not None:
312            output.track_number = track.track_num
313        if album_id:
314            output.album = ItemMapping(
315                media_type=MediaType.ALBUM,
316                item_id=f"{track.band_id}-{album_id}",
317                provider=self.instance_id,
318                name=track.album_title or "",
319            )
320        if track.art_id:
321            output.metadata.add_image(
322                MediaItemImage(
323                    type=ImageType.THUMB,
324                    path=f"https://f4.bcbits.com/img/a{track.art_id}_0.jpg",
325                    provider=self.instance_id,
326                    remotely_accessible=True,
327                )
328            )
329        return output
330
331    def artist_from_api(self, artist: APIArtist) -> MAArtist:
332        """Convert an API Artist object to MA Artist format."""
333        output = MAArtist(
334            item_id=str(artist.id),
335            uri=artist.url,
336            provider=self.instance_id,
337            name=artist.name,
338            provider_mappings={
339                ProviderMapping(
340                    item_id=str(artist.id),
341                    provider_domain=self.domain,
342                    provider_instance=self.instance_id,
343                    url=artist.url,
344                )
345            },
346        )
347        output.metadata.description = f"{artist.url}\n{artist.bio or ''}".strip()
348        output.metadata.add_image(
349            MediaItemImage(
350                type=ImageType.THUMB,
351                path=artist.image_url,
352                provider=self.instance_id,
353                remotely_accessible=True,
354            )
355        )
356        return output
357
358    def album_from_discography_item(
359        self, item: DiscographyItem, *, artist_item_id: str | None = None
360    ) -> MAAlbum:
361        """
362        Convert a raw discography dict to MA Album format.
363
364        Discography items come from the band_details API and contain summary
365        data (title, art_id, release_date string) without full album details.
366        Fields not available from the discography endpoint (url, description)
367        are omitted and populated later when get_album fetches full details.
368
369        :param artist_item_id: Pre-resolved artist item_id; falls back to
370            slug-based resolution if omitted.
371        """
372        band_id = item.get("band_id", 0)
373        item_id = item.get("item_id", 0)
374        album_id = f"{band_id}-{item_id}"
375        # `artist_name` (when set) is the per-album performer; `band_name`
376        # is the page owner. They differ on label-released albums.
377        performer = item.get("artist_name")
378        band_name = item.get("band_name") or ""
379        display_name = performer or band_name
380        artist_item_id = artist_item_id or _resolve_artist_id(
381            band_id=band_id, performer=performer, band_name=band_name
382        )
383
384        # Build art URL from art_id (matches _build_art_url in parsers.py)
385        art_id = item.get("art_id")
386        art_url = f"https://f4.bcbits.com/img/a{art_id}_0.jpg" if art_id else ""
387
388        # Parse year from release_date string like "21 Feb 2020 00:00:00 GMT"
389        year = None
390        release_date = item.get("release_date")
391        if release_date:
392            with suppress(ValueError, TypeError, IndexError):
393                # Format: "21 Feb 2020 00:00:00 GMT" — extract year directly
394                year = int(release_date.split()[2])
395
396        output = MAAlbum(
397            item_id=album_id,
398            provider=self.instance_id,
399            name=item.get("title") or "",
400            artists=UniqueList(
401                [
402                    ItemMapping(
403                        media_type=MediaType.ARTIST,
404                        item_id=artist_item_id,
405                        provider=self.instance_id,
406                        name=display_name,
407                    )
408                ]
409            ),
410            provider_mappings={
411                ProviderMapping(
412                    item_id=album_id,
413                    provider_domain=self.domain,
414                    provider_instance=self.instance_id,
415                )
416            },
417            year=year,
418        )
419        if art_url:
420            output.metadata.add_image(
421                MediaItemImage(
422                    type=ImageType.THUMB,
423                    path=art_url,
424                    provider=self.instance_id,
425                    remotely_accessible=True,
426                )
427            )
428        return output
429
430    def album_from_api(self, album: APIAlbum, *, artist_item_id: str | None = None) -> MAAlbum:
431        """
432        Convert a Bandcamp API album.
433
434        :param artist_item_id: Optional resolved artist item ID.
435        """
436        album_id = f"{album.artist.id}-{album.id}"
437        band_name = album.artist.name
438        display_name = album.tralbum_artist or band_name
439        artist_item_id = artist_item_id or _resolve_artist_id(
440            band_id=album.artist.id, performer=album.tralbum_artist, band_name=band_name
441        )
442        output = MAAlbum(
443            item_id=album_id,
444            provider=self.instance_id,
445            name=album.title,
446            artists=UniqueList(
447                [
448                    ItemMapping(
449                        media_type=MediaType.ARTIST,
450                        item_id=artist_item_id,
451                        provider=self.instance_id,
452                        name=display_name,
453                        image=MediaItemImage(
454                            path=album.art_url,
455                            type=ImageType.THUMB,
456                            provider=self.instance_id,
457                            remotely_accessible=True,
458                        ),
459                    )
460                ]
461            ),
462            provider_mappings={
463                ProviderMapping(
464                    item_id=album_id,
465                    provider_domain=self.domain,
466                    provider_instance=self.instance_id,
467                    url=album.url,
468                )
469            },
470            year=datetime.fromtimestamp(album.release_date, tz=UTC).year
471            if album.release_date
472            else None,
473        )
474        output.metadata.add_image(
475            MediaItemImage(
476                type=ImageType.THUMB,
477                path=album.art_url,
478                provider=self.instance_id,
479                remotely_accessible=True,
480            )
481        )
482        output.metadata.description = f"{album.url}\n{album.about or ''}".strip()
483        return output
484
485    def synthetic_artist(
486        self,
487        band_id: int,
488        performer_name: str,
489        *,
490        url: str | None = None,
491        image_url: str | None = None,
492    ) -> MAArtist:
493        """
494        Build an artist for a performer without a Bandcamp page.
495
496        :param band_id: Hosting Bandcamp artist ID.
497        :param performer_name: Performer display name.
498        :param url: Optional hosting-page URL.
499        :param image_url: Optional artist artwork URL.
500        """
501        item_id = make_artist_id(band_id, performer_name)
502        output = MAArtist(
503            item_id=item_id,
504            provider=self.instance_id,
505            name=performer_name,
506            provider_mappings={
507                ProviderMapping(
508                    item_id=item_id,
509                    provider_domain=self.domain,
510                    provider_instance=self.instance_id,
511                    url=url,
512                )
513            },
514        )
515        if url:
516            output.metadata.description = url
517        if image_url:
518            output.metadata.add_image(
519                MediaItemImage(
520                    type=ImageType.THUMB,
521                    path=image_url,
522                    provider=self.instance_id,
523                    remotely_accessible=True,
524                )
525            )
526        return output
527
528
529def _resolve_artist_id(
530    *,
531    band_id: int | str,
532    performer: str | None,
533    band_name: str | None,
534) -> str:
535    """
536    Resolve the artist item ID for a performer credit.
537
538    :param band_id: Hosting Bandcamp artist ID.
539    :param performer: Performer credit, if present.
540    :param band_name: Hosting artist name.
541    :returns: A real or synthetic Music Assistant artist ID.
542    """
543    if (
544        not performer
545        or not band_name
546        or slugify_performer(performer) == slugify_performer(band_name)
547    ):
548        return str(band_id)
549    return make_artist_id(band_id, performer)
550