music-assistant-server

1"""Manage MediaItems of type Album."""
2
3from __future__ import annotations
4
5import contextlib
6from collections.abc import Iterable
7from dataclasses import dataclass
8from typing import TYPE_CHECKING, Any, cast
9
10import aiohttp
11from music_assistant_models.auth import Scope
12from music_assistant_models.enums import AlbumType, ExternalID, MediaType, ProviderFeature
13from music_assistant_models.errors import (
14    InvalidDataError,
15    MediaNotFoundError,
16    MusicAssistantError,
17    RetriesExhausted,
18)
19from music_assistant_models.helpers import create_safe_string
20from music_assistant_models.media_items import (
21    Album,
22    AlbumSummary,
23    Artist,
24    ItemMapping,
25    MediaItemImage,
26    ProviderMapping,
27    Track,
28    UniqueList,
29)
30
31from music_assistant.constants import DB_TABLE_ALBUM_ARTISTS, DB_TABLE_ALBUM_TRACKS, DB_TABLE_ALBUMS
32from music_assistant.controllers.music.helpers import search_name_match_clause
33from music_assistant.helpers.compare import (
34    AlbumMatchEvidence,
35    album_tracks_have_positions,
36    compare_album_evidence,
37    compare_artists,
38    loose_compare_strings,
39)
40from music_assistant.helpers.database import UNSET
41from music_assistant.helpers.external_ids import barcode_to_upc, is_valid_barcode
42from music_assistant.helpers.json import serialize_to_json
43from music_assistant.models.music_provider import MusicProvider
44
45from .base import MediaControllerBase
46
47if TYPE_CHECKING:
48    from collections.abc import Mapping
49
50    from music_assistant import MusicAssistant
51    from music_assistant.providers.musicbrainz import MusicbrainzProvider
52    from music_assistant.providers.musicbrainz.models import MusicBrainzBarcodeRelease
53
54
55# expected failures from a provider album-track lookup: a missing item or a transient
56# provider/transport outage. Either leaves that tracklist unavailable so the (best-effort,
57# multi-provider) match can continue rather than aborting the whole operation.
58_ALBUM_TRACK_LOOKUP_ERRORS = (
59    MediaNotFoundError,
60    RetriesExhausted,
61    TimeoutError,
62    aiohttp.ClientError,
63)
64
65
66@dataclass
67class _BaseTracksMemo:
68    """Single-slot memo holding the tracklist of one base album, resolved on first use."""
69
70    resolved: bool = False
71    tracks: list[Track] | None = None
72
73
74class AlbumsController(MediaControllerBase[Album]):
75    """Controller managing MediaItems of type Album."""
76
77    db_table = DB_TABLE_ALBUMS
78    media_type = MediaType.ALBUM
79    item_cls = Album
80    summary_item_cls = AlbumSummary
81
82    def __init__(self, mass: MusicAssistant) -> None:
83        """Initialize class."""
84        super().__init__(mass)
85        # register (extra) api handlers
86        api_base = self.api_base
87        self.mass.register_api_command(
88            f"music/{api_base}/album_tracks", self.tracks, required_scope=Scope.LIBRARY_READ
89        )
90        self.mass.register_api_command(
91            f"music/{api_base}/album_versions", self.versions, required_scope=Scope.LIBRARY_READ
92        )
93
94    @property
95    def base_query(self) -> tuple[str, dict[str, Any]]:
96        """Return the base SELECT query for albums and its bound query params."""
97        query = f"""
98        SELECT
99            albums.*,
100            {self._external_ids_query()} AS external_ids,
101            {self._provider_mappings_query()} AS provider_mappings,
102            (SELECT JSON_GROUP_ARRAY(
103                json_object(
104                'item_id', artists.item_id,
105                'provider', 'library',
106                    'name', artists.name,
107                    'sort_name', artists.sort_name,
108                    'media_type', 'artist'
109                )) FROM artists JOIN album_artists on album_artists.album_id = albums.item_id  WHERE artists.item_id = album_artists.artist_id) AS artists
110            FROM albums"""
111        return query, {}
112
113    @property
114    def summary_query(self) -> tuple[str, dict[str, Any]]:
115        """Return the slim SELECT query used for album summary listings."""
116        artists_query = self._artist_mappings_summary_query(DB_TABLE_ALBUM_ARTISTS, "album_id")
117        query = f"""
118        SELECT
119            {self._summary_base_columns()},
120            albums.version,
121            albums.year,
122            albums.album_type,
123            {self._provider_mappings_query()} AS provider_mappings,
124            {artists_query} AS artists
125            FROM albums"""
126        return query, {}
127
128    async def get(
129        self,
130        item_id: str,
131        provider_instance_id_or_domain: str,
132        allow_update_metadata: bool = True,
133        recursive: bool = True,
134    ) -> Album:
135        """Return (full) details for a single media item."""
136        album = await super().get(
137            item_id,
138            provider_instance_id_or_domain,
139            allow_update_metadata=allow_update_metadata,
140        )
141        if not recursive:
142            return album
143
144        # append artist details to full album item (resolve ItemMappings)
145        album_artists: UniqueList[Artist | ItemMapping] = UniqueList()
146        for artist in album.artists:
147            if not isinstance(artist, ItemMapping):
148                album_artists.append(artist)
149                continue
150            with contextlib.suppress(MediaNotFoundError):
151                album_artists.append(
152                    await self.mass.music.artists.get(
153                        artist.item_id, artist.provider, allow_update_metadata=False
154                    )
155                )
156        album.artists = album_artists
157        return album
158
159    async def library_items(  # noqa: PLR0913
160        self,
161        favorite: bool | None = None,
162        search: str | None = None,
163        limit: int = 500,
164        offset: int = 0,
165        order_by: str = "sort_name",
166        provider: str | list[str] | None = None,
167        genre: int | list[int] | None = None,
168        played_only: bool = False,
169        album_types: list[AlbumType] | None = None,
170        *,
171        summary: bool = True,
172        reachable_via: list[str] | None = None,
173        **kwargs: Any,
174    ) -> list[Album]:
175        """
176        Get in-database albums.
177
178        :param favorite: Filter by favorite status.
179        :param search: Filter by search query.
180        :param limit: Maximum number of items to return.
181        :param offset: Number of items to skip.
182        :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
183        :param provider: Filter by provider instance ID (single string or list).
184        :param album_types: Filter by album types.
185        :param genre: Filter by genre id(s).
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        extra_join_parts: list[str] = []
198        artist_table_joined = False
199        # optional album type filter
200        if album_types:
201            extra_query_parts.append("albums.album_type IN :album_types")
202            extra_query_params["album_types"] = [x.value for x in album_types]
203        if order_by and "album_artist_name" in order_by:
204            # join artist table to allow sorting on artist name
205            extra_join_parts.append(
206                "JOIN album_artists ON album_artists.album_id = albums.item_id "
207                "JOIN artists ON artists.item_id = album_artists.artist_id "
208            )
209            artist_table_joined = True
210        if search and " - " in search:
211            # handle combined artist + title search
212            artist_str, title_str = search.split(" - ", 1)
213            search = None
214            title_str = create_safe_string(title_str, True, True)
215            artist_str = create_safe_string(artist_str, True, True)
216            extra_query_parts.append(
217                search_name_match_clause("albums", title_str, "search_title", extra_query_params)
218            )
219            artist_clause = "AND " + search_name_match_clause(
220                "artists", artist_str, "search_artist", extra_query_params
221            )
222            # use join with artists table to filter on artist name
223            extra_join_parts.append(
224                "JOIN album_artists ON album_artists.album_id = albums.item_id "
225                "JOIN artists ON artists.item_id = album_artists.artist_id " + artist_clause
226                if not artist_table_joined
227                else artist_clause
228            )
229            artist_table_joined = True
230        result = await self.get_library_items_by_query(
231            favorite=favorite,
232            search=search,
233            genre_ids=genre,
234            limit=limit,
235            offset=offset,
236            order_by=order_by,
237            provider_filter=self._provider_filter_considering_reachability(provider, reachable_via),
238            extra_query_parts=extra_query_parts,
239            extra_query_params=extra_query_params,
240            extra_join_parts=extra_join_parts,
241            played_only=played_only,
242            in_library_only=True,
243            summary=summary,
244            reachable_via=reachable_via,
245        )
246
247        # Calculate how many more items we need to reach the original limit
248        remaining_limit = limit - len(result)
249
250        if search and len(result) < 25 and not offset and remaining_limit > 0:
251            # append artist items to result
252            search = create_safe_string(search, True, True)
253            artist_clause = "AND " + search_name_match_clause(
254                "artists", search, "search_artist", extra_query_params
255            )
256            extra_join_parts.append(
257                "JOIN album_artists ON album_artists.album_id = albums.item_id "
258                "JOIN artists ON artists.item_id = album_artists.artist_id " + artist_clause
259                if not artist_table_joined
260                else artist_clause
261            )
262            existing_uris = {item.uri for item in result}
263
264            for album in await self.get_library_items_by_query(
265                favorite=favorite,
266                search=None,
267                limit=remaining_limit,
268                order_by=order_by,
269                provider_filter=self._provider_filter_considering_reachability(
270                    provider, reachable_via
271                ),
272                extra_query_parts=extra_query_parts,
273                extra_query_params=extra_query_params,
274                extra_join_parts=extra_join_parts,
275                in_library_only=True,
276                summary=summary,
277                reachable_via=reachable_via,
278            ):
279                # prevent duplicates (when artist is also in the title)
280                if album.uri not in existing_uris:
281                    result.append(album)
282                    # Stop if we've reached the original limit
283                    if len(result) >= limit:
284                        break
285        return result
286
287    async def library_count(
288        self, favorite_only: bool = False, album_types: list[AlbumType] | None = None
289    ) -> int:
290        """
291        Return the number of albums in the library.
292
293        Restricted to the providers the current user is allowed to see when that user
294        has a provider filter set.
295
296        :param favorite_only: Only count albums marked as favorite.
297        :param album_types: Only count albums of these types.
298        """
299        sql_query = f"SELECT item_id FROM {self.db_table}"
300        query_parts: list[str] = []
301        query_params: dict[str, Any] = {}
302        if favorite_only:
303            query_parts.append("favorite = 1")
304        if album_types:
305            query_parts.append("albums.album_type IN :album_types")
306            query_params["album_types"] = [x.value for x in album_types]
307        if provider_filter := self._ensure_provider_filter(None):
308            query_parts.append(
309                self._provider_filter_clause(query_params, provider_filter, in_library_only=True)
310            )
311        if query_parts:
312            sql_query += f" WHERE {' AND '.join(query_parts)}"
313        return await self.mass.music.database.get_count_from_query(sql_query, query_params)
314
315    async def remove_item_from_library(self, item_id: str | int, recursive: bool = True) -> None:
316        """Delete item from the library(database)."""
317        db_id = int(item_id)  # ensure integer
318        # recursively also remove album tracks
319        for db_track in await self.get_library_album_tracks(db_id):
320            if not recursive:
321                raise MusicAssistantError("Album still has tracks linked")
322            with contextlib.suppress(MediaNotFoundError):
323                await self.mass.music.tracks.remove_item_from_library(db_track.item_id)
324        # delete entry(s) from albumtracks table
325        await self.mass.music.database.delete(DB_TABLE_ALBUM_TRACKS, {"album_id": db_id})
326        # delete entry(s) from album artists table
327        await self.mass.music.database.delete(DB_TABLE_ALBUM_ARTISTS, {"album_id": db_id})
328        # delete the album itself from db
329        # this will raise if the item still has references and recursive is false
330        await super().remove_item_from_library(item_id)
331
332    async def set_release_group(
333        self,
334        album_item_id: int,
335        release_group_mbid: str,
336    ) -> None:
337        """
338        Persist a MusicBrainz release-group ID on a library album, idempotently.
339
340        :param album_item_id: Library album item_id (database id).
341        :param release_group_mbid: MusicBrainz release-group UUID to set.
342        """
343        if not release_group_mbid:
344            return
345        try:
346            album = await self.get_library_item(album_item_id)
347        except MusicAssistantError as err:
348            self.logger.debug("set_release_group: cannot load album %s: %s", album_item_id, err)
349            return
350        # Refuse to overwrite — keeps tag-sourced or already-enriched IDs authoritative.
351        if album.get_external_id(ExternalID.MB_RELEASEGROUP):
352            self.logger.debug(
353                "set_release_group: album %s already has MB_RELEASEGROUP — keeping",
354                album_item_id,
355            )
356            return
357        album.add_external_id(ExternalID.MB_RELEASEGROUP, release_group_mbid)
358        await self.update_item_in_library(album_item_id, album)
359        self.logger.debug(
360            "set_release_group: wrote %s onto album %s", release_group_mbid, album_item_id
361        )
362
363    async def tracks(
364        self,
365        item_id: str,
366        provider_instance_id_or_domain: str,
367        in_library_only: bool = False,
368    ) -> list[Track]:
369        """Return album tracks for the given provider album id."""
370        # always check if we have a library item for this album
371        library_album = await self.get_library_item_by_prov_id(
372            item_id, provider_instance_id_or_domain
373        )
374        if not library_album:
375            album_tracks = await self._get_provider_album_tracks(
376                item_id, provider_instance_id_or_domain
377            )
378            # some album-track listings omit the parent album and its image; backfill both
379            # from the provider album so the queue shows the album name and artwork.
380            if album_tracks and (not album_tracks[0].album or not album_tracks[0].image):
381                prov_album = await self.get_provider_item(item_id, provider_instance_id_or_domain)
382                album_mapping = ItemMapping.from_item(prov_album)
383                for track in album_tracks:
384                    if prov_album.image and not track.image:
385                        track.metadata.add_image(prov_album.image)
386                    if track.album is None:
387                        track.album = album_mapping
388            return album_tracks
389
390        # respect the current user's provider filter (if any) for both the
391        # in-library tracks and the live provider fetches below
392        allowed_providers = self._ensure_provider_filter(None)
393        db_items = await self.get_library_album_tracks(
394            library_album.item_id, provider_filter=allowed_providers
395        )
396        result: list[Track] = list(db_items)
397        if in_library_only:
398            # return in-library items only
399            return sorted(db_items, key=lambda x: (x.disc_number, x.track_number))
400
401        # return all (unique) items from all providers
402        # because we are returning the items from all providers combined,
403        # we need to make sure that we don't return duplicates
404        unique_ids: set[str] = {f"{x.disc_number}.{x.track_number}" for x in db_items}
405        unique_ids.update({f"{x.name.lower()}.{x.version.lower()}" for x in db_items})
406        for db_item in db_items:
407            unique_ids.update(x.item_id for x in db_item.provider_mappings)
408        for provider_mapping in library_album.provider_mappings:
409            if (
410                allowed_providers is not None
411                and provider_mapping.provider_instance not in allowed_providers
412            ):
413                continue
414            provider_tracks = await self._get_provider_album_tracks(
415                provider_mapping.item_id, provider_mapping.provider_instance
416            )
417            for provider_track in provider_tracks:
418                # In some cases (looking at you YTM) the disc/track number is not obtained from
419                # library_tracks. Ensure to update the disc/track number when interacting with
420                # album tracks
421                db_track = next(
422                    (
423                        x
424                        for x in db_items
425                        if x.sort_name == provider_track.sort_name
426                        and x.version == provider_track.version
427                    ),
428                    None,
429                )
430                if (
431                    db_track
432                    and db_track.track_number == 0
433                    and db_track.track_number != provider_track.track_number
434                ):
435                    await self._set_album_track(
436                        db_id=int(library_album.item_id),
437                        db_track_id=int(db_track.item_id),
438                        track=provider_track,
439                    )
440                if provider_track.item_id in unique_ids:
441                    continue
442                unique_id = f"{provider_track.disc_number}.{provider_track.track_number}"
443                if unique_id in unique_ids:
444                    continue
445                unique_id = f"{provider_track.name.lower()}.{provider_track.version.lower()}"
446                if unique_id in unique_ids:
447                    continue
448                unique_ids.add(unique_id)
449                provider_track.album = library_album
450                # always prefer album image
451                album_images = [library_album.image] if library_album.image else []
452                track_images: list[MediaItemImage] = provider_track.metadata.images or []
453                provider_track.metadata.images = UniqueList(album_images + track_images)
454                result.append(provider_track)
455        # NOTE: we need to return the results sorted on disc/track here
456        # to ensure the correct order at playback
457        return sorted(result, key=lambda x: (x.disc_number, x.track_number))
458
459    async def versions(
460        self,
461        item_id: str,
462        provider_instance_id_or_domain: str,
463    ) -> UniqueList[Album]:
464        """Return all versions of an album we can find on all providers."""
465        album = await self.get_provider_item(item_id, provider_instance_id_or_domain)
466        streaming_search_query = (
467            f"{album.artists[0].name} - {album.name}" if album.artists else album.name
468        )
469        result: UniqueList[Album] = UniqueList()
470        for provider_id in self.mass.music.get_unique_providers():
471            provider = self.mass.get_provider(provider_id)
472            if not provider or not isinstance(provider, MusicProvider):
473                continue
474            if not self.mass.music.library_supported(provider, MediaType.ALBUM):
475                continue
476            # TODO: filter by artists in db for non-streaming providers
477            search_query = streaming_search_query if provider.is_streaming_provider else album.name
478            result.extend(
479                prov_item
480                for prov_item in await self.search(search_query, provider_id)
481                if loose_compare_strings(album.name, prov_item.name)
482                and compare_artists(prov_item.artists, album.artists, any_match=True)
483                # make sure that the 'base' version is NOT included
484                and not album.provider_mappings.intersection(prov_item.provider_mappings)
485            )
486        return result
487
488    async def get_library_album_tracks(
489        self,
490        item_id: str | int,
491        provider_filter: list[str] | None = None,
492    ) -> list[Track]:
493        """
494        Return in-database album tracks for the given database album.
495
496        :param item_id: The library item ID of the album.
497        :param provider_filter: Optional provider instance ID(s) to limit the result to.
498        """
499        db_id = int(item_id)  # ensure integer
500        # pass the album id as preferred album so the track_album subquery in the
501        # base query returns this album's disc/track numbers for tracks that
502        # appear on multiple albums
503        return await self.mass.music.tracks.get_library_items_by_query(
504            provider_filter=provider_filter,
505            extra_query_parts=[
506                f"tracks.item_id IN (SELECT track_id FROM {DB_TABLE_ALBUM_TRACKS} "
507                "WHERE album_id = :album_id)"
508            ],
509            extra_query_params={"album_id": db_id, "preferred_album_id": db_id},
510        )
511
512    async def add_item_mapping_as_album_to_library(self, item: ItemMapping) -> Album:
513        """
514        Add an ItemMapping as an Album to the library.
515
516        This is only used in special occasions as is basically adds an album
517        to the db without a lot of mandatory data, such as artists.
518        """
519        album = self.album_from_item_mapping(item)
520        return await self.add_item_to_library(album)
521
522    async def match_provider(
523        self, db_album: Album, provider: MusicProvider, strict: bool = True
524    ) -> list[ProviderMapping]:
525        """
526        Try to find a match on the given (streaming) provider for a (database) album.
527
528        Links albums of different providers/qualities together. Sparse provider search
529        results only rule out a confident non-match; a candidate that still looks
530        ambiguous is confirmed against the full provider album, its tracklist and, as a
531        last resort, MusicBrainz before its provider mapping is accepted.
532        """
533        return await self._match_provider(db_album, provider, strict, _BaseTracksMemo())
534
535    async def match_providers(self, db_album: Album) -> None:
536        """
537        Try to find match on all (streaming) providers for the provided (database) album.
538
539        This is used to link objects of different providers/qualities together.
540        """
541        if db_album.provider != "library":
542            return  # Matching only supported for database items
543        if not db_album.artists:
544            return  # guard
545
546        # resolve the base tracklist at most once for the whole match operation
547        base_tracks_memo = _BaseTracksMemo()
548        # try to find match on all providers
549        processed_domains = set()
550        for provider in self.mass.music.providers:
551            if provider.domain in processed_domains:
552                continue
553            if ProviderFeature.SEARCH not in provider.supported_features:
554                continue
555            if not self.mass.music.library_supported(provider, MediaType.ALBUM):
556                continue
557            if not provider.is_streaming_provider:
558                # matching on unique providers is pointless as they push (all) their content to MA
559                continue
560            if match := await self._match_provider(db_album, provider, True, base_tracks_memo):
561                # 100% match, we update the db with the additional provider mapping(s)
562                await self.add_provider_mappings(db_album.item_id, match)
563                processed_domains.add(provider.domain)
564
565    def album_from_item_mapping(self, item: ItemMapping) -> Album:
566        """Create an Album object from an ItemMapping object."""
567        domain, instance_id = None, None
568        if prov := self.mass.get_provider(item.provider):
569            domain = prov.domain
570            instance_id = prov.instance_id
571        return Album.from_dict(
572            {
573                **item.to_dict(),
574                "provider_mappings": [
575                    {
576                        "item_id": item.item_id,
577                        "provider_domain": domain,
578                        "provider_instance": instance_id,
579                        "available": item.available,
580                    }
581                ],
582            }
583        )
584
585    async def _add_library_item(self, item: Album, overwrite_existing: bool = False) -> int:
586        """Add a new record to the database."""
587        if not isinstance(item, Album):  # TODO: Remove this once the codebase is fully typed
588            msg = "Not a valid Album object (ItemMapping can not be added to db)"  # type: ignore[unreachable]
589            raise InvalidDataError(msg)
590        db_id = await self.mass.music.database.insert(
591            self.db_table,
592            {
593                "name": item.name,
594                "sort_name": item.sort_name,
595                "version": item.version,
596                "favorite": item.favorite,
597                "album_type": item.album_type,
598                "year": item.year,
599                "metadata": serialize_to_json(item.metadata),
600                "search_name": create_safe_string(item.name, True, True),
601                "search_sort_name": create_safe_string(item.sort_name or "", True, True),
602                "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
603            },
604        )
605        # update/set external id lookup table
606        await self.set_external_ids(db_id, item.external_ids)
607        # update/set provider_mappings table
608        await self.set_provider_mappings(db_id, item.provider_mappings)
609        # set track artist(s)
610        await self._set_album_artists(db_id, item.artists)
611        self.logger.debug("added %s to database (id: %s)", item.name, db_id)
612        return db_id
613
614    async def _update_library_item(
615        self, item_id: str | int, update: Album, overwrite: bool = False
616    ) -> None:
617        """Update existing record in the database."""
618        db_id = int(item_id)  # ensure integer
619        cur_item = await self.get_library_item(db_id)
620        metadata = update.metadata if overwrite else cur_item.metadata.update(update.metadata)
621        if getattr(update, "album_type", AlbumType.UNKNOWN) != AlbumType.UNKNOWN:
622            album_type = update.album_type
623        else:
624            album_type = cur_item.album_type
625        cur_item.external_ids.update(update.external_ids)
626        name = update.name if overwrite else cur_item.name
627        sort_name = update.sort_name if overwrite else cur_item.sort_name or update.sort_name
628        await self.mass.music.database.update(
629            self.db_table,
630            {"item_id": db_id},
631            {
632                "name": name,
633                "sort_name": sort_name,
634                "version": update.version if overwrite else cur_item.version or update.version,
635                "year": update.year if overwrite else cur_item.year or update.year,
636                "album_type": album_type.value,
637                "metadata": serialize_to_json(metadata),
638                "search_name": create_safe_string(name, True, True),
639                "search_sort_name": create_safe_string(sort_name or "", True, True),
640                "timestamp_added": int(update.date_added.timestamp())
641                if update.date_added
642                else UNSET,
643            },
644        )
645        # update/set external id lookup table
646        await self.set_external_ids(
647            db_id, update.external_ids if overwrite else cur_item.external_ids
648        )
649        # update/set provider_mappings table
650        provider_mappings = (
651            update.provider_mappings
652            if overwrite
653            else {*update.provider_mappings, *cur_item.provider_mappings}
654        )
655        await self.set_provider_mappings(db_id, provider_mappings, overwrite)
656        # set album artist(s)
657        artists = update.artists if overwrite else cur_item.artists + update.artists
658        await self._set_album_artists(db_id, artists, overwrite=overwrite)
659        self.logger.debug("updated %s in database: (id %s)", update.name, db_id)
660
661    async def _get_provider_album_tracks(
662        self, item_id: str, provider_instance_id_or_domain: str
663    ) -> list[Track]:
664        """Return album tracks for the given provider album id."""
665        if prov := self.mass.get_provider(provider_instance_id_or_domain):
666            prov = cast("MusicProvider", prov)
667            return await prov.get_album_tracks(item_id)
668        return []
669
670    async def _confirm_library_candidate(self, db_item: Album, item: Album | ItemMapping) -> bool:
671        """
672        Return True if a library album is the same album as the one being added.
673
674        An edition that cannot be decided on the albums' own metadata is escalated to
675        tracklists and MusicBrainz, so an ambiguous album is linked to the album it
676        belongs to instead of becoming a second library entry.
677        """
678        if not isinstance(item, Album):
679            return await super()._confirm_library_candidate(db_item, item)
680        evidence = compare_album_evidence(db_item, item, strict=True)
681        if evidence != AlbumMatchEvidence.INSUFFICIENT:
682            return evidence == AlbumMatchEvidence.MATCH
683        provider = self.mass.get_provider(item.provider, provider_type=MusicProvider)
684        if provider is None or provider.instance_id != item.provider:
685            # only the exact provider instance the album came from may be fingerprinted,
686            # never a same-domain fallback pointing at a different account/server
687            return False
688        evidence = await self._resolve_album_evidence(
689            db_item, item, provider, True, _BaseTracksMemo()
690        )
691        return evidence == AlbumMatchEvidence.MATCH
692
693    async def _match_provider(
694        self,
695        db_album: Album,
696        provider: MusicProvider,
697        strict: bool,
698        base_tracks_memo: _BaseTracksMemo,
699    ) -> list[ProviderMapping]:
700        """Search one provider and return the mappings of every confirmed album match."""
701        self.logger.debug("Trying to match album %s on provider %s", db_album.name, provider.name)
702        matches: list[ProviderMapping] = []
703        artist_name = db_album.artists[0].name
704        search_str = f"{artist_name} - {db_album.name}"
705        for search_result_item in await self.search(search_str, provider.instance_id):
706            if not search_result_item.available:
707                continue
708            # a sparse search result only rules out a confident non-match; a MATCH or an
709            # ambiguous (INSUFFICIENT) candidate is confirmed against the full album below
710            if (
711                compare_album_evidence(db_album, search_result_item, strict=strict)
712                == AlbumMatchEvidence.NO_MATCH
713            ):
714                continue
715            # search results can be simplified objects, so fetch the full provider album
716            prov_album = await self.get_provider_item(
717                search_result_item.item_id,
718                search_result_item.provider,
719                fallback=search_result_item,
720            )
721            evidence = await self._resolve_album_evidence(
722                db_album, prov_album, provider, strict, base_tracks_memo
723            )
724            if evidence == AlbumMatchEvidence.MATCH:
725                matches.extend(prov_album.provider_mappings)
726        if not matches:
727            self.logger.debug(
728                "Could not find match for Album %s on provider %s",
729                db_album.name,
730                provider.name,
731            )
732        return matches
733
734    async def _resolve_album_evidence(
735        self,
736        db_album: Album,
737        prov_album: Album,
738        provider: MusicProvider,
739        strict: bool,
740        base_tracks_memo: _BaseTracksMemo,
741    ) -> AlbumMatchEvidence:
742        """
743        Return the match evidence for a fully-fetched provider album.
744
745        An ambiguous album is escalated to ordered track fingerprints and, only if those
746        stay inconclusive, to MusicBrainz; a mapping is accepted only on a MATCH.
747
748        :param provider: The exact provider instance the candidate album was matched on;
749            its tracklist is fetched directly so a same-domain fallback can never
750            fingerprint the candidate against a different account/server.
751        """
752        evidence = compare_album_evidence(db_album, prov_album, strict=strict)
753        if evidence != AlbumMatchEvidence.INSUFFICIENT:
754            return evidence
755        # ambiguous metadata: resolve conservatively with ordered track fingerprints
756        base_tracks = await self._resolve_base_album_tracks(db_album, base_tracks_memo)
757        try:
758            compare_tracks = await provider.get_album_tracks(prov_album.item_id)
759        except _ALBUM_TRACK_LOOKUP_ERRORS as err:
760            # the candidate tracklist is unavailable: treat it as absent and let MusicBrainz decide
761            self.logger.debug(
762                "Album tracks unavailable for %s on %s: %s",
763                prov_album.item_id,
764                provider.instance_id,
765                err,
766            )
767            compare_tracks = []
768        evidence = compare_album_evidence(
769            db_album,
770            prov_album,
771            strict=strict,
772            base_tracks=base_tracks,
773            compare_tracks=compare_tracks,
774        )
775        if evidence != AlbumMatchEvidence.INSUFFICIENT:
776            return evidence
777        # tracklists could not resolve it either: consult MusicBrainz as a last resort
778        return await self._musicbrainz_album_evidence(db_album, prov_album)
779
780    async def _resolve_base_album_tracks(
781        self, db_album: Album, base_tracks_memo: _BaseTracksMemo
782    ) -> list[Track] | None:
783        """Return the memoized base tracklist, resolving it once on first use."""
784        if not base_tracks_memo.resolved:
785            base_tracks_memo.tracks = await self._load_base_album_tracks(db_album)
786            base_tracks_memo.resolved = True
787        return base_tracks_memo.tracks
788
789    async def _load_base_album_tracks(self, db_album: Album) -> list[Track] | None:
790        """
791        Return a complete, ordered base tracklist to fingerprint against.
792
793        Iterates the album's existing provider mappings in a deterministic order and
794        returns the first loaded provider's full tracklist whose disc/track positions can
795        be trusted. A provider-sourced tracklist is used rather than the stored library
796        tracks because those can be an incomplete subset (individually added tracks), and
797        an incomplete base would make a track-count difference look like a real conflict.
798        """
799        for mapping in sorted(
800            db_album.provider_mappings,
801            key=lambda mapping: (
802                mapping.provider_domain,
803                mapping.provider_instance,
804                mapping.item_id,
805            ),
806        ):
807            if not mapping.available:
808                continue
809            provider = self.mass.get_provider(mapping.provider_instance, return_unavailable=True)
810            if (
811                provider is None
812                or provider.instance_id != mapping.provider_instance
813                or not provider.available
814            ):
815                # only trust the exact, currently-available provider instance and never a
816                # same-domain fallback pointing at a different account/server
817                continue
818            try:
819                provider_tracks = await self._get_provider_album_tracks(
820                    mapping.item_id, mapping.provider_instance
821                )
822            except _ALBUM_TRACK_LOOKUP_ERRORS as err:
823                # this mapping's tracklist is unavailable: try the next existing mapping
824                self.logger.debug(
825                    "Base album tracks unavailable for %s on %s: %s",
826                    mapping.item_id,
827                    mapping.provider_instance,
828                    err,
829                )
830                continue
831            if album_tracks_have_positions(provider_tracks):
832                return provider_tracks
833        return None
834
835    async def _musicbrainz_album_evidence(
836        self, base_album: Album, compare_album: Album
837    ) -> AlbumMatchEvidence:
838        """
839        Return album match evidence from MusicBrainz release identity, or abstain.
840
841        A barcode that resolves unambiguously to a single specific MusicBrainz release on
842        both albums is strong positive evidence; barcodes belonging to entirely different
843        release groups are negative. A barcode resolving to several releases, a shared
844        release group alone, an unresolved barcode or a lookup failure abstains
845        (INSUFFICIENT) rather than guessing.
846        """
847        base_barcodes = _canonical_album_barcodes(base_album)
848        compare_barcodes = _canonical_album_barcodes(compare_album)
849        if not base_barcodes or not compare_barcodes:
850            return AlbumMatchEvidence.INSUFFICIENT
851        musicbrainz = self.mass.get_provider("musicbrainz")
852        if musicbrainz is None:
853            return AlbumMatchEvidence.INSUFFICIENT
854        musicbrainz = cast("MusicbrainzProvider", musicbrainz)
855        releases_by_barcode: dict[str, list[MusicBrainzBarcodeRelease]] = {}
856        try:
857            for barcode in sorted(base_barcodes | compare_barcodes):
858                releases_by_barcode[barcode] = await musicbrainz.get_releases_by_barcode(barcode)
859        except (RetriesExhausted, InvalidDataError, TimeoutError, aiohttp.ClientError) as err:
860            self.logger.debug(
861                "MusicBrainz barcode lookup failed while matching album %s: %s",
862                base_album.name,
863                err,
864            )
865            return AlbumMatchEvidence.INSUFFICIENT
866        base_release_ids = _unambiguous_release_ids(base_barcodes, releases_by_barcode)
867        compare_release_ids = _unambiguous_release_ids(compare_barcodes, releases_by_barcode)
868        if base_release_ids & compare_release_ids:
869            # both albums carry a barcode that names the same single specific release
870            return AlbumMatchEvidence.MATCH
871        if not all(releases_by_barcode[barcode] for barcode in base_barcodes | compare_barcodes):
872            # an unresolved barcode leaves the release-group sets incomplete, so a disjoint
873            # comparison could wrongly reject regional equivalents: abstain instead
874            return AlbumMatchEvidence.INSUFFICIENT
875        base_group_ids = _release_group_ids(base_barcodes, releases_by_barcode)
876        compare_group_ids = _release_group_ids(compare_barcodes, releases_by_barcode)
877        if base_group_ids.isdisjoint(compare_group_ids):
878            # the barcodes belong to entirely different release groups: different albums
879            return AlbumMatchEvidence.NO_MATCH
880        # a shared release group alone (or an ambiguous barcode) never identifies an edition
881        return AlbumMatchEvidence.INSUFFICIENT
882
883    async def _set_album_artists(
884        self,
885        db_id: int,
886        artists: Iterable[Artist | ItemMapping],
887        overwrite: bool = False,
888    ) -> None:
889        """Store Album Artists."""
890        if overwrite:
891            # on overwrite, clear the album_artists table first
892            await self.mass.music.database.delete(
893                DB_TABLE_ALBUM_ARTISTS,
894                {
895                    "album_id": db_id,
896                },
897            )
898        for artist in artists:
899            await self._set_album_artist(db_id, artist=artist, overwrite=overwrite)
900
901    async def _set_album_artist(
902        self, db_id: int, artist: Artist | ItemMapping, overwrite: bool = False
903    ) -> ItemMapping:
904        """Store Album Artist info."""
905        db_artist: Artist | ItemMapping | None = None
906        if artist.provider == "library":
907            db_artist = artist
908        elif existing := await self.mass.music.artists.get_library_item_by_prov_id(
909            artist.item_id, artist.provider
910        ):
911            db_artist = existing
912
913        if not db_artist or overwrite:
914            # Convert ItemMapping to Artist if needed
915            artist_to_add = (
916                self.mass.music.artists.artist_from_item_mapping(artist)
917                if isinstance(artist, ItemMapping)
918                else artist
919            )
920            db_artist = await self.mass.music.artists.add_item_to_library(
921                artist_to_add, overwrite_existing=overwrite
922            )
923        # write (or update) record in album_artists table
924        await self.mass.music.database.insert_or_replace(
925            DB_TABLE_ALBUM_ARTISTS,
926            {
927                "album_id": db_id,
928                "artist_id": int(db_artist.item_id),
929            },
930        )
931        return ItemMapping.from_item(db_artist)
932
933    async def _set_album_track(self, db_id: int, db_track_id: int, track: Track) -> None:
934        """Store Album Track info."""
935        # write (or update) record in album_tracks table
936        await self.mass.music.database.insert_or_replace(
937            DB_TABLE_ALBUM_TRACKS,
938            {
939                "album_id": db_id,
940                "track_id": db_track_id,
941                "track_number": track.track_number,
942                "disc_number": track.disc_number,
943            },
944        )
945
946    def _parse_summary_row(self, db_row: Mapping[str, Any]) -> AlbumSummary:
947        """Parse a raw summary db row into an AlbumSummary object."""
948        item = cast("AlbumSummary", super()._parse_summary_row(db_row))
949        item.version = db_row["version"] or ""
950        item.year = db_row["year"]
951        item.album_type = AlbumType(db_row["album_type"])
952        item.artists = self._parse_summary_artist_mappings(db_row)
953        return item
954
955
956def _canonical_album_barcodes(album: Album) -> set[str]:
957    """Return an album's valid barcodes in canonical UPC form."""
958    return {
959        barcode_to_upc(value)
960        for external_id_type, value in album.external_ids
961        if external_id_type == ExternalID.BARCODE and is_valid_barcode(value)
962    }
963
964
965def _unambiguous_release_ids(
966    barcodes: set[str], releases_by_barcode: dict[str, list[MusicBrainzBarcodeRelease]]
967) -> set[str]:
968    """Return release ids that at least one of the barcodes resolves to unambiguously."""
969    release_ids: set[str] = set()
970    for barcode in barcodes:
971        resolved = {release.id for release in releases_by_barcode.get(barcode, [])}
972        # only a barcode that maps to exactly one specific release is trustworthy evidence
973        if len(resolved) == 1:
974            release_ids |= resolved
975    return release_ids
976
977
978def _release_group_ids(
979    barcodes: set[str], releases_by_barcode: dict[str, list[MusicBrainzBarcodeRelease]]
980) -> set[str]:
981    """Return every release-group id the barcodes resolve to."""
982    return {
983        release.release_group.id
984        for barcode in barcodes
985        for release in releases_by_barcode.get(barcode, [])
986    }
987