def update(
self,
new_values: MediaItemMetadata,
- allow_overwrite: bool = False,
+ allow_overwrite: bool = True,
) -> MediaItemMetadata:
"""Update metadata (in-place) with new values."""
+ if not new_values:
+ return self
for fld in fields(self):
new_val = getattr(new_values, fld.name)
if new_val is None:
sort_name: str | None = None
uri: str | None = None
version: str = ""
+ available: bool = True
@classmethod
def from_item(cls, item: MediaItem):
"""Create ItemMapping object from regular item."""
- return cls.from_dict(item.to_dict())
+ result = cls.from_dict(item.to_dict())
+ result.available = item.available
+ return result
def __hash__(self):
"""Return custom hash."""
from music_assistant.common.models.media_items import (
Album,
AlbumType,
- Artist,
DbAlbum,
ItemMapping,
MediaType,
Track,
)
-from music_assistant.constants import DB_TABLE_ALBUMS, DB_TABLE_TRACKS, VARIOUS_ARTISTS
+from music_assistant.constants import DB_TABLE_ALBUMS, DB_TABLE_TRACKS
from music_assistant.server.controllers.media.base import MediaControllerBase
from music_assistant.server.helpers.compare import compare_album, loose_compare_strings
)
return db_item
+ async def update(self, item_id: int, update: Album, overwrite: bool = False) -> Album:
+ """Update existing record in the database."""
+ return await self._update_db_item(item_id=item_id, item=update, overwrite=overwrite)
+
async def delete(self, item_id: int, recursive: bool = False) -> None:
"""Delete record from the database."""
# check album tracks
]
)
for prov_item in prov_items
+ # title must (partially) match
if loose_compare_strings(album.name, prov_item.name)
+ # artist must match
+ and album.artists[0].sort_name in {x.sort_name for x in prov_item.artists}
}
# make sure that the 'base' version is NOT included
for prov_version in album.provider_mappings:
return await self._update_db_item(cur_item.item_id, item)
# insert new item
- album_artists = await self._get_album_artists(item, cur_item)
+ album_artists = await self._get_artist_mappings(item, cur_item)
sort_artist = album_artists[0].sort_name if album_artists else ""
new_item = await self.mass.music.database.insert(
self.db_table,
return await self.get_db_item(item_id)
async def _update_db_item(
- self,
- item_id: int,
- item: Album,
+ self, item_id: int, item: Album | ItemMapping, overwrite: bool = False
) -> Album:
"""Update Album record in the database."""
- assert item.provider_mappings, "Item is missing provider mapping(s)"
- assert item.artists, f"Album {item.name} is missing artist"
cur_item = await self.get_db_item(item_id)
- is_file_provider = item.provider.startswith("filesystem")
- metadata = cur_item.metadata.update(item.metadata, is_file_provider)
- provider_mappings = {*cur_item.provider_mappings, *item.provider_mappings}
- if is_file_provider:
- album_artists = await self._get_album_artists(cur_item)
- else:
- album_artists = await self._get_album_artists(cur_item, item)
- cur_item.barcode.update(item.barcode)
- if item.album_type != AlbumType.UNKNOWN:
+ metadata = cur_item.metadata.update(getattr(item, "metadata", None), overwrite)
+ provider_mappings = self._get_provider_mappings(cur_item, item, overwrite)
+ album_artists = await self._get_artist_mappings(cur_item, item, overwrite)
+ if getattr(item, "barcode", None):
+ cur_item.barcode.update(item.barcode)
+ if getattr(item, "album_type", AlbumType.UNKNOWN) != AlbumType.UNKNOWN:
album_type = item.album_type
else:
album_type = cur_item.album_type
-
sort_artist = album_artists[0].sort_name if album_artists else ""
await self.mass.music.database.update(
self.db_table,
{"item_id": item_id},
{
- "name": item.name if is_file_provider else cur_item.name,
- "sort_name": item.sort_name if is_file_provider else cur_item.sort_name,
+ "name": item.name if overwrite else cur_item.name,
+ "sort_name": item.sort_name if overwrite else cur_item.sort_name,
"sort_artist": sort_artist,
- "version": item.version if is_file_provider else cur_item.version,
- "year": item.year or cur_item.year,
+ "version": item.version if overwrite else cur_item.version,
+ "year": item.year if overwrite else cur_item.year or item.year,
"barcode": ";".join(cur_item.barcode),
"album_type": album_type.value,
"artists": serialize_to_json(album_artists) or None,
db_album.name,
provider.name,
)
-
- async def _get_album_artists(
- self,
- db_album: Album,
- updated_album: Album | None = None,
- ) -> list[ItemMapping]:
- """Extract (database) album artist(s) as ItemMapping."""
- album_artists = set()
- for album in (updated_album, db_album):
- if not album:
- continue
- for artist in album.artists:
- album_artists.add(await self._get_artist_mapping(artist))
- # use intermediate set to prevent duplicates
- # filter various artists if multiple artists
- if len(album_artists) > 1:
- album_artists = {x for x in album_artists if (x.name != VARIOUS_ARTISTS)}
- return list(album_artists)
-
- async def _get_artist_mapping(self, artist: Artist | ItemMapping) -> ItemMapping:
- """Extract (database) track artist as ItemMapping."""
- if artist.provider == "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, 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)
)
return db_item
+ async def update(self, item_id: int, update: Artist, overwrite: bool = False) -> Artist:
+ """Update existing record in the database."""
+ return await self._update_db_item(item_id=item_id, item=update, overwrite=overwrite)
+
async def album_artists(
self,
in_library: bool | None = None,
)
return items
- async def _add_db_item(self, item: Artist) -> Artist:
+ async def _add_db_item(self, item: Artist | ItemMapping) -> Artist:
"""Add a new item record to the database."""
- assert isinstance(item, Artist), "Not a full Artist object"
- assert item.provider_mappings, "Item is missing provider mapping(s)"
# enforce various artists name + id
- if compare_strings(item.name, VARIOUS_ARTISTS):
- item.musicbrainz_id = VARIOUS_ARTISTS_ID
- if item.musicbrainz_id == VARIOUS_ARTISTS_ID:
- item.name = VARIOUS_ARTISTS
+ if not isinstance(item, ItemMapping):
+ if compare_strings(item.name, VARIOUS_ARTISTS):
+ item.musicbrainz_id = VARIOUS_ARTISTS_ID
+ if item.musicbrainz_id == VARIOUS_ARTISTS_ID:
+ item.name = VARIOUS_ARTISTS
async with self._db_add_lock:
# always try to grab existing item by musicbrainz_id
cur_item = None
- if item.musicbrainz_id:
- match = {"musicbrainz_id": item.musicbrainz_id}
+ if musicbrainz_id := getattr(item, "musicbrainz_id", None):
+ match = {"musicbrainz_id": musicbrainz_id}
cur_item = await self.mass.music.database.get_row(self.db_table, match)
if not cur_item:
# fallback to exact name match
# insert item
item.timestamp_added = int(utc_timestamp())
item.timestamp_modified = int(utc_timestamp())
+ # edge case: item is an ItemMapping,
+ # try to construct (a half baken) Artist object from it
+ if isinstance(item, ItemMapping):
+ item = Artist.from_dict(item.to_dict())
new_item = await self.mass.music.database.insert(self.db_table, item.to_db_row())
item_id = new_item["item_id"]
# update/set provider_mappings table
return await self.get_db_item(item_id)
async def _update_db_item(
- self,
- item_id: int,
- item: Artist,
+ self, item_id: int, item: Artist | ItemMapping, overwrite: bool = False
) -> Artist:
"""Update Artist record in the database."""
- assert item.provider_mappings, "Item is missing provider mapping(s)"
cur_item = await self.get_db_item(item_id)
- is_file_provider = item.provider.startswith("filesystem")
- metadata = cur_item.metadata.update(item.metadata, is_file_provider)
- provider_mappings = {*cur_item.provider_mappings, *item.provider_mappings}
+ metadata = cur_item.metadata.update(getattr(item, "metadata", None), overwrite)
+ provider_mappings = self._get_provider_mappings(cur_item, item, overwrite)
# enforce various artists name + id
- if compare_strings(item.name, VARIOUS_ARTISTS):
- item.musicbrainz_id = VARIOUS_ARTISTS_ID
- if item.musicbrainz_id == VARIOUS_ARTISTS_ID:
- item.name = VARIOUS_ARTISTS
+ musicbrainz_id = cur_item.musicbrainz_id
+ if (not musicbrainz_id or overwrite) and getattr(item, "musicbrainz_id", None):
+ if compare_strings(item.name, VARIOUS_ARTISTS):
+ item.musicbrainz_id = VARIOUS_ARTISTS_ID
+ if item.musicbrainz_id == VARIOUS_ARTISTS_ID:
+ item.name = VARIOUS_ARTISTS
await self.mass.music.database.update(
self.db_table,
{"item_id": item_id},
{
- "name": item.name if is_file_provider else cur_item.name,
- "sort_name": item.sort_name if is_file_provider else cur_item.sort_name,
- "musicbrainz_id": item.musicbrainz_id or cur_item.musicbrainz_id,
+ "name": item.name if overwrite else cur_item.name,
+ "sort_name": item.sort_name if overwrite else cur_item.sort_name,
+ "musicbrainz_id": musicbrainz_id,
"metadata": serialize_to_json(metadata),
"provider_mappings": serialize_to_json(provider_mappings),
"timestamp_modified": int(utc_timestamp()),
MediaItemType,
PagedItems,
ProviderMapping,
- Track,
media_from_dict,
)
-from music_assistant.constants import DB_TABLE_PROVIDER_MAPPINGS, ROOT_LOGGER_NAME
+from music_assistant.constants import DB_TABLE_PROVIDER_MAPPINGS, ROOT_LOGGER_NAME, VARIOUS_ARTISTS
if TYPE_CHECKING:
+ from music_assistant.common.models.media_items import Album, Artist, Track
from music_assistant.server import MusicAssistant
ItemCls = TypeVar("ItemCls", bound="MediaItemType")
"""Add item to local db and return the database item."""
raise NotImplementedError
+ @abstractmethod
+ async def update(self, item_id: int, update: ItemCls, overwrite: bool = False) -> ItemCls:
+ """Update existing record in the database."""
+
async def delete(self, item_id: int, recursive: bool = False) -> None: # noqa: ARG002
"""Delete record from the database."""
db_item = await self.get_db_item(item_id)
if not fallback:
fallback = await self.get_db_item_by_prov_id(item_id, provider_instance_id_or_domain)
if fallback:
- fallback_result = self.item_cls(
- item_id=item_id,
- provider=provider.instance_id,
- name=fallback.name,
- provider_mappings={
- ProviderMapping(
- item_id=item_id,
- provider_domain=provider.domain,
- provider_instance=provider.instance_id,
- available=False,
- )
- },
- )
- if hasattr(fallback, "version") and hasattr(fallback_result, "version"):
- fallback_result.version = fallback.version
- if hasattr(fallback, "artists") and hasattr(fallback_result, "artists"):
- fallback_result.artists = fallback.artists
- if hasattr(fallback, "album") and hasattr(fallback_result, "album"):
- fallback_result.album = fallback.album
- return fallback_result
+ fallback_item = ItemMapping.from_item(fallback)
+ fallback_item.available = False
+ return fallback_item
raise MediaNotFoundError(
f"{self.media_type.value}://{item_id} not "
"found on provider {provider_instance_id_or_domain}"
"provider_item_id": provider_mapping.item_id,
},
)
+
+ def _get_provider_mappings(
+ self,
+ org_item: ItemCls,
+ update_item: ItemCls | ItemMapping | None = None,
+ overwrite: bool = False,
+ ) -> set[ProviderMapping]:
+ """Get/merge provider mappings for an item."""
+ if not update_item or isinstance(update_item, ItemMapping):
+ return org_item.provider_mappings
+ if overwrite and update_item.provider_mappings:
+ return update_item.provider_mappings
+ return {*org_item.provider_mappings, *update_item.provider_mappings}
+
+ async def _get_artist_mappings(
+ self,
+ org_item: Album | Track,
+ update_item: Album | Track | ItemMapping | None = None,
+ overwrite: bool = False,
+ ) -> list[ItemMapping]:
+ """Extract (database) album/track artist(s) as ItemMapping."""
+ if not update_item or isinstance(update_item, ItemMapping):
+ return org_item.artists
+ if overwrite and update_item.provider_mappings:
+ return update_item.artists
+ item_artists: set[ItemMapping] = set()
+ for item in (org_item, update_item):
+ for artist in item.artists:
+ item_artists.add(await self._get_artist_mapping(artist))
+ # use intermediate set to prevent duplicates
+ # filter various artists if multiple artists
+ if len(item_artists) > 1:
+ item_artists = {x for x in item_artists if (x.name != VARIOUS_ARTISTS)}
+ return list(item_artists)
+
+ async def _get_artist_mapping(self, artist: Artist | ItemMapping) -> ItemMapping:
+ """Extract (database) track artist as ItemMapping."""
+ if artist.provider == "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, artist.provider
+ ):
+ return ItemMapping.from_item(db_artist)
+
+ db_artist = await self.mass.music.artists.add(artist, skip_metadata_lookup=True)
+ return ItemMapping.from_item(db_artist)
)
return db_item
+ async def update(self, item_id: int, update: Playlist, overwrite: bool = False) -> Playlist:
+ """Update existing record in the database."""
+ return await self._update_db_item(item_id=item_id, item=update, overwrite=overwrite)
+
async def tracks(
self,
item_id: str,
if cur_item := await self.mass.music.database.get_row(self.db_table, match):
# update existing
return await self._update_db_item(cur_item["item_id"], item)
-
# insert new item
item.timestamp_added = int(utc_timestamp())
item.timestamp_modified = int(utc_timestamp())
return await self.get_db_item(item_id)
async def _update_db_item(
- self,
- item_id: int,
- item: Playlist,
+ self, item_id: int, item: Playlist, overwrite: bool = True
) -> Playlist:
"""Update Playlist record in the database."""
- assert item.provider_mappings, "Item is missing provider mapping(s)"
cur_item = await self.get_db_item(item_id)
- metadata = cur_item.metadata.update(item.metadata)
- provider_mappings = {*cur_item.provider_mappings, *item.provider_mappings}
+ metadata = cur_item.metadata.update(getattr(item, "metadata", None), overwrite)
+ provider_mappings = self._get_provider_mappings(cur_item, item, overwrite)
await self.mass.music.database.update(
self.db_table,
{"item_id": item_id},
)
return db_item
+ async def update(self, item_id: int, update: Radio, overwrite: bool = False) -> Radio:
+ """Update existing record in the database."""
+ return await self._update_db_item(item_id=item_id, item=update, overwrite=overwrite)
+
async def _add_db_item(self, item: Radio) -> Radio:
"""Add a new item record to the database."""
assert item.provider_mappings, "Item is missing provider mapping(s)"
if cur_item := await self.mass.music.database.get_row(self.db_table, match):
# update existing
return await self._update_db_item(cur_item["item_id"], item)
-
# insert new item
item.timestamp_added = int(utc_timestamp())
item.timestamp_modified = int(utc_timestamp())
# return created object
return await self.get_db_item(item_id)
- async def _update_db_item(
- self,
- item_id: int,
- item: Radio,
- ) -> Radio:
+ async def _update_db_item(self, item_id: int, item: Radio, overwrite: bool = True) -> Radio:
"""Update Radio record in the database."""
- assert item.provider_mappings, "Item is missing provider mapping(s)"
cur_item = await self.get_db_item(item_id)
- metadata = cur_item.metadata.update(item.metadata)
- provider_mappings = {*cur_item.provider_mappings, *item.provider_mappings}
+ metadata = cur_item.metadata.update(getattr(item, "metadata", None), overwrite)
+ provider_mappings = self._get_provider_mappings(cur_item, item, overwrite)
match = {"item_id": item_id}
await self.mass.music.database.update(
self.db_table,
from music_assistant.common.models.errors import MediaNotFoundError, UnsupportedFeaturedException
from music_assistant.common.models.media_items import (
Album,
- Artist,
DbTrack,
ItemMapping,
Track,
async def add(self, item: Track, skip_metadata_lookup: bool = False) -> Track:
"""Add track to local db and return the new database item."""
- assert item.artists
+ assert item.artists, "Artist(s) missing on Track"
# resolve any ItemMapping artists
item.artists = [
await self.mass.music.artists.get_provider_item(
)
return db_item
+ async def update(self, item_id: int, update: Track, overwrite: bool = False) -> Track:
+ """Update existing record in the database."""
+ return await self._update_db_item(item_id=item_id, item=update, overwrite=overwrite)
+
async def versions(
self,
item_id: str,
"""Add a new item record to the database."""
assert isinstance(item, Track), "Not a full Track object"
assert item.artists, "Track is missing artist(s)"
- assert item.provider_mappings, "Track is missing provider id(s)"
+ assert item.provider_mappings, "Track is missing provider mapping(s)"
async with self._db_add_lock:
cur_item = None
return await self._update_db_item(cur_item.item_id, item)
# no existing match found: insert new item
- track_artists = await self._get_track_artists(item)
+ track_artists = await self._get_artist_mappings(item)
track_albums = await self._get_track_albums(item)
sort_artist = track_artists[0].sort_name if track_artists else ""
sort_album = track_albums[0].sort_name if track_albums else ""
return await self.get_db_item(item_id)
async def _update_db_item(
- self,
- item_id: int,
- item: Track,
+ self, item_id: int, item: Track | ItemMapping, overwrite: bool = False
) -> Track:
"""Update Track record in the database, merging data."""
cur_item = await self.get_db_item(item_id)
- is_file_provider = item.provider.startswith("filesystem")
- metadata = cur_item.metadata.update(item.metadata, is_file_provider)
- provider_mappings = {*cur_item.provider_mappings, *item.provider_mappings}
- cur_item.isrc.update(item.isrc)
- if is_file_provider:
- track_artists = await self._get_track_artists(item)
- else:
- track_artists = await self._get_track_artists(cur_item, item)
+ metadata = cur_item.metadata.update(getattr(item, "metadata", None), overwrite)
+ provider_mappings = self._get_provider_mappings(cur_item, item, overwrite)
+ if getattr(item, "isrc", None):
+ cur_item.isrc.update(item.isrc)
+ track_artists = await self._get_artist_mappings(cur_item, item)
track_albums = await self._get_track_albums(cur_item, item)
-
await self.mass.music.database.update(
self.db_table,
{"item_id": item_id},
{
- "name": item.name if is_file_provider else cur_item.name,
- "sort_name": item.sort_name if is_file_provider else cur_item.sort_name,
- "version": item.version if is_file_provider else cur_item.version,
- "duration": item.duration or cur_item.duration,
+ "name": item.name or cur_item.name,
+ "sort_name": item.sort_name or cur_item.sort_name,
+ "version": item.version or cur_item.version,
+ "duration": getattr(item, "duration", None) or cur_item.duration,
"artists": serialize_to_json(track_artists),
"albums": serialize_to_json(track_albums),
"metadata": serialize_to_json(metadata),
self.logger.debug("updated %s in database: %s", item.name, item_id)
return await self.get_db_item(item_id)
- async def _get_track_artists(
- self,
- base_track: Track,
- upd_track: Track | None = None,
- ) -> list[ItemMapping]:
- """Extract all (unique) artists of track as ItemMapping."""
- track_artists = upd_track.artists if upd_track and upd_track.artists else 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_albums(
self,
base_track: Track,
- upd_track: Track | None = None,
+ upd_track: Track | ItemMapping | None = None,
) -> list[TrackAlbumMapping]:
"""Extract all (unique) albums of track as TrackAlbumMapping."""
track_albums: list[TrackAlbumMapping] = []
# existing TrackAlbumMappings are starting point
if base_track.albums:
track_albums = base_track.albums
- elif upd_track and upd_track.albums:
+ elif upd_track and getattr(upd_track, "albums", None):
track_albums = upd_track.albums
# append update item album if needed
- if upd_track and upd_track.album:
+ if upd_track and getattr(upd_track, "album", None):
mapping = await self._get_album_mapping(upd_track.album)
mapping = TrackAlbumMapping.from_dict(
{
):
return ItemMapping.from_item(db_album)
- db_album = await self.mass.music.albums._add_db_item(album)
+ db_album = await self.mass.music.albums.add(album, skip_metadata_lookup=True)
return ItemMapping.from_item(db_album)
-
- async def _get_artist_mapping(self, artist: Artist | ItemMapping) -> ItemMapping:
- """Extract (database) track artist as ItemMapping."""
- if artist.provider == "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, 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)
projects / music-assistant-server.git / commitdiff