/
/
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 search_str = (
711 f"{db_album.artists[0].name} - {db_album.name}" if db_album.artists else db_album.name
712 )
713 for search_result_item in await self.search(search_str, provider.instance_id):
714 if not search_result_item.available:
715 continue
716 # a sparse search result only rules out a confident non-match; a MATCH or an
717 # ambiguous (INSUFFICIENT) candidate is confirmed against the full album below
718 if (
719 compare_album_evidence(db_album, search_result_item, strict=strict)
720 == AlbumMatchEvidence.NO_MATCH
721 ):
722 continue
723 # search results can be simplified objects, so fetch the full provider album
724 prov_album = await self.get_provider_item(
725 search_result_item.item_id,
726 search_result_item.provider,
727 fallback=search_result_item,
728 )
729 evidence = await self._resolve_album_evidence(
730 db_album, prov_album, provider, strict, base_tracks_memo
731 )
732 if evidence == AlbumMatchEvidence.MATCH:
733 matches.extend(prov_album.provider_mappings)
734 if not matches:
735 self.logger.debug(
736 "Could not find match for Album %s on provider %s",
737 db_album.name,
738 provider.name,
739 )
740 return matches
741
742 async def _resolve_album_evidence(
743 self,
744 db_album: Album,
745 prov_album: Album,
746 provider: MusicProvider,
747 strict: bool,
748 base_tracks_memo: _BaseTracksMemo,
749 ) -> AlbumMatchEvidence:
750 """
751 Return the match evidence for a fully-fetched provider album.
752
753 An ambiguous album is escalated to ordered track fingerprints and, only if those
754 stay inconclusive, to MusicBrainz; a mapping is accepted only on a MATCH.
755
756 :param provider: The exact provider instance the candidate album was matched on;
757 its tracklist is fetched directly so a same-domain fallback can never
758 fingerprint the candidate against a different account/server.
759 """
760 evidence = compare_album_evidence(db_album, prov_album, strict=strict)
761 if evidence != AlbumMatchEvidence.INSUFFICIENT:
762 return evidence
763 # ambiguous metadata: resolve conservatively with ordered track fingerprints
764 base_tracks = await self._resolve_base_album_tracks(db_album, base_tracks_memo)
765 try:
766 compare_tracks = await provider.get_album_tracks(prov_album.item_id)
767 except _ALBUM_TRACK_LOOKUP_ERRORS as err:
768 # the candidate tracklist is unavailable: treat it as absent and let MusicBrainz decide
769 self.logger.debug(
770 "Album tracks unavailable for %s on %s: %s",
771 prov_album.item_id,
772 provider.instance_id,
773 err,
774 )
775 compare_tracks = []
776 evidence = compare_album_evidence(
777 db_album,
778 prov_album,
779 strict=strict,
780 base_tracks=base_tracks,
781 compare_tracks=compare_tracks,
782 )
783 if evidence != AlbumMatchEvidence.INSUFFICIENT:
784 return evidence
785 # tracklists could not resolve it either: consult MusicBrainz as a last resort
786 return await self._musicbrainz_album_evidence(db_album, prov_album)
787
788 async def _resolve_base_album_tracks(
789 self, db_album: Album, base_tracks_memo: _BaseTracksMemo
790 ) -> list[Track] | None:
791 """Return the memoized base tracklist, resolving it once on first use."""
792 if not base_tracks_memo.resolved:
793 base_tracks_memo.tracks = await self._load_base_album_tracks(db_album)
794 base_tracks_memo.resolved = True
795 return base_tracks_memo.tracks
796
797 async def _load_base_album_tracks(self, db_album: Album) -> list[Track] | None:
798 """
799 Return a complete, ordered base tracklist to fingerprint against.
800
801 Iterates the album's existing provider mappings in a deterministic order and
802 returns the first loaded provider's full tracklist whose disc/track positions can
803 be trusted. A provider-sourced tracklist is used rather than the stored library
804 tracks because those can be an incomplete subset (individually added tracks), and
805 an incomplete base would make a track-count difference look like a real conflict.
806 """
807 for mapping in sorted(
808 db_album.provider_mappings,
809 key=lambda mapping: (
810 mapping.provider_domain,
811 mapping.provider_instance,
812 mapping.item_id,
813 ),
814 ):
815 if not mapping.available:
816 continue
817 provider = self.mass.get_provider(mapping.provider_instance, return_unavailable=True)
818 if (
819 provider is None
820 or provider.instance_id != mapping.provider_instance
821 or not provider.available
822 ):
823 # only trust the exact, currently-available provider instance and never a
824 # same-domain fallback pointing at a different account/server
825 continue
826 try:
827 provider_tracks = await self._get_provider_album_tracks(
828 mapping.item_id, mapping.provider_instance
829 )
830 except _ALBUM_TRACK_LOOKUP_ERRORS as err:
831 # this mapping's tracklist is unavailable: try the next existing mapping
832 self.logger.debug(
833 "Base album tracks unavailable for %s on %s: %s",
834 mapping.item_id,
835 mapping.provider_instance,
836 err,
837 )
838 continue
839 if album_tracks_have_positions(provider_tracks):
840 return provider_tracks
841 return None
842
843 async def _musicbrainz_album_evidence(
844 self, base_album: Album, compare_album: Album
845 ) -> AlbumMatchEvidence:
846 """
847 Return album match evidence from MusicBrainz release identity, or abstain.
848
849 A barcode that resolves unambiguously to a single specific MusicBrainz release on
850 both albums is strong positive evidence; barcodes belonging to entirely different
851 release groups are negative. A barcode resolving to several releases, a shared
852 release group alone, an unresolved barcode or a lookup failure abstains
853 (INSUFFICIENT) rather than guessing.
854 """
855 base_barcodes = _canonical_album_barcodes(base_album)
856 compare_barcodes = _canonical_album_barcodes(compare_album)
857 if not base_barcodes or not compare_barcodes:
858 return AlbumMatchEvidence.INSUFFICIENT
859 musicbrainz = self.mass.get_provider("musicbrainz")
860 if musicbrainz is None:
861 return AlbumMatchEvidence.INSUFFICIENT
862 musicbrainz = cast("MusicbrainzProvider", musicbrainz)
863 releases_by_barcode: dict[str, list[MusicBrainzBarcodeRelease]] = {}
864 try:
865 for barcode in sorted(base_barcodes | compare_barcodes):
866 releases_by_barcode[barcode] = await musicbrainz.get_releases_by_barcode(barcode)
867 except (RetriesExhausted, InvalidDataError, TimeoutError, aiohttp.ClientError) as err:
868 self.logger.debug(
869 "MusicBrainz barcode lookup failed while matching album %s: %s",
870 base_album.name,
871 err,
872 )
873 return AlbumMatchEvidence.INSUFFICIENT
874 base_release_ids = _unambiguous_release_ids(base_barcodes, releases_by_barcode)
875 compare_release_ids = _unambiguous_release_ids(compare_barcodes, releases_by_barcode)
876 if base_release_ids & compare_release_ids:
877 # both albums carry a barcode that names the same single specific release
878 return AlbumMatchEvidence.MATCH
879 if not all(releases_by_barcode[barcode] for barcode in base_barcodes | compare_barcodes):
880 # an unresolved barcode leaves the release-group sets incomplete, so a disjoint
881 # comparison could wrongly reject regional equivalents: abstain instead
882 return AlbumMatchEvidence.INSUFFICIENT
883 base_group_ids = _release_group_ids(base_barcodes, releases_by_barcode)
884 compare_group_ids = _release_group_ids(compare_barcodes, releases_by_barcode)
885 if base_group_ids.isdisjoint(compare_group_ids):
886 # the barcodes belong to entirely different release groups: different albums
887 return AlbumMatchEvidence.NO_MATCH
888 # a shared release group alone (or an ambiguous barcode) never identifies an edition
889 return AlbumMatchEvidence.INSUFFICIENT
890
891 async def _set_album_artists(
892 self,
893 db_id: int,
894 artists: Iterable[Artist | ItemMapping],
895 overwrite: bool = False,
896 ) -> None:
897 """
898 Store Album Artists.
899
900 An empty set of artists never clears the stored rows: an album that lost its
901 artists disappears from their discography and is skipped by provider matching.
902 """
903 all_artists = list(artists)
904 if not all_artists:
905 if overwrite:
906 # a caller asking to replace all artists with none is a bug,
907 # so keep the stored rows and make the attempt visible
908 self.logger.warning("Ignoring request to clear all artists of album id %s", db_id)
909 return
910 if overwrite:
911 # on overwrite, clear the album_artists table first
912 await self.mass.music.database.delete(
913 DB_TABLE_ALBUM_ARTISTS,
914 {
915 "album_id": db_id,
916 },
917 )
918 for artist in all_artists:
919 await self._set_album_artist(db_id, artist=artist, overwrite=overwrite)
920
921 async def _set_album_artist(
922 self, db_id: int, artist: Artist | ItemMapping, overwrite: bool = False
923 ) -> ItemMapping:
924 """Store Album Artist info."""
925 db_artist: Artist | ItemMapping | None = None
926 if artist.provider == "library":
927 db_artist = artist
928 elif existing := await self.mass.music.artists.get_library_item_by_prov_id(
929 artist.item_id, artist.provider
930 ):
931 db_artist = existing
932
933 if not db_artist or overwrite:
934 # Convert ItemMapping to Artist if needed
935 artist_to_add = (
936 self.mass.music.artists.artist_from_item_mapping(artist)
937 if isinstance(artist, ItemMapping)
938 else artist
939 )
940 db_artist = await self.mass.music.artists.add_item_to_library(
941 artist_to_add, overwrite_existing=overwrite
942 )
943 # write (or update) record in album_artists table
944 await self.mass.music.database.insert_or_replace(
945 DB_TABLE_ALBUM_ARTISTS,
946 {
947 "album_id": db_id,
948 "artist_id": int(db_artist.item_id),
949 },
950 )
951 return ItemMapping.from_item(db_artist)
952
953 async def _set_album_track(self, db_id: int, db_track_id: int, track: Track) -> None:
954 """Store Album Track info."""
955 # write (or update) record in album_tracks table
956 await self.mass.music.database.insert_or_replace(
957 DB_TABLE_ALBUM_TRACKS,
958 {
959 "album_id": db_id,
960 "track_id": db_track_id,
961 "track_number": track.track_number,
962 "disc_number": track.disc_number,
963 },
964 )
965
966 def _parse_summary_row(self, db_row: Mapping[str, Any]) -> AlbumSummary:
967 """Parse a raw summary db row into an AlbumSummary object."""
968 item = cast("AlbumSummary", super()._parse_summary_row(db_row))
969 item.version = db_row["version"] or ""
970 item.year = db_row["year"]
971 item.album_type = AlbumType(db_row["album_type"])
972 item.artists = self._parse_summary_artist_mappings(db_row)
973 return item
974
975
976def _canonical_album_barcodes(album: Album) -> set[str]:
977 """Return an album's valid barcodes in canonical UPC form."""
978 return {
979 barcode_to_upc(value)
980 for external_id_type, value in album.external_ids
981 if external_id_type == ExternalID.BARCODE and is_valid_barcode(value)
982 }
983
984
985def _unambiguous_release_ids(
986 barcodes: set[str], releases_by_barcode: dict[str, list[MusicBrainzBarcodeRelease]]
987) -> set[str]:
988 """Return release ids that at least one of the barcodes resolves to unambiguously."""
989 release_ids: set[str] = set()
990 for barcode in barcodes:
991 resolved = {release.id for release in releases_by_barcode.get(barcode, [])}
992 # only a barcode that maps to exactly one specific release is trustworthy evidence
993 if len(resolved) == 1:
994 release_ids |= resolved
995 return release_ids
996
997
998def _release_group_ids(
999 barcodes: set[str], releases_by_barcode: dict[str, list[MusicBrainzBarcodeRelease]]
1000) -> set[str]:
1001 """Return every release-group id the barcodes resolve to."""
1002 return {
1003 release.release_group.id
1004 for barcode in barcodes
1005 for release in releases_by_barcode.get(barcode, [])
1006 }
1007