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