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 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 media_type == MediaType.ARTIST:
2269            return provider.supports_feature(ProviderFeature.LIBRARY_ARTISTS)
2270        if media_type == MediaType.ALBUM:
2271            return provider.supports_feature(ProviderFeature.LIBRARY_ALBUMS)
2272        if media_type == MediaType.TRACK:
2273            return provider.supports_feature(ProviderFeature.LIBRARY_TRACKS)
2274        if media_type == MediaType.PLAYLIST:
2275            return provider.supports_feature(ProviderFeature.LIBRARY_PLAYLISTS)
2276        if media_type == MediaType.RADIO:
2277            return provider.supports_feature(ProviderFeature.LIBRARY_RADIOS)
2278        if media_type == MediaType.AUDIOBOOK:
2279            return provider.supports_feature(ProviderFeature.LIBRARY_AUDIOBOOKS)
2280        if media_type == MediaType.PODCAST:
2281            return provider.supports_feature(ProviderFeature.LIBRARY_PODCASTS)
2282        return False
2283
2284    def library_edit_supported(self, provider: Provider, media_type: MediaType) -> bool:
2285        """Return whether the provider supports library add/remove for the given media type."""
2286        if provider.type != ProviderType.MUSIC:
2287            return False
2288        if media_type == MediaType.ARTIST:
2289            return provider.supports_feature(ProviderFeature.LIBRARY_ARTISTS_EDIT)
2290        if media_type == MediaType.ALBUM:
2291            return provider.supports_feature(ProviderFeature.LIBRARY_ALBUMS_EDIT)
2292        if media_type == MediaType.TRACK:
2293            return provider.supports_feature(ProviderFeature.LIBRARY_TRACKS_EDIT)
2294        if media_type == MediaType.PLAYLIST:
2295            return provider.supports_feature(ProviderFeature.LIBRARY_PLAYLISTS_EDIT)
2296        if media_type == MediaType.RADIO:
2297            return provider.supports_feature(ProviderFeature.LIBRARY_RADIOS_EDIT)
2298        if media_type == MediaType.AUDIOBOOK:
2299            return provider.supports_feature(ProviderFeature.LIBRARY_AUDIOBOOKS_EDIT)
2300        if media_type == MediaType.PODCAST:
2301            return provider.supports_feature(ProviderFeature.LIBRARY_PODCASTS_EDIT)
2302        return False
2303
2304    def library_favorites_edit_supported(self, provider: Provider, media_type: MediaType) -> bool:
2305        """Return whether the provider supports favorites add/remove for the given media type."""
2306        if provider.type != ProviderType.MUSIC:
2307            return False
2308        if media_type == MediaType.ARTIST:
2309            return provider.supports_feature(ProviderFeature.FAVORITE_ARTISTS_EDIT)
2310        if media_type == MediaType.ALBUM:
2311            return provider.supports_feature(ProviderFeature.FAVORITE_ALBUMS_EDIT)
2312        if media_type == MediaType.TRACK:
2313            return provider.supports_feature(ProviderFeature.FAVORITE_TRACKS_EDIT)
2314        if media_type == MediaType.PLAYLIST:
2315            return provider.supports_feature(ProviderFeature.FAVORITE_PLAYLISTS_EDIT)
2316        if media_type == MediaType.RADIO:
2317            return provider.supports_feature(ProviderFeature.FAVORITE_RADIOS_EDIT)
2318        if media_type == MediaType.AUDIOBOOK:
2319            return provider.supports_feature(ProviderFeature.FAVORITE_AUDIOBOOKS_EDIT)
2320        if media_type == MediaType.PODCAST:
2321            return provider.supports_feature(ProviderFeature.FAVORITE_PODCASTS_EDIT)
2322        return False
2323
2324    def library_sync_back_enabled(self, provider: Provider, media_type: MediaType) -> bool:
2325        """Return whether library sync back is enabled for the provider+media_type."""
2326        conf_value = provider.config.get_value(
2327            CONF_ENTRY_LIBRARY_SYNC_BACK.key, CONF_ENTRY_LIBRARY_SYNC_BACK.default_value
2328        )
2329        return bool(conf_value)
2330
2331    @api_command("music/item_by_name", required_scope=Scope.LIBRARY_READ, allow_impersonation=True)
2332    async def get_item_by_name(
2333        self,
2334        name: str,
2335        artist: str | None = None,
2336        album: str | None = None,
2337        media_type: MediaType | None = None,
2338    ) -> MediaItemType | ItemMapping | None:
2339        """Try to find a media item (such as a playlist) by name."""
2340        return await self._get_item_by_name(name, artist, album, media_type)
2341
2342    @api_command(
2343        "music/verify_item_uri", required_scope=Scope.LIBRARY_READ, allow_impersonation=True
2344    )
2345    async def verify_item_uri(self, uri: str) -> bool:
2346        """
2347        Verify whether a uri points to a valid, accessible item.
2348
2349        :param uri: The uri to verify.
2350        """
2351        return await self._handle_verify_item_uri(uri)
2352
2353    def _apply_user_provider_filter(
2354        self,
2355        providers: Iterable[ProviderInstanceType],
2356    ) -> list[ProviderInstanceType]:
2357        """Filter providers by the current user's music provider filter."""
2358        user = get_current_user()
2359        user_provider_filter = user.provider_filter if user else None
2360        if not user_provider_filter:
2361            return list(providers)
2362        return [
2363            p
2364            for p in providers
2365            if p.type != ProviderType.MUSIC or p.instance_id in user_provider_filter
2366        ]
2367
2368    async def _search_shareable_url(self, search_query: str) -> SearchResults | None:
2369        """
2370        Handle a search query that is a streaming provider public shareable URL.
2371
2372        Returns None if the query is not such a URL and a regular search must be done.
2373        """
2374        try:
2375            media_type, provider_instance_id_or_domain, item_id = await parse_uri(
2376                search_query, validate_id=True
2377            )
2378        except InvalidProviderURI:
2379            return None
2380        except InvalidProviderID as err:
2381            self.logger.warning("%s", str(err))
2382            return SearchResults()
2383        if provider_instance_id_or_domain not in PROVIDERS_WITH_SHAREABLE_URLS:
2384            return None
2385        try:
2386            item = await self.get_item(
2387                media_type=media_type,
2388                item_id=item_id,
2389                provider_instance_id_or_domain=provider_instance_id_or_domain,
2390            )
2391        except MusicAssistantError as err:
2392            self.logger.warning("%s", str(err))
2393            return SearchResults()
2394        if media_type == MediaType.ARTIST:
2395            return SearchResults(artists=[cast("Artist", item)])
2396        if media_type == MediaType.ALBUM:
2397            return SearchResults(albums=[cast("Album", item)])
2398        if media_type == MediaType.TRACK:
2399            return SearchResults(tracks=[cast("Track", item)])
2400        if media_type == MediaType.PLAYLIST:
2401            return SearchResults(playlists=[cast("Playlist", item)])
2402        if media_type == MediaType.AUDIOBOOK:
2403            return SearchResults(audiobooks=[cast("Audiobook", item)])
2404        if media_type == MediaType.PODCAST:
2405            return SearchResults(podcasts=[cast("Podcast", item)])
2406        return SearchResults()
2407
2408    async def _search_provider(
2409        self,
2410        search_query: str,
2411        provider_instance_id_or_domain: str,
2412        media_types: list[MediaType],
2413        limit: int = 10,
2414        skip_item_ids: set[tuple[MediaType, str, str]] | None = None,
2415    ) -> SearchResults | None:
2416        """
2417        Perform search on given provider, returns None if the search failed or timed out.
2418
2419        :param search_query: Search query
2420        :param provider_instance_id_or_domain: instance_id or domain of the provider
2421                                               to perform the search on.
2422        :param media_types: A list of media_types to include.
2423        :param limit: number of items to return in the search (per type).
2424        :param skip_item_ids: Optional set of (media_type, provider_domain, item_id)
2425                              tuples to filter out of the results.
2426        """
2427        prov = self.mass.get_provider(provider_instance_id_or_domain, provider_type=MusicProvider)
2428        if not prov:
2429            return SearchResults()
2430        if ProviderFeature.SEARCH not in prov.supported_features:
2431            return SearchResults()
2432
2433        # create safe search string
2434        search_query = search_query.replace("/", " ").replace("'", "")
2435        # use the per-provider cache so repeated and overlapping searches
2436        # do not hit the provider again
2437        cache_key = f"{search_query}-{'-'.join(sorted([mt.value for mt in media_types]))}-{limit}"
2438        if (
2439            cache := await self.mass.cache.get(
2440                key=cache_key,
2441                provider=prov.instance_id,
2442                category=CACHE_CATEGORY_SEARCH_RESULTS,
2443                base_class=SearchResults,
2444            )
2445        ) is not None:
2446            return filter_search_results(cast("SearchResults", cache), prov.domain, skip_item_ids)
2447        # run the provider search as a separate task (deduplicated by task_id so
2448        # identical concurrent searches share a single provider call) and wait for
2449        # it a limited amount of time only: a slow provider then contributes no
2450        # results now, while its search continues in the background so the result
2451        # is cached and available for a next search request
2452        task = self.mass.create_task(
2453            self._execute_provider_search(prov, search_query, media_types, limit, cache_key),
2454            task_id=f"provider_search_{prov.instance_id}_{cache_key}",
2455        )
2456        try:
2457            async with asyncio.timeout(SEARCH_PROVIDER_SOFT_TIMEOUT):
2458                prov_search_results = await asyncio.shield(task)
2459        except TimeoutError:
2460            self.logger.warning(
2461                "Search on provider %s did not return in time, "
2462                "the search continues in the background",
2463                prov.name,
2464            )
2465            return None
2466        if prov_search_results is None:
2467            return None
2468        return filter_search_results(prov_search_results, prov.domain, skip_item_ids)
2469
2470    async def _execute_provider_search(
2471        self,
2472        prov: MusicProvider,
2473        search_query: str,
2474        media_types: list[MediaType],
2475        limit: int,
2476        cache_key: str,
2477    ) -> SearchResults | None:
2478        """
2479        Execute the actual search on a provider and cache the result.
2480
2481        Returns None if the provider search failed or timed out. All errors are
2482        handled here (and not raised) as this coroutine runs as a background task
2483        that may outlive the request that started it.
2484        """
2485        try:
2486            async with asyncio.timeout(SEARCH_PROVIDER_HARD_TIMEOUT):
2487                result = await prov.search(search_query, media_types, limit)
2488        except TimeoutError:
2489            self.logger.warning("Search on provider %s timed out", prov.name)
2490            return None
2491        except MusicAssistantError as err:
2492            self.logger.warning("Search on provider %s failed: %s", prov.name, str(err))
2493            return None
2494        except Exception as err:
2495            self.logger.error("Search on provider %s failed: %s", prov.name, str(err), exc_info=err)
2496            return None
2497        # only successful results are cached, so failed or timed out
2498        # provider searches are simply retried on a next search
2499        await self._cache_search_results(
2500            cache_key,
2501            result,
2502            # plugin providers do not declare is_streaming_provider,
2503            # treat them as local so their results only get the short expiration
2504            SEARCH_CACHE_EXPIRATION_STREAMING_PROVIDER
2505            if getattr(prov, "is_streaming_provider", False)
2506            else SEARCH_CACHE_EXPIRATION_LOCAL_PROVIDER,
2507            prov.instance_id,
2508        )
2509        return result
2510
2511    async def _cache_search_results(
2512        self, cache_key: str, result: SearchResults, expiration: int, provider: str
2513    ) -> None:
2514        """Store search results in the cache, logging (instead of raising) any cache errors."""
2515        try:
2516            await self.mass.cache.set(
2517                key=cache_key,
2518                data=result.to_dict(),
2519                expiration=expiration,
2520                provider=provider,
2521                category=CACHE_CATEGORY_SEARCH_RESULTS,
2522            )
2523        except Exception as err:
2524            self.logger.warning("Failed to cache search results for %s: %s", provider, str(err))
2525
2526    def _get_covered_media_types(
2527        self, library_results: SearchResults, search_query: str
2528    ) -> set[tuple[MediaType, str]]:
2529        """
2530        Return the (media_type, provider domain/instance) pairs covered by the library.
2531
2532        A pair is considered covered when the library holds a (near) exact name match
2533        for the search query that is mapped to that provider.
2534        """
2535        covered: set[tuple[MediaType, str]] = set()
2536        # extract the artist and title part in case the
2537        # query is formatted as "artist - title"
2538        if " - " in search_query:
2539            artist_part, title_part = search_query.split(" - ", 1)
2540        else:
2541            artist_part, title_part = None, search_query
2542        items: Sequence[MediaItemType | ItemMapping]
2543        for items in (
2544            library_results.artists,
2545            library_results.albums,
2546            library_results.tracks,
2547            library_results.playlists,
2548            library_results.radio,
2549            library_results.audiobooks,
2550            library_results.podcasts,
2551        ):
2552            for item in items:
2553                if compare_strings(item.name, search_query, strict=False):
2554                    pass
2555                elif artist_part and compare_strings(item.name, title_part, strict=False):
2556                    # the item name matches the title part only,
2557                    # so the artist part must match one of the item artists
2558                    if not any(
2559                        compare_strings(artist.name, artist_part, strict=False)
2560                        for artist in getattr(item, "artists", [])
2561                    ):
2562                        continue
2563                else:
2564                    continue
2565                for prov_mapping in cast("MediaItemType", item).provider_mappings:
2566                    if not prov_mapping.available:
2567                        continue
2568                    covered.add((item.media_type, prov_mapping.provider_domain))
2569                    covered.add((item.media_type, prov_mapping.provider_instance))
2570        return covered
2571
2572    def _import_album_tracks_if_enabled(self, album: Album) -> None:
2573        """Import all album tracks into the library for providers that have this enabled."""
2574        for prov_mapping in album.provider_mappings:
2575            # only consider mappings the album was actually added on; additional
2576            # mappings auto-created for other instances of the same provider
2577            # (via match_provider_instances) carry in_library=None and must be skipped
2578            if not prov_mapping.in_library:
2579                continue
2580            provider = self.mass.get_provider(prov_mapping.provider_instance)
2581            if not isinstance(provider, MusicProvider):
2582                continue
2583            if not provider.library_sync_album_tracks_enabled():
2584                continue
2585            self.mass.create_task(provider.import_album_tracks(prov_mapping.item_id, album.name))
2586
2587    async def _get_provider_sound_effects(self, provider: MusicProvider) -> list[SoundEffect]:
2588        """Return all sound effect items from a single provider."""
2589        try:
2590            return [item async for item in provider.get_sound_effects()]
2591        except Exception as err:
2592            self.logger.warning(
2593                "Error while fetching sound effects from %s: %s",
2594                provider.name,
2595                str(err),
2596                exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
2597            )
2598            return []
2599
2600    def _create_provider_sync_handler(
2601        self, provider: MusicProvider, media_type: MediaType
2602    ) -> Callable[[], Awaitable[None]]:
2603        """Create the coroutine used for a managed provider sync task."""
2604
2605        async def run_sync() -> None:
2606            try:
2607                async with self._sync_lock:
2608                    # suppress per-item events during sync; a large library would otherwise
2609                    # emit one (serialized per client) for every item. Subscribers refresh
2610                    # on MUSIC_SYNC_COMPLETED and track progress via TASKS_UPDATED instead.
2611                    token = SUPPRESS_MEDIA_ITEM_UPDATES.set(True)
2612                    try:
2613                        await provider.sync_library(media_type)
2614                    finally:
2615                        SUPPRESS_MEDIA_ITEM_UPDATES.reset(token)
2616            finally:
2617                self.mass.call_later(
2618                    0,
2619                    self._handle_sync_completion_check,
2620                    task_id=MUSIC_SYNC_COMPLETION_CHECK_TASK_ID,
2621                )
2622
2623        return run_sync
2624
2625    def _get_sync_task_id(self, provider: MusicProvider | str, media_type: MediaType) -> str:
2626        """Return deterministic task id for a provider sync."""
2627        provider_instance = (
2628            provider.instance_id if isinstance(provider, MusicProvider) else provider
2629        )
2630        return f"music_sync_{provider_instance}_{media_type.value}"
2631
2632    def _get_sync_task_name(self, provider: MusicProvider, media_type: MediaType) -> str:
2633        """Return display name for a provider sync task."""
2634        return f"Sync {provider.name} {media_type.value}s"
2635
2636    def _get_sync_task_translation_key(self, media_type: MediaType) -> str:
2637        """Return translation key for a provider sync task."""
2638        if media_type == MediaType.ARTIST:
2639            return "sync_provider_artists"
2640        if media_type == MediaType.ALBUM:
2641            return "sync_provider_albums"
2642        if media_type == MediaType.TRACK:
2643            return "sync_provider_tracks"
2644        if media_type == MediaType.PLAYLIST:
2645            return "sync_provider_playlists"
2646        if media_type == MediaType.RADIO:
2647            return "sync_provider_radios"
2648        if media_type == MediaType.AUDIOBOOK:
2649            return "sync_provider_audiobooks"
2650        if media_type == MediaType.PODCAST:
2651            return "sync_provider_podcasts"
2652        return "settings.sync"
2653
2654    def _get_sync_task_metadata(
2655        self, provider: MusicProvider, media_type: MediaType
2656    ) -> TaskMetadata:
2657        """Return metadata for a provider sync task."""
2658        return {
2659            "task_domain": "music_sync",
2660            "provider_domain": provider.domain,
2661            "provider_instance": provider.instance_id,
2662            "provider_name": provider.name,
2663            "media_type": media_type.value,
2664        }
2665
2666    def _handle_sync_completion_check(self) -> None:
2667        """Run follow-up maintenance when no provider sync tasks remain active."""
2668        if self.active_sync_tasks:
2669            return
2670        self.mass.signal_event(EventType.MUSIC_SYNC_COMPLETED)
2671        # freshly synced content is the only source of new duplicates, so the reconciliation
2672        # pass owes the library another walk; it starts once the current one reaches the end,
2673        # since rewinding right now would keep re-examining the same prefix forever
2674        self._set_track_reconciliation_state(self._track_reconciliation_cursor, True)
2675        self._queue_database_cleanup_task()
2676
2677    def _register_database_cleanup_task(self) -> BackgroundTask:
2678        """Register the recurring database cleanup background task."""
2679        utc_hour, utc_minute = local_clock_time_to_utc(5, 0)
2680        desired_schedule = TaskSchedule.daily(hour=utc_hour, minute=utc_minute)
2681        return self.mass.tasks.register_scheduled_task(
2682            task_id=DATABASE_CLEANUP_TASK_ID,
2683            name="Database cleanup",
2684            handler=self._cleanup_database,
2685            schedule=desired_schedule,
2686            translation_key="database_cleanup",
2687            translation_owner=self.translation_owner,
2688            metadata={
2689                "task_domain": "music_database_cleanup",
2690            },
2691            allow_retry=True,
2692        )
2693
2694    def _register_provider_mapping_correction_task(self) -> BackgroundTask:
2695        """Register the recurring provider mapping correction background task."""
2696        utc_hour, utc_minute = local_clock_time_to_utc(4, 0)
2697        desired_schedule = TaskSchedule.daily(every=30, hour=utc_hour, minute=utc_minute)
2698        return self.mass.tasks.register_scheduled_task(
2699            task_id=PROVIDER_MAPPING_CORRECTION_TASK_ID,
2700            name="Correct provider mappings",
2701            handler=self.correct_multi_instance_provider_mappings,
2702            schedule=desired_schedule,
2703            translation_key="correct_provider_mappings",
2704            translation_owner=self.translation_owner,
2705            metadata={
2706                "task_domain": "music_provider_mapping_correction",
2707            },
2708            allow_retry=True,
2709        )
2710
2711    def _register_track_reconciliation_task(self) -> BackgroundTask:
2712        """Register the recurring duplicate track reconciliation background task."""
2713        # runs every hour rather than spread across the day: it is bounded to a small
2714        # batch of candidates per run and never leaves the local database
2715        return self.mass.tasks.register_scheduled_task(
2716            task_id=TRACK_RECONCILIATION_TASK_ID,
2717            name="Reconcile duplicate tracks",
2718            handler=self._reconcile_duplicate_tracks,
2719            schedule=TaskSchedule.hourly(),
2720            translation_key="reconcile_duplicate_tracks",
2721            translation_owner=self.translation_owner,
2722            metadata={
2723                "task_domain": "music_track_reconciliation",
2724            },
2725            allow_retry=True,
2726        )
2727
2728    async def _reconcile_duplicate_tracks(self) -> None:
2729        """Merge a small batch of library tracks that are held twice across providers."""
2730        if self.active_sync_tasks:
2731            # a sync is still filling in albums and mappings, so hold off rather than
2732            # judge duplicates against a half-populated library
2733            update_current_task_progress_text("Waiting for music sync completion")
2734            return
2735        self._start_next_pass_if_due()
2736        if (cursor := self._track_reconciliation_cursor) is None:
2737            # the library has been walked end to end and nothing has been synced since,
2738            # so there is nothing to look for: skip the query rather than scan for a miss
2739            update_current_task_progress_text("No duplicate tracks found")
2740            return
2741        update_current_task_progress_text("Searching for duplicate tracks")
2742        rows = await self.database.get_rows_from_query(
2743            _DUPLICATE_TRACK_CANDIDATES_QUERY,
2744            {
2745                "max_duration_delta": TRACK_RECONCILIATION_MAX_DURATION_DELTA,
2746                "cursor_item_id_1": cursor[0],
2747                "cursor_item_id_2": cursor[1],
2748            },
2749            limit=TRACK_RECONCILIATION_BATCH_SIZE,
2750        )
2751        if not rows:
2752            self._set_track_reconciliation_state(None, self._track_reconciliation_rescan_due)
2753            update_current_task_progress_text("No duplicate tracks found")
2754            return
2755        merged = 0
2756        retry_due = False
2757        examined = cursor
2758        try:
2759            for index, row in enumerate(rows, 1):
2760                update_current_task_progress_from_index(
2761                    index, len(rows), f"Checking duplicate track {index}/{len(rows)}"
2762                )
2763                try:
2764                    if await self._merge_duplicate_track_pair(
2765                        int(row["item_id_1"]), int(row["item_id_2"])
2766                    ):
2767                        merged += 1
2768                except MediaNotFoundError:
2769                    # an earlier merge in this batch already absorbed one of the two rows
2770                    pass
2771                except MusicAssistantError as err:
2772                    # a pair that failed on something transient deserves another look
2773                    retry_due = True
2774                    report_current_task_failure(str(err))
2775                    self.logger.warning(
2776                        "Error while reconciling duplicate tracks %s and %s: %s",
2777                        row["item_id_1"],
2778                        row["item_id_2"],
2779                        str(err),
2780                        exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
2781                    )
2782                examined = (int(row["item_id_1"]), int(row["item_id_2"]))
2783        finally:
2784            # resume after the pair examined last, so candidates this run refused can never
2785            # starve the ones behind them, not even a further pair of the same track that the
2786            # batch boundary cut off. Recording it even when the run is cut short keeps the
2787            # pairs it did not reach for the next run rather than skipping past them.
2788            walked_to_end = len(rows) < TRACK_RECONCILIATION_BATCH_SIZE and examined == (
2789                int(rows[-1]["item_id_1"]),
2790                int(rows[-1]["item_id_2"]),
2791            )
2792            # a merge moves album and artist relations onto the surviving row, which can make
2793            # it a duplicate of a row this walk has already passed, so ask for another pass
2794            self._set_track_reconciliation_state(
2795                None if walked_to_end else examined,
2796                self._track_reconciliation_rescan_due or merged > 0 or retry_due,
2797            )
2798        update_current_task_progress(100, f"Merged {merged} duplicate track(s)")
2799
2800    def _restore_track_reconciliation_state(self) -> None:
2801        """Pick the duplicate track walk back up where the previous run left it."""
2802        cursor = self.mass.config.get_raw_core_config_value(
2803            self.domain, CONF_TRACK_RECONCILIATION_CURSOR, [0, 0]
2804        )
2805        self._track_reconciliation_cursor = (
2806            (int(cursor[0]), int(cursor[1])) if len(cursor) == 2 else None
2807        )
2808        self._track_reconciliation_rescan_due = bool(
2809            self.mass.config.get_raw_core_config_value(
2810                self.domain, CONF_TRACK_RECONCILIATION_RESCAN_DUE, False
2811            )
2812        )
2813
2814    def _set_track_reconciliation_state(
2815        self, cursor: tuple[int, int] | None, rescan_due: bool
2816    ) -> None:
2817        """
2818        Record how far the duplicate track walk has come, surviving a restart.
2819
2820        :param cursor: The pair examined last, or None once the walk reached the end.
2821        :param rescan_due: Whether a completed sync still owes the library another pass.
2822        """
2823        self._track_reconciliation_cursor = cursor
2824        self._track_reconciliation_rescan_due = rescan_due
2825        self.mass.config.set_raw_core_config_value(
2826            self.domain, CONF_TRACK_RECONCILIATION_CURSOR, list(cursor) if cursor else []
2827        )
2828        self.mass.config.set_raw_core_config_value(
2829            self.domain, CONF_TRACK_RECONCILIATION_RESCAN_DUE, rescan_due
2830        )
2831
2832    def _start_next_pass_if_due(self) -> None:
2833        """Rewind the duplicate track walk if a sync has added content and the walk is done."""
2834        # rewinding a walk still in progress would keep re-examining the same first
2835        # candidates, so a pending rescan waits for the current one to reach the end
2836        if not self._track_reconciliation_rescan_due:
2837            return
2838        if self._track_reconciliation_cursor is not None:
2839            return
2840        self._set_track_reconciliation_state((0, 0), False)
2841
2842    async def _albums_agree_on_edition(self, item_id_1: int, item_id_2: int) -> bool:
2843        """
2844        Check that two tracks share an album whose edition matches as well as its title.
2845
2846        :param item_id_1: Library ID of the first track.
2847        :param item_id_2: Library ID of the second track.
2848        """
2849        # the candidate query can only compare album titles, and an edition is held apart
2850        # from the title: without this an original and its remaster or deluxe edition look
2851        # like the same album whenever neither track carries a version of its own
2852        rows = await self.database.get_rows_from_query(
2853            _SHARED_ALBUM_EDITIONS_QUERY,
2854            {"item_id_1": item_id_1, "item_id_2": item_id_2},
2855        )
2856        return any(compare_version(row["version_1"], row["version_2"]) for row in rows)
2857
2858    async def _merge_duplicate_track_pair(self, item_id_1: int, item_id_2: int) -> bool:
2859        """
2860        Merge two candidate rows if they are confirmed to be the same track.
2861
2862        :param item_id_1: Library ID of the lower-numbered candidate row.
2863        :param item_id_2: Library ID of the higher-numbered candidate row.
2864        :return: True when the rows were merged, False when they were left alone.
2865        """
2866        track_1 = await self.tracks.get_library_item(item_id_1)
2867        track_2 = await self.tracks.get_library_item(item_id_2)
2868        # the candidate query already established that both rows sit at the same position on
2869        # an equally titled album, which is the album agreement strict mode looks for, so the
2870        # remaining check is run in non-strict mode. Its version check is reinstated here
2871        # explicitly: without it a remaster, remix or radio edit of equal length would be
2872        # accepted as the original.
2873        if not compare_version(track_1.version, track_2.version):
2874            return False
2875        if not await self._albums_agree_on_edition(item_id_1, item_id_2):
2876            return False
2877        if not compare_track(track_1, track_2, strict=False):
2878            return False
2879        # keep the row that carries the most provider mappings so the fewest mappings and
2880        # relations have to move; equal counts keep the oldest row, which the query orders first
2881        target, source = (
2882            (track_1, track_2)
2883            if len(track_1.provider_mappings) >= len(track_2.provider_mappings)
2884            else (track_2, track_1)
2885        )
2886        self.logger.debug(
2887            "Merging duplicate track %s (id %s) into id %s",
2888            target.name,
2889            source.item_id,
2890            target.item_id,
2891        )
2892        await self.tracks.merge_library_items(target.item_id, source.item_id)
2893        return True
2894
2895    def _queue_database_cleanup_task(self) -> BackgroundTask:
2896        """Queue the post-sync database cleanup as a managed background task."""
2897        self._register_database_cleanup_task()
2898        return self.mass.tasks.run_task(DATABASE_CLEANUP_TASK_ID)
2899
2900    async def _schedule_provider_mediatype_sync(
2901        self, provider: MusicProvider, media_type: MediaType, is_initial: bool = False
2902    ) -> None:
2903        """Schedule Library sync for given provider and media type."""
2904        # handle mediatype specific sync config
2905        conf_key = f"library_sync_{media_type}s"
2906        sync_conf: ConfigValueType = await self.mass.config.get_provider_config_value(
2907            provider.instance_id, conf_key
2908        )
2909        if not sync_conf:
2910            self.mass.tasks.unregister_scheduled_task(self._get_sync_task_id(provider, media_type))
2911            return
2912        self.mass.tasks.register_scheduled_task(
2913            task_id=self._get_sync_task_id(provider, media_type),
2914            name=self._get_sync_task_name(provider, media_type),
2915            handler=self._create_provider_sync_handler(provider, media_type),
2916            schedule=provider.get_default_library_sync_schedule(media_type),
2917            initial_delay=10 if is_initial else None,
2918            translation_key=self._get_sync_task_translation_key(media_type),
2919            translation_args=[provider.name],
2920            translation_owner=self.translation_owner,
2921            metadata=self._get_sync_task_metadata(provider, media_type),
2922            allow_retry=True,
2923        )
2924
2925    async def _get_user_for_provider(
2926        self, provider_mappings_or_instance_id: Iterable[ProviderMapping] | str
2927    ) -> User | None:
2928        """Try to get the MA User based on provider mappings and provider filter."""
2929        all_users = await self.mass.webserver.auth.list_users()
2930        for mapping_or_instance_id in provider_mappings_or_instance_id:
2931            for user in all_users:
2932                if not user.provider_filter:
2933                    continue
2934                if isinstance(mapping_or_instance_id, str):
2935                    if provider_mappings_or_instance_id in user.provider_filter:
2936                        return user
2937                elif mapping_or_instance_id.provider_instance in user.provider_filter:
2938                    return user
2939        return None
2940
2941    async def _upsert_playlog(self, entry: dict[str, Any]) -> None:
2942        """
2943        Write a playlog row, updating the existing row for the item/user if there is one.
2944
2945        Columns left out of the entry keep whatever the existing row holds, and
2946        `user_initiated` is sticky: once a play was explicitly user-initiated it stays that
2947        way for the lifetime of the row, so a later side-effect credit (an autoplay replay,
2948        or a track crediting its album/artist) can never demote it and drop the item out of
2949        the "recently played" recommendations.
2950
2951        The generic `database.upsert()` cannot express either half of that: the sticky OR is
2952        playlog-specific, and it needs an explicit conflict target because the playlog carries
2953        more than one unique constraint.
2954
2955        :param entry: The playlog column values to write, including all of
2956            `PLAYLOG_CONFLICT_KEYS`.
2957        """
2958        columns = list(entry)
2959        updates = [
2960            f"user_initiated = {DB_TABLE_PLAYLOG}.user_initiated OR excluded.user_initiated"
2961            if column == "user_initiated"
2962            else f"{column} = excluded.{column}"
2963            for column in columns
2964            if column not in PLAYLOG_CONFLICT_KEYS
2965        ]
2966        await self.database.execute_write(
2967            f"INSERT INTO {DB_TABLE_PLAYLOG} ({', '.join(columns)}) "
2968            f"VALUES ({', '.join(f':{column}' for column in columns)}) "
2969            f"ON CONFLICT({', '.join(PLAYLOG_CONFLICT_KEYS)}) DO UPDATE SET {', '.join(updates)}",
2970            entry,
2971        )
2972
2973    async def _credit_artist_plays(
2974        self,
2975        artists: Iterable[Artist | ItemMapping],
2976        *,
2977        timestamp: float,
2978        user_ids: list[str],
2979        queue_id: str | None,
2980        skip_ids: set[str],
2981    ) -> None:
2982        """Credit each (library-resolvable) artist with a play, skipping skip_ids."""
2983        for artist in artists:
2984            db_artist = await self.artists.get_library_item_by_prov_id(
2985                artist.item_id, artist.provider
2986            )
2987            if db_artist is None:
2988                continue
2989            if db_artist.item_id in skip_ids:
2990                self.logger.debug("Skipping already-credited artist '%s'", db_artist.name)
2991                continue
2992            await self.database.execute(
2993                f"UPDATE {self.artists.db_table} SET play_count = play_count + 1, "
2994                f"last_played = {timestamp} WHERE item_id = {db_artist.item_id}"
2995            )
2996            self.logger.debug("Credited play for artist '%s'", db_artist.name)
2997            playlog_entry: dict[str, Any] = {
2998                "item_id": db_artist.item_id,
2999                "provider": "library",
3000                "media_type": MediaType.ARTIST.value,
3001                "name": db_artist.name,
3002                "image": serialize_to_json(db_artist.image.to_dict()) if db_artist.image else None,
3003                "fully_played": True,
3004                "seconds_played": None,
3005                "timestamp": timestamp,
3006                "queue_id": queue_id,
3007                "user_initiated": False,
3008            }
3009            for user_id in user_ids:
3010                playlog_entry["userid"] = user_id
3011                await self._upsert_playlog(playlog_entry)
3012
3013    async def _credit_podcast_play(
3014        self,
3015        podcast: Podcast | ItemMapping,
3016        *,
3017        timestamp: float,
3018        user_ids: list[str],
3019        queue_id: str | None,
3020    ) -> None:
3021        """Credit the parent podcast with a play so the show surfaces in recently played."""
3022        # Resolve to the library item first, like _credit_artist_plays does, so an episode's
3023        # parent-podcast credit lands on the same library-scoped row as an explicit play of the
3024        # library show, instead of creating a separate provider-scoped duplicate.
3025        db_podcast = await self.podcasts.get_library_item_by_prov_id(
3026            podcast.item_id, podcast.provider
3027        )
3028        credited_podcast: Podcast | ItemMapping = db_podcast if db_podcast else podcast
3029        playlog_entry: dict[str, Any] = {
3030            "item_id": credited_podcast.item_id,
3031            "provider": "library" if db_podcast else podcast.provider,
3032            "media_type": MediaType.PODCAST.value,
3033            "name": credited_podcast.name,
3034            "image": serialize_to_json(credited_podcast.image.to_dict())
3035            if credited_podcast.image
3036            else None,
3037            "fully_played": True,
3038            "seconds_played": None,
3039            "timestamp": timestamp,
3040            "queue_id": queue_id,
3041            "user_initiated": False,
3042        }
3043        for user_id in user_ids:
3044            playlog_entry["userid"] = user_id
3045            await self._upsert_playlog(playlog_entry)
3046
3047    async def _get_item_by_name(
3048        self,
3049        name: str,
3050        artist: str | None = None,
3051        album: str | None = None,
3052        media_type: MediaType | None = None,
3053    ) -> MediaItemType | ItemMapping | None:
3054        """Try to find a media item (such as a playlist) by name."""
3055        # Future todo: enhance this method with AI capabilities to allow typos and
3056        # natural language.
3057        searchname = name.lower()
3058        allowed_media_types = [
3059            MediaType.PLAYLIST,
3060            MediaType.RADIO,
3061            MediaType.TRACK,
3062            MediaType.ALBUM,
3063            MediaType.ARTIST,
3064            MediaType.AUDIOBOOK,
3065            MediaType.PODCAST,
3066        ]
3067        if media_type in (None, MediaType.UNKNOWN):
3068            media_types = allowed_media_types
3069        elif media_type not in allowed_media_types:
3070            raise InvalidDataError(
3071                f"{media_type} is not a supported media_type. "
3072                f"Supported media_types are {allowed_media_types}"
3073            )
3074        else:
3075            media_types = [media_type]
3076        library_functions = [
3077            self.get_controller(media_type).library_items for media_type in media_types
3078        ]
3079        # prefer (exact) lookup in the library by name
3080        for func in library_functions:
3081            result = await func(search=searchname)
3082            for item in result:
3083                # handle optional artist filter
3084                if (
3085                    artist
3086                    and (artists := getattr(item, "artists", None))
3087                    and not any(x for x in artists if x.name.lower() == artist.lower())
3088                ):
3089                    continue
3090                # handle optional album filter
3091                if (
3092                    album
3093                    and (item_album := getattr(item, "album", None))
3094                    and item_album.name.lower() != album.lower()
3095                ):
3096                    continue
3097                if searchname == item.name.lower():
3098                    return item
3099        # nothing found in the library, fallback to global search
3100        search_name = name
3101        if album and artist:
3102            search_name = f"{artist} - {album} - {name}"
3103        elif album:
3104            search_name = f"{album} - {name}"
3105        elif artist:
3106            search_name = f"{artist} - {name}"
3107        search_results = await self.search(
3108            search_query=search_name,
3109            media_types=[media_type]
3110            if media_type and media_type != MediaType.UNKNOWN
3111            else MediaType.ALL,
3112            limit=8,
3113        )
3114        for results in (
3115            search_results.tracks,
3116            search_results.albums,
3117            search_results.playlists,
3118            search_results.artists,
3119            search_results.radio,
3120            search_results.audiobooks,
3121            search_results.podcasts,
3122        ):
3123            for _item in results:
3124                # simply return the first item because search is already sorted by best match
3125                return _item
3126        return None
3127
3128    async def _handle_verify_item_uri(self, uri: str) -> bool:
3129        user = get_current_user()
3130
3131        try:
3132            media_type, provider_instance_id_or_domain, item_id = await parse_uri(uri)
3133        except InvalidProviderURI, InvalidProviderID:
3134            return False
3135
3136        # fast return for a provider uri which is not part of a user with a provider filter
3137        if (
3138            provider_instance_id_or_domain != "library"
3139            and user
3140            and user.provider_filter
3141            and provider_instance_id_or_domain not in user.provider_filter
3142        ):
3143            return False
3144
3145        # verify that item itself exists
3146        try:
3147            item = await self.get_item(
3148                media_type=media_type,
3149                item_id=item_id,
3150                provider_instance_id_or_domain=provider_instance_id_or_domain,
3151                allow_update_metadata=False,  # no need trigger more methods
3152            )
3153        except MediaNotFoundError, NotImplementedError:
3154            # NotImplementedError: the uri has a valid format, but specifies an unknown media type
3155            return False
3156
3157        # non library item handling for users with no filter, or no user at all
3158        if (
3159            provider_instance_id_or_domain != "library"
3160            or not user
3161            or (user and not user.provider_filter)
3162            or isinstance(item, BrowseFolder)
3163        ):
3164            return True
3165
3166        # library item handling for users with provider filter
3167        for provider_mapping in item.provider_mappings:
3168            if provider_mapping.provider_instance in user.provider_filter:
3169                return True
3170
3171        return False
3172