music-assistant-server

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