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        *,
654        always_include_media_types: list[MediaType] | None = None,
655    ) -> list[ItemMapping]:
656        """
657        Return a list of the last played items.
658
659        :param limit: Maximum number of items to return.
660        :param media_types: Filter by media types.
661        :param userid: Filter by specific user ID.
662        :param queue_id: Filter by specific queue ID.
663        :param fully_played_only: If True, only return fully played items.
664        :param user_initiated_only: If True, only return items initiated by the user.
665        :param played_after_timestamp: If set, only return items played at or after this
666            epoch-seconds timestamp.
667        :param always_include_media_types: Media types to include regardless of
668            user_initiated_only (e.g. podcasts/audiobooks, which have no user-initiated
669            container).
670        """
671        if media_types is None:
672            media_types = MediaType.ALL
673        media_types_str = "(" + ",".join(f'"{x}"' for x in media_types) + ")"
674        available_providers = ("library", *self.get_unique_providers())
675        available_providers_str = "(" + ",".join(f'"{x}"' for x in available_providers) + ")"
676        # user_initiated_only constrains only `media_types`; always_include_media_types are
677        # included regardless (e.g. podcasts/audiobooks have no user-initiated container row).
678        media_type_clause = f"media_type in {media_types_str}"
679        if user_initiated_only:
680            media_type_clause += " AND user_initiated = 1"
681        media_type_clause = f"({media_type_clause})"
682        if always_include_media_types:
683            always_str = "(" + ",".join(f'"{x}"' for x in always_include_media_types) + ")"
684            media_type_clause = f"({media_type_clause} OR media_type in {always_str})"
685        query = (
686            f"SELECT * FROM {DB_TABLE_PLAYLOG} "
687            f"WHERE {media_type_clause} "
688            f"AND provider in {available_providers_str} "
689        )
690        params: dict[str, Any] = {}
691        if fully_played_only:
692            query += "AND fully_played = 1 "
693        if userid:
694            query += "AND userid = :userid "
695            params["userid"] = userid
696        elif user := get_current_user():
697            query += "AND userid = :userid "
698            params["userid"] = user.user_id
699        if queue_id:
700            query += "AND queue_id = :queue_id "
701            params["queue_id"] = queue_id
702        if played_after_timestamp is not None:
703            query += "AND timestamp >= :played_after_timestamp "
704            params["played_after_timestamp"] = played_after_timestamp
705        query += "ORDER BY timestamp DESC"
706        db_rows = await self.mass.music.database.get_rows_from_query(
707            query, params=params or None, limit=limit
708        )
709        result: list[ItemMapping] = []
710        available_providers = ("library", *get_global_cache_value("available_providers", []))
711
712        # Get user provider filter if set
713        user = get_current_user()
714        user_provider_filter = user.provider_filter if user and user.provider_filter else None
715
716        for db_row in db_rows:
717            provider = db_row["provider"]
718            # Apply user provider filter
719            if user_provider_filter and provider not in user_provider_filter:
720                continue
721            result.append(
722                ItemMapping.from_dict(
723                    {
724                        "item_id": db_row["item_id"],
725                        "provider": provider,
726                        "media_type": db_row["media_type"],
727                        "name": db_row["name"],
728                        "image": json_loads(db_row["image"]) if db_row["image"] else None,
729                        "available": provider in available_providers,
730                    }
731                )
732            )
733        return result
734
735    async def recently_played_tracks(
736        self,
737        limit: int,
738        played_after_timestamp: int,
739        userid: str | None = None,
740    ) -> list[RecentPlayedTrack]:
741        """
742        Return recently played, fully played tracks with their recorded artists, newest first.
743
744        :param limit: Maximum number of plays to return.
745        :param played_after_timestamp: Only include plays at or after this epoch-seconds timestamp.
746        :param userid: Restrict to this user (defaults to the current session user, else all users).
747        """
748        query = (
749            f"SELECT item_id, provider, name, image, artists FROM {DB_TABLE_PLAYLOG} "
750            "WHERE media_type = 'track' AND fully_played = 1 "
751            "AND timestamp >= :played_after_timestamp "
752        )
753        params: dict[str, Any] = {"played_after_timestamp": played_after_timestamp}
754        if userid:
755            query += "AND userid = :userid "
756            params["userid"] = userid
757        elif user := get_current_user():
758            query += "AND userid = :userid "
759            params["userid"] = user.user_id
760        query += "ORDER BY timestamp DESC"
761        db_rows = await self.mass.music.database.get_rows_from_query(
762            query, params=params, limit=limit
763        )
764        available_providers = ("library", *get_global_cache_value("available_providers", []))
765        return [
766            RecentPlayedTrack(
767                track=ItemMapping.from_dict(
768                    {
769                        "item_id": db_row["item_id"],
770                        "provider": db_row["provider"],
771                        "media_type": "track",
772                        "name": db_row["name"],
773                        "image": json_loads(db_row["image"]) if db_row["image"] else None,
774                        "available": db_row["provider"] in available_providers,
775                    }
776                ),
777                artists=[ItemMapping.from_dict(artist) for artist in json_loads(db_row["artists"])]
778                if db_row["artists"]
779                else [],
780            )
781            for db_row in db_rows
782        ]
783
784    @api_command("music/recently_added_tracks", required_scope=Scope.LIBRARY_READ)
785    async def recently_added_tracks(self, limit: int = 10) -> list[Track]:
786        """Return a list of the last added tracks."""
787        return await self.tracks.library_items(
788            limit=limit, order_by="timestamp_added_desc", summary=False
789        )
790
791    @api_command("music/in_progress_items", required_scope=Scope.LIBRARY_READ)
792    async def in_progress_items(
793        self, limit: int = 10, all_users: bool = False
794    ) -> list[ItemMapping]:
795        """Return a list of the Audiobooks and PodcastEpisodes that are in progress."""
796        available_providers = ("library", *self.get_unique_providers())
797        available_providers_str = "(" + ",".join(f'"{x}"' for x in available_providers) + ")"
798
799        # An audiobook can be part of the library, in contrast to podcast episodes.
800        # We then need to check the provider mappings table.
801        one_week_ago = int(utc_timestamp()) - (7 * 86400)
802        query = (
803            "SELECT p.item_id, p.media_type, p.name, p.image, p.provider "
804            f"FROM {DB_TABLE_PLAYLOG} p "
805            "WHERE p.media_type IN ('audiobook', 'podcast_episode') "
806            "AND p.fully_played = 0 "
807            "AND p.seconds_played > 0 "
808            f"AND (p.media_type != 'podcast_episode' OR p.timestamp >= {one_week_ago}) "
809        )
810        query += (
811            "AND ( "
812            "CASE WHEN p.provider = 'library' THEN "
813            f"EXISTS (SELECT 1 FROM {DB_TABLE_PROVIDER_MAPPINGS} m "
814            "WHERE m.item_id = p.item_id AND m.media_type = p.media_type "
815        )
816        if not all_users and (user := get_current_user()):
817            filter_for_str = available_providers_str
818            if user.provider_filter:
819                filter_for_str = "(" + ",".join(f'"{x}"' for x in user.provider_filter) + ")"
820            query += (
821                f"AND m.provider_instance IN {filter_for_str} "
822                f"AND m.provider_instance IN {available_providers_str} "
823                ") "
824                f"ELSE (p.provider IN {filter_for_str} AND p.provider IN {available_providers_str})"
825                "END "
826                ") "
827                f"AND p.userid = '{user.user_id}' "
828            )
829        else:
830            # for a library item, we still have to verify via the provider mapping table
831            # that the provider is available
832            query += (
833                f"AND m.provider_instance IN {available_providers_str} "
834                ") "
835                f"ELSE p.provider IN {available_providers_str} "
836                "END "
837                ") "
838            )
839        query += "ORDER BY timestamp DESC"
840
841        db_rows = await self.mass.music.database.get_rows_from_query(query, limit=limit)
842        result: list[ItemMapping] = []
843        for db_row in db_rows:
844            provider = db_row["provider"]
845            result.append(
846                ItemMapping.from_dict(
847                    {
848                        "item_id": db_row["item_id"],
849                        "provider": provider,
850                        "media_type": db_row["media_type"],
851                        "name": db_row["name"],
852                        "image": json_loads(db_row["image"]) if db_row["image"] else None,
853                        "available": provider in available_providers,
854                    }
855                )
856            )
857        return result
858
859    async def get_playlog_provider_item_ids(
860        self, provider_instance_id: str, limit: int = 0, userid: str | None = None
861    ) -> list[tuple[MediaType, str]]:
862        """Return a list of MediaType and provider_item_id of items in playlog of provider."""
863        # check if there is a provider user
864        # this method is not available in the frontend, so no need to check for session users.
865        user: User | None = None
866        if userid:
867            # userid overridden by parameter
868            user = await self.mass.webserver.auth.get_user(userid)
869        elif provider_user := await self._get_user_for_provider(provider_instance_id):
870            # based on configured provider filter we can try to find a user
871            user = provider_user
872
873        query = (
874            f"SELECT * FROM {DB_TABLE_PLAYLOG} "
875            "WHERE media_type in ('audiobook', 'podcast_episode') "
876            f"AND provider in ('library','{provider_instance_id}')"
877        )
878
879        if user:
880            # NOTE: if no user was found, we will return playlog items for all users
881            query += f" AND userid = '{user.user_id}'"
882        db_rows = await self.mass.music.database.get_rows_from_query(query, limit=limit)
883
884        result: list[tuple[MediaType, str]] = []
885        for db_row in db_rows:
886            if db_row["provider"] == "library":
887                # If the provider is library, we need to make sure that the item
888                # is part of the passed provider_instance_id.
889                # A podcast_episode cannot be in the provider_mappings
890                # so these entries must be audiobooks.
891                subquery = (
892                    f"SELECT * FROM {DB_TABLE_PROVIDER_MAPPINGS} "
893                    f"WHERE media_type = 'audiobook' AND item_id = {db_row['item_id']} "
894                    f"AND provider_instance = '{provider_instance_id}'"
895                )
896                subrow = await self.mass.music.database.get_rows_from_query(subquery)
897                if len(subrow) != 1:
898                    continue
899                result.append((MediaType.AUDIOBOOK, subrow[0]["provider_item_id"]))
900                continue
901            # non library - item id is provider_item_id
902            result.append((MediaType(db_row["media_type"]), db_row["item_id"]))
903
904        return result
905
906    @api_command("music/item_by_uri", required_scope=Scope.LIBRARY_READ)
907    async def get_item_by_uri(
908        self, uri: str, allow_update_metadata: bool = False
909    ) -> MediaItemType | BrowseFolder:
910        """Fetch MediaItem by uri."""
911        media_type, provider_instance_id_or_domain, item_id = await parse_uri(uri)
912        return await self.get_item(
913            media_type=media_type,
914            item_id=item_id,
915            provider_instance_id_or_domain=provider_instance_id_or_domain,
916            allow_update_metadata=allow_update_metadata,
917        )
918
919    @api_command("music/sound_effects", required_scope=Scope.LIBRARY_READ)
920    async def sound_effects(self) -> list[SoundEffect]:
921        """Return all sound effect items from providers supporting them."""
922        providers = self._apply_user_provider_filter(
923            self.mass.get_providers_supporting_feature(ProviderFeature.SOUND_EFFECTS)
924        )
925        results_per_provider: list[list[SoundEffect]] = await asyncio.gather(
926            *[
927                self._get_provider_sound_effects(cast("MusicProvider", provider))
928                for provider in providers
929            ]
930        )
931        return [item for sublist in results_per_provider for item in sublist]
932
933    @api_command("music/item", required_scope=Scope.LIBRARY_READ)
934    async def get_item(
935        self,
936        media_type: MediaType,
937        item_id: str,
938        provider_instance_id_or_domain: str,
939        allow_update_metadata: bool = True,
940    ) -> MediaItemType | BrowseFolder:
941        """Get single music item by id and media type."""
942        if provider_instance_id_or_domain == "database":
943            # backwards compatibility - to remove when 2.0 stable is released
944            provider_instance_id_or_domain = "library"
945        provider = self.mass.get_provider(provider_instance_id_or_domain)
946        if media_type in (
947            MediaType.TRACK,
948            MediaType.RADIO,
949            MediaType.SOUND_EFFECT,
950            MediaType.UNKNOWN,  # e.g. plain (HA) URLs, see helpers/uri.py
951        ) and (
952            provider_instance_id_or_domain == "builtin"
953            or (provider and provider.domain == "builtin")
954        ):
955            # handle special case of 'builtin' MusicProvider which allows us to play regular url's
956            builtin_prov = cast("BuiltinProvider", provider or self.mass.get_provider("builtin"))
957            if media_type == MediaType.RADIO:
958                # a radio station must stay a radio station, also when the stream
959                # reports a duration or carries no ICY name
960                return await builtin_prov.get_radio(item_id)
961            if media_type == MediaType.TRACK:
962                # and a track must stay a track, also when the stream carries an
963                # ICY name or reports no duration
964                return await builtin_prov.get_track(item_id)
965            return await builtin_prov.parse_item(item_id, requested_media_type=media_type)
966        if media_type == MediaType.PODCAST_EPISODE:
967            # special case for podcast episodes
968            return await self.podcasts.episode(item_id, provider_instance_id_or_domain)
969        if media_type == MediaType.FOLDER:
970            # special case for folders
971            return BrowseFolder(
972                item_id=item_id,
973                provider=provider_instance_id_or_domain,
974                name=item_id,
975            )
976        if media_type == MediaType.AUDIO_SOURCE:
977            # AudioSources are not library-backed; resolve them through the owning
978            # plugin provider's get_audio_sources() catalog. Returning the live
979            # MediaItem lets play_media create a queue item the standard way.
980            prov = self.mass.get_provider(provider_instance_id_or_domain)
981            if isinstance(prov, PluginProvider):
982                for source in await prov.get_audio_sources():
983                    if source.item_id == item_id:
984                        return source
985            raise MediaNotFoundError(
986                f"AudioSource {provider_instance_id_or_domain}/{item_id} not found"
987            )
988        if media_type == MediaType.SOUND_EFFECT:
989            # Sound effects are not library-backed; resolve them live from the
990            # owning music provider. Returning the live MediaItem lets play_media
991            # create a queue item the standard way.
992            prov = self.mass.get_provider(provider_instance_id_or_domain)
993            if isinstance(prov, MusicProvider) and (
994                ProviderFeature.SOUND_EFFECTS in prov.supported_features
995            ):
996                return await prov.get_sound_effect(item_id)
997            raise MediaNotFoundError(
998                f"SoundEffect {provider_instance_id_or_domain}/{item_id} not found"
999            )
1000        if media_type == MediaType.COLLECTION:
1001            ctrl = self.get_controller_for_collection(item_id)
1002            return await ctrl.get_collection(item_id)
1003        ctrl = self.get_controller(media_type)
1004        return await ctrl.get(
1005            item_id=item_id,
1006            provider_instance_id_or_domain=provider_instance_id_or_domain,
1007            allow_update_metadata=allow_update_metadata,
1008        )
1009
1010    @api_command("music/get_library_item", required_scope=Scope.LIBRARY_READ)
1011    async def get_library_item_by_prov_id(
1012        self,
1013        media_type: MediaType,
1014        item_id: str,
1015        provider_instance_id_or_domain: str,
1016    ) -> MediaItemType | None:
1017        """Get the library item for the given provider item, if present."""
1018        ctrl = self.get_controller(media_type)
1019        return await ctrl.get_library_item_by_prov_id(
1020            item_id=item_id,
1021            provider_instance_id_or_domain=provider_instance_id_or_domain,
1022        )
1023
1024    @api_command("music/favorites/add_item", required_scope=Scope.LIBRARY_WRITE)
1025    async def add_item_to_favorites(
1026        self,
1027        item: str | MediaItemType | ItemMapping,
1028    ) -> None:
1029        """Add an item to the favorites."""
1030        if isinstance(item, str):
1031            # Inspect the URI's media_type first so a stale audio-source URI
1032            # whose plugin is unloaded gives the honest rejection error
1033            # instead of bubbling MediaNotFoundError from get_item_by_uri.
1034            try:
1035                uri_media_type, _, _ = await parse_uri(item)
1036            except InvalidProviderURI, InvalidProviderID:
1037                uri_media_type = None
1038            if uri_media_type in (MediaType.AUDIO_SOURCE, MediaType.SOUND_EFFECT):
1039                raise UnsupportedFeaturedException(
1040                    f"{uri_media_type.value} items can not be favorites"
1041                )
1042            # a favorite URI always resolves to a media item, never a BrowseFolder
1043            item = cast("MediaItemType", await self.get_item_by_uri(item))
1044        if item.media_type in (MediaType.AUDIO_SOURCE, MediaType.SOUND_EFFECT):
1045            # AudioSources and SoundEffects are live provider content (existence
1046            # depends on a loaded provider) and have no stable library identity,
1047            # so they can not be persisted as favorites.
1048            raise UnsupportedFeaturedException(
1049                f"{item.media_type.value} items can not be favorites"
1050            )
1051        # make sure we have a full library item
1052        # a favorite must always be in the library
1053        full_item = cast(
1054            "MediaItemType",
1055            await self.get_item(
1056                item.media_type,
1057                item.item_id,
1058                item.provider,
1059            ),
1060        )
1061        if full_item.provider != "library":
1062            full_item = await self.add_item_to_library(full_item)
1063        # set favorite in library db
1064        ctrl = self.get_controller(item.media_type)
1065        await ctrl.set_favorite(
1066            full_item.item_id,
1067            True,
1068        )
1069        # forward to provider(s) if needed
1070        for prov_mapping in full_item.provider_mappings:
1071            provider = self.mass.get_provider(
1072                prov_mapping.provider_instance, provider_type=MusicProvider
1073            )
1074            if not provider or not self.library_favorites_edit_supported(
1075                provider, full_item.media_type
1076            ):
1077                continue
1078            await provider.set_favorite(prov_mapping.item_id, full_item.media_type, True)
1079
1080    @api_command("music/favorites/remove_item", required_scope=Scope.LIBRARY_WRITE)
1081    async def remove_item_from_favorites(
1082        self,
1083        media_type: MediaType,
1084        library_item_id: str | int,
1085    ) -> None:
1086        """Remove (library) item from the favorites."""
1087        ctrl = self.get_controller(media_type)
1088        await ctrl.set_favorite(
1089            library_item_id,
1090            False,
1091        )
1092        # forward to provider(s) if needed
1093        full_item = await ctrl.get_library_item(library_item_id)
1094        for prov_mapping in full_item.provider_mappings:
1095            provider = self.mass.get_provider(
1096                prov_mapping.provider_instance, provider_type=MusicProvider
1097            )
1098            if not provider or not self.library_favorites_edit_supported(
1099                provider, full_item.media_type
1100            ):
1101                continue
1102            self.mass.create_task(provider.set_favorite(prov_mapping.item_id, media_type, False))
1103
1104    @api_command("music/library/remove_item", required_scope=Scope.LIBRARY_WRITE)
1105    async def remove_item_from_library(
1106        self, media_type: MediaType, library_item_id: str | int, recursive: bool = True
1107    ) -> None:
1108        """
1109        Remove item from the library.
1110
1111        Destructive! Will remove the item and all dependants.
1112        """
1113        ctrl = self.get_controller(media_type)
1114        # remove from provider(s) library
1115        full_item = await ctrl.get_library_item(library_item_id)
1116        for prov_mapping in full_item.provider_mappings:
1117            if not prov_mapping.in_library:
1118                continue
1119            provider = self.mass.get_provider(
1120                prov_mapping.provider_instance, provider_type=MusicProvider
1121            )
1122            if not provider or not self.library_edit_supported(provider, full_item.media_type):
1123                continue
1124            if not self.library_sync_back_enabled(provider, full_item.media_type):
1125                continue
1126            prov_mapping.in_library = False
1127            self.mass.create_task(provider.library_remove(prov_mapping.item_id, media_type))
1128        # remove from library
1129        await ctrl.remove_item_from_library(library_item_id, recursive)
1130
1131    @api_command("music/library/add_item", required_scope=Scope.LIBRARY_WRITE)
1132    async def add_item_to_library(
1133        self, item: str | MediaItemType | ItemMapping, overwrite_existing: bool = False
1134    ) -> MediaItemType:
1135        """Add item (uri or mediaitem) to the library."""
1136        if isinstance(item, ItemMapping):
1137            # handle browse results that are returned as ItemMappings
1138            # uri is always populated post-init, so it is never None here
1139            item = cast("str", item.uri)
1140        # ensure we have a full item
1141        if isinstance(item, str):
1142            # Inspect the URI's media_type first so a stale audio-source URI
1143            # whose plugin is unloaded gives the honest rejection error
1144            # instead of bubbling MediaNotFoundError from get_item_by_uri.
1145            # Mirrors the same guard in add_item_to_favorites.
1146            try:
1147                uri_media_type, _, _ = await parse_uri(item)
1148            except InvalidProviderURI, InvalidProviderID:
1149                uri_media_type = None
1150            if uri_media_type in (MediaType.AUDIO_SOURCE, MediaType.SOUND_EFFECT):
1151                raise UnsupportedFeaturedException(
1152                    f"{uri_media_type.value} items can not be library items"
1153                )
1154            full_item = await self.get_item_by_uri(item)
1155        # For builtin provider (manual URLs), use the provided item directly
1156        # to preserve custom modifications (name, images, etc.)
1157        # For other providers, fetch fresh to ensure data validity
1158        elif item.provider == "builtin":
1159            full_item = item
1160        else:
1161            full_item = await self.get_item(
1162                item.media_type,
1163                item.item_id,
1164                item.provider,
1165            )
1166        full_item = cast("MediaItemType", full_item)
1167        if full_item.media_type in (MediaType.AUDIO_SOURCE, MediaType.SOUND_EFFECT):
1168            # AudioSources and SoundEffects are live provider content (existence
1169            # depends on a loaded provider) and have no stable library identity,
1170            # so they can not be persisted as library items.
1171            raise UnsupportedFeaturedException(
1172                f"{full_item.media_type.value} items can not be library items"
1173            )
1174        # add to provider(s) library first
1175        for prov_mapping in full_item.provider_mappings:
1176            # we optimistically set in library to True to prevent items
1177            # from disappearing when the provider doesn't support library edit
1178            # or 2-way sync is disabled.
1179            prov_mapping.in_library = True
1180            provider = self.mass.get_provider(
1181                prov_mapping.provider_instance, provider_type=MusicProvider
1182            )
1183            if not provider or not self.library_edit_supported(provider, full_item.media_type):
1184                continue
1185            if not self.library_sync_back_enabled(provider, full_item.media_type):
1186                continue
1187            prov_item = deepcopy(full_item) if full_item.provider == "library" else full_item
1188            prov_item.provider = prov_mapping.provider_instance
1189            prov_item.item_id = prov_mapping.item_id
1190            self.mass.create_task(provider.library_add(prov_item))
1191        # add (or overwrite) to library
1192        ctrl = self.get_controller(full_item.media_type)
1193        # ctrl is chosen by media_type, so it matches full_item's runtime type
1194        library_item = await cast("MediaControllerBase[MediaItemType]", ctrl).add_item_to_library(
1195            full_item, overwrite_existing
1196        )
1197        # optionally import all album tracks into the library, mirroring the behavior
1198        # of the library sync (which only triggers on a (scheduled) full sync run)
1199        if full_item.media_type == MediaType.ALBUM:
1200            self._import_album_tracks_if_enabled(cast("Album", library_item))
1201        # perform full metadata scan
1202        await self.mass.metadata.update_metadata(library_item, overwrite_existing)
1203        return library_item
1204
1205    @api_command("music/refresh_item", required_scope=Scope.LIBRARY_MANAGE)
1206    async def refresh_item(  # noqa: PLR0915
1207        self,
1208        media_item: str | MediaItemType,
1209    ) -> MediaItemType | None:
1210        """Try to refresh a mediaitem by requesting it's full object or search for substitutes."""
1211        if isinstance(media_item, str):
1212            # media item uri given
1213            # a refresh URI always resolves to a media item, never a BrowseFolder
1214            media_item = cast("MediaItemType", await self.get_item_by_uri(media_item))
1215
1216        media_type = media_item.media_type
1217        ctrl = self.get_controller(media_type)
1218
1219        # genres are library-only items with no provider mappings, nothing to refresh
1220        if media_type == MediaType.GENRE:
1221            return media_item
1222
1223        library_id = media_item.item_id if media_item.provider == "library" else None
1224
1225        # cache in_library state before the provider fetch overwrites media_item
1226        in_library_cache: dict[tuple[str, str], bool] = {}
1227        for m in media_item.provider_mappings:
1228            if m.in_library is not None:
1229                in_library_cache[(m.provider_instance, m.item_id)] = m.in_library
1230
1231        available_providers = get_global_cache_value("available_providers")
1232        if TYPE_CHECKING:
1233            available_providers = cast("set[str]", available_providers)
1234
1235        # fetch the first (available) provider item
1236        for prov_mapping in sorted(
1237            media_item.provider_mappings, key=lambda x: x.priority, reverse=True
1238        ):
1239            if not self.mass.get_provider(prov_mapping.provider_instance):
1240                # ignore unavailable providers
1241                continue
1242            with suppress(MediaNotFoundError):
1243                media_item = await ctrl.get_provider_item(
1244                    prov_mapping.item_id,
1245                    prov_mapping.provider_instance,
1246                    force_refresh=True,
1247                )
1248                provider = media_item.provider
1249                item_id = media_item.item_id
1250                break
1251        else:
1252            # try to find a substitute using search
1253            searchresult = await self.search(media_item.name, [media_item.media_type], 20)
1254            result: Sequence[MediaItemType | ItemMapping]
1255            if media_item.media_type == MediaType.ARTIST:
1256                result = searchresult.artists
1257            elif media_item.media_type == MediaType.ALBUM:
1258                result = searchresult.albums
1259            elif media_item.media_type == MediaType.TRACK:
1260                result = searchresult.tracks
1261            elif media_item.media_type == MediaType.PLAYLIST:
1262                result = searchresult.playlists
1263            elif media_item.media_type == MediaType.AUDIOBOOK:
1264                result = searchresult.audiobooks
1265            elif media_item.media_type == MediaType.PODCAST:
1266                result = searchresult.podcasts
1267            else:
1268                result = searchresult.radio
1269            for item in result:
1270                if item == media_item or item.provider == "library":
1271                    continue
1272                if item.available:
1273                    provider = item.provider
1274                    item_id = item.item_id
1275                    break
1276            else:
1277                # raise if we didn't find a substitute
1278                raise MediaNotFoundError(f"Could not find a substitute for {media_item.name}")
1279        # fetch full (provider) item
1280        media_item = await ctrl.get_provider_item(item_id, provider, force_refresh=True)
1281        # update library item if needed (including refresh of the metadata etc.)
1282        if library_id is None:
1283            return media_item
1284        # restore in_library state from before the refresh
1285        for prov_mapping in media_item.provider_mappings:
1286            key = (prov_mapping.provider_instance, prov_mapping.item_id)
1287            if prov_mapping.in_library is None and key in in_library_cache:
1288                prov_mapping.in_library = in_library_cache[key]
1289        # ctrl is chosen by media_type, so it matches media_item's runtime type
1290        library_item = await cast(
1291            "MediaControllerBase[MediaItemType]", ctrl
1292        ).update_item_in_library(library_id, media_item, overwrite=True)
1293        if library_item.media_type == MediaType.ALBUM:
1294            # update (local) album tracks
1295            for album_track in await self.albums.tracks(
1296                library_item.item_id, library_item.provider, True
1297            ):
1298                for prov_mapping in album_track.provider_mappings:
1299                    if not (prov := self.mass.get_provider(prov_mapping.provider_instance)):
1300                        continue
1301                    if not isinstance(prov, MusicProvider):
1302                        continue
1303                    if prov.is_streaming_provider:
1304                        continue
1305                    with suppress(MediaNotFoundError):
1306                        prov_track = await prov.get_track(prov_mapping.item_id)
1307                        await self.mass.music.tracks.update_item_in_library(
1308                            album_track.item_id, prov_track
1309                        )
1310        await cast("MediaControllerBase[MediaItemType]", ctrl).match_providers(library_item)
1311        await self.mass.metadata.update_metadata(library_item, force_refresh=True)
1312        return library_item
1313
1314    @api_command("music/mark_played", required_scope=Scope.LIBRARY_WRITE)
1315    async def mark_item_played(
1316        self,
1317        media_item: MediaItemType,
1318        fully_played: bool = True,
1319        seconds_played: int | None = None,
1320        is_playing: bool = False,
1321        userid: str | None = None,
1322        queue_id: str | None = None,
1323        user_initiated: bool = True,
1324        skip_artist_ids: list[str] | None = None,
1325        playback_speed: float | None = None,
1326    ) -> None:
1327        """
1328        Mark item as played in playlog.
1329
1330        :param media_item: The media item to mark as played.
1331        :param fully_played: If True, mark the item as fully played.
1332        :param seconds_played: The number of seconds played.
1333        :param is_playing: If True, the item is currently playing.
1334        :param userid: The user ID to mark the item as played for (instead of the current user).
1335        :param queue_id: The queue ID where the item was played.
1336        :param user_initiated: If True, the playback was initiated by the user (e.g. enqueued).
1337            Sticky once set: a later report can promote a playlog row to user-initiated but
1338            never demote it, so a writer reporting playback it did not itself initiate
1339            (e.g. a provider sync) must pass False.
1340        :param skip_artist_ids: Library artist ids to skip when crediting an album's artists.
1341        :param playback_speed: The current playback speed to persist (audiobooks/podcasts).
1342            If None, any previously stored speed for the item is preserved.
1343        """
1344        timestamp = utc_timestamp()
1345        # we deliberately skip one-off items: sound effects and live inputs whoever owns
1346        # them, and everything the builtin provider plays (except playlists) is a one-off url
1347        if media_item.media_type in (MediaType.SOUND_EFFECT, MediaType.AUDIO_SOURCE):
1348            return
1349        if (
1350            media_item.provider.startswith("builtin")
1351            and media_item.media_type != MediaType.PLAYLIST
1352        ):
1353            return
1354
1355        params = {
1356            "item_id": media_item.item_id,
1357            "provider": media_item.provider,
1358            "media_type": media_item.media_type.value,
1359            "name": media_item.name,
1360            "image": serialize_to_json(media_item.image.to_dict()) if media_item.image else None,
1361            # store lightweight artist mappings so playlog rows can later be matched or
1362            # resolved by artist without an extra provider lookup
1363            "artists": serialize_to_json(
1364                [ItemMapping.from_item(artist).to_dict() for artist in artists]
1365            )
1366            if (artists := getattr(media_item, "artists", None))
1367            else None,
1368            "fully_played": fully_played,
1369            "seconds_played": seconds_played,
1370            "timestamp": timestamp,
1371            "queue_id": queue_id,
1372            "user_initiated": user_initiated,
1373        }
1374        # try to figure out the user that triggered the action
1375        user: User | None = None
1376        if userid:
1377            # userid overridden by parameter
1378            user = await self.mass.webserver.auth.get_user(userid)
1379        elif session_user := get_current_user():
1380            # this is the active session user that triggered the action
1381            user = session_user
1382        elif provider_user := await self._get_user_for_provider(media_item.provider_mappings):
1383            # based on configured provider filter we can try to find a user
1384            user = provider_user
1385
1386        # update generic playlog table (when not playing)
1387        if not is_playing:
1388            if user:
1389                user_ids = [user.user_id]
1390            else:
1391                # NOTE: if no user was found, we will alter the playlog for all users
1392                user_ids = [user.user_id for user in await self.mass.webserver.auth.list_users()]
1393            # Leaving the speed out keeps whatever is already stored for this item/user
1394            # (a provider sync reporting progress has no speed to offer), and falls back to
1395            # the column default of 1.0 for a brand new row.
1396            if playback_speed is not None:
1397                params["playback_speed"] = playback_speed
1398            for user_id in user_ids:
1399                params["userid"] = user_id
1400                await self._upsert_playlog(params)
1401
1402        # Set seconds_played in accordance with fully_played, if the media_item has
1403        # a duration, before it is forwarded to music_providers
1404        if seconds_played is None:
1405            seconds_played = 0
1406            if (
1407                fully_played
1408                and not isinstance(
1409                    media_item, Album | Artist | Genre | Playlist | Podcast | MediaCollection
1410                )
1411                and isinstance(media_item.duration, int)  # for Radio duration can be None
1412            ):
1413                seconds_played = media_item.duration
1414
1415        # forward to provider(s) to sync resume state (e.g. for audiobooks)
1416        for prov_mapping in media_item.provider_mappings:
1417            if (
1418                user
1419                and user.provider_filter
1420                and prov_mapping.provider_instance not in user.provider_filter
1421            ):
1422                continue
1423            if music_prov := self.mass.get_provider(prov_mapping.provider_instance):
1424                if music_prov.type != ProviderType.MUSIC:
1425                    continue
1426                music_prov = cast("MusicProvider", music_prov)
1427                self.mass.create_task(
1428                    music_prov.on_played(
1429                        media_type=media_item.media_type,
1430                        prov_item_id=prov_mapping.item_id,
1431                        fully_played=fully_played,
1432                        position=seconds_played,
1433                        media_item=media_item,
1434                        is_playing=is_playing,
1435                    )
1436                )
1437
1438        # also update playcount in library table (if fully played)
1439        if not fully_played or is_playing:
1440            return
1441        try:
1442            ctrl = self.get_controller(media_item.media_type)
1443        except NotImplementedError:
1444            # skip non-library media types (e.g. AudioSource plugin sources)
1445            return
1446        db_item = await ctrl.get_library_item_by_prov_id(media_item.item_id, media_item.provider)
1447        if db_item:
1448            await self.database.execute(
1449                f"UPDATE {ctrl.db_table} SET play_count = play_count + 1, "
1450                f"last_played = {timestamp} WHERE item_id = {db_item.item_id}"
1451            )
1452            if isinstance(media_item, Track):
1453                self.logger.debug("Credited play for track '%s'", media_item.name)
1454        if isinstance(media_item, Track | Album):
1455            await self._credit_artist_plays(
1456                media_item.artists,
1457                timestamp=timestamp,
1458                user_ids=user_ids,
1459                queue_id=queue_id,
1460                skip_ids=set(skip_artist_ids or ()),
1461            )
1462        if isinstance(media_item, PodcastEpisode) and media_item.podcast:
1463            await self._credit_podcast_play(
1464                media_item.podcast,
1465                timestamp=timestamp,
1466                user_ids=user_ids,
1467                queue_id=queue_id,
1468            )
1469        await self.database.commit()
1470
1471    async def resolve_library_artist_ids(self, artists: Iterable[Artist | ItemMapping]) -> set[str]:
1472        """Resolve the given artist references to their library item ids (when present)."""
1473        ids: set[str] = set()
1474        for artist in artists:
1475            db_artist = await self.artists.get_library_item_by_prov_id(
1476                artist.item_id, artist.provider
1477            )
1478            if db_artist is not None:
1479                ids.add(db_artist.item_id)
1480        return ids
1481
1482    @api_command("music/mark_unplayed", required_scope=Scope.LIBRARY_WRITE)
1483    async def mark_item_unplayed(
1484        self,
1485        media_item: MediaItemType,
1486        userid: str | None = None,
1487    ) -> None:
1488        """
1489        Mark item as unplayed in playlog.
1490
1491        :param media_item: The media item to mark as unplayed.
1492        :param all_users: If True, mark the item as unplayed for all users.
1493        :param userid: The user ID to mark the item as unplayed for (instead of the current user).
1494        """
1495        params = {
1496            "item_id": media_item.item_id,
1497            "provider": media_item.provider,
1498            "media_type": media_item.media_type.value,
1499        }
1500        # try to figure out the user that triggered the action
1501        user: User | None = None
1502        if userid:
1503            # userid overridden by parameter
1504            user = await self.mass.webserver.auth.get_user(userid)
1505        elif session_user := get_current_user():
1506            # this is the active session user that triggered the action
1507            user = session_user
1508        elif provider_user := await self._get_user_for_provider(media_item.provider_mappings):
1509            # based on configured provider filter we can try to find a user
1510            user = provider_user
1511
1512        if user:
1513            user_ids = [user.user_id]
1514        else:
1515            # NOTE: if no user was found, we will alter the playlog for all users
1516            user_ids = [user.user_id for user in await self.mass.webserver.auth.list_users()]
1517        for user_id in user_ids:
1518            params["userid"] = user_id
1519            await self.database.delete(DB_TABLE_PLAYLOG, params)
1520
1521        # forward to provider(s) to sync resume state (e.g. for audiobooks)
1522        for prov_mapping in media_item.provider_mappings:
1523            if (
1524                user
1525                and user.provider_filter
1526                and prov_mapping.provider_instance not in user.provider_filter
1527            ):
1528                continue
1529            if music_prov := self.mass.get_provider(prov_mapping.provider_instance):
1530                if music_prov.type != ProviderType.MUSIC:
1531                    continue
1532                music_prov = cast("MusicProvider", music_prov)
1533                self.mass.create_task(
1534                    music_prov.on_played(
1535                        media_type=media_item.media_type,
1536                        prov_item_id=prov_mapping.item_id,
1537                        fully_played=False,
1538                        position=0,
1539                        media_item=media_item,
1540                    )
1541                )
1542        # also update playcount in library table
1543        ctrl = self.get_controller(media_item.media_type)
1544        db_item = await ctrl.get_library_item_by_prov_id(media_item.item_id, media_item.provider)
1545        if db_item:
1546            await self.database.execute(
1547                f"UPDATE {ctrl.db_table} SET play_count = play_count - 1, "
1548                f"last_played = 0 WHERE item_id = {db_item.item_id}"
1549            )
1550            await self.database.commit()
1551
1552    @api_command("music/track_by_name", required_scope=Scope.LIBRARY_READ)
1553    async def get_track_by_name(
1554        self,
1555        track_name: str,
1556        artist_name: str | None = None,
1557        album_name: str | None = None,
1558        track_version: str | None = None,
1559    ) -> Track | None:
1560        """Get a track by its name, optionally with artist and album."""
1561        if track_version is None:
1562            track_name, version = parse_title_and_version(track_name)
1563        search_query = f"{artist_name} - {track_name}" if artist_name else track_name
1564        search_result = await self.mass.music.search(
1565            search_query=search_query,
1566            media_types=[MediaType.TRACK],
1567        )
1568        for allow_item_mapping in (False, True):
1569            for search_track in search_result.tracks:
1570                if not allow_item_mapping and not isinstance(search_track, Track):
1571                    continue
1572                if not compare_strings(track_name, search_track.name):
1573                    continue
1574                if not compare_version(version, search_track.version):
1575                    continue
1576                # check optional artist(s)
1577                if artist_name and isinstance(search_track, Track):
1578                    for artist in search_track.artists:
1579                        if compare_strings(artist_name, artist.name, False):
1580                            break
1581                    else:
1582                        # no artist match found: abort
1583                        continue
1584                # check optional album
1585                if album_name and isinstance(search_track, Track):
1586                    track_album = search_track.album
1587                    # a track without album info can never match a requested album
1588                    if track_album is None or not compare_strings(
1589                        album_name, track_album.name, False
1590                    ):
1591                        # no album match found: abort
1592                        continue
1593                # if we reach this, we found a match
1594                if not isinstance(search_track, Track):
1595                    # ensure we return an actual Track object
1596                    return await self.mass.music.tracks.get(
1597                        item_id=search_track.item_id,
1598                        provider_instance_id_or_domain=search_track.provider,
1599                    )
1600                return search_track
1601
1602        # try to handle case where something is appended to the title
1603        for splitter in ("•", "-", "|", "(", "["):
1604            if splitter in track_name:
1605                return await self.get_track_by_name(
1606                    track_name=track_name.split(splitter)[0].strip(),
1607                    artist_name=artist_name,
1608                    album_name=None,
1609                    track_version=track_version,
1610                )
1611        # try to handle case where multiple artists are given as single string
1612        if artist_name and (artists := split_artists(artist_name, True)) and len(artists) > 1:
1613            for single_artist in artists:
1614                return await self.get_track_by_name(
1615                    track_name=track_name,
1616                    artist_name=single_artist.split(splitter)[0].strip(),
1617                    album_name=None,
1618                    track_version=track_version,
1619                )
1620        # allow non-exact album match as fallback
1621        if album_name:
1622            return await self.get_track_by_name(
1623                track_name=track_name,
1624                artist_name=artist_name,
1625                album_name=None,
1626                track_version=track_version,
1627            )
1628        # no match found
1629        return None
1630
1631    async def get_resume_position(
1632        self, media_item: Audiobook | PodcastEpisode, userid: str | None = None
1633    ) -> tuple[bool, int]:
1634        """
1635        Get progress (resume point) details for the given audiobook or episode.
1636
1637        This is a separate call to ensure the resume position is always up-to-date
1638        and because many providers have this info present on a dedicated endpoint.
1639
1640        Will be called right before playback starts to ensure the resume position is correct.
1641
1642        Returns a boolean with the fully_played status
1643        and an integer with the resume position in ms.
1644        """
1645        provider_fully_played = False
1646        provider_position_ms = 0
1647        provider_timestamp: datetime | None = None
1648
1649        user: User | None = None
1650        if userid:
1651            # userid overridden by parameter
1652            user = await self.mass.webserver.auth.get_user(userid)
1653        elif session_user := get_current_user():
1654            # this is the active session user that triggered the action
1655            user = session_user
1656        elif provider_user := await self._get_user_for_provider(media_item.provider_mappings):
1657            # based on configured provider filter we can try to find a user
1658            user = provider_user
1659
1660        provider_instances = {x.provider_instance for x in media_item.provider_mappings}
1661        if user and user.provider_filter:
1662            # only if the user has provider filters configured
1663            # otherwise we allow all providers
1664            preferred_provider_instances = provider_instances.intersection(user.provider_filter)
1665        else:
1666            preferred_provider_instances = provider_instances
1667
1668        preferred_providers = [
1669            x
1670            for x in media_item.provider_mappings
1671            if x.provider_instance in preferred_provider_instances
1672        ]
1673
1674        # Try to get position from providers
1675        for prov_mapping in preferred_providers:
1676            if not (
1677                provider := self.mass.get_provider(
1678                    prov_mapping.provider_instance, provider_type=MusicProvider
1679                )
1680            ):
1681                continue
1682            with suppress(NotImplementedError):
1683                (
1684                    provider_fully_played,
1685                    provider_position_ms,
1686                    provider_timestamp,
1687                ) = await provider.get_resume_position(prov_mapping.item_id, media_item.media_type)
1688                break  # Use first provider that returns data
1689
1690        # Get MA's internal position from playlog
1691        ma_fully_played = False
1692        ma_position_ms = 0
1693        ma_timestamp = from_utc_timestamp(0)
1694        params = {
1695            "media_type": media_item.media_type.value,
1696            "item_id": media_item.item_id,
1697            "provider": media_item.provider,
1698        }
1699        if userid:
1700            params["userid"] = userid
1701        elif user:
1702            params["userid"] = user.user_id
1703        if db_entry := await self.database.get_row(DB_TABLE_PLAYLOG, params):
1704            ma_position_ms = db_entry["seconds_played"] * 1000 if db_entry["seconds_played"] else 0
1705            # fully_played is a nullable column; treat an unknown (NULL) value as not played
1706            ma_fully_played = parse_optional_bool(db_entry["fully_played"]) or False
1707            ma_timestamp = from_utc_timestamp(db_entry["timestamp"])
1708
1709        if provider_timestamp is not None and provider_timestamp > ma_timestamp:
1710            return provider_fully_played, provider_position_ms
1711        # Return the higher position to ensure users never lose progress
1712        if ma_position_ms >= provider_position_ms:
1713            return ma_fully_played, ma_position_ms
1714        return provider_fully_played, provider_position_ms
1715
1716    async def get_playback_speed(
1717        self, media_item: Audiobook | PodcastEpisode, userid: str | None = None
1718    ) -> float:
1719        """
1720        Get the stored playback speed for the given audiobook or podcast episode.
1721
1722        Returns 1.0 (normal speed) when no custom speed was stored for the item,
1723        or when no user can be determined to scope the lookup.
1724
1725        :param media_item: The audiobook or podcast episode to look up.
1726        :param userid: The user ID to look up the speed for (instead of the current user).
1727        """
1728        if not userid:
1729            if session_user := get_current_user():
1730                userid = session_user.user_id
1731            elif provider_user := await self._get_user_for_provider(media_item.provider_mappings):
1732                userid = provider_user.user_id
1733            else:
1734                # the speed is stored per user; without one we can't scope the lookup
1735                return 1.0
1736        db_entry = await self.database.get_row(
1737            DB_TABLE_PLAYLOG,
1738            {
1739                "item_id": media_item.item_id,
1740                "provider": media_item.provider,
1741                "media_type": media_item.media_type.value,
1742                "userid": userid,
1743            },
1744        )
1745        if db_entry and (stored_speed := db_entry["playback_speed"]) is not None:
1746            return float(stored_speed)
1747        return 1.0
1748
1749    def get_controller(
1750        self, media_type: MediaType
1751    ) -> (
1752        ArtistsController
1753        | AlbumsController
1754        | TracksController
1755        | RadioController
1756        | PlaylistController
1757        | AudiobooksController
1758        | PodcastsController
1759        | GenreController
1760    ):
1761        """Return controller for MediaType."""
1762        if media_type == MediaType.ARTIST:
1763            return self.artists
1764        if media_type == MediaType.ALBUM:
1765            return self.albums
1766        if media_type == MediaType.TRACK:
1767            return self.tracks
1768        if media_type == MediaType.RADIO:
1769            return self.radio
1770        if media_type == MediaType.PLAYLIST:
1771            return self.playlists
1772        if media_type == MediaType.AUDIOBOOK:
1773            return self.audiobooks
1774        if media_type == MediaType.PODCAST:
1775            return self.podcasts
1776        if media_type == MediaType.PODCAST_EPISODE:
1777            return self.podcasts
1778        if media_type == MediaType.GENRE:
1779            return self.genres
1780        raise NotImplementedError(
1781            f"No media controller available for media type: {media_type.value}"
1782        )
1783
1784    def get_controller_for_collection(
1785        self, item_id: str
1786    ) -> (
1787        ArtistsController
1788        | AlbumsController
1789        | TracksController
1790        | RadioController
1791        | PlaylistController
1792        | AudiobooksController
1793        | PodcastsController
1794        | GenreController
1795    ):
1796        """Return controller for MediaType."""
1797        media_type = get_collection_item_media_type_from_item_id(item_id)
1798        controller = self.get_controller(media_type)
1799        if not isinstance(controller, AudiobooksController):
1800            # currently only supported for audiobooks
1801            raise NotImplementedError(
1802                f"No media controller available for media type: {media_type.value}"
1803            )
1804        return controller
1805
1806    def get_provider_instances(
1807        self, domain: str, return_unavailable: bool = False
1808    ) -> list[MusicProvider]:
1809        """
1810        Return all provider instances for a given domain.
1811
1812        Note that this skips user filters so may only be called from internal code.
1813        """
1814        return cast(
1815            "list[MusicProvider]",
1816            self.mass.get_provider_instances(domain, return_unavailable, ProviderType.MUSIC),
1817        )
1818
1819    def get_unique_providers(self) -> list[str]:
1820        """
1821        Return all unique MusicProvider (instance or domain) ids.
1822
1823        This will return a set of provider instance ids but will only return
1824        a single instance_id per streaming provider domain.
1825
1826        Applies user provider filters (for non-admin users).
1827        """
1828        processed_domains: set[str] = set()
1829        # Get user provider filter if set
1830        user = get_current_user()
1831        user_provider_filter = user.provider_filter if user and user.provider_filter else None
1832        result: list[str] = []
1833        for provider in self.providers:
1834            if provider.is_streaming_provider and provider.domain in processed_domains:
1835                continue
1836            if user_provider_filter and provider.instance_id not in user_provider_filter:
1837                continue
1838            result.append(provider.instance_id)
1839            processed_domains.add(provider.domain)
1840        return result
1841
1842    async def cleanup_provider(self, provider_instance: str) -> None:
1843        """Cleanup provider records from the database."""
1844        deleted_providers = self.mass.config.get_raw_core_config_value(
1845            self.domain, CONF_DELETED_PROVIDERS, []
1846        )
1847        # we add the provider to this hidden config setting just to make sure that
1848        # we can survive this over a restart to make sure that entries are cleaned up
1849        if provider_instance not in deleted_providers:
1850            deleted_providers.append(provider_instance)
1851            self.mass.config.set_raw_core_config_value(
1852                self.domain, CONF_DELETED_PROVIDERS, deleted_providers
1853            )
1854            self.mass.config.save(True)
1855
1856        # always clear cache when a provider is removed
1857        await self.mass.cache.clear()
1858
1859        # cleanup media items from db matched to deleted provider
1860        self.logger.info(
1861            "Removing provider %s from library, this can take a a while...",
1862            provider_instance,
1863        )
1864        errors = 0
1865        # suppress the per-item MEDIA_ITEM_UPDATED events during this bulk removal so we
1866        # don't flood subscribers; they refresh once via the PROVIDERS_UPDATED event
1867        token = SUPPRESS_MEDIA_ITEM_UPDATES.set(True)
1868        try:
1869            for ctrl in (
1870                # order is important here to recursively cleanup bottom up
1871                self.mass.music.radio,
1872                self.mass.music.playlists,
1873                self.mass.music.tracks,
1874                self.mass.music.albums,
1875                self.mass.music.artists,
1876                self.mass.music.podcasts,
1877                self.mass.music.audiobooks,
1878                # run main controllers twice to rule out relations
1879                self.mass.music.tracks,
1880                self.mass.music.albums,
1881                self.mass.music.artists,
1882            ):
1883                query = (
1884                    f"SELECT item_id FROM {DB_TABLE_PROVIDER_MAPPINGS} "
1885                    "WHERE media_type = :media_type "
1886                    "AND provider_instance = :provider_instance"
1887                )
1888                params = {
1889                    "media_type": ctrl.media_type.value,
1890                    "provider_instance": provider_instance,
1891                }
1892                for db_row in await self.database.get_rows_from_query(query, params, limit=100000):
1893                    try:
1894                        await ctrl.remove_provider_mappings(db_row["item_id"], provider_instance)
1895                    except Exception as err:
1896                        # we dont want the whole removal process to stall on one item
1897                        # so in case of an unexpected error, we log and move on.
1898                        self.logger.warning(
1899                            "Error while removing %s: %s",
1900                            db_row["item_id"],
1901                            str(err),
1902                            exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
1903                        )
1904                        errors += 1
1905        finally:
1906            SUPPRESS_MEDIA_ITEM_UPDATES.reset(token)
1907
1908        # remove all orphaned items (not in provider mappings table anymore)
1909        query = (
1910            f"SELECT item_id FROM {DB_TABLE_PROVIDER_MAPPINGS} "
1911            f"WHERE provider_instance = '{provider_instance}'"
1912        )
1913        if remaining_items_count := await self.database.get_count_from_query(query):
1914            errors += remaining_items_count
1915
1916        # cleanup playlog table
1917        await self.mass.music.database.delete(
1918            DB_TABLE_PLAYLOG,
1919            {
1920                "provider": provider_instance,
1921            },
1922        )
1923
1924        if errors == 0:
1925            # cleanup successful, remove from the deleted_providers setting
1926            self.logger.info("Provider %s removed from library", provider_instance)
1927            deleted_providers.remove(provider_instance)
1928            self.mass.config.set_raw_core_config_value(
1929                self.domain, CONF_DELETED_PROVIDERS, deleted_providers
1930            )
1931        else:
1932            self.logger.warning(
1933                "Provider %s was not not fully removed from library", provider_instance
1934            )
1935
1936    async def schedule_provider_sync(self, provider_instance_id: str) -> None:
1937        """Schedule Library sync for given provider."""
1938        if not (
1939            provider := self.mass.get_provider(provider_instance_id, provider_type=MusicProvider)
1940        ):
1941            return
1942        await self.unschedule_provider_sync(provider.instance_id, clear_persisted_state=False)
1943        for media_type in MediaType:
1944            if not self.library_supported(provider, media_type):
1945                continue
1946            await self._schedule_provider_mediatype_sync(provider, media_type, True)
1947
1948    async def unschedule_provider_sync(
1949        self, provider_instance_id: str, clear_persisted_state: bool = True
1950    ) -> None:
1951        """
1952        Unschedule Library sync for given provider and wait for a running sync to stop.
1953
1954        Callers tear down provider state right after this (unloading the provider, or
1955        rescheduling its syncs), so all media types are cancelled first and then awaited
1956        together, keeping the bounded wait to one timeout instead of one per media type.
1957
1958        :param provider_instance_id: The provider instance id to unschedule.
1959        :param clear_persisted_state: Whether to remove persisted schedule state from config.
1960        """
1961        await asyncio.gather(
1962            *(
1963                self.mass.tasks.unregister_scheduled_task_and_wait(
1964                    self._get_sync_task_id(provider_instance_id, media_type),
1965                    clear_persisted_state=clear_persisted_state,
1966                )
1967                for media_type in MediaType
1968            )
1969        )
1970
1971    def get_provider_sync_schedule(
1972        self, provider_instance_id: str, media_type: MediaType
1973    ) -> TaskSchedule | None:
1974        """Return the effective schedule for a provider sync task, if any."""
1975        task_id = self._get_sync_task_id(provider_instance_id, media_type)
1976        with suppress(InvalidDataError):
1977            task = self.mass.tasks.get_task(task_id)
1978            return task.schedule
1979        if not (
1980            provider := self.mass.get_provider(provider_instance_id, provider_type=MusicProvider)
1981        ):
1982            return None
1983        if not self.library_supported(provider, media_type):
1984            return None
1985        return provider.get_default_library_sync_schedule(media_type)
1986
1987    def match_provider_instances(
1988        self,
1989        item: MediaItemType,
1990    ) -> bool:
1991        """Match all provider instances for the given item."""
1992        mappings_added = False
1993        for provider_mapping in list(item.provider_mappings):
1994            if provider_mapping.is_unique:
1995                # unique mapping, no need to map
1996                continue
1997            if not (provider := self.mass.get_provider(provider_mapping.provider_instance)):
1998                continue
1999            if not isinstance(provider, MusicProvider):
2000                continue
2001            if not provider.is_streaming_provider:
2002                continue
2003            provider_instances = self.get_provider_instances(
2004                provider.domain, return_unavailable=True
2005            )
2006            if len(provider_instances) <= 1:
2007                # only a single instance, no need to map
2008                continue
2009            for prov_instance in provider_instances:
2010                if prov_instance.instance_id == provider.instance_id:
2011                    continue
2012                if any(
2013                    pm.provider_instance == prov_instance.instance_id
2014                    for pm in item.provider_mappings
2015                ):
2016                    # mapping already exists
2017                    continue
2018                # create additional mapping for other provider instances of the same provider
2019                item.provider_mappings.add(
2020                    ProviderMapping(
2021                        item_id=provider_mapping.item_id,
2022                        provider_domain=provider.domain,
2023                        provider_instance=prov_instance.instance_id,
2024                        available=provider_mapping.available,
2025                        is_unique=provider_mapping.is_unique,
2026                        audio_format=provider_mapping.audio_format,
2027                        url=provider_mapping.url,
2028                        details=provider_mapping.details,
2029                        in_library=None,
2030                    )
2031                )
2032                mappings_added = True
2033        return mappings_added
2034
2035    @api_command("music/add_provider_mapping", required_scope=Scope.LIBRARY_MANAGE)
2036    async def add_provider_mapping(
2037        self, media_type: MediaType, db_id: str, mapping: ProviderMapping
2038    ) -> None:
2039        """Add provider mapping to the given library item."""
2040        ctrl = self.get_controller(media_type)
2041        await ctrl.add_provider_mappings(db_id, [mapping])
2042
2043    @api_command("music/remove_provider_mapping", required_scope=Scope.LIBRARY_MANAGE)
2044    async def remove_provider_mapping(
2045        self, media_type: MediaType, db_id: str, mapping: ProviderMapping
2046    ) -> None:
2047        """Remove provider mapping from the given library item."""
2048        ctrl = self.get_controller(media_type)
2049        await ctrl.remove_provider_mapping(db_id, mapping.provider_instance, mapping.item_id)
2050
2051    @api_command("music/match_providers", required_scope=Scope.LIBRARY_MANAGE)
2052    async def match_providers(self, media_type: MediaType, db_id: str) -> None:
2053        """Search for mappings on all providers for the given library item."""
2054        ctrl = self.get_controller(media_type)
2055        db_item = await ctrl.get_library_item(db_id)
2056        # ctrl is chosen by media_type, so it matches db_item's runtime type
2057        await cast("MediaControllerBase[MediaItemType]", ctrl).match_providers(db_item)
2058
2059    async def update_provider_mapping(
2060        self,
2061        media_type: MediaType,
2062        db_id: str | int,
2063        provider_instance_id: str,
2064        provider_item_id: str,
2065        *,
2066        available: bool | Any = UNSET,
2067        in_library: bool | Any = UNSET,
2068        is_unique: bool | None | Any = UNSET,
2069        url: str | None | Any = UNSET,
2070        details: str | None | Any = UNSET,
2071        audio_format: AudioFormat | Any = UNSET,
2072    ) -> None:
2073        """Update an existing provider mapping for a library item."""
2074        ctrl = self.get_controller(media_type)
2075        await ctrl.update_provider_mapping(
2076            item_id=db_id,
2077            provider_instance_id=provider_instance_id,
2078            provider_item_id=provider_item_id,
2079            available=available,
2080            in_library=in_library,
2081            is_unique=is_unique,
2082            url=url,
2083            details=details,
2084            audio_format=audio_format,
2085        )
2086
2087    def queue_provider_mapping_correction_task(self) -> BackgroundTask:
2088        """Queue the provider mapping correction as a managed background task."""
2089        self._register_provider_mapping_correction_task()
2090        return self.mass.tasks.run_task(PROVIDER_MAPPING_CORRECTION_TASK_ID)
2091
2092    async def correct_multi_instance_provider_mappings(self) -> None:
2093        """Correct provider mappings for multi-instance providers."""
2094        self.logger.debug("Correcting provider mappings for multi-instance providers...")
2095        multi_instance_providers: set[str] = set()
2096        for provider in self.providers:
2097            if len(self.get_provider_instances(provider.domain)) > 1:
2098                multi_instance_providers.add(provider.instance_id)
2099        if not multi_instance_providers:
2100            return  # no multi-instance providers found, nothing to do
2101
2102        for ctrl in (
2103            self.albums,
2104            self.artists,
2105            self.tracks,
2106            self.playlists,
2107            self.radio,
2108            self.audiobooks,
2109            self.podcasts,
2110        ):
2111            async for db_item in ctrl.iter_library_items(
2112                provider=list(multi_instance_providers), library_items_only=False
2113            ):
2114                if self.match_provider_instances(db_item):
2115                    # ctrl is the per-type controller, so it matches db_item's runtime type
2116                    await cast("MediaControllerBase[MediaItemType]", ctrl).update_item_in_library(
2117                        db_item.item_id, db_item
2118                    )
2119                # prevent overwhelming the event loop
2120                await asyncio.sleep(0.2)
2121        self.logger.debug("Provider mappings correction done")
2122
2123    def library_supported(self, provider: Provider, media_type: MediaType) -> bool:
2124        """Return whether the provider declares LIBRARY support for the given media type."""
2125        if provider.type != ProviderType.MUSIC:
2126            return False
2127        if media_type == MediaType.ARTIST:
2128            return provider.supports_feature(ProviderFeature.LIBRARY_ARTISTS)
2129        if media_type == MediaType.ALBUM:
2130            return provider.supports_feature(ProviderFeature.LIBRARY_ALBUMS)
2131        if media_type == MediaType.TRACK:
2132            return provider.supports_feature(ProviderFeature.LIBRARY_TRACKS)
2133        if media_type == MediaType.PLAYLIST:
2134            return provider.supports_feature(ProviderFeature.LIBRARY_PLAYLISTS)
2135        if media_type == MediaType.RADIO:
2136            return provider.supports_feature(ProviderFeature.LIBRARY_RADIOS)
2137        if media_type == MediaType.AUDIOBOOK:
2138            return provider.supports_feature(ProviderFeature.LIBRARY_AUDIOBOOKS)
2139        if media_type == MediaType.PODCAST:
2140            return provider.supports_feature(ProviderFeature.LIBRARY_PODCASTS)
2141        return False
2142
2143    def library_edit_supported(self, provider: Provider, media_type: MediaType) -> bool:
2144        """Return whether the provider supports library add/remove for the given media type."""
2145        if provider.type != ProviderType.MUSIC:
2146            return False
2147        if media_type == MediaType.ARTIST:
2148            return provider.supports_feature(ProviderFeature.LIBRARY_ARTISTS_EDIT)
2149        if media_type == MediaType.ALBUM:
2150            return provider.supports_feature(ProviderFeature.LIBRARY_ALBUMS_EDIT)
2151        if media_type == MediaType.TRACK:
2152            return provider.supports_feature(ProviderFeature.LIBRARY_TRACKS_EDIT)
2153        if media_type == MediaType.PLAYLIST:
2154            return provider.supports_feature(ProviderFeature.LIBRARY_PLAYLISTS_EDIT)
2155        if media_type == MediaType.RADIO:
2156            return provider.supports_feature(ProviderFeature.LIBRARY_RADIOS_EDIT)
2157        if media_type == MediaType.AUDIOBOOK:
2158            return provider.supports_feature(ProviderFeature.LIBRARY_AUDIOBOOKS_EDIT)
2159        if media_type == MediaType.PODCAST:
2160            return provider.supports_feature(ProviderFeature.LIBRARY_PODCASTS_EDIT)
2161        return False
2162
2163    def library_favorites_edit_supported(self, provider: Provider, media_type: MediaType) -> bool:
2164        """Return whether the provider supports favorites add/remove for the given media type."""
2165        if provider.type != ProviderType.MUSIC:
2166            return False
2167        if media_type == MediaType.ARTIST:
2168            return provider.supports_feature(ProviderFeature.FAVORITE_ARTISTS_EDIT)
2169        if media_type == MediaType.ALBUM:
2170            return provider.supports_feature(ProviderFeature.FAVORITE_ALBUMS_EDIT)
2171        if media_type == MediaType.TRACK:
2172            return provider.supports_feature(ProviderFeature.FAVORITE_TRACKS_EDIT)
2173        if media_type == MediaType.PLAYLIST:
2174            return provider.supports_feature(ProviderFeature.FAVORITE_PLAYLISTS_EDIT)
2175        if media_type == MediaType.RADIO:
2176            return provider.supports_feature(ProviderFeature.FAVORITE_RADIOS_EDIT)
2177        if media_type == MediaType.AUDIOBOOK:
2178            return provider.supports_feature(ProviderFeature.FAVORITE_AUDIOBOOKS_EDIT)
2179        if media_type == MediaType.PODCAST:
2180            return provider.supports_feature(ProviderFeature.FAVORITE_PODCASTS_EDIT)
2181        return False
2182
2183    def library_sync_back_enabled(self, provider: Provider, media_type: MediaType) -> bool:
2184        """Return whether library sync back is enabled for the provider+media_type."""
2185        conf_value = provider.config.get_value(
2186            CONF_ENTRY_LIBRARY_SYNC_BACK.key, CONF_ENTRY_LIBRARY_SYNC_BACK.default_value
2187        )
2188        return bool(conf_value)
2189
2190    @api_command("music/item_by_name", required_scope=Scope.LIBRARY_READ, allow_impersonation=True)
2191    async def get_item_by_name(
2192        self,
2193        name: str,
2194        artist: str | None = None,
2195        album: str | None = None,
2196        media_type: MediaType | None = None,
2197    ) -> MediaItemType | ItemMapping | None:
2198        """Try to find a media item (such as a playlist) by name."""
2199        return await self._get_item_by_name(name, artist, album, media_type)
2200
2201    @api_command(
2202        "music/verify_item_uri", required_scope=Scope.LIBRARY_READ, allow_impersonation=True
2203    )
2204    async def verify_item_uri(self, uri: str) -> bool:
2205        """
2206        Verify whether a uri points to a valid, accessible item.
2207
2208        :param uri: The uri to verify.
2209        """
2210        return await self._handle_verify_item_uri(uri)
2211
2212    def _apply_user_provider_filter(
2213        self,
2214        providers: Iterable[ProviderInstanceType],
2215    ) -> list[ProviderInstanceType]:
2216        """Filter providers by the current user's music provider filter."""
2217        user = get_current_user()
2218        user_provider_filter = user.provider_filter if user else None
2219        if not user_provider_filter:
2220            return list(providers)
2221        return [
2222            p
2223            for p in providers
2224            if p.type != ProviderType.MUSIC or p.instance_id in user_provider_filter
2225        ]
2226
2227    async def _search_shareable_url(self, search_query: str) -> SearchResults | None:
2228        """
2229        Handle a search query that is a streaming provider public shareable URL.
2230
2231        Returns None if the query is not such a URL and a regular search must be done.
2232        """
2233        try:
2234            media_type, provider_instance_id_or_domain, item_id = await parse_uri(
2235                search_query, validate_id=True
2236            )
2237        except InvalidProviderURI:
2238            return None
2239        except InvalidProviderID as err:
2240            self.logger.warning("%s", str(err))
2241            return SearchResults()
2242        if provider_instance_id_or_domain not in PROVIDERS_WITH_SHAREABLE_URLS:
2243            return None
2244        try:
2245            item = await self.get_item(
2246                media_type=media_type,
2247                item_id=item_id,
2248                provider_instance_id_or_domain=provider_instance_id_or_domain,
2249            )
2250        except MusicAssistantError as err:
2251            self.logger.warning("%s", str(err))
2252            return SearchResults()
2253        if media_type == MediaType.ARTIST:
2254            return SearchResults(artists=[cast("Artist", item)])
2255        if media_type == MediaType.ALBUM:
2256            return SearchResults(albums=[cast("Album", item)])
2257        if media_type == MediaType.TRACK:
2258            return SearchResults(tracks=[cast("Track", item)])
2259        if media_type == MediaType.PLAYLIST:
2260            return SearchResults(playlists=[cast("Playlist", item)])
2261        if media_type == MediaType.AUDIOBOOK:
2262            return SearchResults(audiobooks=[cast("Audiobook", item)])
2263        if media_type == MediaType.PODCAST:
2264            return SearchResults(podcasts=[cast("Podcast", item)])
2265        return SearchResults()
2266
2267    async def _search_provider(
2268        self,
2269        search_query: str,
2270        provider_instance_id_or_domain: str,
2271        media_types: list[MediaType],
2272        limit: int = 10,
2273        skip_item_ids: set[tuple[MediaType, str, str]] | None = None,
2274    ) -> SearchResults | None:
2275        """
2276        Perform search on given provider, returns None if the search failed or timed out.
2277
2278        :param search_query: Search query
2279        :param provider_instance_id_or_domain: instance_id or domain of the provider
2280                                               to perform the search on.
2281        :param media_types: A list of media_types to include.
2282        :param limit: number of items to return in the search (per type).
2283        :param skip_item_ids: Optional set of (media_type, provider_domain, item_id)
2284                              tuples to filter out of the results.
2285        """
2286        prov = self.mass.get_provider(provider_instance_id_or_domain, provider_type=MusicProvider)
2287        if not prov:
2288            return SearchResults()
2289        if ProviderFeature.SEARCH not in prov.supported_features:
2290            return SearchResults()
2291
2292        # create safe search string
2293        search_query = search_query.replace("/", " ").replace("'", "")
2294        # use the per-provider cache so repeated and overlapping searches
2295        # do not hit the provider again
2296        cache_key = f"{search_query}-{'-'.join(sorted([mt.value for mt in media_types]))}-{limit}"
2297        if (
2298            cache := await self.mass.cache.get(
2299                key=cache_key,
2300                provider=prov.instance_id,
2301                category=CACHE_CATEGORY_SEARCH_RESULTS,
2302                base_class=SearchResults,
2303            )
2304        ) is not None:
2305            return filter_search_results(cast("SearchResults", cache), prov.domain, skip_item_ids)
2306        # run the provider search as a separate task (deduplicated by task_id so
2307        # identical concurrent searches share a single provider call) and wait for
2308        # it a limited amount of time only: a slow provider then contributes no
2309        # results now, while its search continues in the background so the result
2310        # is cached and available for a next search request
2311        task = self.mass.create_task(
2312            self._execute_provider_search(prov, search_query, media_types, limit, cache_key),
2313            task_id=f"provider_search_{prov.instance_id}_{cache_key}",
2314        )
2315        try:
2316            async with asyncio.timeout(SEARCH_PROVIDER_SOFT_TIMEOUT):
2317                prov_search_results = await asyncio.shield(task)
2318        except TimeoutError:
2319            self.logger.warning(
2320                "Search on provider %s did not return in time, "
2321                "the search continues in the background",
2322                prov.name,
2323            )
2324            return None
2325        if prov_search_results is None:
2326            return None
2327        return filter_search_results(prov_search_results, prov.domain, skip_item_ids)
2328
2329    async def _execute_provider_search(
2330        self,
2331        prov: MusicProvider,
2332        search_query: str,
2333        media_types: list[MediaType],
2334        limit: int,
2335        cache_key: str,
2336    ) -> SearchResults | None:
2337        """
2338        Execute the actual search on a provider and cache the result.
2339
2340        Returns None if the provider search failed or timed out. All errors are
2341        handled here (and not raised) as this coroutine runs as a background task
2342        that may outlive the request that started it.
2343        """
2344        try:
2345            async with asyncio.timeout(SEARCH_PROVIDER_HARD_TIMEOUT):
2346                result = await prov.search(search_query, media_types, limit)
2347        except TimeoutError:
2348            self.logger.warning("Search on provider %s timed out", prov.name)
2349            return None
2350        except MusicAssistantError as err:
2351            self.logger.warning("Search on provider %s failed: %s", prov.name, str(err))
2352            return None
2353        except Exception as err:
2354            self.logger.error("Search on provider %s failed: %s", prov.name, str(err), exc_info=err)
2355            return None
2356        # only successful results are cached, so failed or timed out
2357        # provider searches are simply retried on a next search
2358        await self._cache_search_results(
2359            cache_key,
2360            result,
2361            # plugin providers do not declare is_streaming_provider,
2362            # treat them as local so their results only get the short expiration
2363            SEARCH_CACHE_EXPIRATION_STREAMING_PROVIDER
2364            if getattr(prov, "is_streaming_provider", False)
2365            else SEARCH_CACHE_EXPIRATION_LOCAL_PROVIDER,
2366            prov.instance_id,
2367        )
2368        return result
2369
2370    async def _cache_search_results(
2371        self, cache_key: str, result: SearchResults, expiration: int, provider: str
2372    ) -> None:
2373        """Store search results in the cache, logging (instead of raising) any cache errors."""
2374        try:
2375            await self.mass.cache.set(
2376                key=cache_key,
2377                data=result.to_dict(),
2378                expiration=expiration,
2379                provider=provider,
2380                category=CACHE_CATEGORY_SEARCH_RESULTS,
2381            )
2382        except Exception as err:
2383            self.logger.warning("Failed to cache search results for %s: %s", provider, str(err))
2384
2385    def _get_covered_media_types(
2386        self, library_results: SearchResults, search_query: str
2387    ) -> set[tuple[MediaType, str]]:
2388        """
2389        Return the (media_type, provider domain/instance) pairs covered by the library.
2390
2391        A pair is considered covered when the library holds a (near) exact name match
2392        for the search query that is mapped to that provider.
2393        """
2394        covered: set[tuple[MediaType, str]] = set()
2395        # extract the artist and title part in case the
2396        # query is formatted as "artist - title"
2397        if " - " in search_query:
2398            artist_part, title_part = search_query.split(" - ", 1)
2399        else:
2400            artist_part, title_part = None, search_query
2401        items: Sequence[MediaItemType | ItemMapping]
2402        for items in (
2403            library_results.artists,
2404            library_results.albums,
2405            library_results.tracks,
2406            library_results.playlists,
2407            library_results.radio,
2408            library_results.audiobooks,
2409            library_results.podcasts,
2410        ):
2411            for item in items:
2412                if compare_strings(item.name, search_query, strict=False):
2413                    pass
2414                elif artist_part and compare_strings(item.name, title_part, strict=False):
2415                    # the item name matches the title part only,
2416                    # so the artist part must match one of the item artists
2417                    if not any(
2418                        compare_strings(artist.name, artist_part, strict=False)
2419                        for artist in getattr(item, "artists", [])
2420                    ):
2421                        continue
2422                else:
2423                    continue
2424                for prov_mapping in cast("MediaItemType", item).provider_mappings:
2425                    if not prov_mapping.available:
2426                        continue
2427                    covered.add((item.media_type, prov_mapping.provider_domain))
2428                    covered.add((item.media_type, prov_mapping.provider_instance))
2429        return covered
2430
2431    def _import_album_tracks_if_enabled(self, album: Album) -> None:
2432        """Import all album tracks into the library for providers that have this enabled."""
2433        for prov_mapping in album.provider_mappings:
2434            # only consider mappings the album was actually added on; additional
2435            # mappings auto-created for other instances of the same provider
2436            # (via match_provider_instances) carry in_library=None and must be skipped
2437            if not prov_mapping.in_library:
2438                continue
2439            provider = self.mass.get_provider(prov_mapping.provider_instance)
2440            if not isinstance(provider, MusicProvider):
2441                continue
2442            if not provider.library_sync_album_tracks_enabled():
2443                continue
2444            self.mass.create_task(provider.import_album_tracks(prov_mapping.item_id, album.name))
2445
2446    async def _get_provider_sound_effects(self, provider: MusicProvider) -> list[SoundEffect]:
2447        """Return all sound effect items from a single provider."""
2448        try:
2449            return [item async for item in provider.get_sound_effects()]
2450        except Exception as err:
2451            self.logger.warning(
2452                "Error while fetching sound effects from %s: %s",
2453                provider.name,
2454                str(err),
2455                exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
2456            )
2457            return []
2458
2459    def _create_provider_sync_handler(
2460        self, provider: MusicProvider, media_type: MediaType
2461    ) -> Callable[[], Awaitable[None]]:
2462        """Create the coroutine used for a managed provider sync task."""
2463
2464        async def run_sync() -> None:
2465            try:
2466                async with self._sync_lock:
2467                    # suppress per-item events during sync; a large library would otherwise
2468                    # emit one (serialized per client) for every item. Subscribers refresh
2469                    # on MUSIC_SYNC_COMPLETED and track progress via TASKS_UPDATED instead.
2470                    token = SUPPRESS_MEDIA_ITEM_UPDATES.set(True)
2471                    try:
2472                        await provider.sync_library(media_type)
2473                    finally:
2474                        SUPPRESS_MEDIA_ITEM_UPDATES.reset(token)
2475            finally:
2476                self.mass.call_later(
2477                    0,
2478                    self._handle_sync_completion_check,
2479                    task_id=MUSIC_SYNC_COMPLETION_CHECK_TASK_ID,
2480                )
2481
2482        return run_sync
2483
2484    def _get_sync_task_id(self, provider: MusicProvider | str, media_type: MediaType) -> str:
2485        """Return deterministic task id for a provider sync."""
2486        provider_instance = (
2487            provider.instance_id if isinstance(provider, MusicProvider) else provider
2488        )
2489        return f"music_sync_{provider_instance}_{media_type.value}"
2490
2491    def _get_sync_task_name(self, provider: MusicProvider, media_type: MediaType) -> str:
2492        """Return display name for a provider sync task."""
2493        return f"Sync {provider.name} {media_type.value}s"
2494
2495    def _get_sync_task_translation_key(self, media_type: MediaType) -> str:
2496        """Return translation key for a provider sync task."""
2497        if media_type == MediaType.ARTIST:
2498            return "sync_provider_artists"
2499        if media_type == MediaType.ALBUM:
2500            return "sync_provider_albums"
2501        if media_type == MediaType.TRACK:
2502            return "sync_provider_tracks"
2503        if media_type == MediaType.PLAYLIST:
2504            return "sync_provider_playlists"
2505        if media_type == MediaType.RADIO:
2506            return "sync_provider_radios"
2507        if media_type == MediaType.AUDIOBOOK:
2508            return "sync_provider_audiobooks"
2509        if media_type == MediaType.PODCAST:
2510            return "sync_provider_podcasts"
2511        return "settings.sync"
2512
2513    def _get_sync_task_metadata(
2514        self, provider: MusicProvider, media_type: MediaType
2515    ) -> TaskMetadata:
2516        """Return metadata for a provider sync task."""
2517        return {
2518            "task_domain": "music_sync",
2519            "provider_domain": provider.domain,
2520            "provider_instance": provider.instance_id,
2521            "provider_name": provider.name,
2522            "media_type": media_type.value,
2523        }
2524
2525    def _handle_sync_completion_check(self) -> None:
2526        """Run follow-up maintenance when no provider sync tasks remain active."""
2527        if self.active_sync_tasks:
2528            return
2529        self.mass.signal_event(EventType.MUSIC_SYNC_COMPLETED)
2530        self._queue_database_cleanup_task()
2531
2532    def _register_database_cleanup_task(self) -> BackgroundTask:
2533        """Register the recurring database cleanup background task."""
2534        utc_hour, utc_minute = local_clock_time_to_utc(5, 0)
2535        desired_schedule = TaskSchedule.daily(hour=utc_hour, minute=utc_minute)
2536        return self.mass.tasks.register_scheduled_task(
2537            task_id=DATABASE_CLEANUP_TASK_ID,
2538            name="Database cleanup",
2539            handler=self._cleanup_database,
2540            schedule=desired_schedule,
2541            translation_key="database_cleanup",
2542            translation_owner=self.translation_owner,
2543            metadata={
2544                "task_domain": "music_database_cleanup",
2545            },
2546            allow_retry=True,
2547        )
2548
2549    def _register_provider_mapping_correction_task(self) -> BackgroundTask:
2550        """Register the recurring provider mapping correction background task."""
2551        utc_hour, utc_minute = local_clock_time_to_utc(4, 0)
2552        desired_schedule = TaskSchedule.daily(every=30, hour=utc_hour, minute=utc_minute)
2553        return self.mass.tasks.register_scheduled_task(
2554            task_id=PROVIDER_MAPPING_CORRECTION_TASK_ID,
2555            name="Correct provider mappings",
2556            handler=self.correct_multi_instance_provider_mappings,
2557            schedule=desired_schedule,
2558            translation_key="correct_provider_mappings",
2559            translation_owner=self.translation_owner,
2560            metadata={
2561                "task_domain": "music_provider_mapping_correction",
2562            },
2563            allow_retry=True,
2564        )
2565
2566    def _queue_database_cleanup_task(self) -> BackgroundTask:
2567        """Queue the post-sync database cleanup as a managed background task."""
2568        self._register_database_cleanup_task()
2569        return self.mass.tasks.run_task(DATABASE_CLEANUP_TASK_ID)
2570
2571    async def _schedule_provider_mediatype_sync(
2572        self, provider: MusicProvider, media_type: MediaType, is_initial: bool = False
2573    ) -> None:
2574        """Schedule Library sync for given provider and media type."""
2575        # handle mediatype specific sync config
2576        conf_key = f"library_sync_{media_type}s"
2577        sync_conf: ConfigValueType = await self.mass.config.get_provider_config_value(
2578            provider.instance_id, conf_key
2579        )
2580        if not sync_conf:
2581            self.mass.tasks.unregister_scheduled_task(self._get_sync_task_id(provider, media_type))
2582            return
2583        self.mass.tasks.register_scheduled_task(
2584            task_id=self._get_sync_task_id(provider, media_type),
2585            name=self._get_sync_task_name(provider, media_type),
2586            handler=self._create_provider_sync_handler(provider, media_type),
2587            schedule=provider.get_default_library_sync_schedule(media_type),
2588            initial_delay=10 if is_initial else None,
2589            translation_key=self._get_sync_task_translation_key(media_type),
2590            translation_args=[provider.name],
2591            translation_owner=self.translation_owner,
2592            metadata=self._get_sync_task_metadata(provider, media_type),
2593            allow_retry=True,
2594        )
2595
2596    async def _get_user_for_provider(
2597        self, provider_mappings_or_instance_id: Iterable[ProviderMapping] | str
2598    ) -> User | None:
2599        """Try to get the MA User based on provider mappings and provider filter."""
2600        all_users = await self.mass.webserver.auth.list_users()
2601        for mapping_or_instance_id in provider_mappings_or_instance_id:
2602            for user in all_users:
2603                if not user.provider_filter:
2604                    continue
2605                if isinstance(mapping_or_instance_id, str):
2606                    if provider_mappings_or_instance_id in user.provider_filter:
2607                        return user
2608                elif mapping_or_instance_id.provider_instance in user.provider_filter:
2609                    return user
2610        return None
2611
2612    async def _upsert_playlog(self, entry: dict[str, Any]) -> None:
2613        """
2614        Write a playlog row, updating the existing row for the item/user if there is one.
2615
2616        Columns left out of the entry keep whatever the existing row holds, and
2617        `user_initiated` is sticky: once a play was explicitly user-initiated it stays that
2618        way for the lifetime of the row, so a later side-effect credit (an autoplay replay,
2619        or a track crediting its album/artist) can never demote it and drop the item out of
2620        the "recently played" recommendations.
2621
2622        The generic `database.upsert()` cannot express either half of that: the sticky OR is
2623        playlog-specific, and it needs an explicit conflict target because the playlog carries
2624        more than one unique constraint.
2625
2626        :param entry: The playlog column values to write, including all of
2627            `PLAYLOG_CONFLICT_KEYS`.
2628        """
2629        columns = list(entry)
2630        updates = [
2631            f"user_initiated = {DB_TABLE_PLAYLOG}.user_initiated OR excluded.user_initiated"
2632            if column == "user_initiated"
2633            else f"{column} = excluded.{column}"
2634            for column in columns
2635            if column not in PLAYLOG_CONFLICT_KEYS
2636        ]
2637        await self.database.execute_write(
2638            f"INSERT INTO {DB_TABLE_PLAYLOG} ({', '.join(columns)}) "
2639            f"VALUES ({', '.join(f':{column}' for column in columns)}) "
2640            f"ON CONFLICT({', '.join(PLAYLOG_CONFLICT_KEYS)}) DO UPDATE SET {', '.join(updates)}",
2641            entry,
2642        )
2643
2644    async def _credit_artist_plays(
2645        self,
2646        artists: Iterable[Artist | ItemMapping],
2647        *,
2648        timestamp: float,
2649        user_ids: list[str],
2650        queue_id: str | None,
2651        skip_ids: set[str],
2652    ) -> None:
2653        """Credit each (library-resolvable) artist with a play, skipping skip_ids."""
2654        for artist in artists:
2655            db_artist = await self.artists.get_library_item_by_prov_id(
2656                artist.item_id, artist.provider
2657            )
2658            if db_artist is None:
2659                continue
2660            if db_artist.item_id in skip_ids:
2661                self.logger.debug("Skipping already-credited artist '%s'", db_artist.name)
2662                continue
2663            await self.database.execute(
2664                f"UPDATE {self.artists.db_table} SET play_count = play_count + 1, "
2665                f"last_played = {timestamp} WHERE item_id = {db_artist.item_id}"
2666            )
2667            self.logger.debug("Credited play for artist '%s'", db_artist.name)
2668            playlog_entry: dict[str, Any] = {
2669                "item_id": db_artist.item_id,
2670                "provider": "library",
2671                "media_type": MediaType.ARTIST.value,
2672                "name": db_artist.name,
2673                "image": serialize_to_json(db_artist.image.to_dict()) if db_artist.image else None,
2674                "fully_played": True,
2675                "seconds_played": None,
2676                "timestamp": timestamp,
2677                "queue_id": queue_id,
2678                "user_initiated": False,
2679            }
2680            for user_id in user_ids:
2681                playlog_entry["userid"] = user_id
2682                await self._upsert_playlog(playlog_entry)
2683
2684    async def _credit_podcast_play(
2685        self,
2686        podcast: Podcast | ItemMapping,
2687        *,
2688        timestamp: float,
2689        user_ids: list[str],
2690        queue_id: str | None,
2691    ) -> None:
2692        """Credit the parent podcast with a play so the show surfaces in recently played."""
2693        # Resolve to the library item first, like _credit_artist_plays does, so an episode's
2694        # parent-podcast credit lands on the same library-scoped row as an explicit play of the
2695        # library show, instead of creating a separate provider-scoped duplicate.
2696        db_podcast = await self.podcasts.get_library_item_by_prov_id(
2697            podcast.item_id, podcast.provider
2698        )
2699        credited_podcast: Podcast | ItemMapping = db_podcast if db_podcast else podcast
2700        playlog_entry: dict[str, Any] = {
2701            "item_id": credited_podcast.item_id,
2702            "provider": "library" if db_podcast else podcast.provider,
2703            "media_type": MediaType.PODCAST.value,
2704            "name": credited_podcast.name,
2705            "image": serialize_to_json(credited_podcast.image.to_dict())
2706            if credited_podcast.image
2707            else None,
2708            "fully_played": True,
2709            "seconds_played": None,
2710            "timestamp": timestamp,
2711            "queue_id": queue_id,
2712            "user_initiated": False,
2713        }
2714        for user_id in user_ids:
2715            playlog_entry["userid"] = user_id
2716            await self._upsert_playlog(playlog_entry)
2717
2718    async def _get_item_by_name(
2719        self,
2720        name: str,
2721        artist: str | None = None,
2722        album: str | None = None,
2723        media_type: MediaType | None = None,
2724    ) -> MediaItemType | ItemMapping | None:
2725        """Try to find a media item (such as a playlist) by name."""
2726        # Future todo: enhance this method with AI capabilities to allow typos and
2727        # natural language.
2728        searchname = name.lower()
2729        allowed_media_types = [
2730            MediaType.PLAYLIST,
2731            MediaType.RADIO,
2732            MediaType.TRACK,
2733            MediaType.ALBUM,
2734            MediaType.ARTIST,
2735            MediaType.AUDIOBOOK,
2736            MediaType.PODCAST,
2737        ]
2738        if media_type in (None, MediaType.UNKNOWN):
2739            media_types = allowed_media_types
2740        elif media_type not in allowed_media_types:
2741            raise InvalidDataError(
2742                f"{media_type} is not a supported media_type. "
2743                f"Supported media_types are {allowed_media_types}"
2744            )
2745        else:
2746            media_types = [media_type]
2747        library_functions = [
2748            self.get_controller(media_type).library_items for media_type in media_types
2749        ]
2750        # prefer (exact) lookup in the library by name
2751        for func in library_functions:
2752            result = await func(search=searchname)
2753            for item in result:
2754                # handle optional artist filter
2755                if (
2756                    artist
2757                    and (artists := getattr(item, "artists", None))
2758                    and not any(x for x in artists if x.name.lower() == artist.lower())
2759                ):
2760                    continue
2761                # handle optional album filter
2762                if (
2763                    album
2764                    and (item_album := getattr(item, "album", None))
2765                    and item_album.name.lower() != album.lower()
2766                ):
2767                    continue
2768                if searchname == item.name.lower():
2769                    return item
2770        # nothing found in the library, fallback to global search
2771        search_name = name
2772        if album and artist:
2773            search_name = f"{artist} - {album} - {name}"
2774        elif album:
2775            search_name = f"{album} - {name}"
2776        elif artist:
2777            search_name = f"{artist} - {name}"
2778        search_results = await self.search(
2779            search_query=search_name,
2780            media_types=[media_type]
2781            if media_type and media_type != MediaType.UNKNOWN
2782            else MediaType.ALL,
2783            limit=8,
2784        )
2785        for results in (
2786            search_results.tracks,
2787            search_results.albums,
2788            search_results.playlists,
2789            search_results.artists,
2790            search_results.radio,
2791            search_results.audiobooks,
2792            search_results.podcasts,
2793        ):
2794            for _item in results:
2795                # simply return the first item because search is already sorted by best match
2796                return _item
2797        return None
2798
2799    async def _handle_verify_item_uri(self, uri: str) -> bool:
2800        user = get_current_user()
2801
2802        try:
2803            media_type, provider_instance_id_or_domain, item_id = await parse_uri(uri)
2804        except InvalidProviderURI, InvalidProviderID:
2805            return False
2806
2807        # fast return for a provider uri which is not part of a user with a provider filter
2808        if (
2809            provider_instance_id_or_domain != "library"
2810            and user
2811            and user.provider_filter
2812            and provider_instance_id_or_domain not in user.provider_filter
2813        ):
2814            return False
2815
2816        # verify that item itself exists
2817        try:
2818            item = await self.get_item(
2819                media_type=media_type,
2820                item_id=item_id,
2821                provider_instance_id_or_domain=provider_instance_id_or_domain,
2822                allow_update_metadata=False,  # no need trigger more methods
2823            )
2824        except MediaNotFoundError, NotImplementedError:
2825            # NotImplementedError: the uri has a valid format, but specifies an unknown media type
2826            return False
2827
2828        # non library item handling for users with no filter, or no user at all
2829        if (
2830            provider_instance_id_or_domain != "library"
2831            or not user
2832            or (user and not user.provider_filter)
2833            or isinstance(item, BrowseFolder)
2834        ):
2835            return True
2836
2837        # library item handling for users with provider filter
2838        for provider_mapping in item.provider_mappings:
2839            if provider_mapping.provider_instance in user.provider_filter:
2840                return True
2841
2842        return False
2843