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