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