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