Artist,
ItemMapping,
MediaItemImage,
+ ProviderMapping,
Track,
UniqueList,
)
},
)
+ async def match_provider(
+ self, db_album: Album, provider: MusicProvider, strict: bool = True
+ ) -> list[ProviderMapping]:
+ """
+ Try to find match on (streaming) provider for the provided (database) album.
+
+ This is used to link objects of different providers/qualities together.
+ """
+ self.logger.debug("Trying to match album %s on provider %s", db_album.name, provider.name)
+ matches: list[ProviderMapping] = []
+ artist_name = db_album.artists[0].name
+ search_str = f"{artist_name} - {db_album.name}"
+ search_result = await self.search(search_str, provider.instance_id)
+ for search_result_item in search_result:
+ if not search_result_item.available:
+ continue
+ if not compare_media_item(db_album, search_result_item, strict=strict):
+ continue
+ # we must fetch the full album version, search results can be simplified objects
+ prov_album = await self.get_provider_item(
+ search_result_item.item_id,
+ search_result_item.provider,
+ fallback=search_result_item,
+ )
+ if compare_album(db_album, prov_album, strict=strict):
+ # 100% match
+ matches.extend(prov_album.provider_mappings)
+ if not matches:
+ self.logger.debug(
+ "Could not find match for Album %s on provider %s",
+ db_album.name,
+ provider.name,
+ )
+ return matches
+
async def match_providers(self, db_album: Album) -> None:
"""Try to find match on all (streaming) providers for the provided (database) album.
return # Matching only supported for database items
if not db_album.artists:
return # guard
- artist_name = db_album.artists[0].name
-
- async def find_prov_match(provider: MusicProvider) -> bool:
- self.logger.debug(
- "Trying to match album %s on provider %s", db_album.name, provider.name
- )
- match_found = False
- search_str = f"{artist_name} - {db_album.name}"
- search_result = await self.search(search_str, provider.instance_id)
- for search_result_item in search_result:
- if not search_result_item.available:
- continue
- if not compare_media_item(db_album, search_result_item):
- continue
- # we must fetch the full album version, search results can be simplified objects
- prov_album = await self.get_provider_item(
- search_result_item.item_id,
- search_result_item.provider,
- fallback=search_result_item,
- )
- if compare_album(db_album, prov_album):
- # 100% match, we update the db with the additional provider mapping(s)
- match_found = True
- for provider_mapping in search_result_item.provider_mappings:
- await self.add_provider_mapping(db_album.item_id, provider_mapping)
- db_album.provider_mappings.add(provider_mapping)
- return match_found
# try to find match on all providers
- cur_provider_domains = {x.provider_domain for x in db_album.provider_mappings}
+ processed_domains = set()
for provider in self.mass.music.providers:
- if provider.domain in cur_provider_domains:
+ if provider.domain in processed_domains:
continue
if ProviderFeature.SEARCH not in provider.supported_features:
continue
if not provider.is_streaming_provider:
# matching on unique providers is pointless as they push (all) their content to MA
continue
- if await find_prov_match(provider):
- cur_provider_domains.add(provider.domain)
- else:
- self.logger.debug(
- "Could not find match for Album %s on provider %s",
- db_album.name,
- provider.name,
- )
+ if match := await self.match_provider(db_album, provider):
+ # 100% match, we update the db with the additional provider mapping(s)
+ await self.add_provider_mappings(db_album.item_id, match)
+ processed_domains.add(provider.domain)
def album_from_item_mapping(self, item: ItemMapping) -> Album:
"""Create an Album object from an ItemMapping object."""
MusicAssistantError,
ProviderUnavailableError,
)
-from music_assistant_models.media_items import Album, Artist, ItemMapping, Track
+from music_assistant_models.media_items import Album, Artist, ItemMapping, ProviderMapping, Track
from music_assistant.constants import (
DB_TABLE_ALBUM_ARTISTS,
in_library_only=False,
)
- async def match_providers(self, db_artist: Artist) -> None:
- """Try to find matching artists on all providers for the provided (database) item_id.
-
- This is used to link objects of different providers together.
+ async def match_provider(
+ self, db_artist: Artist, provider: MusicProvider, strict: bool = True
+ ) -> list[ProviderMapping]:
"""
- assert db_artist.provider == "library", "Matching only supported for database items!"
- cur_provider_domains = {x.provider_domain for x in db_artist.provider_mappings}
- for provider in self.mass.music.providers:
- if provider.domain in cur_provider_domains:
- continue
- if ProviderFeature.SEARCH not in provider.supported_features:
- continue
- if not provider.library_supported(MediaType.ARTIST):
- continue
- if not provider.is_streaming_provider:
- # matching on unique providers is pointless as they push (all) their content to MA
- continue
- if await self._match_provider(db_artist, provider):
- cur_provider_domains.add(provider.domain)
- else:
- self.logger.debug(
- "Could not find match for Artist %s on provider %s",
- db_artist.name,
- provider.name,
- )
+ Try to find match on (streaming) provider for the provided (database) artist.
- async def _match_provider(self, db_artist: Artist, provider: MusicProvider) -> bool:
- """Try to find matching artists on given provider for the provided (database) artist."""
+ This is used to link objects of different providers/qualities together.
+ """
self.logger.debug("Trying to match artist %s on provider %s", db_artist.name, provider.name)
+ matches: list[ProviderMapping] = []
# try to get a match with some reference tracks of this artist
ref_tracks = await self.mass.music.artists.tracks(db_artist.item_id, db_artist.provider)
if len(ref_tracks) < 10:
search_str = f"{db_artist.name} - {ref_track.name}"
search_results = await self.mass.music.tracks.search(search_str, provider.domain)
for search_result_item in search_results:
- if not compare_strings(search_result_item.name, ref_track.name, strict=True):
+ if not compare_strings(search_result_item.name, ref_track.name, strict=strict):
continue
# get matching artist from track
for search_item_artist in search_result_item.artists:
- if not compare_strings(search_item_artist.name, db_artist.name, strict=True):
+ if not compare_strings(search_item_artist.name, db_artist.name, strict=strict):
continue
# 100% track match
# get full artist details so we have all metadata
search_item_artist.provider,
fallback=search_item_artist,
)
- # 100% match, we update the db with the additional provider mapping(s)
- for provider_mapping in prov_artist.provider_mappings:
- await self.add_provider_mapping(db_artist.item_id, provider_mapping)
- db_artist.provider_mappings.add(provider_mapping)
- return True
+ # 100% match
+ matches.extend(prov_artist.provider_mappings)
+ if matches:
+ return matches
# try to get a match with some reference albums of this artist
ref_albums = await self.mass.music.artists.albums(db_artist.item_id, db_artist.provider)
if len(ref_albums) < 10:
for search_result_album in search_result_albums:
if not search_result_album.artists:
continue
- if not compare_strings(search_result_album.name, ref_album.name):
+ if not compare_strings(search_result_album.name, ref_album.name, strict=strict):
continue
# artist must match 100%
- if not compare_artist(db_artist, search_result_album.artists[0]):
+ if not compare_artist(db_artist, search_result_album.artists[0], strict=strict):
continue
# 100% match
# get full artist details so we have all metadata
search_result_album.artists[0].provider,
fallback=search_result_album.artists[0],
)
- await self._update_library_item(db_artist.item_id, prov_artist)
- return True
- return False
+ matches.extend(prov_artist.provider_mappings)
+ if matches:
+ return matches
+ if not matches:
+ self.logger.debug(
+ "Could not find match for Artist %s on provider %s",
+ db_artist.name,
+ provider.name,
+ )
+ return matches
+
+ async def match_providers(self, db_artist: Artist) -> None:
+ """Try to find matching artists on all providers for the provided (database) item_id.
+
+ This is used to link objects of different providers together.
+ """
+ if db_artist.provider != "library":
+ return # Matching only supported for database items
+
+ # try to find match on all providers
+
+ cur_provider_domains = {
+ x.provider_domain for x in db_artist.provider_mappings if x.available
+ }
+ for provider in self.mass.music.providers:
+ if provider.domain in cur_provider_domains:
+ continue
+ if ProviderFeature.SEARCH not in provider.supported_features:
+ continue
+ if not provider.library_supported(MediaType.ARTIST):
+ continue
+ if not provider.is_streaming_provider:
+ # matching on unique providers is pointless as they push (all) their content to MA
+ continue
+ if match := await self.match_provider(db_artist, provider):
+ # 100% match, we update the db with the additional provider mapping(s)
+ await self.add_provider_mappings(db_artist.item_id, match)
+ cur_provider_domains.add(provider.domain)
def artist_from_item_mapping(self, item: ItemMapping) -> Artist:
"""Create an Artist object from an ItemMapping object."""
from typing import TYPE_CHECKING, Any
from music_assistant_models.enums import MediaType, ProviderFeature
-from music_assistant_models.media_items import Audiobook, UniqueList
+from music_assistant_models.media_items import Audiobook, ProviderMapping, UniqueList
from music_assistant.constants import DB_TABLE_AUDIOBOOKS, DB_TABLE_PLAYLOG
from music_assistant.controllers.media.base import MediaControllerBase
msg = "Dynamic tracks not supported for Radio MediaItem"
raise NotImplementedError(msg)
- async def match_providers(self, db_audiobook: Audiobook) -> None:
- """Try to find match on all (streaming) providers for the provided (database) audiobook.
+ async def match_provider(
+ self, db_audiobook: Audiobook, provider: MusicProvider, strict: bool = True
+ ) -> list[ProviderMapping]:
+ """
+ Try to find match on (streaming) provider for the provided (database) audiobook.
This is used to link objects of different providers/qualities together.
"""
- if db_audiobook.provider != "library":
- return # Matching only supported for database items
- if not db_audiobook.authors:
- return # guard
- author_name = db_audiobook.authors[0]
-
- async def find_prov_match(provider: MusicProvider) -> bool:
+ self.logger.debug(
+ "Trying to match audiobook %s on provider %s",
+ db_audiobook.name,
+ provider.name,
+ )
+ matches: list[ProviderMapping] = []
+ author_name = db_audiobook.authors[0] if db_audiobook.authors else ""
+ search_str = f"{author_name} - {db_audiobook.name}" if author_name else db_audiobook.name
+ search_result = await self.search(search_str, provider.instance_id)
+ for search_result_item in search_result:
+ if not search_result_item.available:
+ continue
+ if not compare_media_item(db_audiobook, search_result_item, strict=strict):
+ continue
+ # we must fetch the full audiobook version, search results can be simplified objects
+ prov_audiobook = await self.get_provider_item(
+ search_result_item.item_id,
+ search_result_item.provider,
+ fallback=search_result_item,
+ )
+ if compare_audiobook(db_audiobook, prov_audiobook, strict=strict):
+ # 100% match
+ matches.extend(prov_audiobook.provider_mappings)
+ if not matches:
self.logger.debug(
- "Trying to match audiobook %s on provider %s",
+ "Could not find match for Audiobook %s on provider %s",
db_audiobook.name,
provider.name,
)
- match_found = False
- search_str = f"{author_name} - {db_audiobook.name}"
- search_result = await self.search(search_str, provider.instance_id)
- for search_result_item in search_result:
- if not search_result_item.available:
- continue
- if not compare_media_item(db_audiobook, search_result_item):
- continue
- # we must fetch the full audiobook version, search results can be simplified objects
- prov_audiobook = await self.get_provider_item(
- search_result_item.item_id,
- search_result_item.provider,
- fallback=search_result_item,
- )
- if compare_audiobook(db_audiobook, prov_audiobook):
- # 100% match, we update the db with the additional provider mapping(s)
- match_found = True
- for provider_mapping in search_result_item.provider_mappings:
- await self.add_provider_mapping(db_audiobook.item_id, provider_mapping)
- db_audiobook.provider_mappings.add(provider_mapping)
- return match_found
+ return matches
+
+ async def match_providers(self, db_audiobook: Audiobook) -> None:
+ """Try to find match on all (streaming) providers for the provided (database) audiobook.
+
+ This is used to link objects of different providers/qualities together.
+ """
+ if db_audiobook.provider != "library":
+ return # Matching only supported for database items
# try to find match on all providers
cur_provider_domains = {x.provider_domain for x in db_audiobook.provider_mappings}
if not provider.is_streaming_provider:
# matching on unique providers is pointless as they push (all) their content to MA
continue
- if await find_prov_match(provider):
+ if match := await self.match_provider(db_audiobook, provider):
+ # 100% match, we update the db with the additional provider mapping(s)
+ await self.add_provider_mappings(db_audiobook.item_id, match)
cur_provider_domains.add(provider.domain)
- else:
- self.logger.debug(
- "Could not find match for Audiobook %s on provider %s",
- db_audiobook.name,
- provider.name,
- )
async def _set_playlog(self, db_id: int, media_item: Audiobook) -> None:
"""Update/set the playlog table for the given audiobook db item_id."""
self, item_id: str | int, provider_mapping: ProviderMapping
) -> None:
"""Add provider mapping to existing library item."""
+ await self.add_provider_mappings(item_id, [provider_mapping])
+
+ @final
+ async def add_provider_mappings(
+ self, item_id: str | int, provider_mappings: Iterable[ProviderMapping]
+ ) -> None:
+ """
+ Add provider mappings to existing library item.
+
+ :param item_id: The library item ID to add mappings to.
+ :param provider_mappings: The provider mappings to add.
+ """
db_id = int(item_id) # ensure integer
library_item = await self.get_library_item(db_id)
- # ignore if the mapping is already present
- if provider_mapping in library_item.provider_mappings:
- return
- library_item.provider_mappings.add(provider_mapping)
- await self.set_provider_mappings(db_id, library_item.provider_mappings)
+ new_mappings: set[ProviderMapping] = set()
+ for provider_mapping in provider_mappings:
+ # ignore if the mapping is already present
+ if provider_mapping not in library_item.provider_mappings:
+ new_mappings.add(provider_mapping)
+ if new_mappings:
+ library_item.provider_mappings.update(new_mappings)
+ self.mass.music.match_provider_instances(library_item)
+ await self.set_provider_mappings(db_id, library_item.provider_mappings)
+ self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, library_item.uri, library_item)
@final
async def remove_provider_mapping(
):
# try to match the track to the playlist provider
full_track.provider_mappings.update(
- await self.mass.music.tracks.match_provider(playlist_prov, full_track, False)
+ await self.mass.music.tracks.match_provider(
+ full_track, playlist_prov, strict=False
+ )
)
# a track can contain multiple versions on the same provider
from music_assistant_models.enums import MediaType, ProviderFeature
from music_assistant_models.errors import MediaNotFoundError, ProviderUnavailableError
-from music_assistant_models.media_items import Podcast, PodcastEpisode, UniqueList
+from music_assistant_models.media_items import Podcast, PodcastEpisode, ProviderMapping, UniqueList
from music_assistant.constants import DB_TABLE_PLAYLOG, DB_TABLE_PODCASTS
from music_assistant.controllers.media.base import MediaControllerBase
msg = "Dynamic tracks not supported for Podcast MediaItem"
raise NotImplementedError(msg)
- async def match_providers(self, db_podcast: Podcast) -> None:
- """Try to find match on all (streaming) providers for the provided (database) podcast.
+ async def match_provider(
+ self, db_podcast: Podcast, provider: MusicProvider, strict: bool = True
+ ) -> list[ProviderMapping]:
+ """
+ Try to find match on (streaming) provider for the provided (database) podcast.
This is used to link objects of different providers/qualities together.
"""
- if db_podcast.provider != "library":
- return # Matching only supported for database items
-
- async def find_prov_match(provider: MusicProvider) -> bool:
+ self.logger.debug(
+ "Trying to match podcast %s on provider %s",
+ db_podcast.name,
+ provider.name,
+ )
+ matches: list[ProviderMapping] = []
+ search_str = db_podcast.name
+ search_result = await self.search(search_str, provider.instance_id)
+ for search_result_item in search_result:
+ if not search_result_item.available:
+ continue
+ if not compare_media_item(db_podcast, search_result_item, strict=strict):
+ continue
+ # we must fetch the full podcast version, search results can be simplified objects
+ prov_podcast = await self.get_provider_item(
+ search_result_item.item_id,
+ search_result_item.provider,
+ fallback=search_result_item,
+ )
+ if compare_podcast(db_podcast, prov_podcast, strict=strict):
+ # 100% match
+ matches.extend(prov_podcast.provider_mappings)
+ if not matches:
self.logger.debug(
- "Trying to match podcast %s on provider %s",
+ "Could not find match for Podcast %s on provider %s",
db_podcast.name,
provider.name,
)
- match_found = False
- search_str = db_podcast.name
- search_result = await self.search(search_str, provider.instance_id)
- for search_result_item in search_result:
- if not search_result_item.available:
- continue
- if not compare_media_item(db_podcast, search_result_item):
- continue
- # we must fetch the full podcast version, search results can be simplified objects
- prov_podcast = await self.get_provider_item(
- search_result_item.item_id,
- search_result_item.provider,
- fallback=search_result_item,
- )
- if compare_podcast(db_podcast, prov_podcast):
- # 100% match, we update the db with the additional provider mapping(s)
- match_found = True
- for provider_mapping in search_result_item.provider_mappings:
- await self.add_provider_mapping(db_podcast.item_id, provider_mapping)
- db_podcast.provider_mappings.add(provider_mapping)
- return match_found
+ return matches
+
+ async def match_providers(self, db_podcast: Podcast) -> None:
+ """Try to find match on all (streaming) providers for the provided (database) podcast.
+
+ This is used to link objects of different providers/qualities together.
+ """
+ if db_podcast.provider != "library":
+ return # Matching only supported for database items
# try to find match on all providers
cur_provider_domains = {x.provider_domain for x in db_podcast.provider_mappings}
if not provider.is_streaming_provider:
# matching on unique providers is pointless as they push (all) their content to MA
continue
- if await find_prov_match(provider):
+ if match := await self.match_provider(db_podcast, provider):
+ # 100% match, we update the db with the additional provider mapping(s)
+ await self.add_provider_mappings(db_podcast.item_id, match)
cur_provider_domains.add(provider.domain)
- else:
- self.logger.debug(
- "Could not find match for Podcast %s on provider %s",
- db_podcast.name,
- provider.name,
- )
import asyncio
from typing import TYPE_CHECKING
-from music_assistant_models.enums import MediaType
-from music_assistant_models.media_items import Radio, Track
+from music_assistant_models.enums import MediaType, ProviderFeature
+from music_assistant_models.media_items import ProviderMapping, Radio, Track
from music_assistant.constants import DB_TABLE_RADIOS
-from music_assistant.helpers.compare import create_safe_string, loose_compare_strings
+from music_assistant.helpers.compare import (
+ compare_media_item,
+ compare_radio,
+ create_safe_string,
+ loose_compare_strings,
+)
from music_assistant.helpers.json import serialize_to_json
+from music_assistant.models.music_provider import MusicProvider
from .base import MediaControllerBase
msg = "Dynamic tracks not supported for Radio MediaItem"
raise NotImplementedError(msg)
- async def match_providers(self, db_item: Radio) -> None:
- """Try to find match on all (streaming) providers for the provided (database) item.
+ async def match_provider(
+ self, db_radio: Radio, provider: MusicProvider, strict: bool = True
+ ) -> list[ProviderMapping]:
+ """
+ Try to find match on (streaming) provider for the provided (database) radio.
+
+ This is used to link objects of different providers/qualities together.
+ """
+ self.logger.debug(
+ "Trying to match radio %s on provider %s",
+ db_radio.name,
+ provider.name,
+ )
+ matches: list[ProviderMapping] = []
+ search_str = db_radio.name
+ search_result = await self.search(search_str, provider.instance_id)
+ for search_result_item in search_result:
+ if not search_result_item.available:
+ continue
+ if not compare_media_item(db_radio, search_result_item, strict=strict):
+ continue
+ # we must fetch the full radio version, search results can be simplified objects
+ prov_radio = await self.get_provider_item(
+ search_result_item.item_id,
+ search_result_item.provider,
+ fallback=search_result_item,
+ )
+ if compare_radio(db_radio, prov_radio, strict=strict):
+ # 100% match
+ matches.extend(prov_radio.provider_mappings)
+ if not matches:
+ self.logger.debug(
+ "Could not find match for Radio %s on provider %s",
+ db_radio.name,
+ provider.name,
+ )
+ return matches
+
+ async def match_providers(self, db_radio: Radio) -> None:
+ """Try to find match on all (streaming) providers for the provided (database) radio.
This is used to link objects of different providers/qualities together.
"""
- raise NotImplementedError
+ if db_radio.provider != "library":
+ return # Matching only supported for database items
+
+ # try to find match on all providers
+ cur_provider_domains = {x.provider_domain for x in db_radio.provider_mappings}
+ for provider in self.mass.music.providers:
+ if provider.domain in cur_provider_domains:
+ continue
+ if ProviderFeature.SEARCH not in provider.supported_features:
+ continue
+ if not provider.library_supported(MediaType.RADIO):
+ continue
+ if not provider.is_streaming_provider:
+ # matching on unique providers is pointless as they push (all) their content to MA
+ continue
+ if match := await self.match_provider(db_radio, provider):
+ # 100% match, we update the db with the additional provider mapping(s)
+ await self.add_provider_mappings(db_radio.item_id, match)
+ cur_provider_domains.add(provider.domain)
query = f"{DB_TABLE_ALBUMS}.item_id in ({subquery})"
return await self.mass.music.albums._get_library_items_by_query(extra_query_parts=[query])
- async def match_providers(self, ref_track: Track) -> None:
- """Try to find matching track on all providers for the provided (database) track_id.
-
- This is used to link objects of different providers/qualities together.
- """
- track_albums = await self.albums(ref_track.item_id, ref_track.provider)
- for provider in self.mass.music.providers:
- if ProviderFeature.SEARCH not in provider.supported_features:
- continue
- if not provider.is_streaming_provider:
- # matching on unique providers is pointless as they push (all) their content to MA
- continue
- if not provider.library_supported(MediaType.TRACK):
- continue
- provider_matches = await self.match_provider(
- provider, ref_track, strict=True, ref_albums=track_albums
- )
- for provider_mapping in provider_matches:
- # 100% match, we update the db with the additional provider mapping(s)
- await self.add_provider_mapping(ref_track.item_id, provider_mapping)
- ref_track.provider_mappings.add(provider_mapping)
-
async def match_provider(
self,
+ db_track: Track,
provider: MusicProvider,
- ref_track: Track,
strict: bool = True,
ref_albums: list[Album] | None = None,
- ) -> set[ProviderMapping]:
- """Try to find matching track on given provider."""
+ ) -> list[ProviderMapping]:
+ """
+ Try to find match on (streaming) provider for the provided (database) track.
+
+ This is used to link objects of different providers/qualities together.
+ """
if ref_albums is None:
- ref_albums = await self.albums(ref_track.item_id, ref_track.provider)
- if ProviderFeature.SEARCH not in provider.supported_features:
- raise UnsupportedFeaturedException("Provider does not support search")
- if not provider.is_streaming_provider:
- raise UnsupportedFeaturedException("Matching only possible for streaming providers")
- self.logger.debug("Trying to match track %s on provider %s", ref_track.name, provider.name)
- matches: set[ProviderMapping] = set()
- for artist in ref_track.artists:
+ ref_albums = await self.albums(db_track.item_id, db_track.provider)
+ self.logger.debug("Trying to match track %s on provider %s", db_track.name, provider.name)
+ matches: list[ProviderMapping] = []
+ for artist in db_track.artists:
if matches:
break
- search_str = f"{artist.name} - {ref_track.name}"
+ search_str = f"{artist.name} - {db_track.name}"
search_result = await self.search(search_str, provider.domain)
for search_result_item in search_result:
if not search_result_item.available:
continue
# do a basic compare first
- if not compare_media_item(ref_track, search_result_item, strict=False):
+ if not compare_media_item(db_track, search_result_item, strict=False):
continue
# we must fetch the full version, search results can be simplified objects
prov_track = await self.get_provider_item(
search_result_item.provider,
fallback=search_result_item,
)
- if compare_track(ref_track, prov_track, strict=strict, track_albums=ref_albums):
- matches.update(search_result_item.provider_mappings)
+ if compare_track(db_track, prov_track, strict=strict, track_albums=ref_albums):
+ matches.extend(search_result_item.provider_mappings)
if not matches:
self.logger.debug(
"Could not find match for Track %s on provider %s",
- ref_track.name,
+ db_track.name,
provider.name,
)
return matches
+ async def match_providers(self, db_track: Track) -> None:
+ """Try to find matching track on all providers for the provided (database) track_id.
+
+ This is used to link objects of different providers/qualities together.
+ """
+ if db_track.provider != "library":
+ return # Matching only supported for database items
+
+ track_albums = await self.albums(db_track.item_id, db_track.provider)
+ # try to find match on all providers
+ processed_domains = set()
+ for provider in self.mass.music.providers:
+ if provider.domain in processed_domains:
+ continue
+ if ProviderFeature.SEARCH not in provider.supported_features:
+ continue
+ if not provider.library_supported(MediaType.TRACK):
+ continue
+ if not provider.is_streaming_provider:
+ # matching on unique providers is pointless as they push (all) their content to MA
+ continue
+ if match := await self.match_provider(
+ db_track, provider, strict=True, ref_albums=track_albums
+ ):
+ # 100% match, we update the db with the additional provider mapping(s)
+ await self.add_provider_mappings(db_track.item_id, match)
+ processed_domains.add(provider.domain)
+
async def radio_mode_base_tracks(
self,
item_id: str,
# add (or overwrite) to library
ctrl = self.get_controller(full_item.media_type)
library_item = await ctrl.add_item_to_library(full_item, overwrite_existing)
- # perform full metadata scan (and provider match)
+ # perform full metadata scan
await self.mass.metadata.update_metadata(library_item, overwrite_existing)
return library_item
tg.create_task(self.refresh_item(media_item))
@api_command("music/refresh_item")
- async def refresh_item(
+ async def refresh_item( # noqa: PLR0915
self,
media_item: str | MediaItemType,
) -> MediaItemType | None:
await self.mass.music.tracks.update_item_in_library(
album_track.item_id, prov_track
)
-
+ await ctrl.match_providers(library_item)
await self.mass.metadata.update_metadata(library_item, force_refresh=True)
return library_item
)
)
+ @api_command("music/add_provider_mapping")
+ async def add_provider_mapping(
+ self, media_type: MediaType, db_id: str, mapping: ProviderMapping
+ ) -> None:
+ """Add provider mapping to the given library item."""
+ ctrl = self.get_controller(media_type)
+ await ctrl.add_provider_mappings(db_id, [mapping])
+
+ @api_command("music/remove_provider_mapping")
+ async def remove_provider_mapping(
+ self, media_type: MediaType, db_id: str, mapping: ProviderMapping
+ ) -> None:
+ """Remove provider mapping from the given library item."""
+ ctrl = self.get_controller(media_type)
+ await ctrl.remove_provider_mapping(db_id, mapping.provider_instance, mapping.item_id)
+
+ @api_command("music/match_providers")
+ async def match_providers(self, media_type: MediaType, db_id: str) -> None:
+ """Search for mappings on all providers for the given library item."""
+ ctrl = self.get_controller(media_type)
+ db_item = await ctrl.get_library_item(db_id)
+ await ctrl.match_providers(db_item)
+
async def _get_default_recommendations(self) -> list[RecommendationFolder]:
"""Return default recommendations."""
return [
projects / music-assistant-server.git / commitdiff