provider_id: Optional[str] = None,
) -> List[Track]:
"""Return album tracks for the given provider album id."""
- provider = self.mass.music.get_provider(provider_id or provider)
- if not provider:
+ prov = self.mass.music.get_provider(provider_id or provider)
+ if not prov:
return []
- return await provider.get_album_tracks(item_id)
+ return await prov.get_album_tracks(item_id)
async def add_db_item(self, album: Album) -> Album:
"""Add a new album record to the database."""
assert album.provider_ids, "Album is missing provider id(s)"
cur_item = None
- assert album.provider_ids
async with self.mass.database.get_db() as _db:
# always try to grab existing item by musicbrainz_id
if album.musicbrainz_id:
async def _get_album_artist(
self, db_album: Album, updated_album: Optional[Album] = None
) -> ItemMapping | None:
- """Extract album artist as ItemMapping, prefer database ID."""
+ """Extract (database) album artist as ItemMapping."""
for album in (updated_album, db_album):
if not album or not album.artist:
continue
- if isinstance(album.artist, ItemMapping):
- return album.artist
-
if album.artist.provider == ProviderType.DATABASE:
+ if isinstance(album.artist, ItemMapping):
+ return album.artist
return ItemMapping.from_item(album.artist)
- if album.artist.musicbrainz_id:
- album_artist = await self.mass.music.artists.add_db_item(album.artist)
- return ItemMapping.from_item(album_artist)
-
- if album_artist := await self.mass.music.artists.get_db_item_by_prov_id(
+ if db_artist := await self.mass.music.artists.get_db_item_by_prov_id(
album.artist.item_id,
provider=album.artist.provider,
):
- return ItemMapping.from_item(album_artist)
+ return ItemMapping.from_item(db_artist)
+
+ db_artist = await self.mass.music.artists.add_db_item(album.artist)
+ return ItemMapping.from_item(db_artist)
- return ItemMapping.from_item(album.artist)
return None
from music_assistant.helpers.compare import (
compare_album,
+ compare_artist,
compare_strings,
compare_track,
)
searchstr, provider.type
)
for search_result_item in search_results:
- if compare_track(search_result_item, ref_track):
- # get matching artist from track
- for search_item_artist in search_result_item.artists:
- if compare_strings(db_artist.name, search_item_artist.name):
- # 100% album match
- # get full artist details so we have all metadata
- prov_artist = await self.get_provider_item(
- search_item_artist.item_id, search_item_artist.provider
- )
- await self.update_db_item(db_artist.item_id, prov_artist)
- return True
+ if not compare_track(search_result_item, ref_track):
+ continue
+ # get matching artist from track
+ for search_item_artist in search_result_item.artists:
+ if not compare_artist(db_artist, search_result_item.artist):
+ continue
+ # 100% album match
+ # get full artist details so we have all metadata
+ prov_artist = await self.get_provider_item(
+ search_item_artist.item_id, search_item_artist.provider
+ )
+ await self.update_db_item(db_artist.item_id, prov_artist)
+ return True
# try to get a match with some reference albums of this artist
artist_albums = await self.albums(db_artist.item_id, db_artist.provider)
for ref_album in artist_albums:
)
for search_result_item in search_result:
# artist must match 100%
- if not compare_strings(db_artist.name, search_result_item.artist.name):
+ if not compare_artist(db_artist, search_result_item.artist):
continue
- if compare_album(search_result_item, ref_album):
- # 100% album match
- # get full artist details so we have all metadata
- prov_artist = await self.get_provider_item(
- search_result_item.artist.item_id,
- search_result_item.artist.provider,
- )
- await self.update_db_item(db_artist.item_id, prov_artist)
- return True
+ if not compare_album(search_result_item, ref_album):
+ continue
+ # 100% match
+ # get full artist details so we have all metadata
+ prov_artist = await self.get_provider_item(
+ search_result_item.artist.item_id,
+ search_result_item.artist.provider,
+ )
+ await self.update_db_item(db_artist.item_id, prov_artist)
+ return True
return False
from __future__ import annotations
import asyncio
-from typing import List, Optional
+from typing import List, Optional, Union
from music_assistant.helpers.compare import (
compare_artists,
from music_assistant.models.enums import EventType, MediaType, ProviderType
from music_assistant.models.event import MassEvent
from music_assistant.models.media_controller import MediaControllerBase
-from music_assistant.models.media_items import ItemMapping, Track
+from music_assistant.models.media_items import Artist, ItemMapping, Track
class TracksController(MediaControllerBase[Track]):
track_album = await self._get_track_album(cur_item, track)
if overwrite:
provider_ids = track.provider_ids
- track_artists = track.artists
else:
provider_ids = {*cur_item.provider_ids, *track.provider_ids}
- track_artists = cur_item.artists + track.artists
metadata = cur_item.metadata.update(track.metadata, overwrite)
# we store a mapping to artists on the track for easier access/listings
- track_artists = await self._get_track_artists(track, track_artists)
+ track_artists = await self._get_track_artists(cur_item, track)
await self.mass.database.update(
self.db_table,
{"item_id": item_id},
return await self.get_db_item(item_id, db=_db)
async def _get_track_artists(
- self, track: Track, cur_artists: List[ItemMapping] | None = None
+ self, base_track: Track, upd_track: Optional[Track] = None
) -> List[ItemMapping]:
"""Extract all (unique) artists of track as ItemMapping."""
- if cur_artists is None:
- cur_artists = []
- cur_artists.extend(track.artists)
- track_artists: List[ItemMapping] = []
- for item in cur_artists:
- cur_names = {x.name for x in track_artists}
- cur_ids = {x.item_id for x in track_artists}
- track_artist = (
- await self.mass.music.artists.get_db_item_by_prov_id(
- provider_item_id=item.item_id,
- provider=item.provider,
- )
- or item
- )
- if (
- track_artist.name not in cur_names
- and track_artist.item_id not in cur_ids
- ):
- track_artists.append(ItemMapping.from_item(track_artist))
- return track_artists
+ if upd_track and upd_track.artists:
+ track_artists = upd_track.artists
+ else:
+ track_artists = base_track.artists
+ # use intermediate set to clear out duplicates
+ return list({await self._get_artist_mapping(x) for x in track_artists})
async def _get_track_album(
- self, db_track: Track, updated_track: Optional[Track] = None
+ self, base_track: Track, upd_track: Optional[Track] = None
) -> ItemMapping | None:
- """Extract track album as ItemMapping, prefer database ID."""
- for track in (updated_track, db_track):
+ """Extract (database) track album as ItemMapping."""
+ for track in (upd_track, base_track):
if not track or not track.album:
continue
- if isinstance(track.album, ItemMapping):
- return track.album
-
if track.album.provider == ProviderType.DATABASE:
+ if isinstance(track.album, ItemMapping):
+ return track.album
return ItemMapping.from_item(track.album)
- if track.album.musicbrainz_id:
- track_album = await self.mass.music.albums.add_db_item(track.album)
- return ItemMapping.from_item(track_album)
-
- if track_album := await self.mass.music.albums.get_db_item_by_prov_id(
- provider_item_id=track.album.item_id,
+ if db_album := await self.mass.music.albums.get_db_item_by_prov_id(
+ track.album.item_id,
provider=track.album.provider,
):
- return ItemMapping.from_item(track_album)
+ return ItemMapping.from_item(db_album)
+
+ db_album = await self.mass.music.albums.add_db_item(track.album)
+ return ItemMapping.from_item(db_album)
- return ItemMapping.from_item(track.album)
return None
+
+ async def _get_artist_mapping(
+ self, artist: Union[Artist, ItemMapping]
+ ) -> ItemMapping:
+ """Extract (database) track artist as ItemMapping."""
+ if artist.provider == ProviderType.DATABASE:
+ if isinstance(artist, ItemMapping):
+ return artist
+ return ItemMapping.from_item(artist)
+
+ if db_artist := await self.mass.music.artists.get_db_item_by_prov_id(
+ artist.item_id,
+ provider=artist.provider,
+ ):
+ return ItemMapping.from_item(db_artist)
+
+ db_artist = await self.mass.music.artists.add_db_item(artist)
+ return ItemMapping.from_item(db_artist)
from typing import List, Union
from music_assistant.helpers.util import create_clean_string
+from music_assistant.models.enums import AlbumType
from music_assistant.models.media_items import (
Album,
Artist,
def compare_artist(
left_artist: Union[Artist, ItemMapping],
right_artist: Union[Artist, ItemMapping],
- allow_none=False,
) -> bool:
"""Compare two artist items and return True if they match."""
- if allow_none and left_artist is None and right_artist is None:
- return True
if left_artist is None or right_artist is None:
return False
# return early on exact item_id match
return left_artist.musicbrainz_id == right_artist.musicbrainz_id
# fallback to comparing
-
if not left_artist.sort_name:
left_artist.sort_name = create_clean_string(left_artist.name)
if not right_artist.sort_name:
def compare_album(
left_album: Union[Album, ItemMapping],
right_album: Union[Album, ItemMapping],
- allow_none=False,
):
"""Compare two album items and return True if they match."""
- if left_album is None and right_album is None:
- return True
if left_album is None or right_album is None:
return False
# return early on exact item_id match
if (left_album.upc in right_album.upc) or (right_album.upc in left_album.upc):
return True
# prefer match on musicbrainz_id
+ # not present on ItemMapping
if getattr(left_album, "musicbrainz_id", None) and getattr(
right_album, "musicbrainz_id", None
):
return False
if not compare_version(left_album.version, right_album.version):
return False
- # album artist must be either set on both or not at all
+ # compare album artist
+ # Note: Not present on ItemMapping
if hasattr(left_album, "artist") and hasattr(right_album, "artist"):
- if not compare_artist(left_album.artist, right_album.artist, True):
+ if not compare_artist(left_album.artist, right_album.artist):
return False
return left_album.sort_name == right_album.sort_name
"""Compare two track items and return True if they match."""
if left_track is None or right_track is None:
return False
+ # album is required for track linking
+ if left_track.album is None or right_track.album is None:
+ return False
# return early on exact item_id match
if compare_item_id(left_track, right_track):
return True
# track if both tracks are (not) explicit
if not compare_explicit(left_track.metadata, right_track.metadata):
return False
- # exact album match OR (near) exact duration match
- if compare_album(left_track.album, right_track.album, True):
- return True
- if abs(left_track.duration - right_track.duration) <= 2:
- # 100% match, all criteria passed
+ # exact album match = 100% match
+ if compare_album(left_track.album, right_track.album):
return True
- return False
+ # fallback: both albums are compilations and (near-exact) track duration match
+ return (
+ abs(left_track.duration - right_track.duration) <= 1
+ and left_track.album.album_type in (AlbumType.UNKNOWN, AlbumType.COMPILATION)
+ and right_track.album.album_type in (AlbumType.UNKNOWN, AlbumType.COMPILATION)
+ )
projects / music-assistant-server.git / commitdiff