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