import asyncio
import contextlib
from random import choice, random
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
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,
- skip_metadata_lookup: 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,
- skip_metadata_lookup=skip_metadata_lookup,
+ **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
]
return album
- async def add_item_to_library(self, item: Album, skip_metadata_lookup: bool = False) -> Album:
+ async def add_item_to_library(
+ self,
+ item: Album,
+ metadata_lookup: bool = True,
+ add_album_tracks: bool = True,
+ **kwargs: dict[str, Any], # noqa: ARG002
+ ) -> 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.artists:
raise InvalidDataError("Album is missing artist(s)")
# grab additional metadata
- if not skip_metadata_lookup:
+ if not metadata_lookup:
await self.mass.metadata.get_album_metadata(item)
# actually add (or update) the item in the library db
# use the lock to prevent a race condition of the same item being added twice
async with self._db_add_lock:
library_item = await self._add_library_item(item)
# also fetch the same album on all providers
- if not skip_metadata_lookup:
+ if not metadata_lookup:
await self._match(library_item)
library_item = await self.get_library_item(library_item.item_id)
# also add album tracks
- if not skip_metadata_lookup and item.provider != "library":
+ if add_album_tracks and item.provider != "library":
async with asyncio.TaskGroup() as tg:
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, skip_metadata_lookup=skip_metadata_lookup
+ track, metadata_lookup=metadata_lookup
)
)
self.mass.signal_event(
self.mass.register_api_command("music/artists/artist_tracks", self.tracks)
async def add_item_to_library(
- self, item: Artist | ItemMapping, skip_metadata_lookup: bool = False
+ 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):
- skip_metadata_lookup = True
+ metadata_lookup = True
# grab musicbrainz id and additional metadata
- if not skip_metadata_lookup:
+ if not metadata_lookup:
await self.mass.metadata.get_artist_metadata(item)
# actually add (or update) the item in the library db
# use the lock to prevent a race condition of the same item being added twice
async with self._db_add_lock:
library_item = await self._add_library_item(item)
# also fetch same artist on all providers
- if not skip_metadata_lookup:
+ if not metadata_lookup:
await self.match_artist(library_item)
library_item = await self.get_library_item(library_item.item_id)
self.mass.signal_event(
self.logger = logging.getLogger(f"{ROOT_LOGGER_NAME}.music.{self.media_type.value}")
@abstractmethod
- async def add_item_to_library(
- self, item: ItemCls, skip_metadata_lookup: bool = False
- ) -> ItemCls:
+ async def add_item_to_library(self, item: ItemCls, **kwargs: dict[str, Any]) -> ItemCls:
"""Add item to library and return the database item."""
raise NotImplementedError
lazy: bool = True,
details: ItemCls = None,
add_to_library: bool = False,
- skip_metadata_lookup: bool = False,
+ **kwargs: dict[str, Any],
) -> ItemCls:
"""Return (full) details for a single media item."""
if provider_instance_id_or_domain == "database":
# 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,
- skip_metadata_lookup=skip_metadata_lookup,
- task_id=task_id,
+ self.add_item_to_library, item=details, task_id=task_id, **kwargs
)
if not lazy:
await add_task
# try to request the full item
with suppress(MediaNotFoundError, AssertionError, InvalidDataError):
db_artist = await self.mass.music.artists.add_item_to_library(
- artist, skip_metadata_lookup=True
+ artist, metadata_lookup=False
)
return ItemMapping.from_item(db_artist)
# fallback to just the provider item
"music/playlists/remove_playlist_tracks", self.remove_playlist_tracks
)
- async def add_item_to_library(
- self, item: Playlist, skip_metadata_lookup: bool = False
- ) -> Playlist:
+ async def add_item_to_library(self, item: Playlist, metadata_lookup: bool = True) -> Playlist:
"""Add playlist to library and return the new database item."""
if not isinstance(item, Playlist):
raise InvalidDataError(
async for _ in self.tracks(item.item_id, item.provider):
pass
# metadata lookup we need to do after adding it to the db
- if not skip_metadata_lookup:
+ if not metadata_lookup:
await self.mass.metadata.get_playlist_metadata(library_item)
library_item = await self.update_item_in_library(library_item.item_id, library_item)
self.mass.signal_event(
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, skip_metadata_lookup: bool = False) -> Radio:
+ async def add_item_to_library(
+ self, item: Radio, metadata_lookup: bool = True, **kwargs: dict[str, Any] # noqa: ARG002
+ ) -> 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)")
if not item.provider_mappings:
raise InvalidDataError("Radio is missing provider mapping(s)")
- if not skip_metadata_lookup:
+ if not metadata_lookup:
await self.mass.metadata.get_radio_metadata(item)
# actually add (or update) the item in the library db
# use the lock to prevent a race condition of the same item being added twice
import asyncio
import urllib.parse
+from typing import Any
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,
- skip_metadata_lookup: 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,
- skip_metadata_lookup=skip_metadata_lookup,
+ **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,
- skip_metadata_lookup=skip_metadata_lookup,
+ **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,
- skip_metadata_lookup=skip_metadata_lookup,
+ **kwargs,
)
)
track.artists = full_artists
return track
- async def add_item_to_library(self, item: Track, skip_metadata_lookup: bool = False) -> Track:
+ async def add_item_to_library(
+ self, item: Track, metadata_lookup: bool = True, **kwargs: dict[str, Any] # noqa: ARG002
+ ) -> 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)")
for artist in item.album.artists
]
# grab additional metadata
- if not skip_metadata_lookup:
+ if not metadata_lookup:
await self.mass.metadata.get_track_metadata(item)
# fallback track image from album (only if albumtype = single)
if (
async with self._db_add_lock:
library_item = await self._add_library_item(item)
# also fetch same track on all providers (will also get other quality versions)
- if not skip_metadata_lookup:
+ if not metadata_lookup:
await self._match(library_item)
library_item = await self.get_library_item(library_item.item_id)
self.mass.signal_event(
lazy=False,
details=album,
add_to_library=True,
- skip_metadata_lookup=True,
+ metadata_lookup=False,
+ add_album_tracks=False,
)
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):
async def get_album_metadata(self, album: Album) -> None:
"""Get/update rich metadata for an album."""
- # ensure the album has a musicbrainz id or artist
+ # ensure the album has a musicbrainz id or artist(s)
if not (album.mbid or album.artists):
return
# collect metadata from all providers
import asyncio
import logging
+import time
import urllib.parse
from collections.abc import AsyncGenerator
from contextlib import suppress
query_params["seek_position"] = str(seek_position)
if fade_in:
query_params["fade_in"] = "1"
- if query_params:
- url += "?" + urllib.parse.urlencode(query_params)
+ # we add a timestamp as basic checksum
+ # most importantly this is to invalidate any caches
+ # but also to handle edge cases such as single track repeat
+ query_params["ts"] = str(int(time.time()))
+ url += "?" + urllib.parse.urlencode(query_params)
return url
async def create_multi_client_stream_job(
f"<upnp:album>{album}</upnp:album>"
f"<upnp:artist>{artist}</upnp:artist>"
f"<upnp:duration>{int(queue_item.duration)}</upnp:duration>"
- "<upnp:playlistTitle>Music Assistant</upnp:playlistTitle>"
f"<dc:queueItemId>{queue_item.queue_item_id}</dc:queueItemId>"
f"<upnp:albumArtURI>{escape_string(image_url)}</upnp:albumArtURI>"
"<upnp:class>object.item.audioItem.audioBroadcast</upnp:class>"
prov_item.provider_mappings,
)
try:
+ if not library_item and not prov_item.available:
+ # skip unavailable tracks
+ self.logger.debg(
+ "Skipping sync of item %s because it is unavailable", prov_item.uri
+ )
+ continue
if not library_item:
# 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
library_item = await controller.add_item_to_library(
- prov_item, skip_metadata_lookup=True
+ prov_item, metadata_lookup=False
)
elif (
library_item.metadata.checksum and prov_item.metadata.checksum
# make sure that the item exists
# https://github.com/music-assistant/hass-music-assistant/issues/707
library_item = await self.mass.music.tracks.add_item_to_library(
- track, skip_metadata_lookup=True
+ track, metadata_lookup=False
)
subitems.append(library_item)
continue
# make sure that the item exists
# https://github.com/music-assistant/hass-music-assistant/issues/707
library_item = await self.mass.music.playlists.add_item_to_library(
- playlist, skip_metadata_lookup=True
+ playlist, metadata_lookup=False
)
subitems.append(library_item)
continue
if item.ext in TRACK_EXTENSIONS:
# add/update track to db
track = await self._parse_track(item)
- await self.mass.music.tracks.add_item_to_library(
- track, skip_metadata_lookup=True
- )
+ await self.mass.music.tracks.add_item_to_library(track, metadata_lookup=False)
elif item.ext in PLAYLIST_EXTENSIONS:
playlist = await self.get_playlist(item.path)
# add/update] playlist to db
# playlist is always in-library
playlist.favorite = True
await self.mass.music.playlists.add_item_to_library(
- playlist, skip_metadata_lookup=True
+ playlist, metadata_lookup=False
)
except Exception as err: # pylint: disable=broad-except
# we don't want the whole sync to crash on one file so we catch all exceptions here
# much space and bandwidth. Instead we set the filename as value so the image can
# be retrieved later in realtime.
track.metadata.images = [
- MediaItemImage(type=ImageType.THUMB, url=file_item.path, provider=self.instance_id)
+ MediaItemImage(type=ImageType.THUMB, path=file_item.path, provider=self.instance_id)
]
if track.album and not track.album.metadata.images:
album_barcode: str | None = None,
) -> str | None:
"""Retrieve musicbrainz artist id by providing the artist name and albumname or barcode."""
- assert albumname or album_barcode
+ if not (albumname or album_barcode):
+ return None # may not happen, but guard just in case
for searchartist in (
artistname,
re.sub(LUCENE_SPECIAL, r"\\\1", artistname),
track_isrc: str | None = None,
) -> str | None:
"""Retrieve artist id by providing the artist name and trackname or track isrc."""
- assert trackname or track_isrc
+ if not (trackname or track_isrc):
+ return None # may not happen, but guard just in case
searchartist = re.sub(LUCENE_SPECIAL, r"\\\1", artistname)
if track_isrc:
result = await self.get_data(f"isrc/{track_isrc}", inc="artist-credits")
album.year = plex_album.year
if thumb := plex_album.firstAttr("thumb", "parentThumb", "grandparentThumb"):
album.metadata.images = [
- MediaItemImage(type=ImageType.THUMB, url=thumb, provider=self.instance_id)
+ MediaItemImage(type=ImageType.THUMB, path=thumb, provider=self.instance_id)
]
if plex_album.summary:
album.metadata.description = plex_album.summary
album.artists.append(
self._get_item_mapping(
- type=MediaType.ARTIST, url=plex_album.parentKey, provider=plex_album.parentTitle
+ media_type=MediaType.ARTIST,
+ url=plex_album.parentKey,
+ provider=plex_album.parentTitle,
)
)
return album
artist.metadata.description = plex_artist.summary
if thumb := plex_artist.firstAttr("thumb", "parentThumb", "grandparentThumb"):
artist.metadata.images = [
- MediaItemImage(type=ImageType.THUMB, path=thumb, url=self.instance_id)
+ MediaItemImage(type=ImageType.THUMB, path=thumb, provider=self.instance_id)
]
return artist
playlist.metadata.description = plex_playlist.summary
if thumb := plex_playlist.firstAttr("thumb", "parentThumb", "grandparentThumb"):
playlist.metadata.images = [
- MediaItemImage(type=ImageType.THUMB, url=thumb, provider=self.instance_id)
+ MediaItemImage(type=ImageType.THUMB, path=thumb, provider=self.instance_id)
]
playlist.is_editable = True
return playlist
if thumb := plex_track.firstAttr("thumb", "parentThumb", "grandparentThumb"):
track.metadata.images = [
- MediaItemImage(type=ImageType.THUMB, url=thumb, provider=self.instance_id)
+ MediaItemImage(type=ImageType.THUMB, path=thumb, provider=self.instance_id)
]
if plex_track.parentKey:
track.album = self._get_item_mapping(
if media_info.has_cover_image:
media_item.metadata.images = [
- MediaItemImage(type=ImageType.THUMB, path=url, provider="embedded")
+ MediaItemImage(type=ImageType.THUMB, path=url, provider="file")
]
return media_item
"python-slugify==8.0.1",
"mashumaro==3.8.1",
"memory-tempfile==2.2.3",
- "music-assistant-frontend==2.0.9",
+ "music-assistant-frontend==2.0.10",
"pillow==10.0.0",
"unidecode==1.3.6",
"xmltodict==0.13.0",
ifaddr==0.2.0
mashumaro==3.8.1
memory-tempfile==2.2.3
-music-assistant-frontend==2.0.9
+music-assistant-frontend==2.0.10
orjson==3.9.2
pillow==10.0.0
plexapi==4.15.0
projects / music-assistant-server.git / commitdiff