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