music-assistant-server

1"""Model/base for a Music Provider implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from collections.abc import Sequence
8from contextlib import asynccontextmanager
9from contextvars import ContextVar
10from dataclasses import dataclass
11from datetime import datetime
12from typing import TYPE_CHECKING, Final, cast
13
14from music_assistant_models.background_task import TaskSchedule
15from music_assistant_models.enums import ArtistType, MediaType, ProviderFeature
16from music_assistant_models.errors import (
17    AudioError,
18    InvalidDataError,
19    MediaNotFoundError,
20    MusicAssistantError,
21    UnsupportedFeaturedException,
22)
23from music_assistant_models.media_items import (
24    Album,
25    Artist,
26    Audiobook,
27    BrowseFolder,
28    ItemMapping,
29    MediaItemType,
30    Playlist,
31    Podcast,
32    PodcastEpisode,
33    Radio,
34    RecommendationFolder,
35    SearchResults,
36    SoundEffect,
37    Track,
38    UniqueList,
39)
40
41from music_assistant.constants import (
42    CONF_ENTRY_LIBRARY_SYNC_ALBUM_TRACKS,
43    CONF_ENTRY_LIBRARY_SYNC_DELETIONS,
44    CONF_ENTRY_LIBRARY_SYNC_PLAYLIST_TRACKS,
45    PlaylistPlayableItem,
46)
47from music_assistant.controllers.tasks.context import (
48    report_current_task_failure,
49    update_current_task_progress_text,
50)
51
52from .provider import Provider
53
54if TYPE_CHECKING:
55    from collections.abc import AsyncGenerator
56
57    from music_assistant_models.config_entries import ProviderConfig
58    from music_assistant_models.provider import ProviderManifest
59    from music_assistant_models.streamdetails import StreamDetails
60
61    from music_assistant.controllers.music.media.base import (
62        AudiobookSyncDetails,
63        LibraryItemSyncDetails,
64        TrackSyncDetails,
65    )
66    from music_assistant.mass import MusicAssistant
67
68CACHE_CATEGORY_PREV_LIBRARY_IDS: Final[int] = 1
69DEFAULT_MAX_CONCURRENT_STREAMS: Final[int] = 5
70# a provider-wide payload change fails every single item, so only the first failures
71# of a sync run are logged in full to keep the (rotating) log file usable
72MAX_LOGGED_SYNC_FAILURES: Final[int] = 25
73MAX_SYNC_ERROR_DETAIL: Final[int] = 200
74
75LIBRARY_FEATURE_BY_MEDIA_TYPE: Final[dict[MediaType, ProviderFeature]] = {
76    MediaType.ARTIST: ProviderFeature.LIBRARY_ARTISTS,
77    MediaType.ALBUM: ProviderFeature.LIBRARY_ALBUMS,
78    MediaType.TRACK: ProviderFeature.LIBRARY_TRACKS,
79    MediaType.PLAYLIST: ProviderFeature.LIBRARY_PLAYLISTS,
80    MediaType.RADIO: ProviderFeature.LIBRARY_RADIOS,
81    MediaType.AUDIOBOOK: ProviderFeature.LIBRARY_AUDIOBOOKS,
82    MediaType.PODCAST: ProviderFeature.LIBRARY_PODCASTS,
83}
84
85
86@dataclass
87class SyncRunState:
88    """
89    Failure state of one library sync run.
90
91    :param incomplete: True once the run failed to collect an item, which makes its
92        result set an unsafe basis for deleting anything from the library.
93    :param failures: Number of item failures reported by the run so far.
94    """
95
96    incomplete: bool = False
97    failures: int = 0
98
99
100# scoped per run rather than per provider: a standalone import_album_tracks() is
101# launched as its own task, so it must not consume or inflate a running sync's state
102SYNC_RUN_STATE: Final[ContextVar[SyncRunState | None]] = ContextVar(
103    "music_provider_sync_run", default=None
104)
105
106
107def sync_run_state() -> SyncRunState:
108    """Return the state of the sync run in progress, starting one if there is none."""
109    if (state := SYNC_RUN_STATE.get()) is None:
110        state = SyncRunState()
111        SYNC_RUN_STATE.set(state)
112    return state
113
114
115class ProviderStreamLimitError(AudioError):
116    """Raised when a music provider has no source-stream slot available."""
117
118    translation_key = "provider_stream_limit"
119
120    def __init__(self, provider: MusicProvider, wait_timeout: float | None) -> None:
121        """
122        Initialize the provider stream limit error.
123
124        :param provider: Provider instance whose source-stream limit was reached.
125        :param wait_timeout: Seconds spent waiting for a slot, or None for an unbounded wait.
126        """
127        limit = provider.max_concurrent_streams
128        assert limit is not None
129        wait_text = f" after waiting {wait_timeout:g} seconds" if wait_timeout is not None else ""
130        super().__init__(
131            f"{provider.name} has reached its limit of {limit} "
132            f"concurrent source streams{wait_text}.",
133            translation_args=[provider.name, limit],
134        )
135        self.provider_instance = provider.instance_id
136        self.limit = limit
137
138
139def describe_sync_error(err: Exception) -> str:
140    """Return a short description of a sync failure, safe to log and to report to clients."""
141    if isinstance(err, MusicAssistantError):
142        return str(err)
143    # an unexpected error can carry an entire api response as its message, which would end
144    # up in the log and - through the task failure list - in every connected client. report
145    # it by type with a clipped detail and leave the full payload to the debug traceback
146    detail = str(err)
147    if not detail:
148        return type(err).__name__
149    if len(detail) > MAX_SYNC_ERROR_DETAIL:
150        detail = f"{detail[:MAX_SYNC_ERROR_DETAIL]}..."
151    return f"{type(err).__name__}: {detail}"
152
153
154class MusicProvider(Provider):
155    """
156    Base representation of a Music Provider (controller).
157
158    Music Provider implementations should inherit from this base model.
159    """
160
161    def __init__(
162        self,
163        mass: MusicAssistant,
164        manifest: ProviderManifest,
165        config: ProviderConfig,
166        supported_features: set[ProviderFeature] | None = None,
167    ) -> None:
168        """Initialize MusicProvider."""
169        super().__init__(mass, manifest, config, supported_features)
170        max_concurrent_streams = self.max_concurrent_streams
171        if max_concurrent_streams is not None and max_concurrent_streams < 1:
172            raise ValueError("max_concurrent_streams must be at least 1 or None")
173        self._stream_semaphore = (
174            asyncio.BoundedSemaphore(max_concurrent_streams)
175            if max_concurrent_streams is not None
176            else None
177        )
178
179    @property
180    def max_concurrent_streams(self) -> int | None:
181        """
182        Return the number of source streams Music Assistant may run against this provider.
183
184        None means no limit is imposed, which is the correct answer for local and
185        self-hosted sources. Streaming providers get a conservative default of five;
186        override with a lower, evidence-backed value where the service enforces one.
187        Plugin providers (exclusive audio sources) manage their own session exclusivity
188        and are not covered by this limit.
189        """
190        return DEFAULT_MAX_CONCURRENT_STREAMS if self.is_streaming_provider else None
191
192    @property
193    def has_available_stream_slot(self) -> bool:
194        """Return whether a source stream can start without waiting."""
195        return self._stream_semaphore is None or not self._stream_semaphore.locked()
196
197    @asynccontextmanager
198    async def acquire_stream_slot(self, wait_timeout: float | None) -> AsyncGenerator[None]:
199        """
200        Acquire one source-stream slot for the duration of the context.
201
202        :param wait_timeout: Maximum seconds to wait, or None to wait without a timeout.
203        :raises ProviderStreamLimitError: If no slot becomes available before the timeout.
204        """
205        semaphore = self._stream_semaphore
206        if semaphore is None:
207            yield
208            return
209        try:
210            if wait_timeout is None:
211                await semaphore.acquire()
212            else:
213                async with asyncio.timeout(wait_timeout):
214                    await semaphore.acquire()
215        except TimeoutError as err:
216            raise ProviderStreamLimitError(self, wait_timeout) from err
217        try:
218            yield
219        finally:
220            semaphore.release()
221
222    @property
223    def is_streaming_provider(self) -> bool:
224        """
225        Return True if the provider is a streaming provider.
226
227        This literally means that the catalog is not the same as the library contents.
228        For local based providers (files, plex), the catalog is the same as the library content.
229        It also means that data is if this provider is NOT a streaming provider,
230        data cross instances is unique, the catalog and library differs per instance.
231
232        Setting this to True will only query one instance of the provider for search and lookups.
233        Setting this to False will query all instances of this provider for search and lookups.
234        """
235        return True
236
237    @property
238    def supported_media_types(self) -> set[MediaType]:
239        """
240        Return the media types this provider can serve.
241
242        Defaults to the media types the provider declares library support for.
243        Override for providers that can serve (search/stream) media types they
244        cannot list as library items, so they are eligible for search-based
245        lookups such as cross-provider matching and versions.
246        """
247        return {
248            media_type
249            for media_type, feature in LIBRARY_FEATURE_BY_MEDIA_TYPE.items()
250            if feature in self.supported_features
251        }
252
253    @property
254    def unskippable_sync_errors(self) -> tuple[type[Exception], ...]:
255        """
256        Return the errors a library sync must never treat as a skippable item failure.
257
258        Declare the errors this provider raises to signal something a wrapper around its
259        own methods has to act on, such as an expired token that triggers a reauthenticate
260        and a retry. Anything listed here is re-raised instead of skipping the item.
261        """
262        return ()
263
264    @property
265    def supported_artist_types(self) -> set[ArtistType]:
266        """
267        Return all supported artist types by this provider.
268
269        Note, that this property currently is only used, to verify support of artists with
270        ArtistType.AUTHOR or ArtistType.NARRATOR.
271        """
272        return {ArtistType.SINGER}
273
274    async def loaded_in_mass(self) -> None:
275        """Call after the provider has been loaded."""
276
277    async def search(
278        self,
279        search_query: str,
280        media_types: list[MediaType],
281        limit: int = 5,
282    ) -> SearchResults:
283        """
284        Perform search on musicprovider.
285
286        :param search_query: Search query.
287        :param media_types: A list of media_types to include.
288        :param limit: Number of items to return in the search (per type).
289        """
290        if ProviderFeature.SEARCH in self.supported_features:
291            raise NotImplementedError
292        return SearchResults()
293
294    async def get_library_artists(self) -> AsyncGenerator[Artist]:
295        """Retrieve library artists from the provider."""
296        yield  # type: ignore[misc]
297        raise NotImplementedError
298
299    async def get_library_albums(self) -> AsyncGenerator[Album]:
300        """Retrieve library albums from the provider."""
301        yield  # type: ignore[misc]
302        raise NotImplementedError
303
304    async def get_library_tracks(self) -> AsyncGenerator[Track]:
305        """Retrieve library tracks from the provider."""
306        yield  # type: ignore[misc]
307        raise NotImplementedError
308
309    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
310        """Retrieve library/subscribed playlists from the provider."""
311        yield  # type: ignore[misc]
312        raise NotImplementedError
313
314    async def get_library_radios(self) -> AsyncGenerator[Radio]:
315        """Retrieve library/subscribed radio stations from the provider."""
316        yield  # type: ignore[misc]
317        raise NotImplementedError
318
319    async def get_library_audiobooks(self) -> AsyncGenerator[Audiobook]:
320        """Retrieve library/subscribed audiobooks from the provider."""
321        yield  # type: ignore[misc]
322        raise NotImplementedError
323
324    async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
325        """Retrieve library/subscribed podcasts from the provider."""
326        yield  # type: ignore[misc]
327        raise NotImplementedError
328
329    async def get_library_genres(self) -> AsyncGenerator[str]:
330        """Retrieve library genres from the provider."""
331        yield  # type: ignore[misc]
332        raise NotImplementedError
333
334    async def get_artist(self, prov_artist_id: str) -> Artist:
335        """Get full artist details by id."""
336        raise NotImplementedError
337
338    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
339        """
340        Get a list of all albums for the given artist.
341
342        Only called if provider supports ProviderFeature.ARTIST_ALBUMS.
343        """
344        raise NotImplementedError
345
346    async def get_artist_tracks(self, prov_artist_id: str) -> list[Track]:
347        """
348        Get a list of all tracks for the given artist.
349
350        Only called if provider supports ProviderFeature.ARTIST_TRACKS.
351        """
352        raise NotImplementedError
353
354    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
355        """
356        Get a list of most popular tracks for the given artist.
357
358        Only called if provider supports ProviderFeature.ARTIST_TOPTRACKS.
359        """
360        raise NotImplementedError
361
362    async def get_artist_topalbums(self, prov_artist_id: str) -> list[Album]:
363        """
364        Get a list of most popular albums for the given artist.
365
366        Only called if provider supports ProviderFeature.ARTIST_TOPALBUMS.
367        """
368        raise NotImplementedError
369
370    async def get_album(self, prov_album_id: str) -> Album:
371        """Get full album details by id."""
372        raise NotImplementedError
373
374    async def get_track(self, prov_track_id: str) -> Track:
375        """Get full track details by id."""
376        raise NotImplementedError
377
378    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
379        """Get full playlist details by id."""
380        raise NotImplementedError
381
382    async def get_radio(self, prov_radio_id: str) -> Radio:
383        """Get full radio details by id."""
384        raise NotImplementedError
385
386    async def get_audiobook(self, prov_audiobook_id: str) -> Audiobook:
387        """Get full audiobook details by id."""
388        raise NotImplementedError
389
390    async def get_author_audiobooks(self, prov_artist_id: str) -> list[Audiobook]:
391        """
392        Get a list of all audiobooks for the given author.
393
394        Only called if provider supports ProviderFeature.AUTHOR_AUDIOBOOKS.
395        """
396        raise NotImplementedError
397
398    async def get_narrator_audiobooks(self, prov_artist_id: str) -> list[Audiobook]:
399        """
400        Get a list of all audiobooks for the given narrator.
401
402        Only called if provider supports ProviderFeature.NARRATOR_AUDIOBOOKS.
403        """
404        raise NotImplementedError
405
406    async def get_podcast(self, prov_podcast_id: str) -> Podcast:
407        """Get full podcast details by id."""
408        raise NotImplementedError
409
410    async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
411        """Get (full) podcast episode details by id."""
412        raise NotImplementedError
413
414    async def get_sound_effect(self, prov_sound_effect_id: str) -> SoundEffect:
415        """Get full sound effect details by id."""
416        raise NotImplementedError
417
418    async def get_sound_effects(self) -> AsyncGenerator[SoundEffect]:
419        """
420        Get all sound effect items this provider offers.
421
422        Sound effects are not library-backed; they are fetched live from the provider.
423        Only called if provider supports ProviderFeature.SOUND_EFFECTS.
424        """
425        yield  # type: ignore[misc]
426        raise NotImplementedError
427
428    async def get_item_genre_names(self, media_type: MediaType, item_id: str) -> set[str]:
429        """Return genre names for a single item."""
430        raise NotImplementedError
431
432    async def get_album_tracks(
433        self,
434        prov_album_id: str,
435    ) -> list[Track]:
436        """Get album tracks for given album id."""
437        raise NotImplementedError
438
439    async def get_playlist_tracks(
440        self,
441        prov_playlist_id: str,
442        page: int = 0,
443    ) -> Sequence[PlaylistPlayableItem]:
444        """Get all playlist tracks for given playlist id."""
445        raise NotImplementedError
446
447    async def get_dynamic_radio_tracks(self, prov_radio_id: str) -> list[Track]:
448        """
449        Return a fresh batch of tracks for a dynamic radio station.
450
451        Only called for a Radio with `is_dynamic` set. Every call returns a new batch;
452        there is no stable listing and no pagination.
453
454        :param prov_radio_id: The provider's ID of the radio station.
455        """
456        raise NotImplementedError
457
458    async def get_podcast_episodes(
459        self,
460        prov_podcast_id: str,
461    ) -> AsyncGenerator[PodcastEpisode]:
462        """Get all PodcastEpisodes for given podcast id."""
463        yield  # type: ignore[misc]
464        raise NotImplementedError
465
466    async def library_add(self, item: MediaItemType) -> bool:
467        """Add item to provider's library. Return true on success."""
468        if (
469            item.media_type == MediaType.ARTIST
470            and ProviderFeature.LIBRARY_ARTISTS_EDIT in self.supported_features
471        ):
472            raise NotImplementedError
473        if (
474            item.media_type == MediaType.ALBUM
475            and ProviderFeature.LIBRARY_ALBUMS_EDIT in self.supported_features
476        ):
477            raise NotImplementedError
478        if (
479            item.media_type == MediaType.TRACK
480            and ProviderFeature.LIBRARY_TRACKS_EDIT in self.supported_features
481        ):
482            raise NotImplementedError
483        if (
484            item.media_type == MediaType.PLAYLIST
485            and ProviderFeature.LIBRARY_PLAYLISTS_EDIT in self.supported_features
486        ):
487            raise NotImplementedError
488        if (
489            item.media_type == MediaType.RADIO
490            and ProviderFeature.LIBRARY_RADIOS_EDIT in self.supported_features
491        ):
492            raise NotImplementedError
493        if (
494            item.media_type == MediaType.AUDIOBOOK
495            and ProviderFeature.LIBRARY_AUDIOBOOKS_EDIT in self.supported_features
496        ):
497            raise NotImplementedError
498        if (
499            item.media_type == MediaType.PODCAST
500            and ProviderFeature.LIBRARY_PODCASTS_EDIT in self.supported_features
501        ):
502            raise NotImplementedError
503        self.logger.info(
504            "Provider %s does not support library edit, "
505            "the action will only be performed in the local database.",
506            self.name,
507        )
508        return True
509
510    async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
511        """Remove item from provider's library. Return true on success."""
512        if (
513            media_type == MediaType.ARTIST
514            and ProviderFeature.LIBRARY_ARTISTS_EDIT in self.supported_features
515        ):
516            raise NotImplementedError
517        if (
518            media_type == MediaType.ALBUM
519            and ProviderFeature.LIBRARY_ALBUMS_EDIT in self.supported_features
520        ):
521            raise NotImplementedError
522        if (
523            media_type == MediaType.TRACK
524            and ProviderFeature.LIBRARY_TRACKS_EDIT in self.supported_features
525        ):
526            raise NotImplementedError
527        if (
528            media_type == MediaType.PLAYLIST
529            and ProviderFeature.LIBRARY_PLAYLISTS_EDIT in self.supported_features
530        ):
531            raise NotImplementedError
532        if (
533            media_type == MediaType.RADIO
534            and ProviderFeature.LIBRARY_RADIOS_EDIT in self.supported_features
535        ):
536            raise NotImplementedError
537        if (
538            media_type == MediaType.AUDIOBOOK
539            and ProviderFeature.LIBRARY_AUDIOBOOKS_EDIT in self.supported_features
540        ):
541            raise NotImplementedError
542        if (
543            media_type == MediaType.PODCAST
544            and ProviderFeature.LIBRARY_PODCASTS_EDIT in self.supported_features
545        ):
546            raise NotImplementedError
547        self.logger.info(
548            "Provider %s does not support library edit, "
549            "the action will only be performed in the local database.",
550            self.name,
551        )
552        return True
553
554    async def set_favorite(self, prov_item_id: str, media_type: MediaType, favorite: bool) -> None:
555        """
556        Set favorite status for item in provider's library.
557
558        Only called if provider supports ProviderFeature.FAVORITE_*_EDIT.
559
560        Note that this should only be implemented by a provider implementation if
561        the provider differentiates between 'in library' and 'favorited' items.
562        """
563        if (
564            media_type == MediaType.ARTIST
565            and ProviderFeature.FAVORITE_ARTISTS_EDIT in self.supported_features
566        ):
567            raise NotImplementedError
568        if (
569            media_type == MediaType.ALBUM
570            and ProviderFeature.FAVORITE_ALBUMS_EDIT in self.supported_features
571        ):
572            raise NotImplementedError
573        if (
574            media_type == MediaType.TRACK
575            and ProviderFeature.FAVORITE_TRACKS_EDIT in self.supported_features
576        ):
577            raise NotImplementedError
578        if (
579            media_type == MediaType.PLAYLIST
580            and ProviderFeature.FAVORITE_PLAYLISTS_EDIT in self.supported_features
581        ):
582            raise NotImplementedError
583        if (
584            media_type == MediaType.RADIO
585            and ProviderFeature.FAVORITE_RADIOS_EDIT in self.supported_features
586        ):
587            raise NotImplementedError
588        if (
589            media_type == MediaType.AUDIOBOOK
590            and ProviderFeature.FAVORITE_AUDIOBOOKS_EDIT in self.supported_features
591        ):
592            raise NotImplementedError
593        if (
594            media_type == MediaType.PODCAST
595            and ProviderFeature.FAVORITE_PODCASTS_EDIT in self.supported_features
596        ):
597            raise NotImplementedError
598
599    async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
600        """
601        Add track(s) to playlist.
602
603        Only called if provider supports ProviderFeature.PLAYLIST_TRACKS_EDIT.
604        """
605        raise NotImplementedError
606
607    async def remove_playlist_tracks(
608        self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
609    ) -> None:
610        """
611        Remove track(s) from playlist.
612
613        Only called if provider supports ProviderFeature.PLAYLIST_TRACKS_EDIT.
614        """
615        raise NotImplementedError
616
617    async def create_playlist(self, name: str, media_types: set[MediaType]) -> Playlist:
618        """
619        Create a new playlist on provider with given name and targeting media_types.
620
621        Only called if provider supports ProviderFeature.PLAYLIST_CREATE.
622        """
623        raise NotImplementedError
624
625    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
626        """
627        Retrieve a dynamic list of similar tracks based on the provided track.
628
629        Only called if provider supports ProviderFeature.SIMILAR_TRACKS.
630        """
631        raise NotImplementedError
632
633    async def get_similar_artists(self, prov_artist_id: str, limit: int = 25) -> list[Artist]:
634        """
635        Retrieve a dynamic list of similar artists based on the provided artist.
636
637        Only called if provider supports ProviderFeature.SIMILAR_ARTISTS.
638        """
639        raise NotImplementedError
640
641    async def get_resume_position(
642        self, item_id: str, media_type: MediaType
643    ) -> tuple[bool, int, datetime | None]:
644        """
645        Get progress (resume point) details for the given Audiobook or Podcast episode.
646
647        This is a separate call from the regular get_item call to ensure the resume position
648        is always up-to-date and because a lot providers have this info present on a dedicated
649        endpoint.
650
651        Will be called right before playback starts to ensure the resume position is correct.
652
653        Returns a boolean with the fully_played status
654        an integer with the resume position in ms,
655        and an optional timestamp as datetime giving when this resume position was set
656        """
657        raise NotImplementedError
658
659    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
660        """Get streamdetails for a track/radio/chapter/episode."""
661        raise NotImplementedError
662
663    async def get_audio_stream(
664        self, streamdetails: StreamDetails, seek_position: int = 0
665    ) -> AsyncGenerator[bytes]:
666        """
667        Return the (custom) audio stream for the provider item.
668
669        Will only be called when the stream_type is set to CUSTOM.
670        """
671        yield b""
672        raise NotImplementedError
673
674    async def on_streamed(
675        self,
676        streamdetails: StreamDetails,
677    ) -> None:
678        """
679        Handle callback when given streamdetails completed streaming.
680
681        To get the number of seconds streamed, see streamdetails.seconds_streamed.
682        To get the number of seconds seeked/skipped, see streamdetails.seek_position.
683        Note that seconds_streamed is the total streamed seconds, so without seeked time.
684
685        NOTE: Due to internal and player buffering,
686        this may be called in advance of the actual completion.
687        """
688
689    async def on_played(
690        self,
691        media_type: MediaType,
692        prov_item_id: str,
693        fully_played: bool,
694        position: int,
695        media_item: MediaItemType,
696        is_playing: bool = False,
697    ) -> None:
698        """
699        Handle callback when a (playable) media item has been played.
700
701        This is called by the Queue controller when;
702            - a track has been fully played
703            - a track has been stopped (or skipped) after being played
704            - every 30s when a track is playing
705
706        Fully played is True when the track has been played to the end.
707
708        Position is the last known position of the track in seconds, to sync resume state.
709        When fully_played is set to false and position is 0,
710        the user marked the item as unplayed in the UI.
711
712        media_item is the full media item details of the played/playing track.
713
714        is_playing is True when the track is currently playing.
715        """
716
717    async def on_item_updated(self, item: MediaItemType) -> None:
718        """
719        Handle callback when a library item's metadata has been updated.
720
721        Providers can implement this to sync changes to their own storage
722        (e.g. config entries, file tags).
723
724        :param item: The updated library item.
725        """
726
727    async def resolve_image(self, path: str) -> str | bytes:
728        """
729        Resolve an image from an image path.
730
731        This either returns (a generator to get) raw bytes of the image or
732        a string with an http(s) URL or local path that is accessible from the server.
733        """
734        return path
735
736    async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:  # noqa: PLR0911
737        """
738        Browse this provider's items.
739
740        :param path: The path to browse, (e.g. provider_id://artists).
741        """
742        if ProviderFeature.BROWSE not in self.supported_features:
743            # we may NOT use the default implementation if the provider does not support browse
744            raise NotImplementedError
745
746        path_parts = path.split("://")[1].split("/")
747        subpath = path_parts[0] if len(path_parts) > 0 else None
748        sub_subpath = path_parts[1] if len(path_parts) > 1 else None
749        # this reference implementation can be overridden with a provider specific approach
750        if subpath == "artists":
751            if artists := await self.mass.music.artists.library_items(
752                provider=self.instance_id,
753                summary=False,
754            ):
755                return artists
756            # library items not (yet) synced, fallback to direct retrieval
757            return [x async for x in self.get_library_artists()]
758        if subpath == "albums":
759            if albums := await self.mass.music.albums.library_items(
760                provider=self.instance_id,
761                summary=False,
762            ):
763                return albums
764            # library items not (yet) synced, fallback to direct retrieval
765            return [x async for x in self.get_library_albums()]
766        if subpath == "tracks":
767            if tracks := await self.mass.music.tracks.library_items(
768                provider=self.instance_id,
769                summary=False,
770            ):
771                return tracks
772            # library items not (yet) synced, fallback to direct retrieval
773            return [x async for x in self.get_library_tracks()]
774        if subpath == "radios":
775            if radios := await self.mass.music.radio.library_items(
776                provider=self.instance_id,
777                summary=False,
778            ):
779                return radios
780            # library items not (yet) synced, fallback to direct retrieval
781            return [x async for x in self.get_library_radios()]
782        if subpath == "playlists":
783            if playlists := await self.mass.music.playlists.library_items(
784                provider=self.instance_id,
785                summary=False,
786            ):
787                return playlists
788            # library items not (yet) synced, fallback to direct retrieval
789            return [x async for x in self.get_library_playlists()]
790        if subpath == "audiobooks":
791            if audiobooks := await self.mass.music.audiobooks.library_items(
792                provider=self.instance_id,
793                summary=False,
794            ):
795                return audiobooks
796            # library items not (yet) synced, fallback to direct retrieval
797            return [x async for x in self.get_library_audiobooks()]
798        if subpath == "podcasts":
799            if podcasts := await self.mass.music.podcasts.library_items(
800                provider=self.instance_id,
801                summary=False,
802            ):
803                return podcasts
804            # library items not (yet) synced, fallback to direct retrieval
805            return [x async for x in self.get_library_podcasts()]
806        if subpath == "sound_effects":
807            # sound effects are not library-backed, always retrieve them live
808            return [x async for x in self.get_sound_effects()]
809        if subpath == "recommendations" and sub_subpath:
810            # recommendations contents listing
811            return await self.get_recommendation_items(sub_subpath)
812        if subpath == "recommendations":
813            # Main recommendations listing
814            result: list[BrowseFolder] = []
815            recommendations = await self.get_recommendations()
816            for rec in recommendations:
817                result.append(
818                    BrowseFolder(
819                        item_id=rec.item_id,
820                        provider=self.instance_id,
821                        name=rec.name,
822                        is_playable=rec.is_playable,
823                        image=rec.image,
824                        path=f"{path}/{rec.item_id}",
825                    )
826                )
827            return result
828
829        if subpath:
830            # unknown path
831            msg = "Invalid subpath"
832            raise KeyError(msg)
833
834        # no subpath: return main listing
835        folders: list[BrowseFolder] = []
836        if ProviderFeature.LIBRARY_ARTISTS in self.supported_features:
837            folders.append(
838                BrowseFolder(
839                    item_id="artists",
840                    provider=self.instance_id,
841                    path=path + "artists",
842                    name="",
843                    translation_key="artists",
844                    is_playable=True,
845                )
846            )
847        if ProviderFeature.LIBRARY_ALBUMS in self.supported_features:
848            folders.append(
849                BrowseFolder(
850                    item_id="albums",
851                    provider=self.instance_id,
852                    path=path + "albums",
853                    name="",
854                    translation_key="albums",
855                    is_playable=True,
856                )
857            )
858        if ProviderFeature.LIBRARY_TRACKS in self.supported_features:
859            folders.append(
860                BrowseFolder(
861                    item_id="tracks",
862                    provider=self.domain,
863                    path=path + "tracks",
864                    name="",
865                    translation_key="tracks",
866                    is_playable=True,
867                )
868            )
869        if ProviderFeature.LIBRARY_PLAYLISTS in self.supported_features:
870            folders.append(
871                BrowseFolder(
872                    item_id="playlists",
873                    provider=self.instance_id,
874                    path=path + "playlists",
875                    name="",
876                    translation_key="playlists",
877                    is_playable=True,
878                )
879            )
880        if ProviderFeature.LIBRARY_RADIOS in self.supported_features:
881            folders.append(
882                BrowseFolder(
883                    item_id="radios",
884                    provider=self.instance_id,
885                    path=path + "radios",
886                    name="",
887                    translation_key="radios",
888                )
889            )
890        if ProviderFeature.LIBRARY_AUDIOBOOKS in self.supported_features:
891            folders.append(
892                BrowseFolder(
893                    item_id="audiobooks",
894                    provider=self.instance_id,
895                    path=path + "audiobooks",
896                    name="",
897                    translation_key="audiobooks",
898                )
899            )
900        if ProviderFeature.LIBRARY_PODCASTS in self.supported_features:
901            folders.append(
902                BrowseFolder(
903                    item_id="podcasts",
904                    provider=self.instance_id,
905                    path=path + "podcasts",
906                    name="",
907                    translation_key="podcasts",
908                )
909            )
910        if ProviderFeature.SOUND_EFFECTS in self.supported_features:
911            folders.append(
912                BrowseFolder(
913                    item_id="sound_effects",
914                    provider=self.instance_id,
915                    path=path + "sound_effects",
916                    name="",
917                    translation_key="sound_effects",
918                )
919            )
920        if ProviderFeature.RECOMMENDATIONS in self.supported_features:
921            folders.append(
922                BrowseFolder(
923                    item_id="recommendations",
924                    provider=self.instance_id,
925                    path=path + "recommendations",
926                    name="",
927                    translation_key="recommendations",
928                )
929            )
930        if len(folders) == 1:
931            # only one level, return the items directly
932            return await self.browse(folders[0].path)
933        return folders
934
935    async def get_recommendations(self) -> list[RecommendationFolder]:
936        """
937        Get this provider's available recommendation rows, without items.
938
939        Must be fast: return static or cached row descriptors only, without
940        live backend calls. The items for a row are fetched separately
941        through get_recommendation_items.
942
943        Will only be called if ProviderFeature.RECOMMENDATIONS is declared.
944        """
945        if ProviderFeature.RECOMMENDATIONS in self.supported_features:
946            raise NotImplementedError
947        return []
948
949    async def get_recommendation_items(
950        self, item_id: str
951    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
952        """
953        Get the items for a single recommendation row.
954
955        Live backend fetches belong here. Will only be called if
956        ProviderFeature.RECOMMENDATIONS is declared.
957
958        :param item_id: The item_id of the row, as returned by get_recommendations.
959        """
960        if ProviderFeature.RECOMMENDATIONS in self.supported_features:
961            raise NotImplementedError
962        return UniqueList()
963
964    async def sync_library(self, media_type: MediaType) -> None:
965        """Run library sync for this provider."""
966        token = SYNC_RUN_STATE.set(SyncRunState())
967        try:
968            await self._run_library_sync(media_type)
969        finally:
970            SYNC_RUN_STATE.reset(token)
971
972    async def _run_library_sync(self, media_type: MediaType) -> None:
973        """Sync the given media type into the library and process its deletions."""
974        # this reference implementation may be overridden
975        # with a provider specific approach if needed
976
977        if not self.mass.music.library_supported(self, media_type):
978            raise UnsupportedFeaturedException("Library sync not supported for this media type")
979
980        sync_state = sync_run_state()
981        if media_type == MediaType.ARTIST:
982            cur_db_ids = await self._sync_library_artists()
983        elif media_type == MediaType.ALBUM:
984            cur_db_ids = await self._sync_library_albums()
985        elif media_type == MediaType.TRACK:
986            cur_db_ids = await self._sync_library_tracks()
987        elif media_type == MediaType.PLAYLIST:
988            cur_db_ids = await self._sync_library_playlists()
989        elif media_type == MediaType.PODCAST:
990            cur_db_ids = await self._sync_library_podcasts()
991        elif media_type == MediaType.RADIO:
992            cur_db_ids = await self._sync_library_radios()
993        elif media_type == MediaType.AUDIOBOOK:
994            cur_db_ids = await self._sync_library_audiobooks()
995        else:
996            # this should not happen but catch it anyways
997            raise UnsupportedFeaturedException(f"Unexpected media type to sync: {media_type}")
998
999        # process deletions (= no longer in library)
1000        update_current_task_progress_text("Checking library deletions")
1001        controller = self.mass.music.get_controller(media_type)
1002        prev_library_items: list[int] | None
1003        if sync_state.incomplete:
1004            # a skipped item is missing from cur_db_ids just like a deleted one, but it is
1005            # still in the provider's library, so deleting it would throw away valid content
1006            if self.library_sync_deletions_enabled():
1007                summary = f"{sync_state.failures} item(s) could not be synced"
1008                self.logger.warning("Skipping deletions for %s: %s", self.name, summary)
1009                report_current_task_failure(f"Deletions skipped: {summary}")
1010            # merge this run's id's into the stored ones instead of replacing them: that
1011            # keeps both the deletions this run could not tell apart from its own failures
1012            # and the items it saw for the first time, so a later complete run finds either
1013            if prev_library_items := await self.mass.cache.get(
1014                key=media_type.value,
1015                provider=self.instance_id,
1016                category=CACHE_CATEGORY_PREV_LIBRARY_IDS,
1017            ):
1018                cur_db_ids.update(prev_library_items)
1019        elif self.library_sync_deletions_enabled():
1020            if prev_library_items := await self.mass.cache.get(
1021                key=media_type.value,
1022                provider=self.instance_id,
1023                category=CACHE_CATEGORY_PREV_LIBRARY_IDS,
1024            ):
1025                for db_id in prev_library_items:
1026                    if db_id not in cur_db_ids:
1027                        try:
1028                            library_item = await controller.get_library_item(db_id)
1029                        except MediaNotFoundError:
1030                            # edge case: the item is (already) removed from MA library as well
1031                            continue
1032                        # check if we have other provider-mappings (marked as in-library)
1033                        remaining_providers_in_library = {
1034                            x.provider_instance
1035                            for x in library_item.provider_mappings
1036                            if x.provider_instance != self.instance_id and x.in_library
1037                        }
1038                        if not remaining_providers_in_library and not self.is_streaming_provider:
1039                            # for non-streaming providers (local files, library-middlemen
1040                            # like subsonic/jellyfin/plex) an item removed from the provider
1041                            # is actually gone; fully remove it to avoid dangling records
1042                            # that stay visible in artist/album views where in_library is
1043                            # not filtered on
1044                            await controller.remove_item_from_library(db_id)
1045                        else:
1046                            if not remaining_providers_in_library and library_item.favorite:
1047                                # unmark as favorite since no providers have it in library
1048                                await controller.set_favorite(db_id, False)
1049                            # unmark this provider mapping as in_library = False
1050                            # we keep it in the library database so we can keep the metadata
1051                            for prov_map in library_item.provider_mappings:
1052                                if prov_map.provider_instance == self.instance_id:
1053                                    prov_map.in_library = False
1054                            await controller.set_provider_mappings(
1055                                db_id, library_item.provider_mappings
1056                            )
1057                        await asyncio.sleep(0)  # yield to eventloop
1058        # store current list of id's in cache so we can track changes
1059        await self.mass.cache.set(
1060            key=media_type.value,
1061            data=list(cur_db_ids),
1062            provider=self.instance_id,
1063            category=CACHE_CATEGORY_PREV_LIBRARY_IDS,
1064        )
1065        update_current_task_progress_text("Finalizing library sync")
1066
1067    def _update_sync_task_item_status(
1068        self, media_type: MediaType, processed_items: int, item_name: str | None = None
1069    ) -> None:
1070        """Update task text for the item currently being synced."""
1071        message = f"Processed {processed_items} {media_type.value}s"
1072        if item_name:
1073            message = f"{message}: {item_name}"
1074        update_current_task_progress_text(message)
1075
1076    def _handle_sync_item_failure(
1077        self, media_type: MediaType, item_ref: str | None, err: Exception
1078    ) -> None:
1079        """
1080        Log a non-fatal sync failure and record it on the active background task.
1081
1082        :raises Exception: If the provider declared this error unskippable, so that its own
1083            error handling can act on it instead of the item being skipped.
1084        """
1085        if isinstance(err, self.unskippable_sync_errors):
1086            raise err
1087        state = sync_run_state()
1088        state.failures += 1
1089        error_detail = describe_sync_error(err)
1090        if state.failures <= MAX_LOGGED_SYNC_FAILURES:
1091            if isinstance(err, MusicAssistantError):
1092                self.logger.warning(
1093                    "Skipping sync of %s %s - error details: %s",
1094                    media_type.value,
1095                    item_ref,
1096                    error_detail,
1097                )
1098            else:
1099                # not one of our own errors: usually a provider choking on its own api
1100                # payload, but the per-item library writes raise the same way, so log the
1101                # traceback (on debug) to make the actual origin traceable
1102                self.logger.error(
1103                    "Skipping sync of %s %s - unexpected error: %s",
1104                    media_type.value,
1105                    item_ref,
1106                    error_detail,
1107                    exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
1108                )
1109        report_current_task_failure(
1110            f"Failed to sync {media_type.value} {item_ref or '<unknown>'}: {error_detail}"
1111        )
1112
1113    async def _sync_item_genres(
1114        self,
1115        media_type: MediaType,
1116        provider_item_id: str,
1117        library_item_id: int,
1118        fallback_genres: set[str] | None = None,
1119    ) -> None:
1120        try:
1121            genre_names = await self.get_item_genre_names(media_type, provider_item_id)
1122        except NotImplementedError:
1123            if fallback_genres is None:
1124                return
1125            genre_names = fallback_genres
1126
1127        await self.mass.music.genres.sync_media_item_genres(
1128            media_type, library_item_id, set(genre_names)
1129        )
1130
1131    async def _sync_library_artists(self) -> set[int]:
1132        """Sync Library Artists to Music Assistant library."""
1133        self.logger.debug("Start sync of Artists to Music Assistant library.")
1134        cur_db_ids: set[int] = set()
1135        item_count = 0
1136        async for prov_item in self.get_library_artists():
1137            item_count += 1
1138            self._update_sync_task_item_status(MediaType.ARTIST, item_count, prov_item.name)
1139            try:
1140                sync_details = await self.mass.music.artists.get_library_item_sync_details(
1141                    prov_item.provider_mappings,
1142                )
1143                # batch all writes for this item into a single commit
1144                async with self.mass.music.database.deferred_commit():
1145                    if not sync_details:
1146                        # add item to the library
1147                        for prov_map in prov_item.provider_mappings:
1148                            prov_map.in_library = True
1149                        library_item = await self.mass.music.artists.add_item_to_library(prov_item)
1150                        db_id = int(library_item.item_id)
1151                        favorite = library_item.favorite
1152                    elif self._library_item_needs_update(sync_details, prov_item):
1153                        library_item = await self.mass.music.artists.update_item_in_library(
1154                            sync_details.item_id, prov_item
1155                        )
1156                        db_id = int(library_item.item_id)
1157                        favorite = library_item.favorite
1158                    else:
1159                        db_id = sync_details.item_id
1160                        favorite = sync_details.favorite
1161                    cur_db_ids.add(db_id)
1162                    if not favorite and prov_item.favorite:
1163                        # existing library item not favorite but should be
1164                        await self.mass.music.artists.set_favorite(db_id, True)
1165                    fallback_genres = (
1166                        set(prov_item.metadata.genres)
1167                        if prov_item.metadata and prov_item.metadata.genres
1168                        else None
1169                    )
1170                    await self._sync_item_genres(
1171                        MediaType.ARTIST,
1172                        prov_item.item_id,
1173                        db_id,
1174                        fallback_genres,
1175                    )
1176                await asyncio.sleep(0)  # yield to eventloop
1177            except Exception as err:
1178                sync_run_state().incomplete = True
1179                self._handle_sync_item_failure(MediaType.ARTIST, prov_item.uri, err)
1180        return cur_db_ids
1181
1182    def library_sync_album_tracks_enabled(self) -> bool:
1183        """Return whether all tracks of an album should be imported into the library."""
1184        return bool(
1185            self.config.get_value(
1186                CONF_ENTRY_LIBRARY_SYNC_ALBUM_TRACKS.key,
1187                CONF_ENTRY_LIBRARY_SYNC_ALBUM_TRACKS.default_value,
1188            )
1189        )
1190
1191    async def _sync_library_albums(self) -> set[int]:
1192        """Sync Library Albums to Music Assistant library."""
1193        self.logger.debug("Start sync of Albums to Music Assistant library.")
1194        cur_db_ids: set[int] = set()
1195        sync_album_tracks = self.library_sync_album_tracks_enabled()
1196        item_count = 0
1197        async for prov_item in self.get_library_albums():
1198            item_count += 1
1199            self._update_sync_task_item_status(MediaType.ALBUM, item_count, prov_item.name)
1200            try:
1201                sync_details = await self.mass.music.albums.get_library_item_sync_details(
1202                    prov_item.provider_mappings,
1203                )
1204                # batch all writes for this item into a single commit
1205                async with self.mass.music.database.deferred_commit():
1206                    if not sync_details:
1207                        # add item to the library
1208                        for prov_map in prov_item.provider_mappings:
1209                            prov_map.in_library = True
1210                        library_item = await self.mass.music.albums.add_item_to_library(prov_item)
1211                        db_id = int(library_item.item_id)
1212                        favorite = library_item.favorite
1213                    elif self._library_item_needs_update(sync_details, prov_item):
1214                        library_item = await self.mass.music.albums.update_item_in_library(
1215                            sync_details.item_id, prov_item
1216                        )
1217                        db_id = int(library_item.item_id)
1218                        favorite = library_item.favorite
1219                    else:
1220                        db_id = sync_details.item_id
1221                        favorite = sync_details.favorite
1222                    cur_db_ids.add(db_id)
1223                    if not favorite and prov_item.favorite:
1224                        # existing library item not favorite but should be
1225                        await self.mass.music.albums.set_favorite(db_id, True)
1226                    fallback_genres = (
1227                        set(prov_item.metadata.genres)
1228                        if prov_item.metadata and prov_item.metadata.genres
1229                        else None
1230                    )
1231                    await self._sync_item_genres(
1232                        MediaType.ALBUM,
1233                        prov_item.item_id,
1234                        db_id,
1235                        fallback_genres,
1236                    )
1237                await asyncio.sleep(0)  # yield to eventloop
1238            except Exception as err:
1239                sync_run_state().incomplete = True
1240                self._handle_sync_item_failure(MediaType.ALBUM, prov_item.uri, err)
1241                continue
1242            # optionally add album tracks to library. the album is already collected here,
1243            # so failing to import its tracks does not make the album result set incomplete
1244            if sync_album_tracks:
1245                try:
1246                    await self.import_album_tracks(prov_item.item_id, prov_item.name)
1247                except Exception as err:
1248                    self._handle_sync_item_failure(MediaType.ALBUM, prov_item.uri, err)
1249        return cur_db_ids
1250
1251    async def import_album_tracks(self, prov_album_id: str, album_name: str | None = None) -> None:
1252        """
1253        Import all tracks of the given (provider) album into the Music Assistant library.
1254
1255        :param prov_album_id: The provider item id of the album.
1256        :param album_name: Optional album name, used for logging/progress only.
1257        """
1258        self.logger.debug(
1259            "Importing Album Tracks into the Music Assistant library for album %s.",
1260            album_name or prov_album_id,
1261        )
1262        for item_count, prov_track in enumerate(
1263            await self.get_album_tracks(prov_album_id), start=1
1264        ):
1265            self._update_sync_task_item_status(MediaType.TRACK, item_count, prov_track.name)
1266            try:
1267                sync_details = await self.mass.music.tracks.get_library_item_sync_details(
1268                    prov_track.provider_mappings,
1269                )
1270                # batch all writes for this item into a single commit
1271                async with self.mass.music.database.deferred_commit():
1272                    if not sync_details:
1273                        # add item to the library
1274                        for prov_map in prov_track.provider_mappings:
1275                            prov_map.in_library = True
1276                        library_track = await self.mass.music.tracks.add_item_to_library(prov_track)
1277                        db_id = int(library_track.item_id)
1278                    elif not self._check_provider_mappings(sync_details, prov_track, True):
1279                        # existing library track but provider mapping doesn't match
1280                        library_track = await self.mass.music.tracks.update_item_in_library(
1281                            sync_details.item_id, prov_track
1282                        )
1283                        db_id = int(library_track.item_id)
1284                    else:
1285                        db_id = sync_details.item_id
1286                    fallback_genres = (
1287                        set(prov_track.metadata.genres)
1288                        if prov_track.metadata and prov_track.metadata.genres
1289                        else None
1290                    )
1291                    await self._sync_item_genres(
1292                        MediaType.TRACK,
1293                        prov_track.item_id,
1294                        db_id,
1295                        fallback_genres,
1296                    )
1297                await asyncio.sleep(0)  # yield to eventloop
1298            except Exception as err:
1299                self._handle_sync_item_failure(MediaType.TRACK, prov_track.uri, err)
1300
1301    def _validate_audiobook_author_narrator_types(self, prov_item: Audiobook) -> None:
1302        """
1303        Validate of correct artist and artist types.
1304
1305        If a provider supports artists of type Author or Narrator, they have to be part of an audiobook instance.
1306        Otherwise only strings are allowed.
1307        """
1308        if ArtistType.AUTHOR in self.supported_artist_types and not all(
1309            (isinstance(author, Artist) and author.artist_type == ArtistType.AUTHOR)
1310            for author in prov_item.authors
1311        ):
1312            raise InvalidDataError(
1313                f"Provider {self.name} supports ArtistType.AUTHOR, but"
1314                f" item {prov_item.name} does not exclusively provide Artist instances "
1315                "with ArtistType.AUTHOR set."
1316            )
1317        if ArtistType.NARRATOR in self.supported_artist_types and not all(
1318            (isinstance(narrator, Artist) and narrator.artist_type == ArtistType.NARRATOR)
1319            for narrator in prov_item.narrators
1320        ):
1321            raise InvalidDataError(
1322                f"Provider {self.name} supports ArtistType.NARRATOR, but"
1323                f" item {prov_item.name} does not exclusively provide Artist instances "
1324                "with ArtistType.NARRATOR set."
1325            )
1326        if ArtistType.AUTHOR not in self.supported_artist_types and not all(
1327            isinstance(author, str) for author in prov_item.authors
1328        ):
1329            raise InvalidDataError(
1330                f"Provider {self.name} does not support artists of type author, but"
1331                f" item {prov_item.name} does not exclusively provide strings."
1332            )
1333        if ArtistType.NARRATOR not in self.supported_artist_types and not all(
1334            isinstance(narrator, str) for narrator in prov_item.narrators
1335        ):
1336            raise InvalidDataError(
1337                f"Provider {self.name} does not support artists of type narrator, but"
1338                f" item {prov_item.name} does not exclusively provide strings."
1339            )
1340
1341    async def _sync_library_audiobooks(self) -> set[int]:
1342        """Sync Library Audiobooks to Music Assistant library."""
1343        self.logger.debug("Start sync of Audiobooks to Music Assistant library.")
1344        cur_db_ids: set[int] = set()
1345        item_count = 0
1346        async for prov_item in self.get_library_audiobooks():
1347            item_count += 1
1348            self._update_sync_task_item_status(MediaType.AUDIOBOOK, item_count, prov_item.name)
1349            try:
1350                sync_details = cast(
1351                    "AudiobookSyncDetails | None",
1352                    await self.mass.music.audiobooks.get_library_item_sync_details(
1353                        prov_item.provider_mappings,
1354                    ),
1355                )
1356                self._validate_audiobook_author_narrator_types(prov_item)
1357                # batch all writes for this item into a single commit
1358                async with self.mass.music.database.deferred_commit():
1359                    if not sync_details:
1360                        # add item to the library
1361                        for prov_map in prov_item.provider_mappings:
1362                            prov_map.in_library = True
1363                        library_item = await self.mass.music.audiobooks.add_item_to_library(
1364                            prov_item
1365                        )
1366                        db_id = int(library_item.item_id)
1367                        favorite = library_item.favorite
1368                        lib_fully_played = library_item.fully_played
1369                        lib_resume_position_ms = library_item.resume_position_ms
1370                    elif self._library_item_needs_update(sync_details, prov_item):
1371                        library_item = await self.mass.music.audiobooks.update_item_in_library(
1372                            sync_details.item_id, prov_item
1373                        )
1374                        db_id = int(library_item.item_id)
1375                        favorite = library_item.favorite
1376                        lib_fully_played = library_item.fully_played
1377                        lib_resume_position_ms = library_item.resume_position_ms
1378                    else:
1379                        # Detect, if stored authors/narrators are plain strings but the provider
1380                        # now supplies full Artist objects, i.e. artist support changed.
1381                        prov_author = prov_item.authors[0] if prov_item.authors else None
1382                        prov_narrator = prov_item.narrators[0] if prov_item.narrators else None
1383                        if (sync_details.author_is_str and not isinstance(prov_author, str)) or (
1384                            sync_details.narrator_is_str and not isinstance(prov_narrator, str)
1385                        ):
1386                            library_item = await self.mass.music.audiobooks.update_item_in_library(
1387                                sync_details.item_id, prov_item
1388                            )
1389                            db_id = int(library_item.item_id)
1390                            favorite = library_item.favorite
1391                            lib_fully_played = library_item.fully_played
1392                            lib_resume_position_ms = library_item.resume_position_ms
1393                        else:
1394                            db_id = sync_details.item_id
1395                            favorite = sync_details.favorite
1396                            lib_fully_played = sync_details.fully_played
1397                            lib_resume_position_ms = sync_details.resume_position_ms
1398
1399                    cur_db_ids.add(db_id)
1400                    if not favorite and prov_item.favorite:
1401                        # existing library item not favorite but should be
1402                        await self.mass.music.audiobooks.set_favorite(db_id, True)
1403                    # check if resume_position_ms or fully_played changed
1404                    if (
1405                        prov_item.resume_position_ms is not None
1406                        and prov_item.fully_played is not None
1407                        and (
1408                            lib_resume_position_ms != prov_item.resume_position_ms
1409                            or lib_fully_played != prov_item.fully_played
1410                        )
1411                    ):
1412                        await self.mass.music.audiobooks.update_item_in_library(db_id, prov_item)
1413
1414                    fallback_genres = (
1415                        set(prov_item.metadata.genres)
1416                        if prov_item.metadata and prov_item.metadata.genres
1417                        else None
1418                    )
1419                    await self._sync_item_genres(
1420                        MediaType.AUDIOBOOK,
1421                        prov_item.item_id,
1422                        db_id,
1423                        fallback_genres,
1424                    )
1425
1426                await asyncio.sleep(0)  # yield to eventloop
1427            except Exception as err:
1428                sync_run_state().incomplete = True
1429                self._handle_sync_item_failure(MediaType.AUDIOBOOK, prov_item.uri, err)
1430        return cur_db_ids
1431
1432    async def _sync_library_playlists(self) -> set[int]:
1433        """Sync Library Playlists to Music Assistant library."""
1434        self.logger.debug("Start sync of Playlists to Music Assistant library.")
1435        conf_sync_playlist_tracks = self.config.get_value(
1436            CONF_ENTRY_LIBRARY_SYNC_PLAYLIST_TRACKS.key,
1437            CONF_ENTRY_LIBRARY_SYNC_PLAYLIST_TRACKS.default_value,
1438        )
1439        conf_sync_playlist_tracks = cast("list[str]", conf_sync_playlist_tracks)
1440        cur_db_ids: set[int] = set()
1441        item_count = 0
1442        async for prov_item in self.get_library_playlists():
1443            item_count += 1
1444            self._update_sync_task_item_status(MediaType.PLAYLIST, item_count, prov_item.name)
1445            try:
1446                library_item = await self.mass.music.playlists.get_library_item_by_prov_mappings(
1447                    prov_item.provider_mappings,
1448                )
1449                # batch all writes for this item into a single commit
1450                async with self.mass.music.database.deferred_commit():
1451                    if not library_item:
1452                        # add item to the library
1453                        for prov_map in prov_item.provider_mappings:
1454                            prov_map.in_library = True
1455                        library_item = await self.mass.music.playlists.add_item_to_library(
1456                            prov_item
1457                        )
1458                    elif (
1459                        self._library_item_needs_update(library_item, prov_item)
1460                        # or the supported mediatypes changed
1461                        or prov_item.supported_mediatypes != library_item.supported_mediatypes
1462                    ):
1463                        library_item = await self.mass.music.playlists.update_item_in_library(
1464                            library_item.item_id, prov_item
1465                        )
1466                    elif (
1467                        prov_item.is_dynamic
1468                        and not library_item.is_editable
1469                        and (
1470                            prov_item.name != library_item.name
1471                            or prov_item.metadata.images != library_item.metadata.images
1472                        )
1473                    ):
1474                        # the provider is the sole source of truth for non-editable dynamic
1475                        # playlists (e.g. Pandora/personalized-radio stations): overwrite=True
1476                        # replaces the full stored record (not just name/images), which is fine
1477                        # here since there's no local customization on these to lose. Restricted
1478                        # to is_dynamic so static non-editable playlists (e.g. provider
1479                        # "favorites") keep their locally-enriched metadata/images.
1480                        library_item = await self.mass.music.playlists.update_item_in_library(
1481                            library_item.item_id, prov_item, overwrite=True
1482                        )
1483                    cur_db_ids.add(int(library_item.item_id))
1484                    if not library_item.favorite and prov_item.favorite:
1485                        # existing library item not favorite but should be
1486                        await self.mass.music.playlists.set_favorite(library_item.item_id, True)
1487                await asyncio.sleep(0)  # yield to eventloop
1488            except Exception as err:
1489                sync_run_state().incomplete = True
1490                self._handle_sync_item_failure(MediaType.PLAYLIST, prov_item.uri, err)
1491                continue
1492            # optionally sync playlist tracks. the playlist is already collected here, so
1493            # failing on its tracks does not make the playlist result set incomplete
1494            if (
1495                prov_item.name in conf_sync_playlist_tracks
1496                or prov_item.uri in conf_sync_playlist_tracks
1497            ):
1498                try:
1499                    await self._sync_playlist_tracks(prov_item)
1500                except Exception as err:
1501                    self._handle_sync_item_failure(MediaType.PLAYLIST, prov_item.uri, err)
1502        return cur_db_ids
1503
1504    async def _sync_playlist_tracks(self, provider_playlist: Playlist) -> None:
1505        """Sync Playlist Tracks to Music Assistant library."""
1506        self.logger.debug(
1507            "Start sync of Playlist Tracks to Music Assistant library for playlist %s.",
1508            provider_playlist.name,
1509        )
1510        item_count = 0
1511        async for _prov_track in self.iter_playlist_tracks(provider_playlist.item_id):
1512            prov_track: PlaylistPlayableItem | Podcast = _prov_track
1513            item_count += 1
1514            try:
1515                if isinstance(_prov_track, PodcastEpisode):
1516                    # In MA, only full podcasts can be synced to the library
1517                    prov_track = await self.get_podcast(_prov_track.podcast.item_id)
1518                self._update_sync_task_item_status(MediaType.TRACK, item_count, prov_track.name)
1519                controller = self.mass.music.get_controller(prov_track.media_type)
1520                sync_details = await controller.get_library_item_sync_details(
1521                    prov_track.provider_mappings,
1522                )
1523                # batch all writes for this item into a single commit
1524                async with self.mass.music.database.deferred_commit():
1525                    if not sync_details:
1526                        # add item to the library
1527                        for prov_map in prov_track.provider_mappings:
1528                            prov_map.in_library = True
1529                        library_track = await controller.add_item_to_library(prov_track)  # type: ignore[arg-type]
1530                        db_id = int(library_track.item_id)
1531                    elif not self._check_provider_mappings(sync_details, prov_track, True):
1532                        # existing library track but provider mapping doesn't match
1533                        library_track = await controller.update_item_in_library(
1534                            sync_details.item_id,
1535                            prov_track,  # type: ignore[arg-type]
1536                        )
1537                        db_id = int(library_track.item_id)
1538                    else:
1539                        db_id = sync_details.item_id
1540                    fallback_genres = (
1541                        set(prov_track.metadata.genres)
1542                        if prov_track.metadata and prov_track.metadata.genres
1543                        else None
1544                    )
1545                    await self._sync_item_genres(
1546                        MediaType.TRACK,
1547                        prov_track.item_id,
1548                        db_id,
1549                        fallback_genres,
1550                    )
1551                await asyncio.sleep(0)  # yield to eventloop
1552            except Exception as err:
1553                self._handle_sync_item_failure(MediaType.TRACK, prov_track.uri, err)
1554
1555    async def _sync_library_tracks(self) -> set[int]:
1556        """Sync Library Tracks to Music Assistant library."""
1557        self.logger.debug("Start sync of Tracks to Music Assistant library.")
1558        cur_db_ids: set[int] = set()
1559        item_count = 0
1560        async for prov_item in self.get_library_tracks():
1561            item_count += 1
1562            self._update_sync_task_item_status(MediaType.TRACK, item_count, prov_item.name)
1563            try:
1564                sync_details = cast(
1565                    "TrackSyncDetails | None",
1566                    await self.mass.music.tracks.get_library_item_sync_details(
1567                        prov_item.provider_mappings,
1568                    ),
1569                )
1570                if not sync_details and not prov_item.available:
1571                    # skip unavailable tracks
1572                    # TODO: do we want to search for substitutes at this point ?
1573                    self.logger.debug(
1574                        "Skipping sync of track %s because it is unavailable",
1575                        prov_item.uri,
1576                    )
1577                    continue
1578                # batch all writes for this item into a single commit
1579                async with self.mass.music.database.deferred_commit():
1580                    if not sync_details:
1581                        # add item to the library
1582                        for prov_map in prov_item.provider_mappings:
1583                            prov_map.in_library = True
1584                        library_item = await self.mass.music.tracks.add_item_to_library(prov_item)
1585                        db_id = int(library_item.item_id)
1586                        favorite = library_item.favorite
1587                    elif (
1588                        self._library_item_needs_update(sync_details, prov_item)
1589                        # or backfill a missing album(_tracks) link for existing tracks
1590                        or (prov_item.album and not sync_details.has_album)
1591                        # or backfill missing track_artists link(s) for existing tracks
1592                        or (prov_item.artists and not sync_details.has_artists)
1593                    ):
1594                        library_item = await self.mass.music.tracks.update_item_in_library(
1595                            sync_details.item_id, prov_item
1596                        )
1597                        db_id = int(library_item.item_id)
1598                        favorite = library_item.favorite
1599                    else:
1600                        db_id = sync_details.item_id
1601                        favorite = sync_details.favorite
1602                    cur_db_ids.add(db_id)
1603                    if not favorite and prov_item.favorite:
1604                        # existing library item not favorite but should be
1605                        await self.mass.music.tracks.set_favorite(db_id, True)
1606                    fallback_genres = (
1607                        set(prov_item.metadata.genres)
1608                        if prov_item.metadata and prov_item.metadata.genres
1609                        else None
1610                    )
1611                    await self._sync_item_genres(
1612                        MediaType.TRACK,
1613                        prov_item.item_id,
1614                        db_id,
1615                        fallback_genres,
1616                    )
1617                await asyncio.sleep(0)  # yield to eventloop
1618            except Exception as err:
1619                sync_run_state().incomplete = True
1620                self._handle_sync_item_failure(MediaType.TRACK, prov_item.uri, err)
1621        return cur_db_ids
1622
1623    async def _sync_library_podcasts(self) -> set[int]:
1624        """Sync Library Podcasts to Music Assistant library."""
1625        self.logger.debug("Start sync of Podcasts to Music Assistant library.")
1626        cur_db_ids: set[int] = set()
1627        item_count = 0
1628        async for prov_item in self.get_library_podcasts():
1629            item_count += 1
1630            self._update_sync_task_item_status(MediaType.PODCAST, item_count, prov_item.name)
1631            try:
1632                sync_details = await self.mass.music.podcasts.get_library_item_sync_details(
1633                    prov_item.provider_mappings,
1634                )
1635                # batch all writes for this item into a single commit
1636                async with self.mass.music.database.deferred_commit():
1637                    if not sync_details:
1638                        # add item to the library
1639                        for prov_map in prov_item.provider_mappings:
1640                            prov_map.in_library = True
1641                        library_item = await self.mass.music.podcasts.add_item_to_library(prov_item)
1642                        db_id = int(library_item.item_id)
1643                        favorite = library_item.favorite
1644                    elif self._library_item_needs_update(sync_details, prov_item):
1645                        library_item = await self.mass.music.podcasts.update_item_in_library(
1646                            sync_details.item_id, prov_item
1647                        )
1648                        db_id = int(library_item.item_id)
1649                        favorite = library_item.favorite
1650                    else:
1651                        db_id = sync_details.item_id
1652                        favorite = sync_details.favorite
1653                    cur_db_ids.add(db_id)
1654                    if not favorite and prov_item.favorite:
1655                        # existing library item not favorite but should be
1656                        await self.mass.music.podcasts.set_favorite(db_id, True)
1657                    fallback_genres = (
1658                        set(prov_item.metadata.genres)
1659                        if prov_item.metadata and prov_item.metadata.genres
1660                        else None
1661                    )
1662                    await self._sync_item_genres(
1663                        MediaType.PODCAST,
1664                        prov_item.item_id,
1665                        db_id,
1666                        fallback_genres,
1667                    )
1668                await asyncio.sleep(0)  # yield to eventloop
1669            except Exception as err:
1670                sync_run_state().incomplete = True
1671                self._handle_sync_item_failure(MediaType.PODCAST, prov_item.uri, err)
1672                continue
1673            # the podcast is already collected here, so a feed that fails to deliver its
1674            # episodes does not make the podcast result set incomplete
1675            try:
1676                # precache podcast episodes
1677                async for _ in self.mass.music.podcasts.episodes(str(db_id), "library"):
1678                    await asyncio.sleep(0)  # yield to eventloop
1679            except Exception as err:
1680                self._handle_sync_item_failure(MediaType.PODCAST, prov_item.uri, err)
1681        return cur_db_ids
1682
1683    async def _sync_library_radios(self) -> set[int]:
1684        """Sync Library Radios to Music Assistant library."""
1685        self.logger.debug("Start sync of Radios to Music Assistant library.")
1686        cur_db_ids: set[int] = set()
1687        item_count = 0
1688        async for prov_item in self.get_library_radios():
1689            item_count += 1
1690            self._update_sync_task_item_status(MediaType.RADIO, item_count, prov_item.name)
1691            try:
1692                library_item = await self.mass.music.radio.get_library_item_by_prov_mappings(
1693                    prov_item.provider_mappings,
1694                )
1695                # batch all writes for this item into a single commit
1696                async with self.mass.music.database.deferred_commit():
1697                    if not library_item:
1698                        # add item to the library
1699                        for prov_map in prov_item.provider_mappings:
1700                            prov_map.in_library = True
1701                        library_item = await self.mass.music.radio.add_item_to_library(prov_item)
1702                    elif prov_item.is_dynamic and (
1703                        not library_item.is_dynamic
1704                        or prov_item.name != library_item.name
1705                        or prov_item.metadata.images != library_item.metadata.images
1706                    ):
1707                        # must overwrite: merging keeps mappings that serve the wrong tracks
1708                        for prov_map in prov_item.provider_mappings:
1709                            prov_map.in_library = True  # overwrite re-inserts the rows
1710                        library_item = await self.mass.music.radio.update_item_in_library(
1711                            library_item.item_id, prov_item, overwrite=True
1712                        )
1713                    elif self._library_item_needs_update(library_item, prov_item) or (
1714                        library_item.is_dynamic and not prov_item.is_dynamic
1715                    ):
1716                        # a station leaving dynamic mode is no longer provider-owned, so merge
1717                        library_item = await self.mass.music.radio.update_item_in_library(
1718                            library_item.item_id, prov_item
1719                        )
1720                    cur_db_ids.add(int(library_item.item_id))
1721                    if not library_item.favorite and prov_item.favorite:
1722                        # existing library item not favorite but should be
1723                        await self.mass.music.radio.set_favorite(library_item.item_id, True)
1724                await asyncio.sleep(0)  # yield to eventloop
1725
1726            except Exception as err:
1727                sync_run_state().incomplete = True
1728                self._handle_sync_item_failure(MediaType.RADIO, prov_item.uri, err)
1729        return cur_db_ids
1730
1731    # DO NOT OVERRIDE BELOW
1732
1733    def get_default_library_sync_schedule(self, media_type: MediaType) -> TaskSchedule:
1734        """Return the default recurring schedule for library sync tasks of this provider."""
1735        if not self.mass.music.library_supported(self, media_type):
1736            raise UnsupportedFeaturedException(
1737                f"Library sync is not supported for {media_type} on {self.instance_id}"
1738            )
1739        return TaskSchedule.hourly(every=12)
1740
1741    def library_sync_deletions_enabled(self) -> bool:
1742        """Return if Library sync deletions is enabled for this provider."""
1743        conf_value = self.config.get_value(
1744            CONF_ENTRY_LIBRARY_SYNC_DELETIONS.key, CONF_ENTRY_LIBRARY_SYNC_DELETIONS.default_value
1745        )
1746        return bool(conf_value)
1747
1748    async def iter_playlist_tracks(
1749        self,
1750        prov_playlist_id: str,
1751    ) -> AsyncGenerator[PlaylistPlayableItem]:
1752        """Iterate playlist tracks for the given provider playlist id."""
1753        page = 0
1754        while True:
1755            tracks = await self.get_playlist_tracks(
1756                prov_playlist_id,
1757                page=page,
1758            )
1759            if not tracks:
1760                break
1761            for track in tracks:
1762                yield track
1763            page += 1
1764
1765    def _get_library_gen(self, media_type: MediaType) -> AsyncGenerator[MediaItemType]:
1766        """Return library generator for given media_type."""
1767        if media_type == MediaType.ARTIST:
1768            return self.get_library_artists()
1769        if media_type == MediaType.ALBUM:
1770            return self.get_library_albums()
1771        if media_type == MediaType.TRACK:
1772            return self.get_library_tracks()
1773        if media_type == MediaType.PLAYLIST:
1774            return self.get_library_playlists()
1775        if media_type == MediaType.RADIO:
1776            return self.get_library_radios()
1777        if media_type == MediaType.AUDIOBOOK:
1778            return self.get_library_audiobooks()
1779        if media_type == MediaType.PODCAST:
1780            return self.get_library_podcasts()
1781        raise NotImplementedError
1782
1783    def _library_item_needs_update(
1784        self, library_item: MediaItemType | LibraryItemSyncDetails, prov_item: MediaItemType
1785    ) -> bool:
1786        """Return True if the library item needs an update from the given provider item."""
1787        if not self._check_provider_mappings(library_item, prov_item, True):
1788            # provider mapping doesn't match the library item
1789            return True
1790        # the item's date_added changed on the provider
1791        return bool(prov_item.date_added and library_item.date_added != prov_item.date_added)
1792
1793    def _check_provider_mappings(
1794        self,
1795        library_item: MediaItemType | LibraryItemSyncDetails,
1796        provider_item: MediaItemType,
1797        in_library: bool,
1798    ) -> bool:
1799        """Check if provider mapping(s) are consistent between library and provider items."""
1800        for provider_mapping in provider_item.provider_mappings:
1801            if provider_mapping.item_id != provider_item.item_id:
1802                # this should never happen, but guard against it
1803                raise MusicAssistantError("Inconsistent provider mapping item_id found")
1804            if provider_mapping.provider_instance != self.instance_id:
1805                # this should never happen, but guard against it
1806                raise MusicAssistantError("Inconsistent provider mapping instance_id found")
1807            # check if the provider mapping matches the library item
1808            provider_mapping.in_library = in_library
1809            library_mapping = next(
1810                (
1811                    x
1812                    for x in library_item.provider_mappings
1813                    if x.provider_instance == provider_mapping.provider_instance
1814                    and x.item_id == provider_mapping.item_id
1815                ),
1816                None,
1817            )
1818            if not library_mapping:
1819                return False
1820            if provider_mapping.in_library != library_mapping.in_library:
1821                # in-library status doesn't match
1822                return False
1823            if provider_mapping.is_unique != library_mapping.is_unique:
1824                # unique status doesn't match
1825                return False
1826            # check if the library item has all provider instances mappings
1827            is_unique = provider_mapping.is_unique or (not self.is_streaming_provider)
1828            if not is_unique:
1829                # for streaming providers we need to make sure all provider instances
1830                # for this domain are represented in the provider mappings
1831                prov_instances = self.mass.music.get_provider_instances(
1832                    domain=provider_mapping.provider_domain,
1833                    return_unavailable=True,
1834                )
1835                if len(prov_instances) > 1:
1836                    # multiple provider instances for this domain exist
1837                    # make sure the library item has all provider mappings
1838                    for prov_instance in prov_instances:
1839                        if not any(
1840                            x.provider_instance == prov_instance.instance_id
1841                            and x.item_id == provider_mapping.item_id
1842                            for x in library_item.provider_mappings
1843                        ):
1844                            # missing provider mapping for another instance
1845                            # the rest of the core logic will take care of adding it
1846                            # just return False here to trigger that logic
1847                            return False
1848
1849            # final check: availability
1850            return provider_mapping.available == library_mapping.available
1851        return False
1852