music-assistant-server

1"""
2Radio stream artwork lookup for the Metadata Controller.
3
4Provides the RadioArtworkMixin, mixed into the MetaDataController, which resolves
5artist/track artwork for radio streams by matching the station's now-playing
6metadata against the local library and MusicBrainz/online metadata providers.
7"""
8
9from __future__ import annotations
10
11from time import time
12from typing import TYPE_CHECKING, Any, cast
13
14from music_assistant_models.enums import ExternalID, ImageType, ProviderFeature
15from music_assistant_models.errors import (
16    InvalidDataError,
17    MediaNotFoundError,
18    MusicAssistantError,
19    ProviderUnavailableError,
20    ResourceTemporarilyUnavailable,
21)
22from music_assistant_models.media_items import (
23    Album,
24    Artist,
25    ItemMapping,
26    MediaItemImage,
27    MediaItemMetadata,
28    Track,
29)
30from music_assistant_models.streamdetails import StreamMetadata
31from music_assistant_models.unique_list import UniqueList
32
33from music_assistant.helpers.compare import compare_strings, create_safe_string
34from music_assistant.helpers.tags import split_artists
35from music_assistant.helpers.util import parse_title_and_version
36
37from .constants import (
38    AD_DETECTION_PHRASES,
39    CACHE_CATEGORY_RADIO_ARTWORK,
40    CACHE_EXPIRATION_RADIO_ARTWORK,
41    CACHE_EXPIRATION_RADIO_ARTWORK_MISS,
42    CONF_ENABLE_RADIO_METADATA_LOOKUP,
43)
44
45if TYPE_CHECKING:
46    import logging
47
48    from music_assistant_models.streamdetails import StreamDetails
49
50    from music_assistant import MusicAssistant
51    from music_assistant.models.metadata_provider import MetadataProvider
52    from music_assistant.providers.musicbrainz import MusicbrainzProvider
53    from music_assistant.providers.musicbrainz.models import MusicBrainzReleaseGroup
54
55
56class RadioArtworkMixin:
57    """
58    Radio stream artwork functionality for the MetaDataController.
59
60    Expects to be mixed with a class providing ``mass``, ``logger``, ``domain``,
61    the ``providers`` property and the ``get_image_url`` method.
62    """
63
64    if TYPE_CHECKING:
65        mass: MusicAssistant
66        logger: logging.Logger
67        domain: str
68
69        @property
70        def providers(self) -> list[MetadataProvider]: ...  # noqa: D102
71
72        def get_image_url(  # noqa: D102
73            self,
74            image: MediaItemImage,
75            size: int = 0,
76            prefer_proxy: bool = False,
77            image_format: str | None = None,
78            prefer_stream_server: bool = False,
79        ) -> str: ...
80
81    async def get_track_metadata_by_name(
82        self,
83        artist_name: str,
84        track_name: str,
85        album_name: str | None = None,
86    ) -> tuple[MediaItemMetadata | None, str | None, str | None, str | None]:
87        """
88        Search for track/artist metadata by name.
89
90        Checks library first for immediate results, then falls back to
91        MusicBrainz for external metadata lookups.
92
93        :param artist_name: Artist name to search for.
94        :param track_name: Track title to search for.
95        :param album_name: Album announced by the stream, used to refine which artwork is chosen.
96        :returns: Tuple of (metadata, source_description, corrected_artist, corrected_track).
97        """
98        # Clean track name by stripping version suffixes and featuring credits
99        clean_track_name, _ = parse_title_and_version(track_name, strip_for_search=True)
100
101        # Check library track first - fast, no API calls, respects user-curated images
102        if metadata := await self._get_library_track_metadata(artist_name, clean_track_name):
103            return metadata, "library track", artist_name, clean_track_name
104
105        # Use MusicBrainz to get IDs for accurate external metadata lookups
106        musicbrainz_provider = self.mass.get_provider("musicbrainz")
107        if not musicbrainz_provider:
108            # No MusicBrainz, try library artist as fallback
109            if metadata := await self._get_library_artist_metadata(artist_name):
110                return metadata, f"library artist '{artist_name}'", artist_name, clean_track_name
111            return None, None, None, None
112        musicbrainz: MusicbrainzProvider = cast("MusicbrainzProvider", musicbrainz_provider)
113
114        mb_result, swapped = await self._search_musicbrainz_with_variants(
115            musicbrainz, artist_name, clean_track_name
116        )
117
118        if not mb_result:
119            self.logger.debug("No MusicBrainz match for '%s - %s'", artist_name, clean_track_name)
120            # No MB match, try library artist as fallback
121            if metadata := await self._get_library_artist_metadata(artist_name):
122                return metadata, f"library artist '{artist_name}'", artist_name, clean_track_name
123            return None, None, None, None
124
125        mb_artist, mb_release_groups = mb_result
126        if swapped:
127            # Swap the variables so subsequent lookups use the correct order
128            artist_name, clean_track_name = clean_track_name, artist_name
129            self.logger.debug(
130                "MusicBrainz matched with swapped artist/track: '%s - %s'",
131                artist_name,
132                clean_track_name,
133            )
134
135        # Prefer single artwork (exact track art), then fall back to album artwork
136        singles = [rg for rg in mb_release_groups if rg.primary_type == "Single"]
137        albums = [rg for rg in mb_release_groups if rg.primary_type == "Album"]
138
139        # When the station told us the album, move a matching release group to the front so
140        # the cover reflects the broadcast release rather than an arbitrary one. This only
141        # reorders within each type; singles still take precedence over albums.
142        if album_name:
143            singles = self._prioritize_release_groups(singles, album_name)
144            albums = self._prioritize_release_groups(albums, album_name)
145
146        for mb_release_group in singles:
147            if result := await self._get_release_group_artwork(mb_release_group):
148                thumb, provider_name = result
149                return (
150                    thumb,
151                    f"single '{mb_release_group.title}' via {provider_name}",
152                    artist_name,
153                    clean_track_name,
154                )
155
156        if singles:
157            self.logger.debug(
158                "No artwork found for single release of '%s - %s', trying album artwork",
159                artist_name,
160                clean_track_name,
161            )
162
163        for mb_release_group in albums:
164            if result := await self._get_release_group_artwork(mb_release_group):
165                thumb, provider_name = result
166                return (
167                    thumb,
168                    f"album '{mb_release_group.title}' via {provider_name}",
169                    artist_name,
170                    clean_track_name,
171                )
172
173        # Log when falling back to artist artwork
174        self.logger.debug(
175            "No album artwork for '%s - %s', trying artist artwork",
176            artist_name,
177            clean_track_name,
178        )
179
180        # Check library for artist before external lookup
181        if metadata := await self._get_library_artist_metadata(mb_artist.name):
182            return metadata, f"library artist '{mb_artist.name}'", artist_name, clean_track_name
183
184        # Fall back to external artist artwork
185        temp_artist = Artist(
186            item_id="temp",
187            provider="temp",
188            name=mb_artist.name,
189            provider_mappings=set(),
190        )
191        temp_artist.mbid = mb_artist.id
192        for provider in self.providers:
193            if ProviderFeature.ARTIST_METADATA not in provider.supported_features:
194                continue
195            try:
196                if artist_metadata := await provider.get_artist_metadata(temp_artist):
197                    if artist_thumb := self._get_thumb_image(artist_metadata):
198                        return (
199                            artist_thumb,
200                            f"artist '{mb_artist.name}' via {provider.name}",
201                            artist_name,
202                            clean_track_name,
203                        )
204            except (
205                ProviderUnavailableError,
206                ResourceTemporarilyUnavailable,
207                InvalidDataError,
208            ):
209                pass
210
211        return None, None, None, None
212
213    def get_radio_stream_station_image(self, streamdetails: StreamDetails) -> str | None:
214        """
215        Get station image URL from queue current item.
216
217        :param streamdetails: StreamDetails for the radio stream.
218        """
219        if streamdetails.queue_id and (
220            queue := self.mass.player_queues.get(streamdetails.queue_id)
221        ):
222            if queue.current_item and queue.current_item.media_item:
223                if station_image := queue.current_item.media_item.image:
224                    return self.get_image_url(station_image)
225        return None
226
227    @staticmethod
228    def normalize_radio_artist_name(artist_name: str) -> str:
229        """
230        Normalize artist name from radio stream metadata.
231
232        Handles common formats like "Squier, Billy" -> "Billy Squier" while
233        avoiding mangling of names like "Lipps, Inc." or "Portugal. The Man".
234
235        :param artist_name: Raw artist name to normalize.
236        """
237        # Business/title suffixes that should not be flipped
238        no_flip_suffixes = ("inc", "inc.", "ltd", "ltd.", "llc", "corp")
239        # Specific known bands that are 2 words total and split by a comma
240        valid_artist_names = {
241            "hello, goodbye",
242            "wait, what",
243            "goodnight, sunrise",
244            "slaughter beach, dog",
245            "mount, eerie",
246            "american, native",
247        }
248
249        normalized = artist_name.replace("_", " ")
250
251        if "," not in normalized:
252            return normalized
253
254        # Check against known artist exceptions first
255        if normalized.lower() in valid_artist_names:
256            return normalized
257
258        # Don't flip if contains "and" or "&" (e.g., "Crosby, Stills & Nash")
259        if " and " in normalized.lower() or " & " in normalized:
260            return normalized
261
262        parts = normalized.split(",", 1)
263        if len(parts) != 2:
264            return normalized
265
266        before_comma = parts[0].strip()
267        after_comma = parts[1].strip()
268        after_comma_lower = after_comma.lower()
269
270        # Don't flip if suffix is a business/title term
271        if after_comma_lower in no_flip_suffixes:
272            return normalized
273
274        # Flip if suffix is exactly "The" (e.g., "Beatles, The" -> "The Beatles")
275        if after_comma_lower == "the":
276            return f"{after_comma} {before_comma}"
277
278        # Don't flip if 2+ words after comma (e.g., "Portugal, The Man")
279        if len(after_comma.split()) >= 2:
280            return normalized
281
282        # Standard flip (e.g., "Squier, Billy" -> "Billy Squier")
283        return f"{after_comma} {before_comma}"
284
285    async def get_image_url_by_name(
286        self,
287        artist_name: str,
288        track_name: str,
289        fallback_image_url: str | None = None,
290        album_name: str | None = None,
291    ) -> tuple[str | None, str | None, str | None]:
292        """
293        Look up artwork by artist and track name.
294
295        Searches library and external providers for matching artwork.
296        Also returns corrected artist/track names if the search detects
297        swapped metadata (e.g., "Track - Artist" instead of "Artist - Track").
298
299        :param artist_name: Artist name to search for.
300        :param track_name: Track title to search for.
301        :param fallback_image_url: Fallback image URL if no artwork found.
302        :param album_name: Album announced by the stream, used to refine which artwork is chosen.
303        :returns: Tuple of (image_url, corrected_artist, corrected_track).
304        """
305        if " / " in artist_name:
306            artist_name = artist_name.split(" / ", 1)[0].strip()
307        else:
308            artists_tuple = split_artists(artist_name)
309            artist_name = artists_tuple[0] if artists_tuple else artist_name
310
311        if any(phrase in artist_name.lower() for phrase in AD_DETECTION_PHRASES):
312            return fallback_image_url, None, None
313
314        # album_name influences which release group's artwork is chosen, so it must be
315        # part of the cache key, else two albums for the same track would alias.
316        album_key = create_safe_string(album_name) if album_name else ""
317        cache_key = f"{artist_name.lower()}|{track_name.lower()}|{album_key}"
318        cached_result = await self.mass.cache.get(
319            key=cache_key,
320            category=CACHE_CATEGORY_RADIO_ARTWORK,
321        )
322        if cached_result is not None:
323            if cached_result != "":
324                self.logger.debug(
325                    "Radio artwork for '%s - %s': cached",
326                    artist_name,
327                    track_name,
328                )
329                return str(cached_result), None, None
330            self.logger.debug(
331                "Radio artwork for '%s - %s': cached miss",
332                artist_name,
333                track_name,
334            )
335            return fallback_image_url, None, None
336
337        image_url = None
338        corrected_artist = None
339        corrected_track = None
340        try:
341            (
342                metadata,
343                source,
344                corrected_artist,
345                corrected_track,
346            ) = await self.get_track_metadata_by_name(
347                artist_name=artist_name,
348                track_name=track_name,
349                album_name=album_name,
350            )
351            # Use corrected artist/track for logging if available (handles swapped metadata)
352            log_artist = corrected_artist or artist_name
353            log_track = corrected_track or track_name
354            if metadata and metadata.images:
355                image_url = metadata.images[0].path
356                self.logger.debug(
357                    "Radio artwork found for '%s - %s': %s",
358                    log_artist,
359                    log_track,
360                    source,
361                )
362                if "imageproxy" not in image_url:
363                    await self.mass.cache.set(
364                        key=cache_key,
365                        data=image_url,
366                        expiration=CACHE_EXPIRATION_RADIO_ARTWORK,
367                        category=CACHE_CATEGORY_RADIO_ARTWORK,
368                    )
369            else:
370                self.logger.debug(
371                    "Radio artwork for '%s - %s': not found",
372                    log_artist,
373                    log_track,
374                )
375                await self.mass.cache.set(
376                    key=cache_key,
377                    data="",
378                    expiration=CACHE_EXPIRATION_RADIO_ARTWORK_MISS,
379                    category=CACHE_CATEGORY_RADIO_ARTWORK,
380                )
381        except ProviderUnavailableError, ResourceTemporarilyUnavailable, InvalidDataError:
382            pass
383
384        return image_url or fallback_image_url, corrected_artist, corrected_track
385
386    async def update_radio_stream_artwork(self, streamdetails: StreamDetails) -> None:
387        """
388        Fetch and update radio stream artwork.
389
390        :param streamdetails: StreamDetails to update with artwork.
391        """
392        if not self.mass.config.get_raw_core_config_value(
393            self.domain, CONF_ENABLE_RADIO_METADATA_LOOKUP, True
394        ):
395            return
396        if not streamdetails.stream_metadata:
397            return
398        if not streamdetails.stream_metadata.artist or not streamdetails.stream_metadata.title:
399            return
400
401        try:
402            fallback_url = streamdetails.stream_metadata.image_url
403            original_artist = streamdetails.stream_metadata.artist
404            original_title = streamdetails.stream_metadata.title
405            album = streamdetails.stream_metadata.album
406            image_url, corrected_artist, corrected_track = await self.get_image_url_by_name(
407                artist_name=original_artist,
408                track_name=original_title,
409                fallback_image_url=fallback_url,
410                album_name=album,
411            )
412            # Use corrected artist/track if metadata was swapped
413            final_artist = corrected_artist or original_artist
414            final_title = corrected_track or original_title
415            if (
416                image_url != fallback_url
417                or final_artist != original_artist
418                or final_title != original_title
419            ):
420                streamdetails.stream_metadata = StreamMetadata(
421                    title=final_title,
422                    artist=final_artist,
423                    album=album,
424                    image_url=image_url,
425                )
426                streamdetails.stream_metadata_last_updated = time()
427                if streamdetails.queue_id:
428                    self.mass.player_queues.signal_update(streamdetails.queue_id)
429        except MusicAssistantError:
430            pass
431
432    @staticmethod
433    def _prioritize_release_groups(
434        release_groups: list[MusicBrainzReleaseGroup], album_name: str
435    ) -> list[MusicBrainzReleaseGroup]:
436        """
437        Return the release groups reordered with album-name matches first.
438
439        :param release_groups: Release groups to reorder.
440        :param album_name: Album name announced in the stream metadata.
441        """
442        announced = create_safe_string(album_name)
443        if not announced or len(release_groups) < 2:
444            return release_groups
445
446        # loose substring match either way, so an original album still wins when the
447        # station announces a compilation whose title embeds the original album name
448        def matches(release_group: MusicBrainzReleaseGroup) -> bool:
449            title = create_safe_string(release_group.title)
450            return bool(title) and (title in announced or announced in title)
451
452        return sorted(release_groups, key=lambda rg: not matches(rg))
453
454    async def _get_release_group_artwork(
455        self, mb_release_group: MusicBrainzReleaseGroup
456    ) -> tuple[MediaItemMetadata, str] | None:
457        """
458        Try to get thumb artwork for a release group from metadata providers.
459
460        :param mb_release_group: MusicBrainz release group to look up.
461        :returns: Tuple of (metadata, provider_name) or None if not found.
462        """
463        self.logger.debug(
464            "Looking up artwork for release group '%s' (mbid: %s)",
465            mb_release_group.title,
466            mb_release_group.id,
467        )
468        # Create a minimal Album object to pass the MusicBrainz release group ID
469        # to metadata providers for artwork lookup.
470        temp_album = Album(
471            item_id="temp",
472            provider="temp",
473            name=mb_release_group.title,
474            provider_mappings=set(),
475        )
476        temp_album.add_external_id(ExternalID.MB_RELEASEGROUP, mb_release_group.id)
477        if mb_release_group.barcode:
478            temp_album.add_external_id(ExternalID.BARCODE, mb_release_group.barcode)
479        for provider in self.providers:
480            if ProviderFeature.ALBUM_METADATA not in provider.supported_features:
481                continue
482            try:
483                if metadata := await provider.get_album_metadata(temp_album):
484                    if thumb := self._get_thumb_image(metadata):
485                        return thumb, provider.name
486            except (
487                ProviderUnavailableError,
488                ResourceTemporarilyUnavailable,
489                InvalidDataError,
490            ):
491                pass
492        return None
493
494    async def _search_musicbrainz_with_variants(
495        self,
496        musicbrainz: MusicbrainzProvider,
497        artist_name: str,
498        track_name: str,
499    ) -> tuple[Any, bool]:
500        """
501        Search MusicBrainz with fallback variants (swapped, without 'The').
502
503        :param musicbrainz: MusicBrainz provider instance.
504        :param artist_name: Artist name to search for.
505        :param track_name: Track name to search for.
506        :returns: Tuple of (mb_result, swapped) where swapped indicates artist/track were reversed.
507        """
508        # Try original order
509        mb_result = await musicbrainz.get_release_group_by_track_name(artist_name, track_name)
510        if mb_result:
511            return mb_result, False
512
513        # Try swapped (some stations send "Track - Artist")
514        self.logger.debug(
515            "No MusicBrainz match for '%s - %s', trying swapped",
516            artist_name,
517            track_name,
518        )
519        mb_result = await musicbrainz.get_release_group_by_track_name(track_name, artist_name)
520        if mb_result:
521            return mb_result, True
522
523        # Try without "The " prefix
524        artist_no_the = artist_name[4:] if artist_name.lower().startswith("the ") else None
525        track_no_the = track_name[4:] if track_name.lower().startswith("the ") else None
526
527        if artist_no_the:
528            self.logger.debug(
529                "No match, trying without 'The': '%s - %s'", artist_no_the, track_name
530            )
531            mb_result = await musicbrainz.get_release_group_by_track_name(artist_no_the, track_name)
532            if mb_result:
533                return mb_result, False
534
535        if track_no_the:
536            self.logger.debug(
537                "No match, trying swapped without 'The': '%s - %s'", track_no_the, artist_name
538            )
539            mb_result = await musicbrainz.get_release_group_by_track_name(track_no_the, artist_name)
540            if mb_result:
541                return mb_result, True
542
543        return None, False
544
545    def _get_thumb_image(self, metadata: MediaItemMetadata) -> MediaItemMetadata | None:
546        """
547        Extract only THUMB type image from metadata.
548
549        Returns new metadata with only the thumb image, or None if no thumb found.
550        Used for radio artwork where we specifically need artist/album thumbnails,
551        not logos or banners.
552
553        :param metadata: Metadata to extract thumb from.
554        """
555        if not metadata.images:
556            return None
557        for img in metadata.images:
558            if img.type == ImageType.THUMB:
559                return MediaItemMetadata(images=UniqueList([img]))
560        return None
561
562    async def _get_library_track_metadata(
563        self, artist_name: str, track_name: str
564    ) -> MediaItemMetadata | None:
565        """
566        Search library for matching track and return its metadata.
567
568        :param artist_name: Artist name to match.
569        :param track_name: Track title to match.
570        """
571        try:
572            search_query = f"{artist_name} {track_name}"
573            library_tracks = await self.mass.music.tracks.search(search_query, "library", limit=5)
574            for track in library_tracks:
575                if not self._match_artist_name(artist_name, track.artists):
576                    continue
577                if not compare_strings(track_name, track.name, strict=False):
578                    continue
579                if image_url := await self._get_library_item_thumb(track):
580                    return MediaItemMetadata(
581                        images=UniqueList(
582                            [
583                                MediaItemImage(
584                                    type=ImageType.THUMB,
585                                    path=image_url,
586                                    provider="library",
587                                    remotely_accessible=True,
588                                )
589                            ]
590                        )
591                    )
592        except InvalidDataError:
593            pass
594        return None
595
596    async def _get_library_artist_metadata(self, artist_name: str) -> MediaItemMetadata | None:
597        """
598        Search library for matching artist and return its metadata.
599
600        :param artist_name: Artist name to match.
601        """
602        try:
603            library_artists = await self.mass.music.artists.search(artist_name, "library", limit=5)
604            for artist in library_artists:
605                if not compare_strings(artist_name, artist.name, strict=False):
606                    continue
607                if artist.metadata and artist.metadata.images:
608                    for img in artist.metadata.images:
609                        if img.type == ImageType.THUMB:
610                            return MediaItemMetadata(
611                                images=UniqueList(
612                                    [
613                                        MediaItemImage(
614                                            type=ImageType.THUMB,
615                                            path=self.get_image_url(img, prefer_proxy=True),
616                                            provider="library",
617                                            remotely_accessible=True,
618                                        )
619                                    ]
620                                )
621                            )
622        except InvalidDataError:
623            pass
624        return None
625
626    def _match_artist_name(self, search_name: str, artists: list[Artist | ItemMapping]) -> bool:
627        """
628        Check if any artist matches the search name.
629
630        :param search_name: Artist name to search for.
631        :param artists: List of artists to check against.
632        """
633        for artist in artists:
634            if compare_strings(search_name, artist.name, strict=False):
635                return True
636            # Handle "The" prefix variations
637            if compare_strings(f"The {search_name}", artist.name, strict=False):
638                return True
639            if artist.name.lower().startswith("the "):
640                if compare_strings(search_name, artist.name[4:], strict=False):
641                    return True
642        return False
643
644    async def _get_library_item_thumb(self, track: Track) -> str | None:
645        """
646        Get image URL for library track with fallback: track -> album -> artist.
647
648        :param track: Track to get image for.
649        """
650        # Try track image
651        if track.metadata and track.metadata.images:
652            for img in track.metadata.images:
653                if img.type == ImageType.THUMB:
654                    return self.get_image_url(img, prefer_proxy=True)
655
656        # Try album image
657        if track.album:
658            album = track.album
659            if isinstance(album, ItemMapping):
660                try:
661                    full_album = await self.mass.music.albums.get_library_item(album.item_id)
662                    if full_album and full_album.metadata and full_album.metadata.images:
663                        for img in full_album.metadata.images:
664                            if img.type == ImageType.THUMB:
665                                return self.get_image_url(img, prefer_proxy=True)
666                except MediaNotFoundError:
667                    pass
668            elif isinstance(album, Album) and album.metadata and album.metadata.images:
669                for img in album.metadata.images:
670                    if img.type == ImageType.THUMB:
671                        return self.get_image_url(img, prefer_proxy=True)
672
673        # Try artist image
674        for artist in track.artists:
675            if isinstance(artist, ItemMapping):
676                try:
677                    full_artist = await self.mass.music.artists.get_library_item(artist.item_id)
678                    if full_artist and full_artist.metadata and full_artist.metadata.images:
679                        for img in full_artist.metadata.images:
680                            if img.type == ImageType.THUMB:
681                                return self.get_image_url(img, prefer_proxy=True)
682                except MediaNotFoundError:
683                    pass
684            elif isinstance(artist, Artist) and artist.metadata and artist.metadata.images:
685                for img in artist.metadata.images:
686                    if img.type == ImageType.THUMB:
687                        return self.get_image_url(img, prefer_proxy=True)
688
689        return None
690