/
/
/
1"""Manage MediaItems of type Artist."""
2
3from __future__ import annotations
4
5import asyncio
6import contextlib
7from itertools import zip_longest
8from typing import TYPE_CHECKING, Any, Literal, cast, overload
9
10from music_assistant_models.auth import Scope
11from music_assistant_models.enums import (
12 AlbumType,
13 ArtistType,
14 MediaType,
15 ProviderFeature,
16 ProviderType,
17)
18from music_assistant_models.errors import (
19 InvalidDataError,
20 MediaNotFoundError,
21 MusicAssistantError,
22 ProviderUnavailableError,
23)
24from music_assistant_models.helpers import create_safe_string
25from music_assistant_models.media_items import (
26 Album,
27 Artist,
28 ArtistSummary,
29 Audiobook,
30 ItemMapping,
31 MediaCollection,
32 ProviderMapping,
33 Track,
34)
35
36from music_assistant.constants import (
37 DB_TABLE_ALBUM_ARTISTS,
38 DB_TABLE_ARTISTS,
39 DB_TABLE_AUDIOBOOK_ARTISTS,
40 DB_TABLE_TRACK_ARTISTS,
41 VARIOUS_ARTISTS_MBID,
42 VARIOUS_ARTISTS_NAME,
43)
44from music_assistant.controllers.music.helpers import (
45 metadata_for_update,
46 provider_mappings_for_update,
47)
48from music_assistant.helpers.compare import (
49 compare_album,
50 compare_album_name,
51 compare_artist,
52 compare_strings,
53 compare_track,
54)
55from music_assistant.helpers.database import UNSET
56from music_assistant.helpers.json import serialize_to_json
57from music_assistant.models.music_provider import MusicProvider
58
59from .base import MediaControllerBase
60
61if TYPE_CHECKING:
62 from collections.abc import Mapping
63
64 from music_assistant import MusicAssistant
65 from music_assistant.models.metadata_provider import MetadataProvider
66
67
68class ArtistsController(MediaControllerBase[Artist]):
69 """Controller managing MediaItems of type Artist."""
70
71 db_table = DB_TABLE_ARTISTS
72 media_type = MediaType.ARTIST
73 item_cls = Artist
74 summary_item_cls = ArtistSummary
75
76 def __init__(self, mass: MusicAssistant) -> None:
77 """Initialize class."""
78 super().__init__(mass)
79 self._db_add_lock = asyncio.Lock()
80 # register (extra) api handlers
81 api_base = self.api_base
82 self.mass.register_api_command(
83 f"music/{api_base}/artist_albums", self.albums, required_scope=Scope.LIBRARY_READ
84 )
85 self.mass.register_api_command(
86 f"music/{api_base}/artist_tracks", self.tracks, required_scope=Scope.LIBRARY_READ
87 )
88 self.mass.register_api_command(
89 f"music/{api_base}/top_tracks", self.top_tracks, required_scope=Scope.LIBRARY_READ
90 )
91 self.mass.register_api_command(
92 f"music/{api_base}/top_albums", self.top_albums, required_scope=Scope.LIBRARY_READ
93 )
94 self.mass.register_api_command(
95 f"music/{api_base}/artist_audiobooks",
96 self.audiobooks,
97 required_scope=Scope.LIBRARY_READ,
98 )
99 self.mass.register_api_command(
100 f"music/{api_base}/similar_artists",
101 self.similar_artists,
102 required_scope=Scope.LIBRARY_READ,
103 )
104 self.mass.register_api_command(
105 f"music/{api_base}/library_artist_types",
106 self.get_library_artist_types,
107 required_scope=Scope.LIBRARY_READ,
108 )
109
110 @property
111 def summary_query(self) -> tuple[str, dict[str, Any]]:
112 """Return the slim SELECT query used for artist summary listings."""
113 query = f"""
114 SELECT
115 {self._summary_base_columns()},
116 artists.artist_type,
117 {self._provider_mappings_query()} AS provider_mappings
118 FROM artists"""
119 return query, {}
120
121 async def library_count(
122 self,
123 favorite_only: bool = False,
124 album_artists_only: bool = False,
125 artist_type: ArtistType | None = None,
126 ) -> int:
127 """
128 Return the number of artists in the library.
129
130 Restricted to the providers the current user is allowed to see when that user
131 has a provider filter set.
132
133 :param favorite_only: Only count artists marked as favorite.
134 :param album_artists_only: Only count artists that have albums.
135 :param artist_type: Only count artists of this type.
136 """
137 sql_query = f"SELECT item_id FROM {self.db_table}"
138 query_parts = []
139 query_params: dict[str, Any] = {}
140 if artist_type:
141 query_parts.append(f"artist_type = '{artist_type}'")
142 if favorite_only:
143 query_parts.append("favorite = 1")
144 if album_artists_only:
145 query_parts.append(
146 f"item_id in (select {DB_TABLE_ALBUM_ARTISTS}.artist_id "
147 f"FROM {DB_TABLE_ALBUM_ARTISTS})"
148 )
149 if provider_filter := self._ensure_provider_filter(None):
150 query_parts.append(
151 self._provider_filter_clause(query_params, provider_filter, in_library_only=True)
152 )
153 if query_parts:
154 sql_query += f" WHERE {' AND '.join(query_parts)}"
155 return await self.mass.music.database.get_count_from_query(sql_query, query_params)
156
157 async def library_items( # noqa: PLR0913
158 self,
159 favorite: bool | None = None,
160 search: str | None = None,
161 limit: int = 500,
162 offset: int = 0,
163 order_by: str = "sort_name",
164 provider: str | list[str] | None = None,
165 genre: int | list[int] | None = None,
166 played_only: bool = False,
167 album_artists_only: bool = False,
168 artist_type: ArtistType | None = None,
169 *,
170 summary: bool = True,
171 reachable_via: list[str] | None = None,
172 **kwargs: Any,
173 ) -> list[Artist]:
174 """
175 Get in-database (album) artists.
176
177 :param favorite: Filter by favorite status.
178 :param search: Filter by search query.
179 :param limit: Maximum number of items to return.
180 :param offset: Number of items to skip.
181 :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
182 :param provider: Filter by provider instance ID (single string or list).
183 :param album_artists_only: Only return artists that have albums.
184 :param genre: Filter by genre id(s).
185 :param artist_type: The artist's type
186 :param summary: When True (default), return slim summary items containing only the
187 fields needed for a list view. Set to False to get fully hydrated items.
188 :param reachable_via: Restrict results to items with a provider mapping reachable
189 through one of these provider instance ids (OR semantics). See
190 `MediaControllerBase.library_items` for the full semantics.
191 """
192 reachable_via = self._resolve_reachable_via(reachable_via)
193 if reachable_via is not None and not reachable_via:
194 return []
195 extra_query_params: dict[str, Any] = {}
196 extra_query_parts: list[str] = []
197 if artist_type:
198 extra_query_parts = [f"artist_type = '{artist_type}'"]
199 if album_artists_only and artist_type in (None, ArtistType.SINGER):
200 extra_query_parts.append(
201 f"artists.item_id in (select {DB_TABLE_ALBUM_ARTISTS}.artist_id "
202 f"from {DB_TABLE_ALBUM_ARTISTS})"
203 )
204 return await self.get_library_items_by_query(
205 favorite=favorite,
206 search=search,
207 genre_ids=genre,
208 limit=limit,
209 offset=offset,
210 order_by=order_by,
211 provider_filter=self._provider_filter_considering_reachability(provider, reachable_via),
212 extra_query_parts=extra_query_parts,
213 extra_query_params=extra_query_params,
214 played_only=played_only,
215 in_library_only=True,
216 summary=summary,
217 reachable_via=reachable_via,
218 )
219
220 async def tracks(
221 self,
222 item_id: str,
223 provider_instance_id_or_domain: str,
224 provider_filter: str | None = None,
225 ) -> list[Track]:
226 """
227 Return the tracks for a artist.
228
229 For a library item, the in-library tracks are returned, optionally limited to a single
230 provider instance with the provider_filter. For a provider item, that provider's
231 tracks listing is returned (which may be empty if it is not supported).
232
233 :param item_id: The item ID of the artist.
234 :param provider_instance_id_or_domain: The provider instance ID or domain of the artist.
235 :param provider_filter: Optional provider instance ID to limit the (library) result to.
236 """
237 if provider_instance_id_or_domain == "library":
238 return await self.get_library_artist_tracks(item_id, provider_filter=provider_filter)
239 self._validate_provider_filter(provider_instance_id_or_domain, provider_filter)
240 return await self.get_provider_artist_tracks(item_id, provider_instance_id_or_domain)
241
242 async def albums(
243 self,
244 item_id: str,
245 provider_instance_id_or_domain: str,
246 provider_filter: str | None = None,
247 ) -> list[Album]:
248 """
249 Return the albums for an artist.
250
251 For a library item, the in-library albums are returned, optionally limited to a single
252 provider instance with the provider_filter. For a provider item, that provider's
253 albums listing is returned (which may be empty if it is not supported).
254
255 :param item_id: The item ID of the artist.
256 :param provider_instance_id_or_domain: The provider instance ID or domain of the artist.
257 :param provider_filter: Optional provider instance ID to limit the (library) result to.
258 """
259 if provider_instance_id_or_domain == "library":
260 return await self.get_library_artist_albums(item_id, provider_filter=provider_filter)
261 self._validate_provider_filter(provider_instance_id_or_domain, provider_filter)
262 return await self.get_provider_artist_albums(item_id, provider_instance_id_or_domain)
263
264 async def top_tracks(
265 self,
266 item_id: str,
267 provider_instance_id_or_domain: str,
268 provider_filter: str | None = None,
269 ) -> list[Track]:
270 """
271 Return the top/featured tracks for an artist.
272
273 For a library item, the top tracks of all the artist's providers are aggregated (and
274 deduplicated), optionally limited to a single provider instance. For a provider
275 item, that provider's top tracks listing is returned (may be empty if not supported).
276
277 :param item_id: The item ID of the artist.
278 :param provider_instance_id_or_domain: The provider instance ID or domain of the artist.
279 :param provider_filter: Optional provider instance ID to limit the result to.
280 """
281 if provider_instance_id_or_domain == "library":
282 return await self.get_library_artist_toptracks(item_id, provider_filter=provider_filter)
283 self._validate_provider_filter(provider_instance_id_or_domain, provider_filter)
284 return await self.get_provider_artist_toptracks(item_id, provider_instance_id_or_domain)
285
286 async def top_albums(
287 self,
288 item_id: str,
289 provider_instance_id_or_domain: str,
290 provider_filter: str | None = None,
291 ) -> list[Album]:
292 """
293 Return the top/featured albums for an artist.
294
295 For a library item, the top albums of all the artist's providers are aggregated (and
296 deduplicated), optionally limited to a single provider instance. For a provider
297 item, that provider's top albums listing is returned (may be empty if not supported).
298
299 :param item_id: The item ID of the artist.
300 :param provider_instance_id_or_domain: The provider instance ID or domain of the artist.
301 :param provider_filter: Optional provider instance ID to limit the result to.
302 """
303 if provider_instance_id_or_domain == "library":
304 return await self.get_library_artist_topalbums(item_id, provider_filter=provider_filter)
305 self._validate_provider_filter(provider_instance_id_or_domain, provider_filter)
306 return await self.get_provider_artist_topalbums(item_id, provider_instance_id_or_domain)
307
308 async def similar_artists(
309 self,
310 item_id: str,
311 provider_instance_id_or_domain: str,
312 provider_filter: str | None = None,
313 limit: int = 25,
314 ) -> list[Artist]:
315 """
316 Return similar artists for an artist.
317
318 For a library item, the similar artists of all the artist's providers are aggregated
319 (and deduplicated), optionally limited to a single provider instance. For a provider
320 item, that provider's similar artists listing is returned (may be empty if not
321 supported).
322
323 :param item_id: The item ID of the artist.
324 :param provider_instance_id_or_domain: The provider instance ID or domain of the artist.
325 :param provider_filter: Optional provider instance ID to limit the result to.
326 :param limit: Maximum number of similar artists to return.
327 """
328 if provider_instance_id_or_domain == "library":
329 return await self.get_library_artist_similar_artists(
330 item_id, provider_filter=provider_filter, limit=limit
331 )
332 self._validate_provider_filter(provider_instance_id_or_domain, provider_filter)
333 return await self.get_provider_artist_similar_artists(
334 item_id, provider_instance_id_or_domain, limit=limit
335 )
336
337 if TYPE_CHECKING:
338
339 @overload
340 async def audiobooks(
341 self,
342 item_id: str,
343 provider_instance_id_or_domain: str,
344 artist_type: ArtistType = ArtistType.AUTHOR,
345 in_library_only: bool = False,
346 *,
347 collapse_collections: Literal[False] = False,
348 ) -> list[Audiobook]: ...
349
350 @overload
351 async def audiobooks(
352 self,
353 item_id: str,
354 provider_instance_id_or_domain: str,
355 artist_type: ArtistType = ArtistType.AUTHOR,
356 in_library_only: bool = False,
357 *,
358 collapse_collections: Literal[True],
359 ) -> list[Audiobook | MediaCollection[Audiobook]]: ...
360
361 async def audiobooks(
362 self,
363 item_id: str,
364 provider_instance_id_or_domain: str,
365 artist_type: ArtistType = ArtistType.AUTHOR,
366 in_library_only: bool = False,
367 *,
368 collapse_collections: bool = False,
369 ) -> list[Audiobook] | list[Audiobook | MediaCollection[Audiobook]]:
370 """
371 Return audiobooks for an artist.
372
373 Artist_type can be omitted for in-library artists.
374
375 :param collapse_collections: Collapse available collections. Only applies to
376 in-library items; when in_library_only is False, provider items are
377 appended as plain audiobooks alongside the collapsed collections.
378 """
379 if artist_type == ArtistType.SINGER:
380 self.logger.warning("Audiobooks not supported for artist_type SINGER.")
381 return []
382 # always check if we have a library item for this artist
383 library_artist = await self.get_library_item_by_prov_id(
384 item_id, provider_instance_id_or_domain
385 )
386 if library_artist and library_artist.artist_type == ArtistType.SINGER:
387 self.logger.debug(
388 "Ignoring audiobook request for artist of type %s", library_artist.artist_type
389 )
390 return []
391 if not library_artist:
392 if artist_type == ArtistType.AUTHOR:
393 return await self.get_provider_author_audiobooks(
394 item_id, provider_instance_id_or_domain
395 )
396 if artist_type == ArtistType.NARRATOR:
397 return await self.get_provider_narrator_audiobooks(
398 item_id, provider_instance_id_or_domain
399 )
400 return []
401
402 db_items = await self.get_library_author_narrator_audiobooks(
403 library_artist.item_id,
404 artist_type=library_artist.artist_type,
405 collapse_collections=collapse_collections,
406 )
407 result: list[Audiobook] | list[Audiobook | MediaCollection[Audiobook]] = db_items
408 if in_library_only:
409 # return in-library items only
410 return result
411 # return all (unique) items from all providers
412 # initialize unique_ids with db_items to prevent duplicates
413 unique_ids: set[str] = set()
414 for item in db_items:
415 if isinstance(item, MediaCollection):
416 for collection_item in item.items:
417 unique_ids.add(f"{collection_item.name}.{collection_item.version}")
418 else:
419 unique_ids.add(f"{item.name}.{item.version}")
420 unique_providers = self.mass.music.get_unique_providers()
421 audiobook_method = (
422 self.get_provider_author_audiobooks
423 if artist_type == ArtistType.AUTHOR
424 else self.get_provider_narrator_audiobooks
425 )
426 for provider_mapping in library_artist.provider_mappings:
427 if provider_mapping.provider_instance not in unique_providers:
428 continue
429 provider_audiobooks = await audiobook_method(
430 provider_mapping.item_id, provider_mapping.provider_instance
431 )
432 for provider_audiobook in provider_audiobooks:
433 unique_id = f"{provider_audiobook.name}.{provider_audiobook.version}"
434 if unique_id in unique_ids:
435 continue
436 unique_ids.add(unique_id)
437 # prefer db item
438 if db_item := await self.mass.music.audiobooks.get_library_item_by_prov_id(
439 provider_audiobook.item_id, provider_audiobook.provider
440 ):
441 result.append(db_item)
442 elif not in_library_only:
443 result.append(provider_audiobook)
444 return result
445
446 async def get_library_author_narrator_audiobooks(
447 self,
448 item_id: str | int,
449 artist_type: ArtistType,
450 *,
451 collapse_collections: bool = False,
452 ) -> list[Audiobook] | list[Audiobook | MediaCollection[Audiobook]]:
453 """Return all in-library audiobooks for an author/ narrator."""
454 db_id = int(item_id) # ensure integer
455 library_item = await self.get_library_item(db_id)
456 if library_item.artist_type != artist_type:
457 self.logger.debug("Audiobooks only available for artists of type %s", artist_type)
458 return []
459 subquery = (
460 f"SELECT audiobook_id FROM {DB_TABLE_AUDIOBOOK_ARTISTS} WHERE artist_id = :artist_id"
461 )
462 query = f"audiobooks.item_id in ({subquery})"
463 return await self.mass.music.audiobooks.get_library_items_by_query(
464 extra_query_parts=[query],
465 extra_query_params={"artist_id": db_id},
466 collapse_collections=collapse_collections,
467 )
468
469 async def get_provider_author_audiobooks(
470 self,
471 item_id: str,
472 provider_instance_id_or_domain: str,
473 ) -> list[Audiobook]:
474 """Return audiobooks for an author on given provider."""
475 assert provider_instance_id_or_domain != "library"
476 if not (prov := self.mass.get_provider(provider_instance_id_or_domain)):
477 return []
478 prov = cast("MusicProvider", prov)
479 if ProviderFeature.AUTHOR_AUDIOBOOKS in prov.supported_features:
480 return await prov.get_author_audiobooks(item_id)
481 # fallback implementation using the db
482 return await self._get_db_author_narrator_audiobooks(
483 item_id=item_id,
484 provider_instance_id_or_domain=provider_instance_id_or_domain,
485 artist_type=ArtistType.AUTHOR,
486 )
487
488 async def get_provider_narrator_audiobooks(
489 self,
490 item_id: str,
491 provider_instance_id_or_domain: str,
492 ) -> list[Audiobook]:
493 """Return audiobooks for an author on given provider."""
494 assert provider_instance_id_or_domain != "library"
495 if not (prov := self.mass.get_provider(provider_instance_id_or_domain)):
496 return []
497 prov = cast("MusicProvider", prov)
498 if ProviderFeature.NARRATOR_AUDIOBOOKS in prov.supported_features:
499 return await prov.get_narrator_audiobooks(item_id)
500 # fallback implementation using the db
501 return await self._get_db_author_narrator_audiobooks(
502 item_id=item_id,
503 provider_instance_id_or_domain=provider_instance_id_or_domain,
504 artist_type=ArtistType.NARRATOR,
505 )
506
507 async def get_provider_artist_toptracks(
508 self,
509 item_id: str,
510 provider_instance_id_or_domain: str,
511 ) -> list[Track]:
512 """
513 Return the top tracks for an artist on the given provider.
514
515 Each track is resolved to its in-library equivalent where available.
516 """
517 provider = self.mass.get_provider(
518 provider_instance_id_or_domain, provider_type=MusicProvider
519 )
520 if provider is None or not provider.available:
521 return [] # guard against unavailable provider
522 if not provider.supports_feature(ProviderFeature.ARTIST_TOPTRACKS):
523 self.logger.warning(
524 "Provider %s does not support fetching artist top tracks.",
525 provider.name,
526 )
527 return [] # guard against unsupported feature
528 tracks = await provider.get_artist_toptracks(item_id)
529 # resolve to in-library equivalents (in parallel) where available
530 resolved = await asyncio.gather(
531 *(
532 self.mass.music.tracks.get_library_item_by_prov_id(track.item_id, track.provider)
533 for track in tracks
534 )
535 )
536 return [
537 library_track or track for library_track, track in zip(resolved, tracks, strict=True)
538 ]
539
540 async def get_library_artist_toptracks(
541 self,
542 item_id: str | int,
543 provider_filter: str | None = None,
544 ) -> list[Track]:
545 """
546 Return the top tracks for an in-library artist, aggregated across all its providers.
547
548 The result combines (and deduplicates, preserving order) the top tracks from every
549 provider attached to the artist and any metadata/plugin provider implementing the
550 feature. Empty when no provider yields a result.
551
552 :param item_id: The library item ID of the artist.
553 :param provider_filter: Optional provider instance ID to limit the result to.
554 """
555 ref_item = await self.get_library_item(item_id)
556 allowed = self._ensure_provider_filter(provider_filter)
557 # fetch each provider's ranked top tracks in parallel
558 fetches = []
559 # streaming providers attached to the artist (results resolved to library items)
560 for provider_mapping in ref_item.provider_mappings:
561 if allowed is not None and provider_mapping.provider_instance not in allowed:
562 continue
563 music_prov = self.mass.get_provider(
564 provider_mapping.provider_instance, provider_type=MusicProvider
565 )
566 if (
567 music_prov is None
568 or ProviderFeature.ARTIST_TOPTRACKS not in music_prov.supported_features
569 ):
570 continue
571 fetches.append(
572 self.get_provider_artist_toptracks(
573 provider_mapping.item_id, provider_mapping.provider_instance
574 )
575 )
576 # metadata/plugin providers implementing the feature
577 for prov in self.mass.get_providers_supporting_feature(
578 ProviderFeature.ARTIST_TOPTRACKS,
579 priority=(ProviderType.METADATA, ProviderType.PLUGIN),
580 ):
581 if allowed is not None and prov.instance_id not in allowed:
582 continue
583 fetches.append(cast("MetadataProvider", prov).get_artist_toptracks(ref_item))
584 per_provider = await asyncio.gather(*fetches, return_exceptions=True)
585 # drop (and log) any provider that failed so one bad provider can't sink the listing
586 listings: list[list[Track]] = []
587 for listing in per_provider:
588 if isinstance(listing, BaseException):
589 self.logger.warning(
590 "Error fetching top tracks for artist %s from a provider",
591 ref_item.name,
592 exc_info=listing,
593 )
594 continue
595 listings.append(listing)
596 # interleave the providers' rankings by position (zip), deduplicating with the compare
597 # helper (which also matches on version/duration)
598 result: list[Track] = []
599 for row in zip_longest(*listings):
600 for candidate in row:
601 if candidate is None or any(
602 compare_track(existing, candidate) for existing in result
603 ):
604 continue
605 result.append(candidate)
606 return result
607
608 async def get_provider_artist_topalbums(
609 self,
610 item_id: str,
611 provider_instance_id_or_domain: str,
612 ) -> list[Album]:
613 """
614 Return the top/featured albums for an artist on the given provider.
615
616 Each album is resolved to its in-library equivalent where available.
617 """
618 provider = self.mass.get_provider(
619 provider_instance_id_or_domain, provider_type=MusicProvider
620 )
621 if provider is None or not provider.available:
622 return [] # guard against unavailable provider
623 if not provider.supports_feature(ProviderFeature.ARTIST_TOPALBUMS):
624 self.logger.warning(
625 "Provider %s does not support fetching artist top albums.",
626 provider.name,
627 )
628 return [] # guard against unsupported feature
629 albums = await provider.get_artist_topalbums(item_id)
630 # resolve to in-library equivalents (in parallel) where available
631 resolved = await asyncio.gather(
632 *(
633 self.mass.music.albums.get_library_item_by_prov_id(album.item_id, album.provider)
634 for album in albums
635 )
636 )
637 return [
638 library_album or album for library_album, album in zip(resolved, albums, strict=True)
639 ]
640
641 async def get_library_artist_topalbums(
642 self,
643 item_id: str | int,
644 provider_filter: str | None = None,
645 ) -> list[Album]:
646 """
647 Return the top albums for an in-library artist, aggregated across all its providers.
648
649 The result combines (and deduplicates, preserving order) the top albums from every
650 provider attached to the artist and any metadata/plugin provider implementing the
651 feature. Empty when no provider yields a result.
652
653 :param item_id: The library item ID of the artist.
654 :param provider_filter: Optional provider instance ID to limit the result to.
655 """
656 ref_item = await self.get_library_item(item_id)
657 allowed = self._ensure_provider_filter(provider_filter)
658 # fetch each provider's ranked top albums in parallel
659 fetches = []
660 # streaming providers attached to the artist (results resolved to library items)
661 for provider_mapping in ref_item.provider_mappings:
662 if allowed is not None and provider_mapping.provider_instance not in allowed:
663 continue
664 music_prov = self.mass.get_provider(
665 provider_mapping.provider_instance, provider_type=MusicProvider
666 )
667 if (
668 music_prov is None
669 or ProviderFeature.ARTIST_TOPALBUMS not in music_prov.supported_features
670 ):
671 continue
672 fetches.append(
673 self.get_provider_artist_topalbums(
674 provider_mapping.item_id, provider_mapping.provider_instance
675 )
676 )
677 # metadata/plugin providers implementing the feature
678 for prov in self.mass.get_providers_supporting_feature(
679 ProviderFeature.ARTIST_TOPALBUMS,
680 priority=(ProviderType.METADATA, ProviderType.PLUGIN),
681 ):
682 if allowed is not None and prov.instance_id not in allowed:
683 continue
684 fetches.append(cast("MetadataProvider", prov).get_artist_topalbums(ref_item))
685 per_provider = await asyncio.gather(*fetches, return_exceptions=True)
686 # drop (and log) any provider that failed so one bad provider can't sink the listing
687 listings: list[list[Album]] = []
688 for listing in per_provider:
689 if isinstance(listing, BaseException):
690 self.logger.warning(
691 "Error fetching top albums for artist %s from a provider",
692 ref_item.name,
693 exc_info=listing,
694 )
695 continue
696 listings.append(listing)
697 # interleave the providers' rankings by position (zip), deduplicating with the compare
698 # helper (which also matches on version/duration)
699 result: list[Album] = []
700 for row in zip_longest(*listings):
701 for candidate in row:
702 if candidate is None or any(
703 compare_album(existing, candidate) for existing in result
704 ):
705 continue
706 result.append(candidate)
707 return result
708
709 async def get_provider_artist_tracks(
710 self,
711 item_id: str,
712 provider_instance_id_or_domain: str,
713 ) -> list[Track]:
714 """Return all tracks for an artist on given provider."""
715 provider = self.mass.get_provider(
716 provider_instance_id_or_domain, provider_type=MusicProvider
717 )
718 if provider is None or not provider.available:
719 return [] # guard against unavailable provider
720 if provider.supports_feature(ProviderFeature.ARTIST_TRACKS):
721 return await provider.get_artist_tracks(item_id)
722 # fallback: enumerate (and dedupe) the tracks of all the artist's albums on the provider
723 result: list[Track] = []
724 unique_ids: set[str] = set()
725 for album in await self.get_provider_artist_albums(item_id, provider_instance_id_or_domain):
726 for track in await self.mass.music.albums.tracks(album.item_id, album.provider):
727 unique_id = f"{track.name}.{track.version}"
728 if unique_id in unique_ids:
729 continue
730 unique_ids.add(unique_id)
731 result.append(track)
732 return result
733
734 async def get_library_artist_tracks(
735 self,
736 item_id: str | int,
737 provider_filter: str | None = None,
738 ) -> list[Track]:
739 """Return all in-library tracks for an artist, optionally limited to a single provider."""
740 db_id = int(item_id) # ensure integer
741 library_item = await self.get_library_item(db_id)
742 if library_item.artist_type != ArtistType.SINGER:
743 self.logger.debug("Tracks only available for artists of type ARTIST")
744 return []
745 subquery = f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = :artist_id"
746 query = f"tracks.item_id in ({subquery})"
747 return await self.mass.music.tracks.get_library_items_by_query(
748 extra_query_parts=[query],
749 extra_query_params={"artist_id": db_id},
750 provider_filter=self._ensure_provider_filter(provider_filter),
751 in_library_only=True,
752 )
753
754 async def get_provider_artist_albums(
755 self,
756 item_id: str,
757 provider_instance_id_or_domain: str,
758 ) -> list[Album]:
759 """Return albums for an artist on given provider."""
760 provider = self.mass.get_provider(
761 provider_instance_id_or_domain, provider_type=MusicProvider
762 )
763 if provider is None or not provider.available:
764 return [] # guard against unavailable provider
765 if not provider.supports_feature(ProviderFeature.ARTIST_ALBUMS):
766 self.logger.warning(
767 "Provider %s does not support fetching all artist albums.",
768 provider.name,
769 )
770 return [] # guard against unsupported feature
771 return await provider.get_artist_albums(item_id)
772
773 async def get_library_artist_albums(
774 self,
775 item_id: str | int,
776 provider_filter: str | None = None,
777 ) -> list[Album]:
778 """Return all in-library albums for an artist, optionally limited to a single provider."""
779 db_id = int(item_id) # ensure integer
780 library_item = await self.get_library_item(db_id)
781 if library_item.artist_type != ArtistType.SINGER:
782 self.logger.debug("Albums only available for artists of type ARTIST")
783 return []
784 subquery = f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = :artist_id"
785 query = f"albums.item_id in ({subquery})"
786 return await self.mass.music.albums.get_library_items_by_query(
787 extra_query_parts=[query],
788 extra_query_params={"artist_id": db_id},
789 provider_filter=self._ensure_provider_filter(provider_filter),
790 in_library_only=True,
791 )
792
793 async def get_provider_artist_similar_artists(
794 self,
795 item_id: str,
796 provider_instance_id_or_domain: str,
797 limit: int = 25,
798 ) -> list[Artist]:
799 """
800 Return similar artists for an artist on the given provider.
801
802 Each artist is resolved to its in-library equivalent where available.
803 """
804 provider = self.mass.get_provider(
805 provider_instance_id_or_domain, provider_type=MusicProvider
806 )
807 if provider is None or not provider.available:
808 return [] # guard against unavailable provider
809 if not provider.supports_feature(ProviderFeature.SIMILAR_ARTISTS):
810 self.logger.warning(
811 "Provider %s does not support fetching similar artists.",
812 provider.name,
813 )
814 return [] # guard against unsupported feature
815 artists = await provider.get_similar_artists(item_id, limit=limit)
816 # resolve to in-library equivalents (in parallel) where available
817 resolved = await asyncio.gather(
818 *(
819 self.get_library_item_by_prov_id(artist.item_id, artist.provider)
820 for artist in artists
821 )
822 )
823 return [
824 library_artist or artist
825 for library_artist, artist in zip(resolved, artists, strict=True)
826 ]
827
828 async def get_library_artist_similar_artists(
829 self,
830 item_id: str | int,
831 provider_filter: str | None = None,
832 limit: int = 25,
833 ) -> list[Artist]:
834 """
835 Return similar artists for an in-library artist, aggregated across all its providers.
836
837 The result combines (and deduplicates, preserving order) the similar artists from
838 every provider attached to the artist and any metadata/plugin provider implementing
839 the feature. Empty when no provider yields a result.
840
841 :param item_id: The library item ID of the artist.
842 :param provider_filter: Optional provider instance ID to limit the result to.
843 :param limit: Maximum number of similar artists to return.
844 """
845 ref_item = await self.get_library_item(item_id)
846 allowed = self._ensure_provider_filter(provider_filter)
847 # fetch each provider's similar artists in parallel
848 fetches = []
849 # streaming providers attached to the artist (results resolved to library items)
850 for provider_mapping in ref_item.provider_mappings:
851 if allowed is not None and provider_mapping.provider_instance not in allowed:
852 continue
853 music_prov = self.mass.get_provider(
854 provider_mapping.provider_instance, provider_type=MusicProvider
855 )
856 if (
857 music_prov is None
858 or ProviderFeature.SIMILAR_ARTISTS not in music_prov.supported_features
859 ):
860 continue
861 fetches.append(
862 self.get_provider_artist_similar_artists(
863 provider_mapping.item_id, provider_mapping.provider_instance, limit=limit
864 )
865 )
866 # metadata/plugin providers implementing the feature
867 for prov in self.mass.get_providers_supporting_feature(
868 ProviderFeature.SIMILAR_ARTISTS,
869 priority=(ProviderType.METADATA, ProviderType.PLUGIN),
870 ):
871 if allowed is not None and prov.instance_id not in allowed:
872 continue
873 fetches.append(
874 cast("MetadataProvider", prov).get_similar_artists(ref_item, limit=limit)
875 )
876 per_provider = await asyncio.gather(*fetches, return_exceptions=True)
877 # drop (and log) any provider that failed so one bad provider can't sink the listing
878 listings: list[list[Artist]] = []
879 for listing in per_provider:
880 if isinstance(listing, BaseException):
881 self.logger.warning(
882 "Error fetching similar artists for %s from a provider",
883 ref_item.name,
884 exc_info=listing,
885 )
886 continue
887 listings.append(listing)
888 # interleave the providers' results by position (zip), deduplicating with the compare
889 # helper, and cap to the requested limit
890 result: list[Artist] = []
891 for row in zip_longest(*listings):
892 for candidate in row:
893 if candidate is None or any(
894 compare_artist(existing, candidate) for existing in result
895 ):
896 continue
897 result.append(candidate)
898 return result[:limit]
899
900 async def get_library_artist_types(self) -> list[ArtistType]:
901 """Get all supported in-library artist types."""
902 artist_types: list[ArtistType] = []
903 query = f"SELECT DISTINCT artist_type FROM {DB_TABLE_ARTISTS}"
904 rows = await self.mass.music.database.get_rows_from_query(query)
905 for row in rows:
906 artist_types.append(ArtistType(row["artist_type"]))
907 return artist_types
908
909 async def remove_item_from_library(self, item_id: str | int, recursive: bool = True) -> None:
910 """Delete record from the database."""
911 db_id = int(item_id) # ensure integer
912 library_item = await self.get_library_item(db_id)
913
914 if library_item.artist_type == ArtistType.SINGER:
915 await self._remove_music_artist_from_library(db_id=db_id, recursive=recursive)
916 elif library_item.artist_type in (ArtistType.AUTHOR, ArtistType.NARRATOR):
917 await self._remove_author_narrator_from_library(db_id=db_id, recursive=recursive)
918 else:
919 raise MusicAssistantError(f"Unknown artist_type {library_item.artist_type}.")
920
921 # delete the artist itself from db
922 # this will raise if the item still has references and recursive is false
923 await super().remove_item_from_library(db_id)
924
925 async def match_provider(
926 self, db_artist: Artist, provider: MusicProvider, strict: bool = True
927 ) -> list[ProviderMapping]:
928 """
929 Try to find match on (streaming) provider for the provided (database) artist.
930
931 This is used to link objects of different providers/qualities together.
932
933 :param strict: How strictly the candidate artist itself must match; the reference
934 track/album only ever has to corroborate it, never match exactly.
935 """
936 self.logger.debug("Trying to match artist %s on provider %s", db_artist.name, provider.name)
937 # try to get a match with some reference tracks of this artist
938 ref_tracks = await self.mass.music.artists.tracks(db_artist.item_id, db_artist.provider)
939 if len(ref_tracks) < 10:
940 # fetch reference tracks from provider(s) attached to the artist
941 for provider_mapping in db_artist.provider_mappings:
942 with contextlib.suppress(ProviderUnavailableError, MediaNotFoundError):
943 ref_tracks += await self.mass.music.artists.tracks(
944 provider_mapping.item_id, provider_mapping.provider_instance
945 )
946 for ref_track in ref_tracks:
947 search_str = f"{db_artist.name} - {ref_track.name}"
948 search_results = await self.mass.music.tracks.search(search_str, provider.domain)
949 for search_result_item in search_results:
950 # the reference track must corroborate the candidate, not merely share its title
951 if not compare_track(ref_track, search_result_item, strict=False):
952 continue
953 # get matching artist from track
954 for search_item_artist in search_result_item.artists:
955 if matches := await self._confirm_artist_match(
956 db_artist, search_item_artist, strict
957 ):
958 return matches
959 # try to get a match with some reference albums of this artist
960 ref_albums = await self.mass.music.artists.albums(db_artist.item_id, db_artist.provider)
961 if len(ref_albums) < 10:
962 # fetch reference albums from provider(s) attached to the artist
963 for provider_mapping in db_artist.provider_mappings:
964 with contextlib.suppress(ProviderUnavailableError, MediaNotFoundError):
965 ref_albums += await self.mass.music.artists.albums(
966 provider_mapping.item_id, provider_mapping.provider_instance
967 )
968 for ref_album in ref_albums:
969 if ref_album.album_type == AlbumType.COMPILATION:
970 continue
971 if not ref_album.artists:
972 continue
973 search_str = f"{db_artist.name} - {ref_album.name}"
974 search_result_albums = await self.mass.music.albums.search(search_str, provider.domain)
975 for search_result_album in search_result_albums:
976 # only the album's identity matters here: a different edition is still the
977 # same record by the same artist, so the credits below decide the match
978 if not compare_album_name(search_result_album.name, ref_album.name):
979 continue
980 for search_album_artist in search_result_album.artists:
981 if matches := await self._confirm_artist_match(
982 db_artist, search_album_artist, strict
983 ):
984 return matches
985 self.logger.debug(
986 "Could not find match for Artist %s on provider %s",
987 db_artist.name,
988 provider.name,
989 )
990 return []
991
992 async def match_providers(self, db_artist: Artist) -> None:
993 """
994 Try to find matching artists on all providers for the provided (database) item_id.
995
996 This is used to link objects of different providers together.
997 """
998 if db_artist.provider != "library":
999 return # Matching only supported for database items
1000
1001 # try to find match on all providers
1002
1003 cur_provider_domains = {
1004 x.provider_domain for x in db_artist.provider_mappings if x.available
1005 }
1006 for provider in self.mass.music.providers:
1007 if provider.domain in cur_provider_domains:
1008 continue
1009 if ProviderFeature.SEARCH not in provider.supported_features:
1010 continue
1011 if MediaType.ARTIST not in provider.supported_media_types:
1012 continue
1013 if not provider.is_streaming_provider:
1014 # matching on unique providers is pointless as they push (all) their content to MA
1015 continue
1016 if match := await self.match_provider(db_artist, provider):
1017 # 100% match, we update the db with the additional provider mapping(s)
1018 await self.add_provider_mappings(db_artist.item_id, match)
1019 cur_provider_domains.add(provider.domain)
1020
1021 def artist_from_item_mapping(self, item: ItemMapping) -> Artist:
1022 """Create an Artist object from an ItemMapping object."""
1023 domain, instance_id = None, None
1024 if prov := self.mass.get_provider(item.provider):
1025 domain = prov.domain
1026 instance_id = prov.instance_id
1027 return Artist.from_dict(
1028 {
1029 **item.to_dict(),
1030 "provider_mappings": [
1031 {
1032 "item_id": item.item_id,
1033 "provider_domain": domain,
1034 "provider_instance": instance_id,
1035 "available": item.available,
1036 }
1037 ],
1038 }
1039 )
1040
1041 def _validate_provider_filter(
1042 self, provider_instance_id_or_domain: str, provider_filter: str | None
1043 ) -> None:
1044 """Raise when a provider filter is set that does not match the requested provider."""
1045 if provider_filter is not None and provider_filter != provider_instance_id_or_domain:
1046 raise MusicAssistantError(
1047 f"provider_filter '{provider_filter}' does not match the requested "
1048 f"provider '{provider_instance_id_or_domain}'"
1049 )
1050
1051 async def _confirm_artist_match(
1052 self, db_artist: Artist, candidate: Artist | ItemMapping, strict: bool
1053 ) -> list[ProviderMapping]:
1054 """
1055 Return the provider mappings of a candidate artist that confirms as the given artist.
1056
1057 :param candidate: The artist as credited on a search result, which may be a simplified
1058 object without external ids.
1059 """
1060 if not compare_artist(db_artist, candidate, strict=strict):
1061 return []
1062 # only the full artist carries the external ids and artist type that can still reject
1063 # the candidate, so a credit the provider cannot resolve confirms nothing; a credit
1064 # that resolves to a library item is already owned by another artist
1065 with contextlib.suppress(MediaNotFoundError):
1066 prov_artist = await self.get_provider_item(candidate.item_id, candidate.provider)
1067 if prov_artist.provider != "library" and compare_artist(
1068 db_artist, prov_artist, strict=strict
1069 ):
1070 return list(prov_artist.provider_mappings)
1071 return []
1072
1073 async def _add_library_item(
1074 self, item: Artist | ItemMapping, overwrite_existing: bool = False
1075 ) -> int:
1076 """Add a new item record to the database."""
1077 # If item is an ItemMapping, convert it
1078 if isinstance(item, ItemMapping):
1079 item = self.artist_from_item_mapping(item)
1080 # enforce various artists name + id
1081 if compare_strings(item.name, VARIOUS_ARTISTS_NAME):
1082 item.mbid = VARIOUS_ARTISTS_MBID
1083 if item.mbid == VARIOUS_ARTISTS_MBID:
1084 item.name = VARIOUS_ARTISTS_NAME
1085 # no existing item matched: insert item
1086 db_id = await self.mass.music.database.insert(
1087 self.db_table,
1088 {
1089 "name": item.name,
1090 "sort_name": item.sort_name,
1091 "favorite": item.favorite,
1092 "metadata": serialize_to_json(item.metadata),
1093 "search_name": create_safe_string(item.name, True, True),
1094 "search_sort_name": create_safe_string(item.sort_name or "", True, True),
1095 "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
1096 "artist_type": item.artist_type,
1097 },
1098 )
1099 # update/set external id lookup table
1100 await self.set_external_ids(db_id, item.external_ids)
1101 # update/set provider_mappings table
1102 await self.set_provider_mappings(db_id, item.provider_mappings)
1103 self.logger.debug("added %s to database (id: %s)", item.name, db_id)
1104 return db_id
1105
1106 async def _update_library_item(
1107 self, item_id: str | int, update: Artist | ItemMapping, overwrite: bool = False
1108 ) -> None:
1109 """Update existing record in the database."""
1110 db_id = int(item_id) # ensure integer
1111 cur_item = await self.get_library_item(db_id)
1112 if isinstance(update, ItemMapping):
1113 # NOTE that artist is the only mediatype where its accepted we
1114 # receive an itemmapping from streaming providers
1115 update = self.artist_from_item_mapping(update)
1116 metadata = cur_item.metadata
1117 else:
1118 metadata = metadata_for_update(cur_item.metadata, update.metadata, overwrite)
1119 cur_item.external_ids.update(update.external_ids)
1120 # enforce various artists name + id
1121 mbid = cur_item.mbid
1122 if (not mbid or overwrite) and getattr(update, "mbid", None):
1123 if compare_strings(update.name, VARIOUS_ARTISTS_NAME):
1124 update.mbid = VARIOUS_ARTISTS_MBID
1125 if update.mbid == VARIOUS_ARTISTS_MBID:
1126 update.name = VARIOUS_ARTISTS_NAME
1127
1128 name = update.name if overwrite else cur_item.name
1129 sort_name = update.sort_name if overwrite else cur_item.sort_name or update.sort_name
1130 await self.mass.music.database.update(
1131 self.db_table,
1132 {"item_id": db_id},
1133 {
1134 "name": name,
1135 "sort_name": sort_name,
1136 "metadata": serialize_to_json(metadata),
1137 "search_name": create_safe_string(name, True, True),
1138 "search_sort_name": create_safe_string(sort_name or "", True, True),
1139 "timestamp_added": int(update.date_added.timestamp())
1140 if update.date_added
1141 else UNSET,
1142 "artist_type": update.artist_type,
1143 },
1144 )
1145 self.logger.debug("updated %s in database: %s", update.name, db_id)
1146 # update/set external id lookup table
1147 await self.set_external_ids(
1148 db_id, update.external_ids if overwrite else cur_item.external_ids
1149 )
1150 # update/set provider_mappings table
1151 provider_mappings = provider_mappings_for_update(
1152 cur_item.provider_mappings, update.provider_mappings, overwrite
1153 )
1154 await self.set_provider_mappings(db_id, provider_mappings, overwrite)
1155 self.logger.debug("updated %s in database: (id %s)", update.name, db_id)
1156
1157 async def _validate_library_item_merge(self, target: Artist, source: Artist) -> None:
1158 """Validate that two artists have the same role."""
1159 await super()._validate_library_item_merge(target, source)
1160 if target.artist_type != source.artist_type:
1161 msg = (
1162 f"Cannot merge artist '{source.name}' into '{target.name}': "
1163 "artists must have the same role."
1164 )
1165 raise InvalidDataError(msg)
1166
1167 async def _remove_music_artist_from_library(self, db_id: int, recursive: bool) -> None:
1168 # recursively also remove artist albums
1169 for db_row in await self.mass.music.database.get_rows_from_query(
1170 f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = :artist_id",
1171 {"artist_id": db_id},
1172 limit=5000,
1173 ):
1174 if not recursive:
1175 raise MusicAssistantError("Artist still has albums linked")
1176 with contextlib.suppress(MediaNotFoundError):
1177 await self.mass.music.albums.remove_item_from_library(db_row["album_id"])
1178 # recursively also remove artist tracks
1179 for db_row in await self.mass.music.database.get_rows_from_query(
1180 f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = :artist_id",
1181 {"artist_id": db_id},
1182 limit=5000,
1183 ):
1184 if not recursive:
1185 raise MusicAssistantError("Artist still has tracks linked")
1186 with contextlib.suppress(MediaNotFoundError):
1187 await self.mass.music.tracks.remove_item_from_library(db_row["track_id"])
1188
1189 async def _remove_author_narrator_from_library(self, db_id: int, recursive: bool) -> None:
1190 # recursively also remove author/ narrator audiobooks
1191 for db_row in await self.mass.music.database.get_rows_from_query(
1192 f"SELECT audiobook_id FROM {DB_TABLE_AUDIOBOOK_ARTISTS} WHERE artist_id = :artist_id",
1193 {"artist_id": db_id},
1194 limit=5000,
1195 ):
1196 if not recursive:
1197 raise MusicAssistantError("Artist still has audiobooks linked")
1198 with contextlib.suppress(MediaNotFoundError):
1199 await self.mass.music.audiobooks.remove_item_from_library(db_row["audiobook_id"])
1200
1201 async def _get_db_author_narrator_audiobooks(
1202 self, item_id: str, provider_instance_id_or_domain: str, artist_type: ArtistType
1203 ) -> list[Audiobook]:
1204 if db_author_narrator := await self.mass.music.artists.get_library_item_by_prov_id(
1205 item_id,
1206 provider_instance_id_or_domain,
1207 ):
1208 if db_author_narrator.artist_type != artist_type:
1209 self.logger.debug("Artist type must be %s.", artist_type)
1210 return []
1211 db_artist_id = int(db_author_narrator.item_id) # ensure integer
1212 subquery = f"SELECT audiobook_id FROM {DB_TABLE_AUDIOBOOK_ARTISTS} WHERE artist_id = :artist_id"
1213 query = f"audiobooks.item_id in ({subquery})"
1214 return await self.mass.music.audiobooks.get_library_items_by_query(
1215 extra_query_parts=[query],
1216 extra_query_params={"artist_id": db_artist_id},
1217 provider_filter=[provider_instance_id_or_domain],
1218 )
1219 return []
1220
1221 def _parse_summary_row(self, db_row: Mapping[str, Any]) -> ArtistSummary:
1222 """Parse a raw summary db row into an ArtistSummary object."""
1223 item = cast("ArtistSummary", super()._parse_summary_row(db_row))
1224 item.artist_type = ArtistType(db_row["artist_type"])
1225 return item
1226