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