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