music-assistant-server

1"""MusicController: Orchestrates all data from music providers and sync to internal database."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from collections.abc import Awaitable, Callable, Coroutine, Iterable, Sequence
8from contextlib import suppress
9from copy import deepcopy
10from datetime import datetime
11from itertools import zip_longest
12from typing import TYPE_CHECKING, Any, NamedTuple, cast
13
14from music_assistant_models.auth import Scope
15from music_assistant_models.background_task import BackgroundTask, TaskMetadata, TaskSchedule
16from music_assistant_models.config_entries import (
17    ConfigActionResult,
18    ConfigEntry,
19    ConfigValueType,
20)
21from music_assistant_models.enums import (
22    ConfigEntryType,
23    EventType,
24    MediaType,
25    ProviderFeature,
26    ProviderType,
27    TaskStatus,
28)
29from music_assistant_models.errors import (
30    InvalidDataError,
31    InvalidProviderID,
32    InvalidProviderURI,
33    MediaNotFoundError,
34    MusicAssistantError,
35    UnsupportedFeaturedException,
36)
37from music_assistant_models.helpers import get_global_cache_value
38from music_assistant_models.media_items import (
39    Album,
40    Artist,
41    AudioFormat,
42    BrowseFolder,
43    Genre,
44    ItemMapping,
45    MediaItemType,
46    Playlist,
47    Podcast,
48    PodcastEpisode,
49    ProviderMapping,
50    SearchResults,
51    SoundEffect,
52    Track,
53)
54from music_assistant_models.media_items.media_item import MediaCollection
55
56from music_assistant.constants import (
57    CONF_ENTRY_LIBRARY_SYNC_BACK,
58    DB_TABLE_PLAYLOG,
59    DB_TABLE_PROVIDER_MAPPINGS,
60    PROVIDERS_WITH_SHAREABLE_URLS,
61)
62from music_assistant.controllers.music.constants import (
63    CACHE_CATEGORY_SEARCH_RESULTS,
64    CONF_DELETED_PROVIDERS,
65    CONF_RESET_DB,
66    DATABASE_CLEANUP_TASK_ID,
67    DB_SCHEMA_VERSION,
68    MUSIC_SYNC_COMPLETION_CHECK_TASK_ID,
69    PROVIDER_MAPPING_CORRECTION_TASK_ID,
70    SEARCH_CACHE_EXPIRATION_COMBINED,
71    SEARCH_CACHE_EXPIRATION_LOCAL_PROVIDER,
72    SEARCH_CACHE_EXPIRATION_STREAMING_PROVIDER,
73    SEARCH_PROVIDER_HARD_TIMEOUT,
74    SEARCH_PROVIDER_SOFT_TIMEOUT,
75)
76from music_assistant.controllers.music.database import (
77    PLAYLOG_CONFLICT_KEYS,
78    MusicDatabaseSetupMixin,
79)
80from music_assistant.controllers.music.helpers import filter_search_results, sort_search_result
81from music_assistant.controllers.music.media.albums import AlbumsController
82from music_assistant.controllers.music.media.artists import ArtistsController
83from music_assistant.controllers.music.media.audiobooks import AudiobooksController
84from music_assistant.controllers.music.media.base import SUPPRESS_MEDIA_ITEM_UPDATES
85from music_assistant.controllers.music.media.genres import GenreController
86from music_assistant.controllers.music.media.playlists import PlaylistController
87from music_assistant.controllers.music.media.podcasts import PodcastsController
88from music_assistant.controllers.music.media.radio import RadioController
89from music_assistant.controllers.music.media.tracks import TracksController
90from music_assistant.controllers.music.recency import RecencyEngine
91from music_assistant.controllers.music.recommendations.controller import (
92    RecommendationsController,
93)
94from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
95from music_assistant.helpers.api import api_command
96from music_assistant.helpers.collections import get_collection_item_media_type_from_item_id
97from music_assistant.helpers.compare import compare_strings, compare_version
98from music_assistant.helpers.database import UNSET, DatabaseConnection
99from music_assistant.helpers.datetime import (
100    from_utc_timestamp,
101    local_clock_time_to_utc,
102    utc_timestamp,
103)
104from music_assistant.helpers.json import json_loads, serialize_to_json
105from music_assistant.helpers.tags import split_artists
106from music_assistant.helpers.uri import parse_uri
107from music_assistant.helpers.util import parse_optional_bool, parse_title_and_version
108from music_assistant.models.core_controller import CoreController
109from music_assistant.models.music_provider import MusicProvider
110from music_assistant.models.plugin import PluginProvider
111
112if TYPE_CHECKING:
113    from music_assistant_models.auth import User
114    from music_assistant_models.config_entries import CoreConfig
115    from music_assistant_models.media_items import Audiobook
116
117    from music_assistant import MusicAssistant
118    from music_assistant.controllers.music.media.base import MediaControllerBase
119    from music_assistant.helpers.json import SerializableType
120    from music_assistant.models import ProviderInstanceType
121    from music_assistant.models.provider import Provider
122    from music_assistant.providers.builtin import BuiltinProvider
123
124
125class RecentPlayedTrack(NamedTuple):
126    """A recently played track from the playlog, with the artists recorded at play time."""
127
128    track: ItemMapping
129    artists: list[ItemMapping]
130
131
132class MusicController(MusicDatabaseSetupMixin, CoreController):
133    """Several helpers around the musicproviders."""
134
135    domain: str = "music"
136    config: CoreConfig
137
138    def __init__(self, mass: MusicAssistant) -> None:
139        """Initialize class."""
140        super().__init__(mass)
141        self.cache = self.mass.cache
142        self.artists = ArtistsController(self.mass)
143        self.albums = AlbumsController(self.mass)
144        self.tracks = TracksController(self.mass)
145        self.radio = RadioController(self.mass)
146        self.playlists = PlaylistController(self.mass)
147        self.audiobooks = AudiobooksController(self.mass)
148        self.podcasts = PodcastsController(self.mass)
149        self.genres = GenreController(self.mass)
150        self.recommendations = RecommendationsController(self.mass)
151        self.recency = RecencyEngine(self.mass)
152        self._database: DatabaseConnection | None = None
153        self._sync_lock = asyncio.Lock()
154        self.manifest.name = "Music controller"
155        self.manifest.description = (
156            "Music Assistant's core controller which manages all music from all providers."
157        )
158        self.manifest.icon = "archive-music"
159
160    @property
161    def database(self) -> DatabaseConnection:
162        """Return the database connection."""
163        if self._database is None:
164            raise RuntimeError("Database not initialized")
165        return self._database
166
167    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
168        """Return all Config Entries for this core module (if any)."""
169        return (
170            ConfigEntry(
171                key=CONF_RESET_DB,
172                type=ConfigEntryType.ACTION,
173                category="generic",
174                advanced=True,
175            ),
176        )
177
178    async def handle_config_action(
179        self, action: str
180    ) -> tuple[ConfigEntry, ...] | ConfigActionResult | None:
181        """Handle a one-shot action button press and report its outcome."""
182        if action == CONF_RESET_DB:
183            await self._reset_database()
184            await self.mass.cache.clear()
185            await self.start_sync()
186            return ConfigActionResult(translation_key=f"{CONF_RESET_DB}.result")
187        return await super().handle_config_action(action)
188
189    async def setup(self, config: CoreConfig) -> None:
190        """Async initialize of module."""
191        self.config = config
192        # setup library database
193        await self._setup_database()
194        # make sure to finish any removal jobs
195        for removed_provider in cast(
196            "list[str]",
197            self.mass.config.get_raw_core_config_value(self.domain, CONF_DELETED_PROVIDERS, []),
198        ):
199            await self.cleanup_provider(removed_provider)
200
201    async def post_setup(self) -> None:
202        """Handle logic after all core controllers have been set up."""
203        self._register_database_cleanup_task()
204        self._register_provider_mapping_correction_task()
205        self.genres.register_scheduled_scan_task()
206
207    async def close(self) -> None:
208        """Cleanup on exit."""
209        if self._database:
210            await self._database.close()
211
212    async def get_diagnostics(self) -> dict[str, SerializableType]:
213        """Return diagnostics info for this controller to include in diagnostics reports."""
214        return {
215            "db_schema_version": DB_SCHEMA_VERSION,
216            "sync_tasks_active": len(self.active_sync_tasks),
217        }
218
219    async def on_provider_loaded(self, provider: MusicProvider) -> None:
220        """Handle logic when a provider is loaded."""
221        await self.schedule_provider_sync(provider.instance_id)
222
223    async def on_provider_unload(self, provider: MusicProvider) -> None:
224        """
225        Handle logic when a provider is (about to get) unloaded.
226
227        Sync tasks are unscheduled by MusicAssistant.unload_provider itself, which also
228        decides whether their persisted state is kept (reload) or cleared (removal).
229        """
230
231    @property
232    def providers(self) -> list[MusicProvider]:
233        """
234        Return all loaded/running MusicProviders (instances).
235
236        Note that this applies user provider filters (for all user types).
237        """
238        return cast(
239            "list[MusicProvider]",
240            [
241                x
242                for x in self._apply_user_provider_filter(self.mass.providers)
243                if x.type == ProviderType.MUSIC
244            ],
245        )
246
247    @api_command("music/sync", required_scope=Scope.LIBRARY_MANAGE)
248    async def start_sync(
249        self,
250        media_types: list[MediaType] | None = None,
251        providers: list[str] | None = None,
252    ) -> list[BackgroundTask]:
253        """
254        Start running the sync of (all or selected) musicproviders.
255
256        media_types: only sync these media types. None for all.
257        providers: only sync these provider instances. None for all.
258        """
259        tasks: list[BackgroundTask] = []
260        if media_types is None:
261            media_types = MediaType.ALL
262        if providers is None:
263            providers = [x.instance_id for x in self.providers]
264
265        for media_type in media_types:
266            for provider in self.providers:
267                if provider.instance_id not in providers:
268                    continue
269                if not self.library_supported(provider, media_type):
270                    continue
271                # handle mediatype specific sync config
272                conf_key = f"library_sync_{media_type}s"
273                sync_conf: ConfigValueType = await self.mass.config.get_provider_config_value(
274                    provider.instance_id, conf_key
275                )
276                if not sync_conf:
277                    continue
278                await self._schedule_provider_mediatype_sync(provider, media_type, True)
279                task_id = self._get_sync_task_id(provider, media_type)
280                try:
281                    tasks.append(self.mass.tasks.run_task(task_id))
282                except InvalidDataError:
283                    tasks.append(
284                        self.mass.tasks.run_background_task(
285                            task_id=task_id,
286                            name=self._get_sync_task_name(provider, media_type),
287                            handler=self._create_provider_sync_handler(provider, media_type),
288                            translation_key=self._get_sync_task_translation_key(media_type),
289                            translation_args=[provider.name],
290                            translation_owner=self.translation_owner,
291                            user_id=(user.user_id if (user := get_current_user()) else None),
292                            metadata=self._get_sync_task_metadata(provider, media_type),
293                            allow_retry=True,
294                            priority=True,
295                        )
296                    )
297        return tasks
298
299    @property
300    def active_sync_tasks(self) -> list[BackgroundTask]:
301        """Return provider sync tasks that are currently pending or running."""
302        return [
303            task
304            for task in self.mass.tasks.get_tasks_by_metadata(task_domain="music_sync")
305            if task.status in (TaskStatus.PENDING, TaskStatus.RUNNING)
306        ]
307
308    @api_command("music/search", required_scope=Scope.LIBRARY_READ, allow_impersonation=True)
309    async def search(
310        self,
311        search_query: str,
312        media_types: list[MediaType] = MediaType.ALL,
313        limit: int = 25,
314        library_only: bool = False,
315        providers: list[str] | None = None,
316    ) -> SearchResults:
317        """
318        Perform global search for media items on all providers.
319
320        :param search_query: Search query.
321        :param media_types: A list of media_types to include.
322        :param limit: number of items to return in the search (per type).
323        :param library_only: Deprecated - use providers=["library"] instead.
324        :param providers: Optionally restrict the search to the given providers
325            (by instance id or domain), where the special value "library" selects
326            the library. Omit to search the library and all available providers.
327        """
328        if not search_query.strip():
329            # several providers reject an empty query with a hard error
330            return SearchResults()
331        if not media_types:
332            media_types = MediaType.ALL
333        if library_only and providers is None:
334            # handle deprecated library_only flag
335            providers = ["library"]
336        # resolve the search targets: all (unique) music providers plus plugin
337        # providers with search support, optionally filtered by the providers argument
338        plugin_search_providers = [
339            p.instance_id
340            for p in self.mass.get_providers_supporting_feature(
341                ProviderFeature.SEARCH,
342                priority=(ProviderType.PLUGIN,),
343            )
344        ]
345        all_search_providers = sorted(self.get_unique_providers() + plugin_search_providers)
346        if providers is None:
347            include_library = True
348            search_providers = all_search_providers
349        else:
350            include_library = "library" in providers
351            requested_providers = set(providers)
352            search_providers = [
353                instance_id
354                for instance_id in all_search_providers
355                if (prov := self.mass.get_provider(instance_id))
356                and (prov.instance_id in requested_providers or prov.domain in requested_providers)
357            ]
358        # use cache to avoid repeated searches
359        cache_key = (
360            f"{search_query}-{'-'.join(sorted([mt.value for mt in media_types]))}-{limit}-"
361            f"{int(include_library)}-{','.join(search_providers)}"
362        )
363        if cache := await self.mass.cache.get(
364            key=cache_key,
365            provider=self.domain,
366            category=CACHE_CATEGORY_SEARCH_RESULTS,
367            base_class=SearchResults,
368        ):
369            return cast("SearchResults", cache)
370        # Check if the search query is a streaming provider public shareable URL
371        if (url_result := await self._search_shareable_url(search_query)) is not None:
372            return url_result
373        # handle normal global search by querying the library and all providers
374        # the library is always searched first: it is fast and its results are used
375        # to deduplicate provider results and to skip provider searches for media
376        # types that already have a (near) exact match in the library
377        library_results = await self.search_library(search_query, media_types, limit=limit)
378        results_per_provider: list[SearchResults] = []
379        if include_library:
380            results_per_provider.append(library_results)
381        all_results_complete = True
382        if search_providers:
383            # create a set of all provider item ids already in library
384            # this way we can avoid returning duplicates in the search results
385            all_prov_item_ids = {
386                (item.media_type, prov_mapping.provider_domain, prov_mapping.item_id)
387                for items in (
388                    library_results.artists,
389                    library_results.albums,
390                    library_results.tracks,
391                    library_results.playlists,
392                    library_results.audiobooks,
393                    library_results.podcasts,
394                )
395                for item in items
396                for prov_mapping in cast("MediaItemType", item).provider_mappings
397            }
398            # only apply the exact match shortcut on a regular global search;
399            # an explicit providers selection must always search those providers
400            covered_media_types = (
401                self._get_covered_media_types(library_results, search_query)
402                if providers is None
403                else set()
404            )
405            provider_searches: list[Coroutine[Any, Any, SearchResults | None]] = []
406            for provider_instance in search_providers:
407                if not (prov := self.mass.get_provider(provider_instance)):
408                    continue
409                # skip media types for which the library already holds a (near)
410                # exact match that is mapped to this provider: searching the
411                # provider again for that media type will not add anything new
412                prov_media_types = [
413                    mt
414                    for mt in media_types
415                    if (mt, prov.domain) not in covered_media_types
416                    and (mt, prov.instance_id) not in covered_media_types
417                ]
418                if not prov_media_types:
419                    continue
420                provider_searches.append(
421                    self._search_provider(
422                        search_query,
423                        provider_instance,
424                        prov_media_types,
425                        limit=limit,
426                        skip_item_ids=all_prov_item_ids,
427                    )
428                )
429            # include results from all (unique) music providers
430            # one failing provider must not break the entire search,
431            # so exceptions are logged and excluded from the results
432            gather_results = await asyncio.gather(*provider_searches, return_exceptions=True)
433            for res in gather_results:
434                if isinstance(res, SearchResults):
435                    results_per_provider.append(res)
436                    continue
437                # a provider that failed or timed out contributes no results
438                all_results_complete = False
439                if isinstance(res, BaseException):
440                    self.logger.error("Search on provider failed", exc_info=res)
441        # return result from all providers while keeping index
442        # so the result is sorted as each provider delivered
443        result = SearchResults(
444            artists=[
445                item
446                for sublist in zip_longest(*[x.artists for x in results_per_provider])
447                for item in sublist
448                if item is not None
449            ][:limit],
450            albums=[
451                item
452                for sublist in zip_longest(*[x.albums for x in results_per_provider])
453                for item in sublist
454                if item is not None
455            ][:limit],
456            genres=[
457                item
458                for sublist in zip_longest(*[x.genres for x in results_per_provider])
459                for item in sublist
460                if item is not None
461            ][:limit],
462            tracks=[
463                item
464                for sublist in zip_longest(*[x.tracks for x in results_per_provider])
465                for item in sublist
466                if item is not None
467            ][:limit],
468            playlists=[
469                item
470                for sublist in zip_longest(*[x.playlists for x in results_per_provider])
471                for item in sublist
472                if item is not None
473            ][:limit],
474            radio=[
475                item
476                for sublist in zip_longest(*[x.radio for x in results_per_provider])
477                for item in sublist
478                if item is not None
479            ][:limit],
480            audiobooks=[
481                item
482                for sublist in zip_longest(*[x.audiobooks for x in results_per_provider])
483                for item in sublist
484                if item is not None
485            ][:limit],
486            podcasts=[
487                item
488                for sublist in zip_longest(*[x.podcasts for x in results_per_provider])
489                for item in sublist
490                if item is not None
491            ][:limit],
492            sound_effects=[
493                item
494                for sublist in zip_longest(*[x.sound_effects for x in results_per_provider])
495                for item in sublist
496                if item is not None
497            ][:limit],
498        )
499
500        # the search results should already be sorted by relevance
501        # but we apply one extra round of sorting and that is to put exact name
502        # matches and library items first
503        for field in (
504            "artists",
505            "albums",
506            "genres",
507            "tracks",
508            "playlists",
509            "radio",
510            "audiobooks",
511            "podcasts",
512            "sound_effects",
513        ):
514            setattr(result, field, sort_search_result(search_query, getattr(result, field)))
515        # only cache the combined result if all providers contributed,
516        # so a failed or timed out provider is retried on a next search
517        if all_results_complete:
518            await self._cache_search_results(
519                cache_key, result, SEARCH_CACHE_EXPIRATION_COMBINED, self.domain
520            )
521        return result
522
523    async def search_library(
524        self,
525        search_query: str,
526        media_types: list[MediaType],
527        limit: int = 10,
528    ) -> SearchResults:
529        """
530        Perform search on the library.
531
532        :param search_query: Search query
533        :param media_types: A list of media_types to include.
534        :param limit: number of items to return in the search (per type).
535        """
536        result_fields: dict[MediaType, str] = {
537            MediaType.ARTIST: "artists",
538            MediaType.ALBUM: "albums",
539            MediaType.GENRE: "genres",
540            MediaType.TRACK: "tracks",
541            MediaType.PLAYLIST: "playlists",
542            MediaType.RADIO: "radio",
543            MediaType.AUDIOBOOK: "audiobooks",
544            MediaType.PODCAST: "podcasts",
545        }
546        result = SearchResults()
547        # search all media types in parallel, each is an independent db query
548        searchable_media_types = [x for x in media_types if x in result_fields]
549        search_results = await asyncio.gather(
550            *[
551                self.get_controller(media_type).search(search_query, "library", limit=limit)
552                for media_type in searchable_media_types
553            ]
554        )
555        for media_type, items in zip(searchable_media_types, search_results, strict=True):
556            if items:
557                setattr(result, result_fields[media_type], items)
558        return result
559
560    @api_command("music/browse", required_scope=Scope.LIBRARY_READ)
561    async def browse(
562        self, path: str | None = None
563    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
564        """Browse Music providers."""
565        if not path or path == "root":
566            # root level; folder per provider that declares BROWSE
567            root_items: list[MediaItemType | BrowseFolder] = []
568            providers_with_browse = self.mass.get_providers_supporting_feature(
569                ProviderFeature.BROWSE
570            )
571            for prov in self._apply_user_provider_filter(providers_with_browse):
572                root_items.append(
573                    BrowseFolder(
574                        item_id="root",
575                        provider=prov.domain,
576                        path=f"{prov.instance_id}://",
577                        uri=f"{prov.instance_id}://",
578                        name=prov.name,
579                    )
580                )
581            # AudioSource providers surface at root like regular providers; a
582            # provider with a single user-initiable source is promoted to that
583            # source directly so it's playable in one tap.
584            audio_source_providers = self.mass.get_providers_supporting_feature(
585                ProviderFeature.AUDIO_SOURCE
586            )
587            for prov in self._apply_user_provider_filter(audio_source_providers):
588                if not isinstance(prov, PluginProvider):
589                    continue
590                initiable = [
591                    source for source in await prov.get_audio_sources() if source.can_initiate
592                ]
593                if not initiable:
594                    continue
595                if len(initiable) == 1:
596                    root_items.append(initiable[0])
597                else:
598                    root_items.append(
599                        BrowseFolder(
600                            item_id="root",
601                            provider=prov.domain,
602                            path=f"{prov.instance_id}://",
603                            uri=f"{prov.instance_id}://",
604                            name=prov.name,
605                        )
606                    )
607            return root_items
608
609        # provider level
610        prepend_items: list[BrowseFolder] = []
611        provider_instance, sub_path = path.split("://", 1)
612        browse_prov = self.mass.get_provider(provider_instance)
613        # handle regular provider listing, always add back folder first
614        if not browse_prov or not sub_path:
615            prepend_items.append(
616                BrowseFolder(item_id="root", provider="library", path="root", name="..")
617            )
618            if not browse_prov:
619                return prepend_items
620        else:
621            back_path = f"{provider_instance}://" + "/".join(sub_path.split("/")[:-1])
622            prepend_items.append(
623                BrowseFolder(
624                    item_id="back",
625                    provider=provider_instance,
626                    path=back_path,
627                    name="..",
628                )
629            )
630        # AudioSource providers don't implement browse(); list their initiable sources directly
631        if (
632            isinstance(browse_prov, PluginProvider)
633            and ProviderFeature.AUDIO_SOURCE in browse_prov.supported_features
634        ):
635            initiable_items: list[MediaItemType | BrowseFolder] = [
636                source for source in await browse_prov.get_audio_sources() if source.can_initiate
637            ]
638            return [*prepend_items, *initiable_items]
639        # limit -1 to account for the prepended items
640        prov_items = await cast("MusicProvider", browse_prov).browse(path=path)
641        return [*prepend_items, *prov_items]
642
643    @api_command("music/recently_played_items", required_scope=Scope.LIBRARY_READ)
644    async def recently_played(
645        self,
646        limit: int = 10,
647        media_types: list[MediaType] | None = None,
648        userid: str | None = None,
649        queue_id: str | None = None,
650        fully_played_only: bool = True,
651        user_initiated_only: bool = False,
652        played_after_timestamp: int | None = None,
653        providers: list[str] | None = None,
654        *,
655        always_include_media_types: list[MediaType] | None = None,
656    ) -> list[ItemMapping]:
657        """
658        Return a list of the last played items.
659
660        :param limit: Maximum number of items to return.
661        :param media_types: Filter by media types.
662        :param userid: Filter by specific user ID.
663        :param queue_id: Filter by specific queue ID.
664        :param fully_played_only: If True, only return fully played items.
665        :param user_initiated_only: If True, only return items initiated by the user.
666        :param played_after_timestamp: If set, only return items played at or after this
667            epoch-seconds timestamp.
668        :param providers: Restrict results to items reachable through one of these provider
669            instance ids (OR semantics). None applies no filter; an explicit empty list
670            returns no items.
671        :param always_include_media_types: Media types to include regardless of
672            user_initiated_only (e.g. podcasts/audiobooks, which have no user-initiated
673            container).
674        """
675        if providers is not None and not providers:
676            return []
677        if media_types is None:
678            media_types = MediaType.ALL
679        media_types_str = "(" + ",".join(f'"{x}"' for x in media_types) + ")"
680        available_providers = ("library", *self.get_active_provider_instances())
681        available_providers_str = "(" + ",".join(f'"{x}"' for x in available_providers) + ")"
682        # user_initiated_only constrains only `media_types`; always_include_media_types are
683        # included regardless (e.g. podcasts/audiobooks have no user-initiated container row).
684        media_type_clause = f"p.media_type in {media_types_str}"
685        if user_initiated_only:
686            media_type_clause += " AND p.user_initiated = 1"
687        media_type_clause = f"({media_type_clause})"
688        if always_include_media_types:
689            always_str = "(" + ",".join(f'"{x}"' for x in always_include_media_types) + ")"
690            media_type_clause = f"({media_type_clause} OR p.media_type in {always_str})"
691
692        params: dict[str, Any] = {}
693        user = get_current_user()
694        # a library row only needs resolving through its provider mappings when a filter
695        # (explicit or user-scoped) is actually active; otherwise every library row is
696        # kept, matching this method's unfiltered behavior.
697        if providers is not None or (user and user.provider_filter):
698            requested_clause = ""
699            direct_requested_clause = ""
700            if providers is not None:
701                params["requested_providers"] = providers
702                requested_clause = " AND m.provider_instance IN :requested_providers"
703                direct_requested_clause = " AND p.provider IN :requested_providers"
704            provider_clause = (
705                "(CASE WHEN p.provider = 'library' THEN "
706                f"EXISTS (SELECT 1 FROM {DB_TABLE_PROVIDER_MAPPINGS} m "
707                "WHERE m.item_id = p.item_id AND m.media_type = p.media_type "
708                f"AND m.available = 1 "
709                f"AND m.provider_instance IN {available_providers_str}{requested_clause}) "
710                f"ELSE (p.provider IN {available_providers_str}{direct_requested_clause}) END)"
711            )
712        else:
713            provider_clause = f"p.provider IN {available_providers_str}"
714        query = (
715            f"SELECT p.* FROM {DB_TABLE_PLAYLOG} p WHERE {media_type_clause} AND {provider_clause} "
716        )
717        if fully_played_only:
718            query += "AND p.fully_played = 1 "
719        if userid:
720            query += "AND p.userid = :userid "
721            params["userid"] = userid
722        elif user:
723            query += "AND p.userid = :userid "
724            params["userid"] = user.user_id
725        if queue_id:
726            query += "AND p.queue_id = :queue_id "
727            params["queue_id"] = queue_id
728        if played_after_timestamp is not None:
729            query += "AND p.timestamp >= :played_after_timestamp "
730            params["played_after_timestamp"] = played_after_timestamp
731        query += "ORDER BY p.timestamp DESC"
732        db_rows = await self.mass.music.database.get_rows_from_query(
733            query, params=params or None, limit=limit
734        )
735        result: list[ItemMapping] = []
736        available_providers = ("library", *get_global_cache_value("available_providers", []))
737        for db_row in db_rows:
738            provider = db_row["provider"]
739            result.append(
740                ItemMapping.from_dict(
741                    {
742                        "item_id": db_row["item_id"],
743                        "provider": provider,
744                        "media_type": db_row["media_type"],
745                        "name": db_row["name"],
746                        "image": json_loads(db_row["image"]) if db_row["image"] else None,
747                        "available": provider in available_providers,
748                    }
749                )
750            )
751        return result
752
753    async def recently_played_tracks(
754        self,
755        limit: int,
756        played_after_timestamp: int,
757        userid: str | None = None,
758    ) -> list[RecentPlayedTrack]:
759        """
760        Return recently played, fully played tracks with their recorded artists, newest first.
761
762        :param limit: Maximum number of plays to return.
763        :param played_after_timestamp: Only include plays at or after this epoch-seconds timestamp.
764        :param userid: Restrict to this user (defaults to the current session user, else all users).
765        """
766        query = (
767            f"SELECT item_id, provider, name, image, artists FROM {DB_TABLE_PLAYLOG} "
768            "WHERE media_type = 'track' AND fully_played = 1 "
769            "AND timestamp >= :played_after_timestamp "
770        )
771        params: dict[str, Any] = {"played_after_timestamp": played_after_timestamp}
772        if userid:
773            query += "AND userid = :userid "
774            params["userid"] = userid
775        elif user := get_current_user():
776            query += "AND userid = :userid "
777            params["userid"] = user.user_id
778        query += "ORDER BY timestamp DESC"
779        db_rows = await self.mass.music.database.get_rows_from_query(
780            query, params=params, limit=limit
781        )
782        available_providers = ("library", *get_global_cache_value("available_providers", []))
783        return [
784            RecentPlayedTrack(
785                track=ItemMapping.from_dict(
786                    {
787                        "item_id": db_row["item_id"],
788                        "provider": db_row["provider"],
789                        "media_type": "track",
790                        "name": db_row["name"],
791                        "image": json_loads(db_row["image"]) if db_row["image"] else None,
792                        "available": db_row["provider"] in available_providers,
793                    }
794                ),
795                artists=[ItemMapping.from_dict(artist) for artist in json_loads(db_row["artists"])]
796                if db_row["artists"]
797                else [],
798            )
799            for db_row in db_rows
800        ]
801
802    @api_command("music/recently_added_tracks", required_scope=Scope.LIBRARY_READ)
803    async def recently_added_tracks(self, limit: int = 10) -> list[Track]:
804        """Return a list of the last added tracks."""
805        return await self.tracks.library_items(
806            limit=limit, order_by="timestamp_added_desc", summary=False
807        )
808
809    @api_command("music/in_progress_items", required_scope=Scope.LIBRARY_READ)
810    async def in_progress_items(
811        self, limit: int = 10, all_users: bool = False, providers: list[str] | None = None
812    ) -> list[ItemMapping]:
813        """
814        Return a list of the Audiobooks and PodcastEpisodes that are in progress.
815
816        :param limit: Maximum number of items to return.
817        :param all_users: If True, include in-progress items across all users, not just
818            the current session's user.
819        :param providers: Restrict results to items reachable through one of these provider
820            instance ids (OR semantics). None applies no filter; an explicit empty list
821            returns no items.
822        """
823        if providers is not None and not providers:
824            return []
825        available_providers = ("library", *self.get_active_provider_instances())
826        available_providers_str = "(" + ",".join(f'"{x}"' for x in available_providers) + ")"
827        params: dict[str, Any] = {}
828        requested_clause = ""
829        direct_requested_clause = ""
830        if providers is not None:
831            params["requested_providers"] = providers
832            requested_clause = " AND m.provider_instance IN :requested_providers"
833            direct_requested_clause = " AND p.provider IN :requested_providers"
834
835        # An audiobook can be part of the library, in contrast to podcast episodes.
836        # We then need to check the provider mappings table.
837        one_week_ago = int(utc_timestamp()) - (7 * 86400)
838        query = (
839            "SELECT p.item_id, p.media_type, p.name, p.image, p.provider "
840            f"FROM {DB_TABLE_PLAYLOG} p "
841            "WHERE p.media_type IN ('audiobook', 'podcast_episode') "
842            "AND p.fully_played = 0 "
843            "AND p.seconds_played > 0 "
844            f"AND (p.media_type != 'podcast_episode' OR p.timestamp >= {one_week_ago}) "
845        )
846        query += (
847            "AND ( "
848            "CASE WHEN p.provider = 'library' THEN "
849            f"EXISTS (SELECT 1 FROM {DB_TABLE_PROVIDER_MAPPINGS} m "
850            "WHERE m.item_id = p.item_id AND m.media_type = p.media_type "
851            "AND m.available = 1 "
852        )
853        if not all_users and (user := get_current_user()):
854            filter_for_str = available_providers_str
855            if user.provider_filter:
856                filter_for_str = "(" + ",".join(f'"{x}"' for x in user.provider_filter) + ")"
857            query += (
858                f"AND m.provider_instance IN {filter_for_str} "
859                f"AND m.provider_instance IN {available_providers_str}"
860                f"{requested_clause} "
861                ") "
862                f"ELSE (p.provider IN {filter_for_str} AND p.provider IN {available_providers_str}"
863                f"{direct_requested_clause})"
864                "END "
865                ") "
866                f"AND p.userid = '{user.user_id}' "
867            )
868        else:
869            # for a library item, we still have to verify via the provider mapping table
870            # that the provider is available
871            query += (
872                f"AND m.provider_instance IN {available_providers_str}"
873                f"{requested_clause} "
874                ") "
875                f"ELSE p.provider IN {available_providers_str}"
876                f"{direct_requested_clause} "
877                "END "
878                ") "
879            )
880        query += "ORDER BY timestamp DESC"
881
882        db_rows = await self.mass.music.database.get_rows_from_query(
883            query, params=params or None, limit=limit
884        )
885        result: list[ItemMapping] = []
886        for db_row in db_rows:
887            provider = db_row["provider"]
888            result.append(
889                ItemMapping.from_dict(
890                    {
891                        "item_id": db_row["item_id"],
892                        "provider": provider,
893                        "media_type": db_row["media_type"],
894                        "name": db_row["name"],
895                        "image": json_loads(db_row["image"]) if db_row["image"] else None,
896                        "available": provider in available_providers,
897                    }
898                )
899            )
900        return result
901
902    async def get_playlog_provider_item_ids(
903        self, provider_instance_id: str, limit: int = 0, userid: str | None = None
904    ) -> list[tuple[MediaType, str]]:
905        """Return a list of MediaType and provider_item_id of items in playlog of provider."""
906        # check if there is a provider user
907        # this method is not available in the frontend, so no need to check for session users.
908        user: User | None = None
909        if userid:
910            # userid overridden by parameter
911            user = await self.mass.webserver.auth.get_user(userid)
912        elif provider_user := await self._get_user_for_provider(provider_instance_id):
913            # based on configured provider filter we can try to find a user
914            user = provider_user
915
916        query = (
917            f"SELECT * FROM {DB_TABLE_PLAYLOG} "
918            "WHERE media_type in ('audiobook', 'podcast_episode') "
919            f"AND provider in ('library','{provider_instance_id}')"
920        )
921
922        if user:
923            # NOTE: if no user was found, we will return playlog items for all users
924            query += f" AND userid = '{user.user_id}'"
925        db_rows = await self.mass.music.database.get_rows_from_query(query, limit=limit)
926
927        result: list[tuple[MediaType, str]] = []
928        for db_row in db_rows:
929            if db_row["provider"] == "library":
930                # If the provider is library, we need to make sure that the item
931                # is part of the passed provider_instance_id.
932                # A podcast_episode cannot be in the provider_mappings
933                # so these entries must be audiobooks.
934                subquery = (
935                    f"SELECT * FROM {DB_TABLE_PROVIDER_MAPPINGS} "
936                    f"WHERE media_type = 'audiobook' AND item_id = {db_row['item_id']} "
937                    f"AND provider_instance = '{provider_instance_id}'"
938                )
939                subrow = await self.mass.music.database.get_rows_from_query(subquery)
940                if len(subrow) != 1:
941                    continue
942                result.append((MediaType.AUDIOBOOK, subrow[0]["provider_item_id"]))
943                continue
944            # non library - item id is provider_item_id
945            result.append((MediaType(db_row["media_type"]), db_row["item_id"]))
946
947        return result
948
949    @api_command("music/item_by_uri", required_scope=Scope.LIBRARY_READ)
950    async def get_item_by_uri(
951        self, uri: str, allow_update_metadata: bool = False
952    ) -> MediaItemType | BrowseFolder:
953        """Fetch MediaItem by uri."""
954        media_type, provider_instance_id_or_domain, item_id = await parse_uri(uri)
955        return await self.get_item(
956            media_type=media_type,
957            item_id=item_id,
958            provider_instance_id_or_domain=provider_instance_id_or_domain,
959            allow_update_metadata=allow_update_metadata,
960        )
961
962    @api_command("music/sound_effects", required_scope=Scope.LIBRARY_READ)
963    async def sound_effects(self) -> list[SoundEffect]:
964        """Return all sound effect items from providers supporting them."""
965        providers = self._apply_user_provider_filter(
966            self.mass.get_providers_supporting_feature(ProviderFeature.SOUND_EFFECTS)
967        )
968        results_per_provider: list[list[SoundEffect]] = await asyncio.gather(
969            *[
970                self._get_provider_sound_effects(cast("MusicProvider", provider))
971                for provider in providers
972            ]
973        )
974        return [item for sublist in results_per_provider for item in sublist]
975
976    @api_command("music/item", required_scope=Scope.LIBRARY_READ)
977    async def get_item(
978        self,
979        media_type: MediaType,
980        item_id: str,
981        provider_instance_id_or_domain: str,
982        allow_update_metadata: bool = True,
983    ) -> MediaItemType | BrowseFolder:
984        """Get single music item by id and media type."""
985        if provider_instance_id_or_domain == "database":
986            # backwards compatibility - to remove when 2.0 stable is released
987            provider_instance_id_or_domain = "library"
988        provider = self.mass.get_provider(provider_instance_id_or_domain)
989        if media_type in (
990            MediaType.TRACK,
991            MediaType.RADIO,
992            MediaType.SOUND_EFFECT,
993            MediaType.UNKNOWN,  # e.g. plain (HA) URLs, see helpers/uri.py
994        ) and (
995            provider_instance_id_or_domain == "builtin"
996            or (provider and provider.domain == "builtin")
997        ):
998            # handle special case of 'builtin' MusicProvider which allows us to play regular url's
999            builtin_prov = cast("BuiltinProvider", provider or self.mass.get_provider("builtin"))
1000            if media_type == MediaType.RADIO:
1001                # a radio station must stay a radio station, also when the stream
1002                # reports a duration or carries no ICY name
1003                return await builtin_prov.get_radio(item_id)
1004            if media_type == MediaType.TRACK:
1005                # and a track must stay a track, also when the stream carries an
1006                # ICY name or reports no duration
1007                return await builtin_prov.get_track(item_id)
1008            return await builtin_prov.parse_item(item_id, requested_media_type=media_type)
1009        if media_type == MediaType.PODCAST_EPISODE:
1010            # special case for podcast episodes
1011            return await self.podcasts.episode(item_id, provider_instance_id_or_domain)
1012        if media_type == MediaType.FOLDER:
1013            # special case for folders
1014            return BrowseFolder(
1015                item_id=item_id,
1016                provider=provider_instance_id_or_domain,
1017                name=item_id,
1018            )
1019        if media_type == MediaType.AUDIO_SOURCE:
1020            # AudioSources are not library-backed; resolve them through the owning
1021            # plugin provider's get_audio_sources() catalog. Returning the live
1022            # MediaItem lets play_media create a queue item the standard way.
1023            prov = self.mass.get_provider(provider_instance_id_or_domain)
1024            if isinstance(prov, PluginProvider):
1025                for source in await prov.get_audio_sources():
1026                    if source.item_id == item_id:
1027                        return source
1028            raise MediaNotFoundError(
1029                f"AudioSource {provider_instance_id_or_domain}/{item_id} not found"
1030            )
1031        if media_type == MediaType.SOUND_EFFECT:
1032            # Sound effects are not library-backed; resolve them live from the
1033            # owning music provider. Returning the live MediaItem lets play_media
1034            # create a queue item the standard way.
1035            prov = self.mass.get_provider(provider_instance_id_or_domain)
1036            if isinstance(prov, MusicProvider) and (
1037                ProviderFeature.SOUND_EFFECTS in prov.supported_features
1038            ):
1039                return await prov.get_sound_effect(item_id)
1040            raise MediaNotFoundError(
1041                f"SoundEffect {provider_instance_id_or_domain}/{item_id} not found"
1042            )
1043        if media_type == MediaType.COLLECTION:
1044            ctrl = self.get_controller_for_collection(item_id)
1045            return await ctrl.get_collection(item_id)
1046        ctrl = self.get_controller(media_type)
1047        return await ctrl.get(
1048            item_id=item_id,
1049            provider_instance_id_or_domain=provider_instance_id_or_domain,
1050            allow_update_metadata=allow_update_metadata,
1051        )
1052
1053    @api_command("music/get_library_item", required_scope=Scope.LIBRARY_READ)
1054    async def get_library_item_by_prov_id(
1055        self,
1056        media_type: MediaType,
1057        item_id: str,
1058        provider_instance_id_or_domain: str,
1059    ) -> MediaItemType | None:
1060        """Get the library item for the given provider item, if present."""
1061        ctrl = self.get_controller(media_type)
1062        return await ctrl.get_library_item_by_prov_id(
1063            item_id=item_id,
1064            provider_instance_id_or_domain=provider_instance_id_or_domain,
1065        )
1066
1067    @api_command("music/favorites/add_item", required_scope=Scope.LIBRARY_WRITE)
1068    async def add_item_to_favorites(
1069        self,
1070        item: str | MediaItemType | ItemMapping,
1071    ) -> None:
1072        """Add an item to the favorites."""
1073        if isinstance(item, str):
1074            # Inspect the URI's media_type first so a stale audio-source URI
1075            # whose plugin is unloaded gives the honest rejection error
1076            # instead of bubbling MediaNotFoundError from get_item_by_uri.
1077            try:
1078                uri_media_type, _, _ = await parse_uri(item)
1079            except InvalidProviderURI, InvalidProviderID:
1080                uri_media_type = None
1081            if uri_media_type in (MediaType.AUDIO_SOURCE, MediaType.SOUND_EFFECT):
1082                raise UnsupportedFeaturedException(
1083                    f"{uri_media_type.value} items can not be favorites"
1084                )
1085            # a favorite URI always resolves to a media item, never a BrowseFolder
1086            item = cast("MediaItemType", await self.get_item_by_uri(item))
1087        if item.media_type in (MediaType.AUDIO_SOURCE, MediaType.SOUND_EFFECT):
1088            # AudioSources and SoundEffects are live provider content (existence
1089            # depends on a loaded provider) and have no stable library identity,
1090            # so they can not be persisted as favorites.
1091            raise UnsupportedFeaturedException(
1092                f"{item.media_type.value} items can not be favorites"
1093            )
1094        # make sure we have a full library item
1095        # a favorite must always be in the library
1096        full_item = cast(
1097            "MediaItemType",
1098            await self.get_item(
1099                item.media_type,
1100                item.item_id,
1101                item.provider,
1102            ),
1103        )
1104        if full_item.provider != "library":
1105            full_item = await self.add_item_to_library(full_item)
1106        # set favorite in library db
1107        ctrl = self.get_controller(item.media_type)
1108        await ctrl.set_favorite(
1109            full_item.item_id,
1110            True,
1111        )
1112        # forward to provider(s) if needed
1113        for prov_mapping in full_item.provider_mappings:
1114            provider = self.mass.get_provider(
1115                prov_mapping.provider_instance, provider_type=MusicProvider
1116            )
1117            if not provider or not self.library_favorites_edit_supported(
1118                provider, full_item.media_type
1119            ):
1120                continue
1121            await provider.set_favorite(prov_mapping.item_id, full_item.media_type, True)
1122
1123    @api_command("music/favorites/remove_item", required_scope=Scope.LIBRARY_WRITE)
1124    async def remove_item_from_favorites(
1125        self,
1126        media_type: MediaType,
1127        library_item_id: str | int,
1128    ) -> None:
1129        """Remove (library) item from the favorites."""
1130        ctrl = self.get_controller(media_type)
1131        await ctrl.set_favorite(
1132            library_item_id,
1133            False,
1134        )
1135        # forward to provider(s) if needed
1136        full_item = await ctrl.get_library_item(library_item_id)
1137        for prov_mapping in full_item.provider_mappings:
1138            provider = self.mass.get_provider(
1139                prov_mapping.provider_instance, provider_type=MusicProvider
1140            )
1141            if not provider or not self.library_favorites_edit_supported(
1142                provider, full_item.media_type
1143            ):
1144                continue
1145            self.mass.create_task(provider.set_favorite(prov_mapping.item_id, media_type, False))
1146
1147    @api_command("music/library/remove_item", required_scope=Scope.LIBRARY_WRITE)
1148    async def remove_item_from_library(
1149        self, media_type: MediaType, library_item_id: str | int, recursive: bool = True
1150    ) -> None:
1151        """
1152        Remove item from the library.
1153
1154        Destructive! Will remove the item and all dependants.
1155        """
1156        ctrl = self.get_controller(media_type)
1157        # remove from provider(s) library
1158        full_item = await ctrl.get_library_item(library_item_id)
1159        for prov_mapping in full_item.provider_mappings:
1160            if not prov_mapping.in_library:
1161                continue
1162            provider = self.mass.get_provider(
1163                prov_mapping.provider_instance, provider_type=MusicProvider
1164            )
1165            if not provider or not self.library_edit_supported(provider, full_item.media_type):
1166                continue
1167            if not self.library_sync_back_enabled(provider, full_item.media_type):
1168                continue
1169            prov_mapping.in_library = False
1170            self.mass.create_task(provider.library_remove(prov_mapping.item_id, media_type))
1171        # remove from library
1172        await ctrl.remove_item_from_library(library_item_id, recursive)
1173
1174    @api_command("music/library/add_item", required_scope=Scope.LIBRARY_WRITE)
1175    async def add_item_to_library(
1176        self, item: str | MediaItemType | ItemMapping, overwrite_existing: bool = False
1177    ) -> MediaItemType:
1178        """Add item (uri or mediaitem) to the library."""
1179        if isinstance(item, ItemMapping):
1180            # handle browse results that are returned as ItemMappings
1181            # uri is always populated post-init, so it is never None here
1182            item = cast("str", item.uri)
1183        # ensure we have a full item
1184        if isinstance(item, str):
1185            # Inspect the URI's media_type first so a stale audio-source URI
1186            # whose plugin is unloaded gives the honest rejection error
1187            # instead of bubbling MediaNotFoundError from get_item_by_uri.
1188            # Mirrors the same guard in add_item_to_favorites.
1189            try:
1190                uri_media_type, _, _ = await parse_uri(item)
1191            except InvalidProviderURI, InvalidProviderID:
1192                uri_media_type = None
1193            if uri_media_type in (MediaType.AUDIO_SOURCE, MediaType.SOUND_EFFECT):
1194                raise UnsupportedFeaturedException(
1195                    f"{uri_media_type.value} items can not be library items"
1196                )
1197            full_item = await self.get_item_by_uri(item)
1198        # For builtin provider (manual URLs), use the provided item directly
1199        # to preserve custom modifications (name, images, etc.)
1200        # For other providers, fetch fresh to ensure data validity
1201        elif item.provider == "builtin":
1202            full_item = item
1203        else:
1204            full_item = await self.get_item(
1205                item.media_type,
1206                item.item_id,
1207                item.provider,
1208            )
1209        full_item = cast("MediaItemType", full_item)
1210        if full_item.media_type in (MediaType.AUDIO_SOURCE, MediaType.SOUND_EFFECT):
1211            # AudioSources and SoundEffects are live provider content (existence
1212            # depends on a loaded provider) and have no stable library identity,
1213            # so they can not be persisted as library items.
1214            raise UnsupportedFeaturedException(
1215                f"{full_item.media_type.value} items can not be library items"
1216            )
1217        # add to provider(s) library first
1218        for prov_mapping in full_item.provider_mappings:
1219            # we optimistically set in library to True to prevent items
1220            # from disappearing when the provider doesn't support library edit
1221            # or 2-way sync is disabled.
1222            prov_mapping.in_library = True
1223            provider = self.mass.get_provider(
1224                prov_mapping.provider_instance, provider_type=MusicProvider
1225            )
1226            if not provider or not self.library_edit_supported(provider, full_item.media_type):
1227                continue
1228            if not self.library_sync_back_enabled(provider, full_item.media_type):
1229                continue
1230            prov_item = deepcopy(full_item) if full_item.provider == "library" else full_item
1231            prov_item.provider = prov_mapping.provider_instance
1232            prov_item.item_id = prov_mapping.item_id
1233            self.mass.create_task(provider.library_add(prov_item))
1234        # add (or overwrite) to library
1235        ctrl = self.get_controller(full_item.media_type)
1236        # ctrl is chosen by media_type, so it matches full_item's runtime type
1237        library_item = await cast("MediaControllerBase[MediaItemType]", ctrl).add_item_to_library(
1238            full_item, overwrite_existing
1239        )
1240        # optionally import all album tracks into the library, mirroring the behavior
1241        # of the library sync (which only triggers on a (scheduled) full sync run)
1242        if full_item.media_type == MediaType.ALBUM:
1243            self._import_album_tracks_if_enabled(cast("Album", library_item))
1244        # perform full metadata scan
1245        await self.mass.metadata.update_metadata(library_item, overwrite_existing)
1246        return library_item
1247
1248    @api_command("music/refresh_item", required_scope=Scope.LIBRARY_MANAGE)
1249    async def refresh_item(  # noqa: PLR0915
1250        self,
1251        media_item: str | MediaItemType,
1252    ) -> MediaItemType | None:
1253        """Try to refresh a mediaitem by requesting it's full object or search for substitutes."""
1254        if isinstance(media_item, str):
1255            # media item uri given
1256            # a refresh URI always resolves to a media item, never a BrowseFolder
1257            media_item = cast("MediaItemType", await self.get_item_by_uri(media_item))
1258
1259        media_type = media_item.media_type
1260        ctrl = self.get_controller(media_type)
1261
1262        # genres are library-only items with no provider mappings, nothing to refresh
1263        if media_type == MediaType.GENRE:
1264            return media_item
1265
1266        library_id = media_item.item_id if media_item.provider == "library" else None
1267
1268        # cache in_library state before the provider fetch overwrites media_item
1269        in_library_cache: dict[tuple[str, str], bool] = {}
1270        for m in media_item.provider_mappings:
1271            if m.in_library is not None:
1272                in_library_cache[(m.provider_instance, m.item_id)] = m.in_library
1273
1274        available_providers = get_global_cache_value("available_providers")
1275        if TYPE_CHECKING:
1276            available_providers = cast("set[str]", available_providers)
1277
1278        # fetch the first (available) provider item
1279        for prov_mapping in sorted(
1280            media_item.provider_mappings, key=lambda x: x.priority, reverse=True
1281        ):
1282            if not self.mass.get_provider(prov_mapping.provider_instance):
1283                # ignore unavailable providers
1284                continue
1285            with suppress(MediaNotFoundError):
1286                media_item = await ctrl.get_provider_item(
1287                    prov_mapping.item_id,
1288                    prov_mapping.provider_instance,
1289                    force_refresh=True,
1290                )
1291                provider = media_item.provider
1292                item_id = media_item.item_id
1293                break
1294        else:
1295            # try to find a substitute using search
1296            searchresult = await self.search(media_item.name, [media_item.media_type], 20)
1297            result: Sequence[MediaItemType | ItemMapping]
1298            if media_item.media_type == MediaType.ARTIST:
1299                result = searchresult.artists
1300            elif media_item.media_type == MediaType.ALBUM:
1301                result = searchresult.albums
1302            elif media_item.media_type == MediaType.TRACK:
1303                result = searchresult.tracks
1304            elif media_item.media_type == MediaType.PLAYLIST:
1305                result = searchresult.playlists
1306            elif media_item.media_type == MediaType.AUDIOBOOK:
1307                result = searchresult.audiobooks
1308            elif media_item.media_type == MediaType.PODCAST:
1309                result = searchresult.podcasts
1310            else:
1311                result = searchresult.radio
1312            for item in result:
1313                if item == media_item or item.provider == "library":
1314                    continue
1315                if item.available:
1316                    provider = item.provider
1317                    item_id = item.item_id
1318                    break
1319            else:
1320                # raise if we didn't find a substitute
1321                raise MediaNotFoundError(f"Could not find a substitute for {media_item.name}")
1322        # fetch full (provider) item
1323        media_item = await ctrl.get_provider_item(item_id, provider, force_refresh=True)
1324        # update library item if needed (including refresh of the metadata etc.)
1325        if library_id is None:
1326            return media_item
1327        # restore in_library state from before the refresh
1328        for prov_mapping in media_item.provider_mappings:
1329            key = (prov_mapping.provider_instance, prov_mapping.item_id)
1330            if prov_mapping.in_library is None and key in in_library_cache:
1331                prov_mapping.in_library = in_library_cache[key]
1332        # ctrl is chosen by media_type, so it matches media_item's runtime type
1333        library_item = await cast(
1334            "MediaControllerBase[MediaItemType]", ctrl
1335        ).update_item_in_library(library_id, media_item, overwrite=True)
1336        if library_item.media_type == MediaType.ALBUM:
1337            # update (local) album tracks
1338            for album_track in await self.albums.tracks(
1339                library_item.item_id, library_item.provider, True
1340            ):
1341                for prov_mapping in album_track.provider_mappings:
1342                    if not (prov := self.mass.get_provider(prov_mapping.provider_instance)):
1343                        continue
1344                    if not isinstance(prov, MusicProvider):
1345                        continue
1346                    if prov.is_streaming_provider:
1347                        continue
1348                    with suppress(MediaNotFoundError):
1349                        prov_track = await prov.get_track(prov_mapping.item_id)
1350                        await self.mass.music.tracks.update_item_in_library(
1351                            album_track.item_id, prov_track
1352                        )
1353        await cast("MediaControllerBase[MediaItemType]", ctrl).match_providers(library_item)
1354        await self.mass.metadata.update_metadata(library_item, force_refresh=True)
1355        return library_item
1356
1357    @api_command("music/mark_played", required_scope=Scope.LIBRARY_WRITE)
1358    async def mark_item_played(
1359        self,
1360        media_item: MediaItemType,
1361        fully_played: bool = True,
1362        seconds_played: int | None = None,
1363        is_playing: bool = False,
1364        userid: str | None = None,
1365        queue_id: str | None = None,
1366        user_initiated: bool = True,
1367        skip_artist_ids: list[str] | None = None,
1368        playback_speed: float | None = None,
1369    ) -> None:
1370        """
1371        Mark item as played in playlog.
1372
1373        :param media_item: The media item to mark as played.
1374        :param fully_played: If True, mark the item as fully played.
1375        :param seconds_played: The number of seconds played.
1376        :param is_playing: If True, the item is currently playing.
1377        :param userid: The user ID to mark the item as played for (instead of the current user).
1378        :param queue_id: The queue ID where the item was played.
1379        :param user_initiated: If True, the playback was initiated by the user (e.g. enqueued).
1380            Sticky once set: a later report can promote a playlog row to user-initiated but
1381            never demote it, so a writer reporting playback it did not itself initiate
1382            (e.g. a provider sync) must pass False.
1383        :param skip_artist_ids: Library artist ids to skip when crediting an album's artists.
1384        :param playback_speed: The current playback speed to persist (audiobooks/podcasts).
1385            If None, any previously stored speed for the item is preserved.
1386        """
1387        timestamp = utc_timestamp()
1388        # we deliberately skip one-off items: sound effects and live inputs whoever owns
1389        # them, and everything the builtin provider plays (except playlists) is a one-off url
1390        if media_item.media_type in (MediaType.SOUND_EFFECT, MediaType.AUDIO_SOURCE):
1391            return
1392        if (
1393            media_item.provider.startswith("builtin")
1394            and media_item.media_type != MediaType.PLAYLIST
1395        ):
1396            return
1397
1398        params = {
1399            "item_id": media_item.item_id,
1400            "provider": media_item.provider,
1401            "media_type": media_item.media_type.value,
1402            "name": media_item.name,
1403            "image": serialize_to_json(media_item.image.to_dict()) if media_item.image else None,
1404            # store lightweight artist mappings so playlog rows can later be matched or
1405            # resolved by artist without an extra provider lookup
1406            "artists": serialize_to_json(
1407                [ItemMapping.from_item(artist).to_dict() for artist in artists]
1408            )
1409            if (artists := getattr(media_item, "artists", None))
1410            else None,
1411            "fully_played": fully_played,
1412            "seconds_played": seconds_played,
1413            "timestamp": timestamp,
1414            "queue_id": queue_id,
1415            "user_initiated": user_initiated,
1416        }
1417        # try to figure out the user that triggered the action
1418        user: User | None = None
1419        if userid:
1420            # userid overridden by parameter
1421            user = await self.mass.webserver.auth.get_user(userid)
1422        elif session_user := get_current_user():
1423            # this is the active session user that triggered the action
1424            user = session_user
1425        elif provider_user := await self._get_user_for_provider(media_item.provider_mappings):
1426            # based on configured provider filter we can try to find a user
1427            user = provider_user
1428
1429        # update generic playlog table (when not playing)
1430        if not is_playing:
1431            if user:
1432                user_ids = [user.user_id]
1433            else:
1434                # NOTE: if no user was found, we will alter the playlog for all users
1435                user_ids = [user.user_id for user in await self.mass.webserver.auth.list_users()]
1436            # Leaving the speed out keeps whatever is already stored for this item/user
1437            # (a provider sync reporting progress has no speed to offer), and falls back to
1438            # the column default of 1.0 for a brand new row.
1439            if playback_speed is not None:
1440                params["playback_speed"] = playback_speed
1441            for user_id in user_ids:
1442                params["userid"] = user_id
1443                await self._upsert_playlog(params)
1444
1445        # Set seconds_played in accordance with fully_played, if the media_item has
1446        # a duration, before it is forwarded to music_providers
1447        if seconds_played is None:
1448            seconds_played = 0
1449            if (
1450                fully_played
1451                and not isinstance(
1452                    media_item, Album | Artist | Genre | Playlist | Podcast | MediaCollection
1453                )
1454                and isinstance(media_item.duration, int)  # for Radio duration can be None
1455            ):
1456                seconds_played = media_item.duration
1457
1458        # forward to provider(s) to sync resume state (e.g. for audiobooks)
1459        for prov_mapping in media_item.provider_mappings:
1460            if (
1461                user
1462                and user.provider_filter
1463                and prov_mapping.provider_instance not in user.provider_filter
1464            ):
1465                continue
1466            if music_prov := self.mass.get_provider(prov_mapping.provider_instance):
1467                if music_prov.type != ProviderType.MUSIC:
1468                    continue
1469                music_prov = cast("MusicProvider", music_prov)
1470                self.mass.create_task(
1471                    music_prov.on_played(
1472                        media_type=media_item.media_type,
1473                        prov_item_id=prov_mapping.item_id,
1474                        fully_played=fully_played,
1475                        position=seconds_played,
1476                        media_item=media_item,
1477                        is_playing=is_playing,
1478                    )
1479                )
1480
1481        # also update playcount in library table (if fully played)
1482        if not fully_played or is_playing:
1483            return
1484        try:
1485            ctrl = self.get_controller(media_item.media_type)
1486        except NotImplementedError:
1487            # skip non-library media types (e.g. AudioSource plugin sources)
1488            return
1489        db_item = await ctrl.get_library_item_by_prov_id(media_item.item_id, media_item.provider)
1490        if db_item:
1491            await self.database.execute(
1492                f"UPDATE {ctrl.db_table} SET play_count = play_count + 1, "
1493                f"last_played = {timestamp} WHERE item_id = {db_item.item_id}"
1494            )
1495            if isinstance(media_item, Track):
1496                self.logger.debug("Credited play for track '%s'", media_item.name)
1497        if isinstance(media_item, Track | Album):
1498            await self._credit_artist_plays(
1499                media_item.artists,
1500                timestamp=timestamp,
1501                user_ids=user_ids,
1502                queue_id=queue_id,
1503                skip_ids=set(skip_artist_ids or ()),
1504            )
1505        if isinstance(media_item, PodcastEpisode) and media_item.podcast:
1506            await self._credit_podcast_play(
1507                media_item.podcast,
1508                timestamp=timestamp,
1509                user_ids=user_ids,
1510                queue_id=queue_id,
1511            )
1512        await self.database.commit()
1513
1514    async def resolve_library_artist_ids(self, artists: Iterable[Artist | ItemMapping]) -> set[str]:
1515        """Resolve the given artist references to their library item ids (when present)."""
1516        ids: set[str] = set()
1517        for artist in artists:
1518            db_artist = await self.artists.get_library_item_by_prov_id(
1519                artist.item_id, artist.provider
1520            )
1521            if db_artist is not None:
1522                ids.add(db_artist.item_id)
1523        return ids
1524
1525    @api_command("music/mark_unplayed", required_scope=Scope.LIBRARY_WRITE)
1526    async def mark_item_unplayed(
1527        self,
1528        media_item: MediaItemType,
1529        userid: str | None = None,
1530    ) -> None:
1531        """
1532        Mark item as unplayed in playlog.
1533
1534        :param media_item: The media item to mark as unplayed.
1535        :param all_users: If True, mark the item as unplayed for all users.
1536        :param userid: The user ID to mark the item as unplayed for (instead of the current user).
1537        """
1538        params = {
1539            "item_id": media_item.item_id,
1540            "provider": media_item.provider,
1541            "media_type": media_item.media_type.value,
1542        }
1543        # try to figure out the user that triggered the action
1544        user: User | None = None
1545        if userid:
1546            # userid overridden by parameter
1547            user = await self.mass.webserver.auth.get_user(userid)
1548        elif session_user := get_current_user():
1549            # this is the active session user that triggered the action
1550            user = session_user
1551        elif provider_user := await self._get_user_for_provider(media_item.provider_mappings):
1552            # based on configured provider filter we can try to find a user
1553            user = provider_user
1554
1555        if user:
1556            user_ids = [user.user_id]
1557        else:
1558            # NOTE: if no user was found, we will alter the playlog for all users
1559            user_ids = [user.user_id for user in await self.mass.webserver.auth.list_users()]
1560        for user_id in user_ids:
1561            params["userid"] = user_id
1562            await self.database.delete(DB_TABLE_PLAYLOG, params)
1563
1564        # forward to provider(s) to sync resume state (e.g. for audiobooks)
1565        for prov_mapping in media_item.provider_mappings:
1566            if (
1567                user
1568                and user.provider_filter
1569                and prov_mapping.provider_instance not in user.provider_filter
1570            ):
1571                continue
1572            if music_prov := self.mass.get_provider(prov_mapping.provider_instance):
1573                if music_prov.type != ProviderType.MUSIC:
1574                    continue
1575                music_prov = cast("MusicProvider", music_prov)
1576                self.mass.create_task(
1577                    music_prov.on_played(
1578                        media_type=media_item.media_type,
1579                        prov_item_id=prov_mapping.item_id,
1580                        fully_played=False,
1581                        position=0,
1582                        media_item=media_item,
1583                    )
1584                )
1585        # also update playcount in library table
1586        ctrl = self.get_controller(media_item.media_type)
1587        db_item = await ctrl.get_library_item_by_prov_id(media_item.item_id, media_item.provider)
1588        if db_item:
1589            await self.database.execute(
1590                f"UPDATE {ctrl.db_table} SET play_count = play_count - 1, "
1591                f"last_played = 0 WHERE item_id = {db_item.item_id}"
1592            )
1593            await self.database.commit()
1594
1595    @api_command("music/track_by_name", required_scope=Scope.LIBRARY_READ)
1596    async def get_track_by_name(
1597        self,
1598        track_name: str,
1599        artist_name: str | None = None,
1600        album_name: str | None = None,
1601        track_version: str | None = None,
1602    ) -> Track | None:
1603        """Get a track by its name, optionally with artist and album."""
1604        if track_version is None:
1605            track_name, version = parse_title_and_version(track_name)
1606        search_query = f"{artist_name} - {track_name}" if artist_name else track_name
1607        search_result = await self.mass.music.search(
1608            search_query=search_query,
1609            media_types=[MediaType.TRACK],
1610        )
1611        for allow_item_mapping in (False, True):
1612            for search_track in search_result.tracks:
1613                if not allow_item_mapping and not isinstance(search_track, Track):
1614                    continue
1615                if not compare_strings(track_name, search_track.name):
1616                    continue
1617                if not compare_version(version, search_track.version):
1618                    continue
1619                # check optional artist(s)
1620                if artist_name and isinstance(search_track, Track):
1621                    for artist in search_track.artists:
1622                        if compare_strings(artist_name, artist.name, False):
1623                            break
1624                    else:
1625                        # no artist match found: abort
1626                        continue
1627                # check optional album
1628                if album_name and isinstance(search_track, Track):
1629                    track_album = search_track.album
1630                    # a track without album info can never match a requested album
1631                    if track_album is None or not compare_strings(
1632                        album_name, track_album.name, False
1633                    ):
1634                        # no album match found: abort
1635                        continue
1636                # if we reach this, we found a match
1637                if not isinstance(search_track, Track):
1638                    # ensure we return an actual Track object
1639                    return await self.mass.music.tracks.get(
1640                        item_id=search_track.item_id,
1641                        provider_instance_id_or_domain=search_track.provider,
1642                    )
1643                return search_track
1644
1645        # try to handle case where something is appended to the title
1646        for splitter in ("•", "-", "|", "(", "["):
1647            if splitter in track_name:
1648                return await self.get_track_by_name(
1649                    track_name=track_name.split(splitter)[0].strip(),
1650                    artist_name=artist_name,
1651                    album_name=None,
1652                    track_version=track_version,
1653                )
1654        # try to handle case where multiple artists are given as single string
1655        if artist_name and (artists := split_artists(artist_name, True)) and len(artists) > 1:
1656            for single_artist in artists:
1657                return await self.get_track_by_name(
1658                    track_name=track_name,
1659                    artist_name=single_artist.split(splitter)[0].strip(),
1660                    album_name=None,
1661                    track_version=track_version,
1662                )
1663        # allow non-exact album match as fallback
1664        if album_name:
1665            return await self.get_track_by_name(
1666                track_name=track_name,
1667                artist_name=artist_name,
1668                album_name=None,
1669                track_version=track_version,
1670            )
1671        # no match found
1672        return None
1673
1674    async def get_resume_position(
1675        self, media_item: Audiobook | PodcastEpisode, userid: str | None = None
1676    ) -> tuple[bool, int]:
1677        """
1678        Get progress (resume point) details for the given audiobook or episode.
1679
1680        This is a separate call to ensure the resume position is always up-to-date
1681        and because many providers have this info present on a dedicated endpoint.
1682
1683        Will be called right before playback starts to ensure the resume position is correct.
1684
1685        Returns a boolean with the fully_played status
1686        and an integer with the resume position in ms.
1687        """
1688        provider_fully_played = False
1689        provider_position_ms = 0
1690        provider_timestamp: datetime | None = None
1691
1692        user: User | None = None
1693        if userid:
1694            # userid overridden by parameter
1695            user = await self.mass.webserver.auth.get_user(userid)
1696        elif session_user := get_current_user():
1697            # this is the active session user that triggered the action
1698            user = session_user
1699        elif provider_user := await self._get_user_for_provider(media_item.provider_mappings):
1700            # based on configured provider filter we can try to find a user
1701            user = provider_user
1702
1703        provider_instances = {x.provider_instance for x in media_item.provider_mappings}
1704        if user and user.provider_filter:
1705            # only if the user has provider filters configured
1706            # otherwise we allow all providers
1707            preferred_provider_instances = provider_instances.intersection(user.provider_filter)
1708        else:
1709            preferred_provider_instances = provider_instances
1710
1711        preferred_providers = [
1712            x
1713            for x in media_item.provider_mappings
1714            if x.provider_instance in preferred_provider_instances
1715        ]
1716
1717        # Try to get position from providers
1718        for prov_mapping in preferred_providers:
1719            if not (
1720                provider := self.mass.get_provider(
1721                    prov_mapping.provider_instance, provider_type=MusicProvider
1722                )
1723            ):
1724                continue
1725            with suppress(NotImplementedError):
1726                (
1727                    provider_fully_played,
1728                    provider_position_ms,
1729                    provider_timestamp,
1730                ) = await provider.get_resume_position(prov_mapping.item_id, media_item.media_type)
1731                break  # Use first provider that returns data
1732
1733        # Get MA's internal position from playlog
1734        ma_fully_played = False
1735        ma_position_ms = 0
1736        ma_timestamp = from_utc_timestamp(0)
1737        params = {
1738            "media_type": media_item.media_type.value,
1739            "item_id": media_item.item_id,
1740            "provider": media_item.provider,
1741        }
1742        if userid:
1743            params["userid"] = userid
1744        elif user:
1745            params["userid"] = user.user_id
1746        if db_entry := await self.database.get_row(DB_TABLE_PLAYLOG, params):
1747            ma_position_ms = db_entry["seconds_played"] * 1000 if db_entry["seconds_played"] else 0
1748            # fully_played is a nullable column; treat an unknown (NULL) value as not played
1749            ma_fully_played = parse_optional_bool(db_entry["fully_played"]) or False
1750            ma_timestamp = from_utc_timestamp(db_entry["timestamp"])
1751
1752        if provider_timestamp is not None and provider_timestamp > ma_timestamp:
1753            return provider_fully_played, provider_position_ms
1754        # Return the higher position to ensure users never lose progress
1755        if ma_position_ms >= provider_position_ms:
1756            return ma_fully_played, ma_position_ms
1757        return provider_fully_played, provider_position_ms
1758
1759    async def get_playback_speed(
1760        self, media_item: Audiobook | PodcastEpisode, userid: str | None = None
1761    ) -> float:
1762        """
1763        Get the stored playback speed for the given audiobook or podcast episode.
1764
1765        Returns 1.0 (normal speed) when no custom speed was stored for the item,
1766        or when no user can be determined to scope the lookup.
1767
1768        :param media_item: The audiobook or podcast episode to look up.
1769        :param userid: The user ID to look up the speed for (instead of the current user).
1770        """
1771        if not userid:
1772            if session_user := get_current_user():
1773                userid = session_user.user_id
1774            elif provider_user := await self._get_user_for_provider(media_item.provider_mappings):
1775                userid = provider_user.user_id
1776            else:
1777                # the speed is stored per user; without one we can't scope the lookup
1778                return 1.0
1779        db_entry = await self.database.get_row(
1780            DB_TABLE_PLAYLOG,
1781            {
1782                "item_id": media_item.item_id,
1783                "provider": media_item.provider,
1784                "media_type": media_item.media_type.value,
1785                "userid": userid,
1786            },
1787        )
1788        if db_entry and (stored_speed := db_entry["playback_speed"]) is not None:
1789            return float(stored_speed)
1790        return 1.0
1791
1792    def get_controller(
1793        self, media_type: MediaType
1794    ) -> (
1795        ArtistsController
1796        | AlbumsController
1797        | TracksController
1798        | RadioController
1799        | PlaylistController
1800        | AudiobooksController
1801        | PodcastsController
1802        | GenreController
1803    ):
1804        """Return controller for MediaType."""
1805        if media_type == MediaType.ARTIST:
1806            return self.artists
1807        if media_type == MediaType.ALBUM:
1808            return self.albums
1809        if media_type == MediaType.TRACK:
1810            return self.tracks
1811        if media_type == MediaType.RADIO:
1812            return self.radio
1813        if media_type == MediaType.PLAYLIST:
1814            return self.playlists
1815        if media_type == MediaType.AUDIOBOOK:
1816            return self.audiobooks
1817        if media_type == MediaType.PODCAST:
1818            return self.podcasts
1819        if media_type == MediaType.PODCAST_EPISODE:
1820            return self.podcasts
1821        if media_type == MediaType.GENRE:
1822            return self.genres
1823        raise NotImplementedError(
1824            f"No media controller available for media type: {media_type.value}"
1825        )
1826
1827    def get_controller_for_collection(
1828        self, item_id: str
1829    ) -> (
1830        ArtistsController
1831        | AlbumsController
1832        | TracksController
1833        | RadioController
1834        | PlaylistController
1835        | AudiobooksController
1836        | PodcastsController
1837        | GenreController
1838    ):
1839        """Return controller for MediaType."""
1840        media_type = get_collection_item_media_type_from_item_id(item_id)
1841        controller = self.get_controller(media_type)
1842        if not isinstance(controller, AudiobooksController):
1843            # currently only supported for audiobooks
1844            raise NotImplementedError(
1845                f"No media controller available for media type: {media_type.value}"
1846            )
1847        return controller
1848
1849    def get_provider_instances(
1850        self, domain: str, return_unavailable: bool = False
1851    ) -> list[MusicProvider]:
1852        """
1853        Return all provider instances for a given domain.
1854
1855        Note that this skips user filters so may only be called from internal code.
1856        """
1857        return cast(
1858            "list[MusicProvider]",
1859            self.mass.get_provider_instances(domain, return_unavailable, ProviderType.MUSIC),
1860        )
1861
1862    def get_unique_providers(self) -> list[str]:
1863        """
1864        Return all unique MusicProvider (instance or domain) ids.
1865
1866        This will return a set of provider instance ids but will only return
1867        a single instance_id per streaming provider domain.
1868
1869        Applies user provider filters (for non-admin users).
1870        """
1871        processed_domains: set[str] = set()
1872        # Get user provider filter if set
1873        user = get_current_user()
1874        user_provider_filter = user.provider_filter if user and user.provider_filter else None
1875        result: list[str] = []
1876        for provider in self.providers:
1877            if provider.is_streaming_provider and provider.domain in processed_domains:
1878                continue
1879            if user_provider_filter and provider.instance_id not in user_provider_filter:
1880                continue
1881            result.append(provider.instance_id)
1882            processed_domains.add(provider.domain)
1883        return result
1884
1885    def get_active_provider_instances(self) -> list[str]:
1886        """
1887        Return the instance ids of all currently loaded, available MusicProviders.
1888
1889        Unlike `get_unique_providers`, this keeps every instance of a streaming
1890        provider's domain instead of collapsing to one per domain, so a caller
1891        validating a specific requested provider instance id isn't shadowed by
1892        another instance of the same domain. Applies the current user's provider
1893        filter (via the `providers` property) and excludes providers that are
1894        loaded but not currently available.
1895        """
1896        return [provider.instance_id for provider in self.providers if provider.available]
1897
1898    async def cleanup_provider(self, provider_instance: str) -> None:
1899        """Cleanup provider records from the database."""
1900        deleted_providers = self.mass.config.get_raw_core_config_value(
1901            self.domain, CONF_DELETED_PROVIDERS, []
1902        )
1903        # we add the provider to this hidden config setting just to make sure that
1904        # we can survive this over a restart to make sure that entries are cleaned up
1905        if provider_instance not in deleted_providers:
1906            deleted_providers.append(provider_instance)
1907            self.mass.config.set_raw_core_config_value(
1908                self.domain, CONF_DELETED_PROVIDERS, deleted_providers
1909            )
1910            self.mass.config.save(True)
1911
1912        # always clear cache when a provider is removed
1913        await self.mass.cache.clear()
1914
1915        # cleanup media items from db matched to deleted provider
1916        self.logger.info(
1917            "Removing provider %s from library, this can take a a while...",
1918            provider_instance,
1919        )
1920        errors = 0
1921        # suppress the per-item MEDIA_ITEM_UPDATED events during this bulk removal so we
1922        # don't flood subscribers; they refresh once via the PROVIDERS_UPDATED event
1923        token = SUPPRESS_MEDIA_ITEM_UPDATES.set(True)
1924        try:
1925            for ctrl in (
1926                # order is important here to recursively cleanup bottom up
1927                self.mass.music.radio,
1928                self.mass.music.playlists,
1929                self.mass.music.tracks,
1930                self.mass.music.albums,
1931                self.mass.music.artists,
1932                self.mass.music.podcasts,
1933                self.mass.music.audiobooks,
1934                # run main controllers twice to rule out relations
1935                self.mass.music.tracks,
1936                self.mass.music.albums,
1937                self.mass.music.artists,
1938            ):
1939                query = (
1940                    f"SELECT item_id FROM {DB_TABLE_PROVIDER_MAPPINGS} "
1941                    "WHERE media_type = :media_type "
1942                    "AND provider_instance = :provider_instance"
1943                )
1944                params = {
1945                    "media_type": ctrl.media_type.value,
1946                    "provider_instance": provider_instance,
1947                }
1948                for db_row in await self.database.get_rows_from_query(query, params, limit=100000):
1949                    try:
1950                        await ctrl.remove_provider_mappings(db_row["item_id"], provider_instance)
1951                    except Exception as err:
1952                        # we dont want the whole removal process to stall on one item
1953                        # so in case of an unexpected error, we log and move on.
1954                        self.logger.warning(
1955                            "Error while removing %s: %s",
1956                            db_row["item_id"],
1957                            str(err),
1958                            exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
1959                        )
1960                        errors += 1
1961        finally:
1962            SUPPRESS_MEDIA_ITEM_UPDATES.reset(token)
1963
1964        # remove all orphaned items (not in provider mappings table anymore)
1965        query = (
1966            f"SELECT item_id FROM {DB_TABLE_PROVIDER_MAPPINGS} "
1967            f"WHERE provider_instance = '{provider_instance}'"
1968        )
1969        if remaining_items_count := await self.database.get_count_from_query(query):
1970            errors += remaining_items_count
1971
1972        # cleanup playlog table
1973        await self.mass.music.database.delete(
1974            DB_TABLE_PLAYLOG,
1975            {
1976                "provider": provider_instance,
1977            },
1978        )
1979
1980        if errors == 0:
1981            # cleanup successful, remove from the deleted_providers setting
1982            self.logger.info("Provider %s removed from library", provider_instance)
1983            deleted_providers.remove(provider_instance)
1984            self.mass.config.set_raw_core_config_value(
1985                self.domain, CONF_DELETED_PROVIDERS, deleted_providers
1986            )
1987        else:
1988            self.logger.warning(
1989                "Provider %s was not not fully removed from library", provider_instance
1990            )
1991
1992    async def schedule_provider_sync(self, provider_instance_id: str) -> None:
1993        """Schedule Library sync for given provider."""
1994        if not (
1995            provider := self.mass.get_provider(provider_instance_id, provider_type=MusicProvider)
1996        ):
1997            return
1998        await self.unschedule_provider_sync(provider.instance_id, clear_persisted_state=False)
1999        for media_type in MediaType:
2000            if not self.library_supported(provider, media_type):
2001                continue
2002            await self._schedule_provider_mediatype_sync(provider, media_type, True)
2003
2004    async def unschedule_provider_sync(
2005        self, provider_instance_id: str, clear_persisted_state: bool = True
2006    ) -> None:
2007        """
2008        Unschedule Library sync for given provider and wait for a running sync to stop.
2009
2010        Callers tear down provider state right after this (unloading the provider, or
2011        rescheduling its syncs), so all media types are cancelled first and then awaited
2012        together, keeping the bounded wait to one timeout instead of one per media type.
2013
2014        :param provider_instance_id: The provider instance id to unschedule.
2015        :param clear_persisted_state: Whether to remove persisted schedule state from config.
2016        """
2017        await asyncio.gather(
2018            *(
2019                self.mass.tasks.unregister_scheduled_task_and_wait(
2020                    self._get_sync_task_id(provider_instance_id, media_type),
2021                    clear_persisted_state=clear_persisted_state,
2022                )
2023                for media_type in MediaType
2024            )
2025        )
2026
2027    def get_provider_sync_schedule(
2028        self, provider_instance_id: str, media_type: MediaType
2029    ) -> TaskSchedule | None:
2030        """Return the effective schedule for a provider sync task, if any."""
2031        task_id = self._get_sync_task_id(provider_instance_id, media_type)
2032        with suppress(InvalidDataError):
2033            task = self.mass.tasks.get_task(task_id)
2034            return task.schedule
2035        if not (
2036            provider := self.mass.get_provider(provider_instance_id, provider_type=MusicProvider)
2037        ):
2038            return None
2039        if not self.library_supported(provider, media_type):
2040            return None
2041        return provider.get_default_library_sync_schedule(media_type)
2042
2043    def match_provider_instances(
2044        self,
2045        item: MediaItemType,
2046    ) -> bool:
2047        """Match all provider instances for the given item."""
2048        mappings_added = False
2049        for provider_mapping in list(item.provider_mappings):
2050            if provider_mapping.is_unique:
2051                # unique mapping, no need to map
2052                continue
2053            if not (provider := self.mass.get_provider(provider_mapping.provider_instance)):
2054                continue
2055            if not isinstance(provider, MusicProvider):
2056                continue
2057            if not provider.is_streaming_provider:
2058                continue
2059            provider_instances = self.get_provider_instances(
2060                provider.domain, return_unavailable=True
2061            )
2062            if len(provider_instances) <= 1:
2063                # only a single instance, no need to map
2064                continue
2065            for prov_instance in provider_instances:
2066                if prov_instance.instance_id == provider.instance_id:
2067                    continue
2068                if any(
2069                    pm.provider_instance == prov_instance.instance_id
2070                    for pm in item.provider_mappings
2071                ):
2072                    # mapping already exists
2073                    continue
2074                # create additional mapping for other provider instances of the same provider
2075                item.provider_mappings.add(
2076                    ProviderMapping(
2077                        item_id=provider_mapping.item_id,
2078                        provider_domain=provider.domain,
2079                        provider_instance=prov_instance.instance_id,
2080                        available=provider_mapping.available,
2081                        is_unique=provider_mapping.is_unique,
2082                        audio_format=provider_mapping.audio_format,
2083                        url=provider_mapping.url,
2084                        details=provider_mapping.details,
2085                        in_library=None,
2086                    )
2087                )
2088                mappings_added = True
2089        return mappings_added
2090
2091    @api_command("music/add_provider_mapping", required_scope=Scope.LIBRARY_MANAGE)
2092    async def add_provider_mapping(
2093        self, media_type: MediaType, db_id: str, mapping: ProviderMapping
2094    ) -> None:
2095        """Add provider mapping to the given library item."""
2096        ctrl = self.get_controller(media_type)
2097        await ctrl.add_provider_mappings(db_id, [mapping])
2098
2099    @api_command("music/remove_provider_mapping", required_scope=Scope.LIBRARY_MANAGE)
2100    async def remove_provider_mapping(
2101        self, media_type: MediaType, db_id: str, mapping: ProviderMapping
2102    ) -> None:
2103        """Remove provider mapping from the given library item."""
2104        ctrl = self.get_controller(media_type)
2105        await ctrl.remove_provider_mapping(db_id, mapping.provider_instance, mapping.item_id)
2106
2107    @api_command("music/match_providers", required_scope=Scope.LIBRARY_MANAGE)
2108    async def match_providers(self, media_type: MediaType, db_id: str) -> None:
2109        """Search for mappings on all providers for the given library item."""
2110        ctrl = self.get_controller(media_type)
2111        db_item = await ctrl.get_library_item(db_id)
2112        # ctrl is chosen by media_type, so it matches db_item's runtime type
2113        await cast("MediaControllerBase[MediaItemType]", ctrl).match_providers(db_item)
2114
2115    async def update_provider_mapping(
2116        self,
2117        media_type: MediaType,
2118        db_id: str | int,
2119        provider_instance_id: str,
2120        provider_item_id: str,
2121        *,
2122        available: bool | Any = UNSET,
2123        in_library: bool | Any = UNSET,
2124        is_unique: bool | None | Any = UNSET,
2125        url: str | None | Any = UNSET,
2126        details: str | None | Any = UNSET,
2127        audio_format: AudioFormat | Any = UNSET,
2128    ) -> None:
2129        """Update an existing provider mapping for a library item."""
2130        ctrl = self.get_controller(media_type)
2131        await ctrl.update_provider_mapping(
2132            item_id=db_id,
2133            provider_instance_id=provider_instance_id,
2134            provider_item_id=provider_item_id,
2135            available=available,
2136            in_library=in_library,
2137            is_unique=is_unique,
2138            url=url,
2139            details=details,
2140            audio_format=audio_format,
2141        )
2142
2143    def queue_provider_mapping_correction_task(self) -> BackgroundTask:
2144        """Queue the provider mapping correction as a managed background task."""
2145        self._register_provider_mapping_correction_task()
2146        return self.mass.tasks.run_task(PROVIDER_MAPPING_CORRECTION_TASK_ID)
2147
2148    async def correct_multi_instance_provider_mappings(self) -> None:
2149        """Correct provider mappings for multi-instance providers."""
2150        self.logger.debug("Correcting provider mappings for multi-instance providers...")
2151        multi_instance_providers: set[str] = set()
2152        for provider in self.providers:
2153            if len(self.get_provider_instances(provider.domain)) > 1:
2154                multi_instance_providers.add(provider.instance_id)
2155        if not multi_instance_providers:
2156            return  # no multi-instance providers found, nothing to do
2157
2158        for ctrl in (
2159            self.albums,
2160            self.artists,
2161            self.tracks,
2162            self.playlists,
2163            self.radio,
2164            self.audiobooks,
2165            self.podcasts,
2166        ):
2167            async for db_item in ctrl.iter_library_items(
2168                provider=list(multi_instance_providers), library_items_only=False
2169            ):
2170                if self.match_provider_instances(db_item):
2171                    # ctrl is the per-type controller, so it matches db_item's runtime type
2172                    await cast("MediaControllerBase[MediaItemType]", ctrl).update_item_in_library(
2173                        db_item.item_id, db_item
2174                    )
2175                # prevent overwhelming the event loop
2176                await asyncio.sleep(0.2)
2177        self.logger.debug("Provider mappings correction done")
2178
2179    def library_supported(self, provider: Provider, media_type: MediaType) -> bool:
2180        """Return whether the provider declares LIBRARY support for the given media type."""
2181        if provider.type != ProviderType.MUSIC:
2182            return False
2183        if media_type == MediaType.ARTIST:
2184            return provider.supports_feature(ProviderFeature.LIBRARY_ARTISTS)
2185        if media_type == MediaType.ALBUM:
2186            return provider.supports_feature(ProviderFeature.LIBRARY_ALBUMS)
2187        if media_type == MediaType.TRACK:
2188            return provider.supports_feature(ProviderFeature.LIBRARY_TRACKS)
2189        if media_type == MediaType.PLAYLIST:
2190            return provider.supports_feature(ProviderFeature.LIBRARY_PLAYLISTS)
2191        if media_type == MediaType.RADIO:
2192            return provider.supports_feature(ProviderFeature.LIBRARY_RADIOS)
2193        if media_type == MediaType.AUDIOBOOK:
2194            return provider.supports_feature(ProviderFeature.LIBRARY_AUDIOBOOKS)
2195        if media_type == MediaType.PODCAST:
2196            return provider.supports_feature(ProviderFeature.LIBRARY_PODCASTS)
2197        return False
2198
2199    def library_edit_supported(self, provider: Provider, media_type: MediaType) -> bool:
2200        """Return whether the provider supports library add/remove for the given media type."""
2201        if provider.type != ProviderType.MUSIC:
2202            return False
2203        if media_type == MediaType.ARTIST:
2204            return provider.supports_feature(ProviderFeature.LIBRARY_ARTISTS_EDIT)
2205        if media_type == MediaType.ALBUM:
2206            return provider.supports_feature(ProviderFeature.LIBRARY_ALBUMS_EDIT)
2207        if media_type == MediaType.TRACK:
2208            return provider.supports_feature(ProviderFeature.LIBRARY_TRACKS_EDIT)
2209        if media_type == MediaType.PLAYLIST:
2210            return provider.supports_feature(ProviderFeature.LIBRARY_PLAYLISTS_EDIT)
2211        if media_type == MediaType.RADIO:
2212            return provider.supports_feature(ProviderFeature.LIBRARY_RADIOS_EDIT)
2213        if media_type == MediaType.AUDIOBOOK:
2214            return provider.supports_feature(ProviderFeature.LIBRARY_AUDIOBOOKS_EDIT)
2215        if media_type == MediaType.PODCAST:
2216            return provider.supports_feature(ProviderFeature.LIBRARY_PODCASTS_EDIT)
2217        return False
2218
2219    def library_favorites_edit_supported(self, provider: Provider, media_type: MediaType) -> bool:
2220        """Return whether the provider supports favorites add/remove for the given media type."""
2221        if provider.type != ProviderType.MUSIC:
2222            return False
2223        if media_type == MediaType.ARTIST:
2224            return provider.supports_feature(ProviderFeature.FAVORITE_ARTISTS_EDIT)
2225        if media_type == MediaType.ALBUM:
2226            return provider.supports_feature(ProviderFeature.FAVORITE_ALBUMS_EDIT)
2227        if media_type == MediaType.TRACK:
2228            return provider.supports_feature(ProviderFeature.FAVORITE_TRACKS_EDIT)
2229        if media_type == MediaType.PLAYLIST:
2230            return provider.supports_feature(ProviderFeature.FAVORITE_PLAYLISTS_EDIT)
2231        if media_type == MediaType.RADIO:
2232            return provider.supports_feature(ProviderFeature.FAVORITE_RADIOS_EDIT)
2233        if media_type == MediaType.AUDIOBOOK:
2234            return provider.supports_feature(ProviderFeature.FAVORITE_AUDIOBOOKS_EDIT)
2235        if media_type == MediaType.PODCAST:
2236            return provider.supports_feature(ProviderFeature.FAVORITE_PODCASTS_EDIT)
2237        return False
2238
2239    def library_sync_back_enabled(self, provider: Provider, media_type: MediaType) -> bool:
2240        """Return whether library sync back is enabled for the provider+media_type."""
2241        conf_value = provider.config.get_value(
2242            CONF_ENTRY_LIBRARY_SYNC_BACK.key, CONF_ENTRY_LIBRARY_SYNC_BACK.default_value
2243        )
2244        return bool(conf_value)
2245
2246    @api_command("music/item_by_name", required_scope=Scope.LIBRARY_READ, allow_impersonation=True)
2247    async def get_item_by_name(
2248        self,
2249        name: str,
2250        artist: str | None = None,
2251        album: str | None = None,
2252        media_type: MediaType | None = None,
2253    ) -> MediaItemType | ItemMapping | None:
2254        """Try to find a media item (such as a playlist) by name."""
2255        return await self._get_item_by_name(name, artist, album, media_type)
2256
2257    @api_command(
2258        "music/verify_item_uri", required_scope=Scope.LIBRARY_READ, allow_impersonation=True
2259    )
2260    async def verify_item_uri(self, uri: str) -> bool:
2261        """
2262        Verify whether a uri points to a valid, accessible item.
2263
2264        :param uri: The uri to verify.
2265        """
2266        return await self._handle_verify_item_uri(uri)
2267
2268    def _apply_user_provider_filter(
2269        self,
2270        providers: Iterable[ProviderInstanceType],
2271    ) -> list[ProviderInstanceType]:
2272        """Filter providers by the current user's music provider filter."""
2273        user = get_current_user()
2274        user_provider_filter = user.provider_filter if user else None
2275        if not user_provider_filter:
2276            return list(providers)
2277        return [
2278            p
2279            for p in providers
2280            if p.type != ProviderType.MUSIC or p.instance_id in user_provider_filter
2281        ]
2282
2283    async def _search_shareable_url(self, search_query: str) -> SearchResults | None:
2284        """
2285        Handle a search query that is a streaming provider public shareable URL.
2286
2287        Returns None if the query is not such a URL and a regular search must be done.
2288        """
2289        try:
2290            media_type, provider_instance_id_or_domain, item_id = await parse_uri(
2291                search_query, validate_id=True
2292            )
2293        except InvalidProviderURI:
2294            return None
2295        except InvalidProviderID as err:
2296            self.logger.warning("%s", str(err))
2297            return SearchResults()
2298        if provider_instance_id_or_domain not in PROVIDERS_WITH_SHAREABLE_URLS:
2299            return None
2300        try:
2301            item = await self.get_item(
2302                media_type=media_type,
2303                item_id=item_id,
2304                provider_instance_id_or_domain=provider_instance_id_or_domain,
2305            )
2306        except MusicAssistantError as err:
2307            self.logger.warning("%s", str(err))
2308            return SearchResults()
2309        if media_type == MediaType.ARTIST:
2310            return SearchResults(artists=[cast("Artist", item)])
2311        if media_type == MediaType.ALBUM:
2312            return SearchResults(albums=[cast("Album", item)])
2313        if media_type == MediaType.TRACK:
2314            return SearchResults(tracks=[cast("Track", item)])
2315        if media_type == MediaType.PLAYLIST:
2316            return SearchResults(playlists=[cast("Playlist", item)])
2317        if media_type == MediaType.AUDIOBOOK:
2318            return SearchResults(audiobooks=[cast("Audiobook", item)])
2319        if media_type == MediaType.PODCAST:
2320            return SearchResults(podcasts=[cast("Podcast", item)])
2321        return SearchResults()
2322
2323    async def _search_provider(
2324        self,
2325        search_query: str,
2326        provider_instance_id_or_domain: str,
2327        media_types: list[MediaType],
2328        limit: int = 10,
2329        skip_item_ids: set[tuple[MediaType, str, str]] | None = None,
2330    ) -> SearchResults | None:
2331        """
2332        Perform search on given provider, returns None if the search failed or timed out.
2333
2334        :param search_query: Search query
2335        :param provider_instance_id_or_domain: instance_id or domain of the provider
2336                                               to perform the search on.
2337        :param media_types: A list of media_types to include.
2338        :param limit: number of items to return in the search (per type).
2339        :param skip_item_ids: Optional set of (media_type, provider_domain, item_id)
2340                              tuples to filter out of the results.
2341        """
2342        prov = self.mass.get_provider(provider_instance_id_or_domain, provider_type=MusicProvider)
2343        if not prov:
2344            return SearchResults()
2345        if ProviderFeature.SEARCH not in prov.supported_features:
2346            return SearchResults()
2347
2348        # create safe search string
2349        search_query = search_query.replace("/", " ").replace("'", "")
2350        # use the per-provider cache so repeated and overlapping searches
2351        # do not hit the provider again
2352        cache_key = f"{search_query}-{'-'.join(sorted([mt.value for mt in media_types]))}-{limit}"
2353        if (
2354            cache := await self.mass.cache.get(
2355                key=cache_key,
2356                provider=prov.instance_id,
2357                category=CACHE_CATEGORY_SEARCH_RESULTS,
2358                base_class=SearchResults,
2359            )
2360        ) is not None:
2361            return filter_search_results(cast("SearchResults", cache), prov.domain, skip_item_ids)
2362        # run the provider search as a separate task (deduplicated by task_id so
2363        # identical concurrent searches share a single provider call) and wait for
2364        # it a limited amount of time only: a slow provider then contributes no
2365        # results now, while its search continues in the background so the result
2366        # is cached and available for a next search request
2367        task = self.mass.create_task(
2368            self._execute_provider_search(prov, search_query, media_types, limit, cache_key),
2369            task_id=f"provider_search_{prov.instance_id}_{cache_key}",
2370        )
2371        try:
2372            async with asyncio.timeout(SEARCH_PROVIDER_SOFT_TIMEOUT):
2373                prov_search_results = await asyncio.shield(task)
2374        except TimeoutError:
2375            self.logger.warning(
2376                "Search on provider %s did not return in time, "
2377                "the search continues in the background",
2378                prov.name,
2379            )
2380            return None
2381        if prov_search_results is None:
2382            return None
2383        return filter_search_results(prov_search_results, prov.domain, skip_item_ids)
2384
2385    async def _execute_provider_search(
2386        self,
2387        prov: MusicProvider,
2388        search_query: str,
2389        media_types: list[MediaType],
2390        limit: int,
2391        cache_key: str,
2392    ) -> SearchResults | None:
2393        """
2394        Execute the actual search on a provider and cache the result.
2395
2396        Returns None if the provider search failed or timed out. All errors are
2397        handled here (and not raised) as this coroutine runs as a background task
2398        that may outlive the request that started it.
2399        """
2400        try:
2401            async with asyncio.timeout(SEARCH_PROVIDER_HARD_TIMEOUT):
2402                result = await prov.search(search_query, media_types, limit)
2403        except TimeoutError:
2404            self.logger.warning("Search on provider %s timed out", prov.name)
2405            return None
2406        except MusicAssistantError as err:
2407            self.logger.warning("Search on provider %s failed: %s", prov.name, str(err))
2408            return None
2409        except Exception as err:
2410            self.logger.error("Search on provider %s failed: %s", prov.name, str(err), exc_info=err)
2411            return None
2412        # only successful results are cached, so failed or timed out
2413        # provider searches are simply retried on a next search
2414        await self._cache_search_results(
2415            cache_key,
2416            result,
2417            # plugin providers do not declare is_streaming_provider,
2418            # treat them as local so their results only get the short expiration
2419            SEARCH_CACHE_EXPIRATION_STREAMING_PROVIDER
2420            if getattr(prov, "is_streaming_provider", False)
2421            else SEARCH_CACHE_EXPIRATION_LOCAL_PROVIDER,
2422            prov.instance_id,
2423        )
2424        return result
2425
2426    async def _cache_search_results(
2427        self, cache_key: str, result: SearchResults, expiration: int, provider: str
2428    ) -> None:
2429        """Store search results in the cache, logging (instead of raising) any cache errors."""
2430        try:
2431            await self.mass.cache.set(
2432                key=cache_key,
2433                data=result.to_dict(),
2434                expiration=expiration,
2435                provider=provider,
2436                category=CACHE_CATEGORY_SEARCH_RESULTS,
2437            )
2438        except Exception as err:
2439            self.logger.warning("Failed to cache search results for %s: %s", provider, str(err))
2440
2441    def _get_covered_media_types(
2442        self, library_results: SearchResults, search_query: str
2443    ) -> set[tuple[MediaType, str]]:
2444        """
2445        Return the (media_type, provider domain/instance) pairs covered by the library.
2446
2447        A pair is considered covered when the library holds a (near) exact name match
2448        for the search query that is mapped to that provider.
2449        """
2450        covered: set[tuple[MediaType, str]] = set()
2451        # extract the artist and title part in case the
2452        # query is formatted as "artist - title"
2453        if " - " in search_query:
2454            artist_part, title_part = search_query.split(" - ", 1)
2455        else:
2456            artist_part, title_part = None, search_query
2457        items: Sequence[MediaItemType | ItemMapping]
2458        for items in (
2459            library_results.artists,
2460            library_results.albums,
2461            library_results.tracks,
2462            library_results.playlists,
2463            library_results.radio,
2464            library_results.audiobooks,
2465            library_results.podcasts,
2466        ):
2467            for item in items:
2468                if compare_strings(item.name, search_query, strict=False):
2469                    pass
2470                elif artist_part and compare_strings(item.name, title_part, strict=False):
2471                    # the item name matches the title part only,
2472                    # so the artist part must match one of the item artists
2473                    if not any(
2474                        compare_strings(artist.name, artist_part, strict=False)
2475                        for artist in getattr(item, "artists", [])
2476                    ):
2477                        continue
2478                else:
2479                    continue
2480                for prov_mapping in cast("MediaItemType", item).provider_mappings:
2481                    if not prov_mapping.available:
2482                        continue
2483                    covered.add((item.media_type, prov_mapping.provider_domain))
2484                    covered.add((item.media_type, prov_mapping.provider_instance))
2485        return covered
2486
2487    def _import_album_tracks_if_enabled(self, album: Album) -> None:
2488        """Import all album tracks into the library for providers that have this enabled."""
2489        for prov_mapping in album.provider_mappings:
2490            # only consider mappings the album was actually added on; additional
2491            # mappings auto-created for other instances of the same provider
2492            # (via match_provider_instances) carry in_library=None and must be skipped
2493            if not prov_mapping.in_library:
2494                continue
2495            provider = self.mass.get_provider(prov_mapping.provider_instance)
2496            if not isinstance(provider, MusicProvider):
2497                continue
2498            if not provider.library_sync_album_tracks_enabled():
2499                continue
2500            self.mass.create_task(provider.import_album_tracks(prov_mapping.item_id, album.name))
2501
2502    async def _get_provider_sound_effects(self, provider: MusicProvider) -> list[SoundEffect]:
2503        """Return all sound effect items from a single provider."""
2504        try:
2505            return [item async for item in provider.get_sound_effects()]
2506        except Exception as err:
2507            self.logger.warning(
2508                "Error while fetching sound effects from %s: %s",
2509                provider.name,
2510                str(err),
2511                exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
2512            )
2513            return []
2514
2515    def _create_provider_sync_handler(
2516        self, provider: MusicProvider, media_type: MediaType
2517    ) -> Callable[[], Awaitable[None]]:
2518        """Create the coroutine used for a managed provider sync task."""
2519
2520        async def run_sync() -> None:
2521            try:
2522                async with self._sync_lock:
2523                    # suppress per-item events during sync; a large library would otherwise
2524                    # emit one (serialized per client) for every item. Subscribers refresh
2525                    # on MUSIC_SYNC_COMPLETED and track progress via TASKS_UPDATED instead.
2526                    token = SUPPRESS_MEDIA_ITEM_UPDATES.set(True)
2527                    try:
2528                        await provider.sync_library(media_type)
2529                    finally:
2530                        SUPPRESS_MEDIA_ITEM_UPDATES.reset(token)
2531            finally:
2532                self.mass.call_later(
2533                    0,
2534                    self._handle_sync_completion_check,
2535                    task_id=MUSIC_SYNC_COMPLETION_CHECK_TASK_ID,
2536                )
2537
2538        return run_sync
2539
2540    def _get_sync_task_id(self, provider: MusicProvider | str, media_type: MediaType) -> str:
2541        """Return deterministic task id for a provider sync."""
2542        provider_instance = (
2543            provider.instance_id if isinstance(provider, MusicProvider) else provider
2544        )
2545        return f"music_sync_{provider_instance}_{media_type.value}"
2546
2547    def _get_sync_task_name(self, provider: MusicProvider, media_type: MediaType) -> str:
2548        """Return display name for a provider sync task."""
2549        return f"Sync {provider.name} {media_type.value}s"
2550
2551    def _get_sync_task_translation_key(self, media_type: MediaType) -> str:
2552        """Return translation key for a provider sync task."""
2553        if media_type == MediaType.ARTIST:
2554            return "sync_provider_artists"
2555        if media_type == MediaType.ALBUM:
2556            return "sync_provider_albums"
2557        if media_type == MediaType.TRACK:
2558            return "sync_provider_tracks"
2559        if media_type == MediaType.PLAYLIST:
2560            return "sync_provider_playlists"
2561        if media_type == MediaType.RADIO:
2562            return "sync_provider_radios"
2563        if media_type == MediaType.AUDIOBOOK:
2564            return "sync_provider_audiobooks"
2565        if media_type == MediaType.PODCAST:
2566            return "sync_provider_podcasts"
2567        return "settings.sync"
2568
2569    def _get_sync_task_metadata(
2570        self, provider: MusicProvider, media_type: MediaType
2571    ) -> TaskMetadata:
2572        """Return metadata for a provider sync task."""
2573        return {
2574            "task_domain": "music_sync",
2575            "provider_domain": provider.domain,
2576            "provider_instance": provider.instance_id,
2577            "provider_name": provider.name,
2578            "media_type": media_type.value,
2579        }
2580
2581    def _handle_sync_completion_check(self) -> None:
2582        """Run follow-up maintenance when no provider sync tasks remain active."""
2583        if self.active_sync_tasks:
2584            return
2585        self.mass.signal_event(EventType.MUSIC_SYNC_COMPLETED)
2586        self._queue_database_cleanup_task()
2587
2588    def _register_database_cleanup_task(self) -> BackgroundTask:
2589        """Register the recurring database cleanup background task."""
2590        utc_hour, utc_minute = local_clock_time_to_utc(5, 0)
2591        desired_schedule = TaskSchedule.daily(hour=utc_hour, minute=utc_minute)
2592        return self.mass.tasks.register_scheduled_task(
2593            task_id=DATABASE_CLEANUP_TASK_ID,
2594            name="Database cleanup",
2595            handler=self._cleanup_database,
2596            schedule=desired_schedule,
2597            translation_key="database_cleanup",
2598            translation_owner=self.translation_owner,
2599            metadata={
2600                "task_domain": "music_database_cleanup",
2601            },
2602            allow_retry=True,
2603        )
2604
2605    def _register_provider_mapping_correction_task(self) -> BackgroundTask:
2606        """Register the recurring provider mapping correction background task."""
2607        utc_hour, utc_minute = local_clock_time_to_utc(4, 0)
2608        desired_schedule = TaskSchedule.daily(every=30, hour=utc_hour, minute=utc_minute)
2609        return self.mass.tasks.register_scheduled_task(
2610            task_id=PROVIDER_MAPPING_CORRECTION_TASK_ID,
2611            name="Correct provider mappings",
2612            handler=self.correct_multi_instance_provider_mappings,
2613            schedule=desired_schedule,
2614            translation_key="correct_provider_mappings",
2615            translation_owner=self.translation_owner,
2616            metadata={
2617                "task_domain": "music_provider_mapping_correction",
2618            },
2619            allow_retry=True,
2620        )
2621
2622    def _queue_database_cleanup_task(self) -> BackgroundTask:
2623        """Queue the post-sync database cleanup as a managed background task."""
2624        self._register_database_cleanup_task()
2625        return self.mass.tasks.run_task(DATABASE_CLEANUP_TASK_ID)
2626
2627    async def _schedule_provider_mediatype_sync(
2628        self, provider: MusicProvider, media_type: MediaType, is_initial: bool = False
2629    ) -> None:
2630        """Schedule Library sync for given provider and media type."""
2631        # handle mediatype specific sync config
2632        conf_key = f"library_sync_{media_type}s"
2633        sync_conf: ConfigValueType = await self.mass.config.get_provider_config_value(
2634            provider.instance_id, conf_key
2635        )
2636        if not sync_conf:
2637            self.mass.tasks.unregister_scheduled_task(self._get_sync_task_id(provider, media_type))
2638            return
2639        self.mass.tasks.register_scheduled_task(
2640            task_id=self._get_sync_task_id(provider, media_type),
2641            name=self._get_sync_task_name(provider, media_type),
2642            handler=self._create_provider_sync_handler(provider, media_type),
2643            schedule=provider.get_default_library_sync_schedule(media_type),
2644            initial_delay=10 if is_initial else None,
2645            translation_key=self._get_sync_task_translation_key(media_type),
2646            translation_args=[provider.name],
2647            translation_owner=self.translation_owner,
2648            metadata=self._get_sync_task_metadata(provider, media_type),
2649            allow_retry=True,
2650        )
2651
2652    async def _get_user_for_provider(
2653        self, provider_mappings_or_instance_id: Iterable[ProviderMapping] | str
2654    ) -> User | None:
2655        """Try to get the MA User based on provider mappings and provider filter."""
2656        all_users = await self.mass.webserver.auth.list_users()
2657        for mapping_or_instance_id in provider_mappings_or_instance_id:
2658            for user in all_users:
2659                if not user.provider_filter:
2660                    continue
2661                if isinstance(mapping_or_instance_id, str):
2662                    if provider_mappings_or_instance_id in user.provider_filter:
2663                        return user
2664                elif mapping_or_instance_id.provider_instance in user.provider_filter:
2665                    return user
2666        return None
2667
2668    async def _upsert_playlog(self, entry: dict[str, Any]) -> None:
2669        """
2670        Write a playlog row, updating the existing row for the item/user if there is one.
2671
2672        Columns left out of the entry keep whatever the existing row holds, and
2673        `user_initiated` is sticky: once a play was explicitly user-initiated it stays that
2674        way for the lifetime of the row, so a later side-effect credit (an autoplay replay,
2675        or a track crediting its album/artist) can never demote it and drop the item out of
2676        the "recently played" recommendations.
2677
2678        The generic `database.upsert()` cannot express either half of that: the sticky OR is
2679        playlog-specific, and it needs an explicit conflict target because the playlog carries
2680        more than one unique constraint.
2681
2682        :param entry: The playlog column values to write, including all of
2683            `PLAYLOG_CONFLICT_KEYS`.
2684        """
2685        columns = list(entry)
2686        updates = [
2687            f"user_initiated = {DB_TABLE_PLAYLOG}.user_initiated OR excluded.user_initiated"
2688            if column == "user_initiated"
2689            else f"{column} = excluded.{column}"
2690            for column in columns
2691            if column not in PLAYLOG_CONFLICT_KEYS
2692        ]
2693        await self.database.execute_write(
2694            f"INSERT INTO {DB_TABLE_PLAYLOG} ({', '.join(columns)}) "
2695            f"VALUES ({', '.join(f':{column}' for column in columns)}) "
2696            f"ON CONFLICT({', '.join(PLAYLOG_CONFLICT_KEYS)}) DO UPDATE SET {', '.join(updates)}",
2697            entry,
2698        )
2699
2700    async def _credit_artist_plays(
2701        self,
2702        artists: Iterable[Artist | ItemMapping],
2703        *,
2704        timestamp: float,
2705        user_ids: list[str],
2706        queue_id: str | None,
2707        skip_ids: set[str],
2708    ) -> None:
2709        """Credit each (library-resolvable) artist with a play, skipping skip_ids."""
2710        for artist in artists:
2711            db_artist = await self.artists.get_library_item_by_prov_id(
2712                artist.item_id, artist.provider
2713            )
2714            if db_artist is None:
2715                continue
2716            if db_artist.item_id in skip_ids:
2717                self.logger.debug("Skipping already-credited artist '%s'", db_artist.name)
2718                continue
2719            await self.database.execute(
2720                f"UPDATE {self.artists.db_table} SET play_count = play_count + 1, "
2721                f"last_played = {timestamp} WHERE item_id = {db_artist.item_id}"
2722            )
2723            self.logger.debug("Credited play for artist '%s'", db_artist.name)
2724            playlog_entry: dict[str, Any] = {
2725                "item_id": db_artist.item_id,
2726                "provider": "library",
2727                "media_type": MediaType.ARTIST.value,
2728                "name": db_artist.name,
2729                "image": serialize_to_json(db_artist.image.to_dict()) if db_artist.image else None,
2730                "fully_played": True,
2731                "seconds_played": None,
2732                "timestamp": timestamp,
2733                "queue_id": queue_id,
2734                "user_initiated": False,
2735            }
2736            for user_id in user_ids:
2737                playlog_entry["userid"] = user_id
2738                await self._upsert_playlog(playlog_entry)
2739
2740    async def _credit_podcast_play(
2741        self,
2742        podcast: Podcast | ItemMapping,
2743        *,
2744        timestamp: float,
2745        user_ids: list[str],
2746        queue_id: str | None,
2747    ) -> None:
2748        """Credit the parent podcast with a play so the show surfaces in recently played."""
2749        # Resolve to the library item first, like _credit_artist_plays does, so an episode's
2750        # parent-podcast credit lands on the same library-scoped row as an explicit play of the
2751        # library show, instead of creating a separate provider-scoped duplicate.
2752        db_podcast = await self.podcasts.get_library_item_by_prov_id(
2753            podcast.item_id, podcast.provider
2754        )
2755        credited_podcast: Podcast | ItemMapping = db_podcast if db_podcast else podcast
2756        playlog_entry: dict[str, Any] = {
2757            "item_id": credited_podcast.item_id,
2758            "provider": "library" if db_podcast else podcast.provider,
2759            "media_type": MediaType.PODCAST.value,
2760            "name": credited_podcast.name,
2761            "image": serialize_to_json(credited_podcast.image.to_dict())
2762            if credited_podcast.image
2763            else None,
2764            "fully_played": True,
2765            "seconds_played": None,
2766            "timestamp": timestamp,
2767            "queue_id": queue_id,
2768            "user_initiated": False,
2769        }
2770        for user_id in user_ids:
2771            playlog_entry["userid"] = user_id
2772            await self._upsert_playlog(playlog_entry)
2773
2774    async def _get_item_by_name(
2775        self,
2776        name: str,
2777        artist: str | None = None,
2778        album: str | None = None,
2779        media_type: MediaType | None = None,
2780    ) -> MediaItemType | ItemMapping | None:
2781        """Try to find a media item (such as a playlist) by name."""
2782        # Future todo: enhance this method with AI capabilities to allow typos and
2783        # natural language.
2784        searchname = name.lower()
2785        allowed_media_types = [
2786            MediaType.PLAYLIST,
2787            MediaType.RADIO,
2788            MediaType.TRACK,
2789            MediaType.ALBUM,
2790            MediaType.ARTIST,
2791            MediaType.AUDIOBOOK,
2792            MediaType.PODCAST,
2793        ]
2794        if media_type in (None, MediaType.UNKNOWN):
2795            media_types = allowed_media_types
2796        elif media_type not in allowed_media_types:
2797            raise InvalidDataError(
2798                f"{media_type} is not a supported media_type. "
2799                f"Supported media_types are {allowed_media_types}"
2800            )
2801        else:
2802            media_types = [media_type]
2803        library_functions = [
2804            self.get_controller(media_type).library_items for media_type in media_types
2805        ]
2806        # prefer (exact) lookup in the library by name
2807        for func in library_functions:
2808            result = await func(search=searchname)
2809            for item in result:
2810                # handle optional artist filter
2811                if (
2812                    artist
2813                    and (artists := getattr(item, "artists", None))
2814                    and not any(x for x in artists if x.name.lower() == artist.lower())
2815                ):
2816                    continue
2817                # handle optional album filter
2818                if (
2819                    album
2820                    and (item_album := getattr(item, "album", None))
2821                    and item_album.name.lower() != album.lower()
2822                ):
2823                    continue
2824                if searchname == item.name.lower():
2825                    return item
2826        # nothing found in the library, fallback to global search
2827        search_name = name
2828        if album and artist:
2829            search_name = f"{artist} - {album} - {name}"
2830        elif album:
2831            search_name = f"{album} - {name}"
2832        elif artist:
2833            search_name = f"{artist} - {name}"
2834        search_results = await self.search(
2835            search_query=search_name,
2836            media_types=[media_type]
2837            if media_type and media_type != MediaType.UNKNOWN
2838            else MediaType.ALL,
2839            limit=8,
2840        )
2841        for results in (
2842            search_results.tracks,
2843            search_results.albums,
2844            search_results.playlists,
2845            search_results.artists,
2846            search_results.radio,
2847            search_results.audiobooks,
2848            search_results.podcasts,
2849        ):
2850            for _item in results:
2851                # simply return the first item because search is already sorted by best match
2852                return _item
2853        return None
2854
2855    async def _handle_verify_item_uri(self, uri: str) -> bool:
2856        user = get_current_user()
2857
2858        try:
2859            media_type, provider_instance_id_or_domain, item_id = await parse_uri(uri)
2860        except InvalidProviderURI, InvalidProviderID:
2861            return False
2862
2863        # fast return for a provider uri which is not part of a user with a provider filter
2864        if (
2865            provider_instance_id_or_domain != "library"
2866            and user
2867            and user.provider_filter
2868            and provider_instance_id_or_domain not in user.provider_filter
2869        ):
2870            return False
2871
2872        # verify that item itself exists
2873        try:
2874            item = await self.get_item(
2875                media_type=media_type,
2876                item_id=item_id,
2877                provider_instance_id_or_domain=provider_instance_id_or_domain,
2878                allow_update_metadata=False,  # no need trigger more methods
2879            )
2880        except MediaNotFoundError, NotImplementedError:
2881            # NotImplementedError: the uri has a valid format, but specifies an unknown media type
2882            return False
2883
2884        # non library item handling for users with no filter, or no user at all
2885        if (
2886            provider_instance_id_or_domain != "library"
2887            or not user
2888            or (user and not user.provider_filter)
2889            or isinstance(item, BrowseFolder)
2890        ):
2891            return True
2892
2893        # library item handling for users with provider filter
2894        for provider_mapping in item.provider_mappings:
2895            if provider_mapping.provider_instance in user.provider_filter:
2896                return True
2897
2898        return False
2899