import asyncio
import contextlib
from random import choice, random
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
from music_assistant.common.helpers.datetime import utc_timestamp
from music_assistant.common.helpers.json import serialize_to_json
lazy: bool = True,
details: Album | ItemMapping = None,
add_to_library: bool = False,
- **kwargs: dict[str, Any],
) -> Album:
"""Return (full) details for a single media item."""
album = await super().get(
lazy=lazy,
details=details,
add_to_library=add_to_library,
- **kwargs,
)
# append full artist details to full album item
album.artists = [
lazy=lazy,
details=item,
add_to_library=add_to_library,
- **kwargs,
)
for item in album.artists
]
self,
item: Album,
metadata_lookup: bool = True,
- add_album_tracks: bool = True,
- **kwargs: dict[str, Any], # noqa: ARG002
+ add_album_tracks: bool = False,
) -> Album:
"""Add album to library and return the database item."""
if not isinstance(item, Album):
raise InvalidDataError("Not a valid Album object (ItemMapping can not be added to db)")
if not item.provider_mappings:
raise InvalidDataError("Album is missing provider mapping(s)")
- # resolve any ItemMapping artists
- item.artists = [
- await self.mass.music.artists.get_provider_item(
- artist.item_id, artist.provider, fallback=artist
- )
- if isinstance(artist, ItemMapping)
- else artist
- for artist in item.artists
- ]
- if not item.artists:
- raise InvalidDataError("Album is missing artist(s)")
# grab additional metadata
if not metadata_lookup:
await self.mass.metadata.get_album_metadata(item)
for track in await self._get_provider_album_tracks(item.item_id, item.provider):
track.album = library_item
tg.create_task(
- self.mass.music.tracks.add_item_to_library(
- track, metadata_lookup=metadata_lookup
- )
+ self.mass.music.tracks.add_item_to_library(track, metadata_lookup=False)
)
self.mass.signal_event(
EventType.MEDIA_ITEM_ADDED,
self,
item: Artist | ItemMapping,
metadata_lookup: bool = True,
- **kwargs: dict[str, Any], # noqa: ARG002
) -> Artist:
"""Add artist to library and return the database item."""
if isinstance(item, ItemMapping):
self.logger = logging.getLogger(f"{ROOT_LOGGER_NAME}.music.{self.media_type.value}")
@abstractmethod
- async def add_item_to_library(self, item: ItemCls, **kwargs: dict[str, Any]) -> ItemCls:
+ async def add_item_to_library(self, item: ItemCls, metadata_lookup: bool = True) -> ItemCls:
"""Add item to library and return the database item."""
raise NotImplementedError
lazy: bool = True,
details: ItemCls = None,
add_to_library: bool = False,
- **kwargs: dict[str, Any],
) -> ItemCls:
"""Return (full) details for a single media item."""
if provider_instance_id_or_domain == "database":
# only if we really need to wait for the result (e.g. to prevent race conditions),
# we can set lazy to false and we await the job to complete.
task_id = f"add_{self.media_type.value}.{details.provider}.{details.item_id}"
- add_task = self.mass.create_task(
- self.add_item_to_library, item=details, task_id=task_id, **kwargs
- )
+ add_task = self.mass.create_task(self.add_item_to_library, item=details, task_id=task_id)
if not lazy:
await add_task
return add_task.result()
return ItemMapping.from_item(db_artist)
# try to request the full item
+ artist = await self.mass.music.artists.get_provider_item(
+ artist.item_id, artist.provider, fallback=artist
+ )
with suppress(MediaNotFoundError, AssertionError, InvalidDataError):
db_artist = await self.mass.music.artists.add_item_to_library(
artist, metadata_lookup=False
)
return ItemMapping.from_item(db_artist)
# fallback to just the provider item
- artist = await self.mass.music.artists.get_provider_item(
- artist.item_id, artist.provider, fallback=artist
- )
+ # this can happen for unavailable items
if isinstance(artist, ItemMapping):
- # this can happen for unavailable items
return artist
return ItemMapping.from_item(artist)
from __future__ import annotations
import asyncio
-from typing import Any
from music_assistant.common.helpers.datetime import utc_timestamp
from music_assistant.common.helpers.json import serialize_to_json
# return the aggregated result
return all_versions.values()
- async def add_item_to_library(
- self, item: Radio, metadata_lookup: bool = True, **kwargs: dict[str, Any] # noqa: ARG002
- ) -> Radio:
+ async def add_item_to_library(self, item: Radio, metadata_lookup: bool = True) -> Radio:
"""Add radio to library and return the new database item."""
if not isinstance(item, Radio):
raise InvalidDataError("Not a valid Radio object (ItemMapping can not be added to db)")
import asyncio
import urllib.parse
-from typing import Any
+from contextlib import suppress
from music_assistant.common.helpers.datetime import utc_timestamp
from music_assistant.common.helpers.json import serialize_to_json
details: Track = None,
album_uri: str | None = None,
add_to_library: bool = False,
- **kwargs: dict[str, Any],
) -> Track:
"""Return (full) details for a single media item."""
track = await super().get(
lazy=lazy,
details=details,
add_to_library=add_to_library,
- **kwargs,
)
# append full album details to full track item
try:
lazy=lazy,
details=None if isinstance(track.album, ItemMapping) else track.album,
add_to_library=add_to_library,
- **kwargs,
)
elif provider_instance_id_or_domain == "library":
# grab the first album this track is attached to
lazy=lazy,
details=None if isinstance(artist, ItemMapping) else artist,
add_to_library=add_to_library,
- **kwargs,
)
)
track.artists = full_artists
return track
- async def add_item_to_library(
- self, item: Track, metadata_lookup: bool = True, **kwargs: dict[str, Any] # noqa: ARG002
- ) -> Track:
+ async def add_item_to_library(self, item: Track, metadata_lookup: bool = True) -> Track:
"""Add track to library and return the new database item."""
if not isinstance(item, Track):
raise InvalidDataError("Not a valid Track object (ItemMapping can not be added to db)")
raise InvalidDataError("Track is missing artist(s)")
if not item.provider_mappings:
raise InvalidDataError("Track is missing provider mapping(s)")
- # resolve any ItemMapping artists
- item.artists = [
- await self.mass.music.artists.get_provider_item(
- artist.item_id, artist.provider, fallback=artist
- )
- if isinstance(artist, ItemMapping)
- else artist
- for artist in item.artists
- ]
- # resolve ItemMapping album
- if isinstance(item.album, ItemMapping):
- item.album = await self.mass.music.albums.get_provider_item(
- item.album.item_id, item.album.provider, fallback=item.album
- )
- if isinstance(item.album, ItemMapping):
- self.logger.warning(
- "Unable to resolve Album for track %s, "
- "track will be added to the library without album",
- item.uri,
- )
- if item.album and not isinstance(item.album, ItemMapping):
- item.album.artists = [
- await self.mass.music.artists.get_provider_item(
- artist.item_id, artist.provider, fallback=artist
- )
- if isinstance(artist, ItemMapping)
- else artist
- for artist in item.album.artists
- ]
# grab additional metadata
if not metadata_lookup:
await self.mass.metadata.get_track_metadata(item)
async def _set_track_album(self, db_id: int, album: Album, disc_number: int, track_number: int):
"""Store AlbumTrack info."""
+ db_album = None
if album.provider == "library":
db_album = album
+ elif existing := await self.mass.music.artists.get_library_item_by_prov_id(
+ album.item_id, album.provider
+ ):
+ db_album = existing
else:
- db_album = await self.mass.music.albums.get(
- item_id=album.item_id,
- provider_instance_id_or_domain=album.provider,
- lazy=False,
- details=album,
- add_to_library=True,
- metadata_lookup=False,
- add_album_tracks=False,
+ # not an existing album, we need to fetch and add it
+ if isinstance(album, ItemMapping):
+ album = await self.mass.music.albums.get_provider_item(
+ album.item_id, album.provider, fallback=album
+ )
+ with suppress(MediaNotFoundError, AssertionError, InvalidDataError):
+ db_album = await self.mass.music.albums.add_item_to_library(
+ album, metadata_lookup=False, add_album_tracks=False
+ )
+
+ if not db_album:
+ self.logger.warning(
+ "Unable to resolve Album for track %s, "
+ "track will be added to the library without album",
+ album.uri,
)
+ return
album_mapping = {"track_id": db_id, "album_id": int(db_album.item_id)}
if db_row := await self.mass.music.database.get_row(DB_TABLE_ALBUM_TRACKS, album_mapping):
# update existing
# create full db item
# note that we skip the metadata lookup purely to speed up the sync
# the additional metadata is then lazy retrieved afterwards
+
prov_item.favorite = True
+ extra_kwargs = (
+ {"add_album_tracks": True} if media_type == MediaType.ALBUM else {}
+ )
library_item = await controller.add_item_to_library(
- prov_item, metadata_lookup=False
+ prov_item, metadata_lookup=False, **extra_kwargs
)
elif (
library_item.metadata.checksum and prov_item.metadata.checksum
projects / music-assistant-server.git / commitdiff