music-assistant-server

1"""Main Spotify provider implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import os
7import shutil
8import time
9from collections import OrderedDict
10from collections.abc import AsyncGenerator, Sequence
11from contextlib import suppress
12from dataclasses import dataclass
13from datetime import datetime
14from pathlib import Path
15from typing import Any, cast
16
17import aiohttp
18from music_assistant_models.config_entries import ConfigEntry
19from music_assistant_models.enums import (
20    ConfigEntryType,
21    ImageType,
22    MediaType,
23    ProviderFeature,
24    StreamType,
25)
26from music_assistant_models.errors import (
27    AudioError,
28    LoginFailed,
29    MediaNotFoundError,
30    ProviderUnavailableError,
31    RateLimited,
32    ResourceTemporarilyUnavailable,
33    UnsupportedFeaturedException,
34)
35from music_assistant_models.media_items import (
36    Album,
37    Artist,
38    Audiobook,
39    BrowseFolder,
40    ItemMapping,
41    MediaItemImage,
42    MediaItemType,
43    Playlist,
44    Podcast,
45    PodcastEpisode,
46    ProviderMapping,
47    SearchResults,
48    Track,
49    UniqueList,
50)
51from music_assistant_models.media_items.metadata import MediaItemChapter
52from music_assistant_models.streamdetails import StreamDetails
53from orjson import JSONDecodeError
54
55from music_assistant.constants import CONF_ENTRY_UNOFFICIAL_PROVIDER
56from music_assistant.controllers.cache import use_cache
57from music_assistant.helpers.app_vars import app_var
58from music_assistant.helpers.json import SerializableType, json_loads
59from music_assistant.helpers.throttle_retry import ThrottlerManager, throttle_with_retries
60from music_assistant.helpers.util import lock
61from music_assistant.models.music_provider import MusicProvider, ProviderStreamLimitError
62from music_assistant.providers.spotify_connect.base import (
63    AUDIO_QUALITY_LOSSLESS,
64    AUDIO_QUALITY_OPTIONS,
65)
66
67from .backends import LibrespotBackend, SoloistBackend, SpotifyPlaybackBackend
68from .constants import (
69    BACKEND_SOLOIST,
70    CONF_ACCOUNT_ID,
71    CONF_AUDIO_QUALITY,
72    CONF_CLIENT_ID,
73    CONF_PLAYBACK_BACKEND,
74    CONF_REFRESH_TOKEN_DEV,
75    CONF_REFRESH_TOKEN_GLOBAL,
76    CONF_SPOTIFY_NORMALIZATION,
77    CONF_SYNC_AUDIOBOOK_PROGRESS,
78    CONF_SYNC_PODCAST_PROGRESS,
79    CREDENTIALS_FILE,
80    LIKED_SONGS_FAKE_PLAYLIST_ID_PREFIX,
81    SOLOIST_DATA_DIR_NAME,
82)
83from .helpers import get_spotify_token
84from .parsers import (
85    parse_album,
86    parse_artist,
87    parse_audiobook,
88    parse_playlist,
89    parse_podcast,
90    parse_podcast_episode,
91    parse_track,
92)
93
94_PLAYLIST_PAGINATION_STATE_LIMIT = 32
95
96
97class NotModifiedError(Exception):
98    """Exception raised when a resource has not been modified."""
99
100
101@dataclass(slots=True)
102class _PlaylistPaginationState:
103    """Hold the synchronization and metadata snapshot for one playlist endpoint."""
104
105    lock: asyncio.Lock
106    snapshot: dict[str, Any] | None = None
107
108
109class SpotifyProvider(MusicProvider):
110    """Implementation of a Spotify MusicProvider."""
111
112    # Global session (MA's client ID) - always present
113    _auth_info_global: dict[str, Any] | None = None
114    # Developer session (user's custom client ID) - optional
115    _auth_info_dev: dict[str, Any] | None = None
116    _sp_user: dict[str, Any] | None = None
117    _audiobooks_supported = False
118    _playlist_pagination_states: OrderedDict[tuple[str, bool], _PlaylistPaginationState]
119    # True if user has configured a custom client ID with valid authentication
120    dev_session_active: bool = False
121    throttler: ThrottlerManager
122    backend: SpotifyPlaybackBackend
123
124    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
125        """
126        Return Config entries to setup this provider.
127
128        Authentication is handled by the setup flow (see setup_flow.py); only the genuine
129        options are configurable here.
130        """
131        # audiobook progress sync is only offered where the account's region supports audiobooks
132        audiobooks_supported = bool(getattr(self, "audiobooks_supported", False))
133        return (
134            CONF_ENTRY_UNOFFICIAL_PROVIDER,
135            ConfigEntry(
136                key=CONF_SPOTIFY_NORMALIZATION,
137                type=ConfigEntryType.BOOLEAN,
138                default_value=True,
139                required=False,
140                # librespot hands over Spotify's own file untouched, so there is
141                # nothing on that backend to normalize with
142                hidden=self.get_setup_value(CONF_PLAYBACK_BACKEND) != BACKEND_SOLOIST,
143            ),
144            ConfigEntry(
145                key=CONF_AUDIO_QUALITY,
146                type=ConfigEntryType.STRING,
147                default_value=AUDIO_QUALITY_LOSSLESS,
148                required=False,
149                options=AUDIO_QUALITY_OPTIONS,
150                # librespot streams Spotify's own file untouched, so there is
151                # nothing to choose there
152                hidden=self.get_setup_value(CONF_PLAYBACK_BACKEND) != BACKEND_SOLOIST,
153            ),
154            ConfigEntry(
155                key=CONF_SYNC_PODCAST_PROGRESS,
156                type=ConfigEntryType.BOOLEAN,
157                default_value=True,
158                category="sync_options",
159            ),
160            ConfigEntry(
161                key=CONF_SYNC_AUDIOBOOK_PROGRESS,
162                type=ConfigEntryType.BOOLEAN,
163                default_value=False,
164                category="sync_options",
165                hidden=not audiobooks_supported,
166            ),
167        )
168
169    async def handle_async_init(self) -> None:
170        """Handle async initialization of the provider."""
171        self.cache_dir = os.path.join(self.mass.cache_path, self.instance_id)
172        self._playlist_pagination_states = OrderedDict()
173        # Default throttler for global session (heavy rate limited)
174        self.throttler = ThrottlerManager(rate_limit=1, period=2)
175
176        # playback authorization is independent of the Web API tokens
177        self.backend = self._create_backend()
178        await self.backend.setup()
179        try:
180            # try login which will raise if it fails (logs in global session)
181            await self.login()
182
183            # Check if user has a custom client ID with valid dev token
184            client_id = self.get_setup_value(CONF_CLIENT_ID)
185            dev_token = self.get_setup_value(CONF_REFRESH_TOKEN_DEV)
186
187            if client_id and dev_token and self._sp_user:
188                await self.login_dev()
189                # Verify user matches
190                userinfo = await self._get_data("me", use_global_session=False)
191                if userinfo["id"] != self._sp_user["id"]:
192                    raise LoginFailed(
193                        "Developer session must use the same Spotify account as the main session."
194                    )
195                # loosen the throttler when a custom client id is used
196                self.throttler = ThrottlerManager(rate_limit=45, period=30)
197                self.dev_session_active = True
198                self.logger.info("Developer Spotify session active.")
199
200            self._audiobooks_supported = await self._test_audiobook_support()
201            if not self._audiobooks_supported:
202                self.logger.info(
203                    "Audiobook support disabled: Audiobooks are not available in your region. "
204                    "See https://support.spotify.com/us/authors/article/audiobooks-availability/ "
205                    "for supported countries."
206                )
207            # login material the other backend left behind is of no further use:
208            # remove it — only now that the load succeeded, so a failed load (and
209            # its config rollback) still has the working credential
210            await asyncio.to_thread(self._remove_unused_playback_credentials)
211        except BaseException:
212            # a failed load is never registered, so unload() will not run:
213            # release whatever the backend acquired (e.g. the shared pulse
214            # capture server) before propagating
215            with suppress(Exception):
216                await self.backend.unload()
217            raise
218
219    async def unload(self, is_removed: bool = False) -> None:
220        """Handle close/cleanup of the provider."""
221        try:
222            if (backend := getattr(self, "backend", None)) is not None:
223                await backend.unload()
224        finally:
225            if is_removed:
226                # Both hold reusable login material - the soloist session in the
227                # storage dir, librespot's credential in the cache - so a removed
228                # instance keeps neither, even if the teardown above failed.
229                await asyncio.to_thread(self._remove_login_material)
230
231    @property
232    def spotify_normalization_configured(self) -> bool:
233        """
234        Return whether the configuration asks Spotify to normalize this audio.
235
236        Only the soloist backend can: librespot hands over Spotify's file
237        untouched, so its audio arrives at the master's own level.
238        """
239        return self._soloist_backend is not None and bool(
240            # the default is stated here too: get_value answers with the argument,
241            # not the entry's default, if the key was never parsed into the config
242            self.config.get_value(CONF_SPOTIFY_NORMALIZATION, True)
243        )
244
245    def delivers_normalized_audio(self, streamdetails: StreamDetails) -> bool:
246        """
247        Return whether Spotify's own loudness normalization handles this audio.
248
249        The session serving this item's queue answers for itself. The engine reads
250        its settings only at startup, so a setting changed mid-playback must not make
251        the streams core normalize on top of what the engine is still doing - it
252        takes effect on the next playback instead.
253
254        :param streamdetails: Stream details of the item being asked about.
255        """
256        backend = self._soloist_backend
257        if backend is not None and (live := backend.session_normalizes(streamdetails)) is not None:
258            return live
259        return self.spotify_normalization_configured
260
261    def delivers_crossfaded_audio(self, streamdetails: StreamDetails) -> bool | None:
262        """
263        Return whether Spotify's own engine crossfades this playback.
264
265        Only the soloist backend can: it keeps one session across items and is handed
266        the queue's crossfade duration when that session starts, so the overlap lives
267        in the audio it delivers. librespot fetches every track on its own. The session
268        serving this item's queue answers for the item's own boundaries, because it read
269        that duration once - a setting changed mid-playback applies to the next session,
270        not to the audio this one is still serving.
271
272        :param streamdetails: Stream details of the item being asked about.
273        """
274        backend = self._soloist_backend
275        return backend.session_crossfades(streamdetails) if backend is not None else False
276
277    @property
278    def max_concurrent_streams(self) -> int:
279        """
280        Return how many source streams Music Assistant may run against this provider.
281
282        Two on either playback backend: a Spotify account tolerates two
283        concurrent librespot fetches (main + playback), and on the Soloist
284        backend the item that is ending and the item that continues from the
285        same session are two streams reading it in turn.
286        """
287        return 2
288
289    @property
290    def audiobooks_supported(self) -> bool:
291        """Check if audiobooks are supported for this user/region."""
292        return self._audiobooks_supported
293
294    @property
295    def audiobook_progress_sync_enabled(self) -> bool:
296        """Check if audiobook progress sync is enabled."""
297        return bool(self.config.get_value(CONF_SYNC_AUDIOBOOK_PROGRESS, False))
298
299    @property
300    def podcast_progress_sync_enabled(self) -> bool:
301        """Check if played status sync is enabled."""
302        value = self.config.get_value(CONF_SYNC_PODCAST_PROGRESS, True)
303        return bool(value) if value is not None else True
304
305    @property
306    def supported_features(self) -> set[ProviderFeature]:
307        """Return the features supported by this Provider."""
308        features = self._supported_features.copy()
309        # Add audiobook features if enabled
310        if self.audiobooks_supported:
311            features.add(ProviderFeature.LIBRARY_AUDIOBOOKS)
312            features.add(ProviderFeature.LIBRARY_AUDIOBOOKS_EDIT)
313        return features
314
315    @property
316    def account_id(self) -> str | None:
317        """Return the Spotify user id of the logged-in account, if known."""
318        return str(self._sp_user["id"]) if self._sp_user else None
319
320    @property
321    def instance_name_postfix(self) -> str | None:
322        """Return a (default) instance name postfix for this provider instance."""
323        if self._sp_user:
324            return str(self._sp_user["display_name"])
325        return None
326
327    async def get_diagnostics(self) -> dict[str, SerializableType]:
328        """Return diagnostics info for this provider to include in diagnostics reports."""
329        return {
330            "logged_in": self._sp_user is not None,
331            "token_expires_in_sec": (
332                round(self._auth_info_global["expires_at"] - time.time())
333                if self._auth_info_global
334                else None
335            ),
336            "dev_session_active": self.dev_session_active,
337            "playback_backend": str(self.get_setup_value(CONF_PLAYBACK_BACKEND) or "librespot"),
338            "audiobooks_supported": self._audiobooks_supported,
339            **(await self.backend.get_diagnostics() if hasattr(self, "backend") else {}),
340        }
341
342    ## Library retrieval methods (generators)
343    async def get_library_artists(self) -> AsyncGenerator[Artist]:
344        """Retrieve library artists from spotify."""
345        endpoint = "me/following"
346        while True:
347            spotify_artists = await self._get_data(
348                endpoint,
349                type="artist",
350                limit=50,
351            )
352            for item in spotify_artists["artists"]["items"]:
353                if item and item["id"]:
354                    yield parse_artist(item, self)
355            if spotify_artists["artists"]["next"]:
356                endpoint = spotify_artists["artists"]["next"]
357                endpoint = endpoint.replace("https://api.spotify.com/v1/", "")
358            else:
359                break
360
361    async def get_library_albums(self) -> AsyncGenerator[Album]:
362        """Retrieve library albums from the provider."""
363        async for item in self._get_all_items("me/albums"):
364            if item["album"] and item["album"]["id"]:
365                yield parse_album(item["album"], self)
366
367    async def get_library_tracks(self) -> AsyncGenerator[Track]:
368        """Retrieve library tracks from the provider."""
369        async for item in self._get_all_items("me/tracks"):
370            if item and item["track"] and item["track"]["id"]:
371                yield parse_track(item["track"], self)
372
373    async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
374        """Retrieve library podcasts from spotify."""
375        async for item in self._get_all_items("me/shows"):
376            if item["show"] and item["show"]["id"]:
377                show_obj = item["show"]
378                # Filter out audiobooks - they have a distinctive description format
379                description = show_obj.get("description", "")
380                if description.startswith("Author(s):") and "Narrator(s):" in description:
381                    continue
382                yield parse_podcast(show_obj, self)
383
384    async def get_library_audiobooks(self) -> AsyncGenerator[Audiobook]:
385        """Retrieve library audiobooks from spotify."""
386        if not self.audiobooks_supported:
387            return
388        async for item in self._get_all_items("me/audiobooks"):
389            if item and item["id"]:
390                # Parse the basic audiobook
391                audiobook = parse_audiobook(item, self)
392                # Add chapters from Spotify API data
393                await self._add_audiobook_chapters(audiobook)
394                yield audiobook
395
396    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
397        """
398        Retrieve playlists from the provider.
399
400        Note: We use the global session here because playlists like "Daily Mix"
401        are only returned when using the non-dev (global) token.
402        """
403        yield await self._get_liked_songs_playlist()
404        async for item in self._get_all_items("me/playlists", use_global_session=True):
405            if item and item["id"]:
406                yield parse_playlist(item, self)
407
408    async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
409        """
410        Browse Spotify items, including curated sections (new releases, genres & moods).
411
412        :param path: The path to browse (e.g. provider_id:// or provider_id://new-releases).
413        """
414        path_parts = path.split("://")[1].split("/") if "://" in path else []
415        subpath = path_parts[0] if path_parts else None
416        sub_subpath = path_parts[1] if len(path_parts) > 1 else None
417        locale = self.mass.metadata.locale
418
419        if subpath == "new-releases":
420            return await self._get_new_releases()
421
422        if subpath == "categories" and sub_subpath:
423            return await self._get_category_playlists(sub_subpath, locale)
424
425        if subpath == "categories":
426            return await self._get_categories(locale)
427
428        # For root path, add curated folders on top of standard library folders.
429        # At the root the path always ends in "://", so curated paths can be appended directly.
430        if not subpath:
431            curated: list[BrowseFolder] = [
432                BrowseFolder(
433                    item_id="new-releases",
434                    provider=self.instance_id,
435                    path=f"{path}new-releases",
436                    name="New Releases",
437                    translation_key="new_releases",
438                    is_playable=True,
439                ),
440                BrowseFolder(
441                    item_id="categories",
442                    provider=self.instance_id,
443                    path=f"{path}categories",
444                    name="Genres & Moods",
445                    translation_key="genres_and_moods",
446                    is_playable=False,
447                ),
448            ]
449            standard = await super().browse(path)
450            return [*curated, *standard]
451
452        return await super().browse(path)
453
454    @use_cache()
455    async def search(
456        self, search_query: str, media_types: list[MediaType] | None = None, limit: int = 5
457    ) -> SearchResults:
458        """
459        Perform search on musicprovider.
460
461        :param search_query: Search query.
462        :param media_types: A list of media_types to include.
463        :param limit: Number of items to return in the search (per type).
464        """
465        searchresult = SearchResults()
466        if media_types is None:
467            return searchresult
468
469        searchtype = self._build_search_types(media_types)
470        if not searchtype:
471            return searchresult
472
473        search_query = search_query.replace("'", "")
474        offset = 0
475        page_limit = min(limit, 10)
476
477        while True:
478            api_result = await self._get_data(
479                "search", q=search_query, type=searchtype, limit=page_limit, offset=offset
480            )
481            items_received = self._process_search_results(api_result, searchresult)
482
483            offset += page_limit
484            if offset >= limit or items_received < page_limit:
485                break
486
487        return searchresult
488
489    @use_cache()
490    async def get_artist(self, prov_artist_id: str) -> Artist:
491        """Get full artist details by id."""
492        artist_obj = await self._get_data(f"artists/{prov_artist_id}")
493        return parse_artist(artist_obj, self)
494
495    @use_cache()
496    async def get_album(self, prov_album_id: str) -> Album:
497        """Get full album details by id."""
498        album_obj = await self._get_data(f"albums/{prov_album_id}")
499        return parse_album(album_obj, self)
500
501    @use_cache()
502    async def get_track(self, prov_track_id: str) -> Track:
503        """Get full track details by id."""
504        track_obj = await self._get_data(f"tracks/{prov_track_id}")
505        return parse_track(track_obj, self)
506
507    @use_cache()
508    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
509        """Get full playlist details by id."""
510        if prov_playlist_id == self._get_liked_songs_playlist_id():
511            return await self._get_liked_songs_playlist()
512
513        # Check cache to see if this playlist requires global token
514        use_global = await self._playlist_requires_global_token(prov_playlist_id)
515        if use_global:
516            playlist_obj = await self._get_data(
517                f"playlists/{prov_playlist_id}", use_global_session=True
518            )
519            return parse_playlist(playlist_obj, self)
520
521        # Try with dev token first (if available), fallback to global on 400 error
522        # Some playlists like Spotify-owned (Daily Mix) or Liked Songs only work with global token
523        try:
524            playlist_obj = await self._get_data(f"playlists/{prov_playlist_id}")
525            return parse_playlist(playlist_obj, self)
526        except MediaNotFoundError:
527            if self.dev_session_active:
528                # Remember that this playlist requires global token
529                await self._set_playlist_requires_global_token(prov_playlist_id)
530                playlist_obj = await self._get_data(
531                    f"playlists/{prov_playlist_id}", use_global_session=True
532                )
533                return parse_playlist(playlist_obj, self)
534            raise
535
536    @use_cache()
537    async def get_podcast(self, prov_podcast_id: str) -> Podcast:
538        """Get full podcast details by id."""
539        podcast_obj = await self._get_data(f"shows/{prov_podcast_id}")
540        if not podcast_obj:
541            raise MediaNotFoundError(f"Podcast not found: {prov_podcast_id}")
542        return parse_podcast(podcast_obj, self)
543
544    @use_cache()
545    async def get_audiobook(self, prov_audiobook_id: str) -> Audiobook:
546        """Get full audiobook details by id."""
547        if not self.audiobooks_supported:
548            raise UnsupportedFeaturedException("Audiobooks are not supported with this account")
549
550        audiobook_obj = await self._get_data(f"audiobooks/{prov_audiobook_id}")
551        if not audiobook_obj:
552            raise MediaNotFoundError(f"Audiobook not found: {prov_audiobook_id}")
553
554        # Parse basic audiobook without chapters first
555        audiobook = parse_audiobook(audiobook_obj, self)
556
557        # Add chapters from Spotify API data
558        await self._add_audiobook_chapters(audiobook)
559
560        # Note: Resume position will be handled by MA's internal system
561        # which calls get_resume_position() when needed
562
563        return audiobook
564
565    async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
566        """Get all podcast episodes."""
567        podcast = await self.get_podcast(prov_podcast_id)
568
569        # Get (cached) episode data
570        episodes_data = await self._get_podcast_episodes_data(prov_podcast_id)
571
572        # API lists newest-first; number down so bigger position = newer
573        total = len(episodes_data)
574        for idx, episode_data in enumerate(episodes_data):
575            episode = parse_podcast_episode(episode_data, self, podcast)
576            episode.position = total - idx
577
578            # Set played status if sync is enabled and resume data exists
579            if self.podcast_progress_sync_enabled and "resume_point" in episode_data:
580                resume_point = episode_data["resume_point"]
581                fully_played = resume_point.get("fully_played", False)
582                position_ms = resume_point.get("resume_position_ms", 0)
583
584                episode.fully_played = fully_played or None
585                episode.resume_position_ms = position_ms if position_ms > 0 else None
586
587            yield episode
588
589    @use_cache(86400)  # 24 hours
590    async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
591        """Get full podcast episode details by id."""
592        episode_obj = await self._get_data(f"episodes/{prov_episode_id}", market="from_token")
593        if not episode_obj:
594            raise MediaNotFoundError(f"Episode not found: {prov_episode_id}")
595        return parse_podcast_episode(episode_obj, self)
596
597    async def get_resume_position(
598        self, item_id: str, media_type: MediaType
599    ) -> tuple[bool, int, datetime | None]:
600        """Get resume position for episode/audiobook from Spotify."""
601        if media_type == MediaType.PODCAST_EPISODE:
602            if not self.podcast_progress_sync_enabled:
603                raise NotImplementedError("Spotify podcast resume sync disabled in settings")
604
605            try:
606                episode_obj = await self._get_data(f"episodes/{item_id}", market="from_token")
607            except MediaNotFoundError:
608                raise NotImplementedError("Episode not found on Spotify")
609            except (ResourceTemporarilyUnavailable, aiohttp.ClientError) as e:
610                self.logger.debug(f"Error fetching episode {item_id}: {e}")
611                raise NotImplementedError("Unable to fetch episode data from Spotify")
612
613            if (
614                not episode_obj
615                or "resume_point" not in episode_obj
616                or not episode_obj["resume_point"]
617            ):
618                raise NotImplementedError("No resume point data from Spotify")
619
620            resume_point = episode_obj["resume_point"]
621            fully_played = resume_point.get("fully_played", False)
622            position_ms = resume_point.get("resume_position_ms", 0)
623            return fully_played, position_ms, None
624
625        if media_type == MediaType.AUDIOBOOK:
626            if not self.audiobooks_supported:
627                raise NotImplementedError("Audiobook support is disabled")
628            if not self.audiobook_progress_sync_enabled:
629                raise NotImplementedError("Spotify audiobook resume sync disabled in settings")
630
631            try:
632                chapters_data = await self._get_audiobook_chapters_data(item_id)
633                if not chapters_data:
634                    raise NotImplementedError("No chapters data available")
635
636                total_position_ms = 0
637                fully_played = True
638
639                for chapter in chapters_data:
640                    resume_point = chapter.get("resume_point", {})
641                    chapter_fully_played = resume_point.get("fully_played", False)
642                    chapter_position_ms = resume_point.get("resume_position_ms", 0)
643
644                    if chapter_fully_played:
645                        total_position_ms += chapter.get("duration_ms", 0)
646                    elif chapter_position_ms > 0:
647                        total_position_ms += chapter_position_ms
648                        fully_played = False
649                        break
650                    else:
651                        fully_played = False
652                        break
653
654                return fully_played, total_position_ms, None
655
656            except (MediaNotFoundError, ResourceTemporarilyUnavailable, aiohttp.ClientError) as e:
657                self.logger.debug(f"Failed to get audiobook resume position for {item_id}: {e}")
658                raise NotImplementedError("Unable to get audiobook resume position from Spotify")
659
660        else:
661            raise NotImplementedError(f"Resume position not supported for {media_type}")
662
663    async def on_played(
664        self,
665        media_type: MediaType,
666        prov_item_id: str,
667        fully_played: bool,
668        position: int,
669        media_item: MediaItemType,
670        is_playing: bool = False,
671    ) -> None:
672        """
673        Call when an episode/audiobook is played in MA.
674
675        MA automatically handles internal position tracking - this method is for
676        provider-specific actions like syncing to external services.
677        """
678        if media_type == MediaType.PODCAST_EPISODE:
679            if not isinstance(media_item, PodcastEpisode):
680                return
681
682            # Log the playback for monitoring/debugging
683            safe_position = position or 0
684            if media_item.duration > 0:
685                completion_percentage = (safe_position / media_item.duration) * 100
686            else:
687                completion_percentage = 0
688
689            self.logger.debug(
690                f"Episode played: {prov_item_id} at {safe_position}s "
691                f"({completion_percentage:.1f}%, fully_played: {fully_played})"
692            )
693
694            # Note: No API exists to sync playback position back to Spotify for episodes
695            # MA handles all internal position tracking automatically
696
697        elif media_type == MediaType.AUDIOBOOK:
698            if not isinstance(media_item, Audiobook):
699                return
700
701            # Log the playback for monitoring/debugging
702            safe_position = position or 0
703            if media_item.duration > 0:
704                completion_percentage = (safe_position / media_item.duration) * 100
705            else:
706                completion_percentage = 0
707
708            self.logger.debug(
709                f"Audiobook played: {prov_item_id} at {safe_position}s "
710                f"({completion_percentage:.1f}%, fully_played: {fully_played})"
711            )
712
713            # Note: No API exists to sync playback position back to Spotify for audiobooks
714            # MA handles all internal position tracking automatically
715
716            # The resume position will be automatically updated by MA's internal tracking
717            # and will be retrieved via get_audiobook() which combines MA + Spotify positions
718
719    @use_cache(86400 * 365, allow_expired_cache=True)  # 1 year - album track listings are immutable
720    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
721        """Get all album tracks for given album id."""
722        return [
723            parse_track(item, self)
724            async for item in self._get_all_items(f"albums/{prov_album_id}/tracks")
725            if item["id"]
726        ]
727
728    @use_cache(3600 * 3, allow_expired_cache=True)  # 3 hours
729    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
730        """Get playlist tracks."""
731        is_liked_songs = prov_playlist_id == self._get_liked_songs_playlist_id()
732        uri = "me/tracks" if is_liked_songs else f"playlists/{prov_playlist_id}/items"
733
734        # Liked songs always require global session
735        # For other playlists, call get_playlist first to trigger the fallback logic
736        # and populate the cache for which token to use
737        if is_liked_songs:
738            use_global = True
739        else:
740            # This call is cached and will determine/cache if global token is needed
741            await self.get_playlist(prov_playlist_id)
742            use_global = await self._playlist_requires_global_token(prov_playlist_id)
743
744        page_size = 50
745        offset = page * page_size
746        known_global = use_global
747
748        while True:
749            try:
750                meta = await self._get_playlist_pagination_meta(uri, page, use_global)
751                cache_checksum = meta["etag"]
752                total = meta["total"]
753
754                # Spotify has started returning 5xx for offset >= total on some
755                # playlists (notably algorithmic ones like Daily Mix). The retry
756                # storm that follows surfaces as "No playable items found".
757                if total and offset >= total:
758                    spotify_result = {"total": total, "items": []}
759                else:
760                    spotify_result = await self._get_data_with_caching(
761                        uri,
762                        cache_checksum,
763                        limit=page_size,
764                        offset=offset,
765                        use_global_session=use_global,
766                    )
767                break
768            except MediaNotFoundError:
769                if use_global or not self.dev_session_active:
770                    raise
771                # Development Mode exposes metadata but restricts items for non-owned playlists.
772                use_global = True
773
774        if use_global and not known_global:
775            await self._set_playlist_requires_global_token(prov_playlist_id)
776
777        result: list[Track] = []
778        total = spotify_result.get("total", 0)
779        items = spotify_result.get("items", [])
780        # playlists/{id}/items is transitioning from item["track"] to item["item"]
781        # during Spotify's Feb 2026 rollout, so accept either shape.
782        for index, item in enumerate(items, 1):
783            # Spotify wraps/recycles items for offsets beyond the playlist size,
784            # so we need to break when we've reached the total.
785            if (offset + index) > total:
786                break
787            track_data = item and (item.get("item") or item.get("track"))
788            if not (track_data and track_data.get("id")):
789                continue
790            track = parse_track(track_data, self)
791            track.position = offset + index
792            result.append(track)
793        return result
794
795    @use_cache(86400 * 14, allow_expired_cache=True)  # 14 days
796    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
797        """Get a list of all albums for the given artist."""
798        try:
799            return [
800                parse_album(item, self)
801                async for item in self._get_all_items(
802                    f"artists/{prov_artist_id}/albums?include_groups=album,single,compilation",
803                    limit=10,
804                )
805                if (item and item["id"])
806            ]
807        except MediaNotFoundError:
808            self.logger.warning("Unable to fetch albums for artist %s", prov_artist_id)
809            return []
810
811    @use_cache(86400 * 14, allow_expired_cache=True)  # 14 days
812    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
813        """Get a list of 10 most popular tracks for the given artist."""
814        try:
815            artist = await self.get_artist(prov_artist_id)
816            endpoint = f"artists/{prov_artist_id}/top-tracks"
817            items = await self._get_data(endpoint)
818            return [
819                parse_track(item, self, artist=artist)
820                for item in items["tracks"]
821                if (item and item["id"])
822            ]
823        except MediaNotFoundError:
824            self.logger.warning(
825                "Top tracks search for artist %s appears to have been removed by Spotify for this account.",
826                prov_artist_id,
827            )
828            return []
829
830    async def library_add(self, item: MediaItemType) -> bool:
831        """Add item to library."""
832        uri_type_map = {
833            MediaType.ARTIST: "artist",
834            MediaType.ALBUM: "album",
835            MediaType.TRACK: "track",
836            MediaType.PLAYLIST: "playlist",
837            MediaType.PODCAST: "show",
838            MediaType.AUDIOBOOK: "audiobook",
839        }
840        if item.media_type == MediaType.AUDIOBOOK and not self.audiobooks_supported:
841            return False
842        uri_type = uri_type_map.get(item.media_type)
843        if not uri_type:
844            return False
845        uri = f"spotify:{uri_type}:{item.item_id}"
846        await self._put_data("me/library", uris=uri)
847        return True
848
849    async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
850        """Remove item from library."""
851        uri_type_map = {
852            MediaType.ARTIST: "artist",
853            MediaType.ALBUM: "album",
854            MediaType.TRACK: "track",
855            MediaType.PLAYLIST: "playlist",
856            MediaType.PODCAST: "show",
857            MediaType.AUDIOBOOK: "audiobook",
858        }
859        if media_type == MediaType.AUDIOBOOK and not self.audiobooks_supported:
860            return False
861        uri_type = uri_type_map.get(media_type)
862        if not uri_type:
863            return False
864        uri = f"spotify:{uri_type}:{prov_item_id}"
865        await self._delete_data("me/library", uris=uri)
866        return True
867
868    async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
869        """Add track(s) to playlist."""
870        track_uris = [f"spotify:track:{track_id}" for track_id in prov_track_ids]
871        data = {"uris": track_uris}
872        await self._post_data(f"playlists/{prov_playlist_id}/items", data=data)
873
874    async def remove_playlist_tracks(
875        self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
876    ) -> None:
877        """Remove track(s) from playlist."""
878        track_uris = []
879        for pos in positions_to_remove:
880            uri = f"playlists/{prov_playlist_id}/items"
881            spotify_result = await self._get_data(uri, limit=1, offset=pos - 1)
882            for item in spotify_result["items"]:
883                track_data = item and (item.get("item") or item.get("track"))
884                if not (track_data and track_data.get("id")):
885                    continue
886                track_uris.append({"uri": f"spotify:track:{track_data['id']}"})
887        data = {"items": track_uris}
888        await self._delete_data(f"playlists/{prov_playlist_id}/items", data=data)
889
890    async def create_playlist(self, name: str, media_types: set[MediaType]) -> Playlist:
891        """Create a new playlist on provider with given name."""
892        data = {"name": name, "public": False}
893        new_playlist = await self._post_data("me/playlists", data=data)
894        self._fix_create_playlist_api_bug(new_playlist)
895        return parse_playlist(new_playlist, self)
896
897    @use_cache(86400 * 14, allow_expired_cache=True)  # 14 days
898    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
899        """Retrieve a dynamic list of tracks based on the provided item."""
900        # Recommendations endpoint is only available on global session (not developer API)
901        # https://developer.spotify.com/blog/2024-11-27-changes-to-the-web-api
902        endpoint = "recommendations"
903        items = await self._get_data(
904            endpoint, seed_tracks=prov_track_id, limit=limit, use_global_session=True
905        )
906        return [parse_track(item, self) for item in items["tracks"] if (item and item["id"])]
907
908    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
909        """Return content details for the given track/episode/audiobook when it will be streamed."""
910        if media_type == MediaType.AUDIOBOOK and self.audiobooks_supported:
911            chapters_data = await self._get_audiobook_chapters_data(item_id)
912            if not chapters_data:
913                raise MediaNotFoundError(f"No chapters found for audiobook {item_id}")
914
915            # Calculate total duration and convert to seconds for StreamDetails
916            total_duration_ms = sum(chapter.get("duration_ms", 0) for chapter in chapters_data)
917            duration_seconds = total_duration_ms // 1000
918
919            # Create chapter URIs for streaming
920            chapter_uris = []
921            for chapter in chapters_data:
922                chapter_id = chapter["id"]
923                chapter_uri = f"spotify:episode:{chapter_id}"
924                chapter_uris.append(chapter_uri)
925
926            return StreamDetails(
927                item_id=item_id,
928                provider=self.instance_id,
929                media_type=MediaType.AUDIOBOOK,
930                # what Spotify serves, for display; the bytes that actually
931                # arrive are described by decoded_audio_format
932                audio_format=self.backend.source_audio_format(MediaType.AUDIOBOOK),
933                decoded_audio_format=self.backend.handoff_audio_format,
934                stream_type=StreamType.CUSTOM,
935                is_realtime=self.backend.is_realtime,
936                allow_seek=True,
937                can_seek=True,
938                duration=duration_seconds,
939                data={"chapters": chapter_uris, "chapters_data": chapters_data},
940            )
941
942        # For all other media types (tracks, podcast episodes)
943        return StreamDetails(
944            item_id=item_id,
945            provider=self.instance_id,
946            media_type=media_type,
947            audio_format=self.backend.source_audio_format(media_type),
948            decoded_audio_format=self.backend.handoff_audio_format,
949            stream_type=StreamType.CUSTOM,
950            is_realtime=self.backend.is_realtime,
951            allow_seek=True,
952            can_seek=True,
953        )
954
955    async def get_audio_stream(
956        self, streamdetails: StreamDetails, seek_position: int = 0
957    ) -> AsyncGenerator[bytes]:
958        """Get audio stream from Spotify via librespot."""
959        if streamdetails.media_type == MediaType.AUDIOBOOK and isinstance(streamdetails.data, dict):
960            chapter_uris = streamdetails.data.get("chapters", [])
961            chapters_data = streamdetails.data.get("chapters_data", [])
962
963            # Calculate which chapter to start from based on seek_position
964            seek_position_ms = seek_position * 1000
965            current_seek_ms = seek_position_ms
966            start_chapter = 0
967
968            if seek_position > 0 and chapters_data:
969                accumulated_duration_ms = 0
970
971                for i, chapter_data in enumerate(chapters_data):
972                    chapter_duration_ms = chapter_data.get("duration_ms", 0)
973
974                    if accumulated_duration_ms + chapter_duration_ms > seek_position_ms:
975                        start_chapter = i
976                        current_seek_ms = seek_position_ms - accumulated_duration_ms
977                        break
978                    accumulated_duration_ms += chapter_duration_ms
979                else:
980                    start_chapter = len(chapter_uris) - 1
981                    current_seek_ms = 0
982
983            # Convert back to seconds for librespot
984            current_seek_seconds = int(current_seek_ms // 1000)
985
986            # Stream chapters starting from the calculated position
987            consecutive_failures = 0
988            for i in range(start_chapter, len(chapter_uris)):
989                chapter_uri = chapter_uris[i]
990                chapter_seek = current_seek_seconds if i == start_chapter else 0
991
992                try:
993                    chunk_count = 0
994                    async for chunk in self.backend.stream_spotify_uri(
995                        chapter_uri, chapter_seek, streamdetails=streamdetails
996                    ):
997                        yield chunk
998                        chunk_count += 1
999                    if chunk_count > 0:
1000                        consecutive_failures = 0
1001                except ProviderStreamLimitError:
1002                    # capacity, not a broken chapter: skipping ahead would burn
1003                    # chapters and end as a plain error, which costs the item its
1004                    # availability and the caller its chance to wait or reselect
1005                    raise
1006                except Exception as e:
1007                    self.logger.warning("Chapter %s streaming failed", i + 1)
1008                    consecutive_failures += 1
1009                    if consecutive_failures >= 3:
1010                        raise AudioError("Audiobook streaming failed") from e
1011                    continue
1012        else:
1013            # Handle normal tracks and podcast episodes
1014            media_type = (
1015                "episode" if streamdetails.media_type == MediaType.PODCAST_EPISODE else "track"
1016            )
1017            spotify_uri = f"spotify:{media_type}:{streamdetails.item_id}"
1018            async for chunk in self.backend.stream_spotify_uri(
1019                spotify_uri, seek_position, streamdetails=streamdetails
1020            ):
1021                yield chunk
1022
1023    @lock
1024    async def login(self, force_refresh: bool = False) -> dict[str, Any]:
1025        """
1026        Log-in Spotify global session and return Auth/token info.
1027
1028        This uses MA's global client ID which has full API access but heavy rate limits.
1029        """
1030        # return the cached access token while it is still valid (refreshed before expiry)
1031        if (
1032            not force_refresh
1033            and self._auth_info_global
1034            and (self._auth_info_global["expires_at"] > (time.time() + 600))
1035        ):
1036            return self._auth_info_global
1037        # read the refresh token from the persisted store rather than the in-memory config copy,
1038        # which can lag a rotation and would make us refresh with a stale (revoked) token
1039        if not (refresh_token := self._stored_refresh_token(CONF_REFRESH_TOKEN_GLOBAL)):
1040            raise LoginFailed("Authentication required")
1041
1042        try:
1043            auth_info = await get_spotify_token(
1044                self.mass.http_session,
1045                app_var("spotify_client_id"),  # Always use MA's global client ID
1046                refresh_token,
1047                "global",
1048            )
1049            self.logger.debug("Successfully refreshed global access token")
1050        except LoginFailed as err:
1051            if "revoked" in str(err) or "invalid_grant" in str(err):
1052                # Spotify rotates the refresh token on refresh and revokes the previous one.
1053                # If the stored token was rotated while this refresh was in flight, the token
1054                # we tried is merely stale, so keep the newer one instead of forcing re-auth.
1055                if not self._refresh_token_superseded(CONF_REFRESH_TOKEN_GLOBAL, refresh_token):
1056                    self._update_setup_data(CONF_REFRESH_TOKEN_GLOBAL, None)
1057                    if self.available:
1058                        self.unload_with_error(err)
1059            elif self.available:
1060                self.mass.create_task(self.mass.unload_provider_with_error(self.instance_id, err))
1061            raise
1062
1063        # make sure that our updated creds get stored in memory + config
1064        self._auth_info_global = auth_info
1065        # Spotify revokes the previous refresh token only when it rotates one, so on rotation
1066        # persist immediately to ensure the new token survives a crash within the debounced-save
1067        # window and avoids a forced re-auth; an unchanged token uses the normal debounced save.
1068        token_rotated = auth_info["refresh_token"] != refresh_token
1069        self._update_setup_data(
1070            CONF_REFRESH_TOKEN_GLOBAL,
1071            auth_info["refresh_token"],
1072            immediate=token_rotated,
1073        )
1074
1075        # get logged-in user info
1076        if not self._sp_user:
1077            self._sp_user = userinfo = await self._get_data(
1078                "me", auth_info=auth_info, use_global_session=True
1079            )
1080            if country := userinfo.get("country"):
1081                self.mass.metadata.set_default_preferred_language(country)
1082            if self.get_setup_value(CONF_ACCOUNT_ID) != userinfo["id"]:
1083                # instances configured before the account was recorded fill it in here,
1084                # so the setup flow can spot a duplicate account without loading them
1085                self._update_setup_data(CONF_ACCOUNT_ID, userinfo["id"])
1086            self.logger.info("Successfully logged in to Spotify as %s", userinfo["display_name"])
1087        return auth_info
1088
1089    @lock
1090    async def login_dev(self, force_refresh: bool = False) -> dict[str, Any]:
1091        """
1092        Log-in Spotify developer session and return Auth/token info.
1093
1094        This uses the user's custom client ID which has less rate limits but limited API access.
1095        """
1096        # return the cached access token while it is still valid (refreshed before expiry)
1097        if (
1098            not force_refresh
1099            and self._auth_info_dev
1100            and (self._auth_info_dev["expires_at"] > (time.time() + 600))
1101        ):
1102            return self._auth_info_dev
1103        # read the refresh token from the persisted store rather than the in-memory config copy,
1104        # which can lag a rotation and would make us refresh with a stale (revoked) token
1105        refresh_token = self._stored_refresh_token(CONF_REFRESH_TOKEN_DEV)
1106        client_id = self.get_setup_value(CONF_CLIENT_ID)
1107        if not refresh_token or not client_id:
1108            raise LoginFailed("Developer authentication not configured")
1109
1110        try:
1111            auth_info = await get_spotify_token(
1112                self.mass.http_session,
1113                cast("str", client_id),
1114                refresh_token,
1115                "developer",
1116            )
1117            self.logger.debug("Successfully refreshed developer access token")
1118        except LoginFailed as err:
1119            if "revoked" in str(err) or "invalid_grant" in str(err):
1120                # Spotify rotates the refresh token on refresh and revokes the previous one.
1121                # If the stored token was rotated while this refresh was in flight, the token
1122                # we tried is merely stale, so keep the newer one instead of forcing re-auth.
1123                if not self._refresh_token_superseded(CONF_REFRESH_TOKEN_DEV, refresh_token):
1124                    self._update_setup_data(CONF_REFRESH_TOKEN_DEV, None)
1125                    self._update_setup_data(CONF_CLIENT_ID, None)
1126            # Don't unload - we can still use the global session
1127            self.dev_session_active = False
1128            self.logger.warning(str(err))
1129            raise
1130
1131        # make sure that our updated creds get stored in memory + config
1132        self._auth_info_dev = auth_info
1133        # Spotify revokes the previous refresh token only when it rotates one, so on rotation
1134        # persist immediately to ensure the new token survives a crash within the debounced-save
1135        # window and avoids a forced re-auth; an unchanged token uses the normal debounced save.
1136        token_rotated = auth_info["refresh_token"] != refresh_token
1137        self._update_setup_data(
1138            CONF_REFRESH_TOKEN_DEV,
1139            auth_info["refresh_token"],
1140            immediate=token_rotated,
1141        )
1142
1143        self.logger.info("Successfully logged in to Spotify developer session")
1144        return auth_info
1145
1146    def _build_search_types(self, media_types: list[MediaType]) -> str:
1147        """Build comma-separated search types string from media types."""
1148        searchtypes = []
1149        if MediaType.ARTIST in media_types:
1150            searchtypes.append("artist")
1151        if MediaType.ALBUM in media_types:
1152            searchtypes.append("album")
1153        if MediaType.TRACK in media_types:
1154            searchtypes.append("track")
1155        if MediaType.PLAYLIST in media_types:
1156            searchtypes.append("playlist")
1157        if MediaType.PODCAST in media_types:
1158            searchtypes.append("show")
1159        if MediaType.AUDIOBOOK in media_types and self.audiobooks_supported:
1160            searchtypes.append("audiobook")
1161        return ",".join(searchtypes)
1162
1163    def _process_search_results(
1164        self, api_result: dict[str, Any], searchresult: SearchResults
1165    ) -> int:
1166        """
1167        Process API search results and update searchresult object.
1168
1169        Returns the total number of items received.
1170        """
1171        items_received = 0
1172
1173        if "artists" in api_result:
1174            artists = [
1175                parse_artist(item, self)
1176                for item in api_result["artists"]["items"]
1177                if (item and item["id"] and item["name"])
1178            ]
1179            searchresult.artists = [*searchresult.artists, *artists]
1180            items_received += len(api_result["artists"]["items"])
1181
1182        if "albums" in api_result:
1183            albums = [
1184                parse_album(item, self)
1185                for item in api_result["albums"]["items"]
1186                if (item and item["id"])
1187            ]
1188            searchresult.albums = [*searchresult.albums, *albums]
1189            items_received += len(api_result["albums"]["items"])
1190
1191        if "tracks" in api_result:
1192            tracks = [
1193                parse_track(item, self)
1194                for item in api_result["tracks"]["items"]
1195                if (item and item["id"])
1196            ]
1197            searchresult.tracks = [*searchresult.tracks, *tracks]
1198            items_received += len(api_result["tracks"]["items"])
1199
1200        if "playlists" in api_result:
1201            playlists = [
1202                parse_playlist(item, self)
1203                for item in api_result["playlists"]["items"]
1204                if (item and item["id"])
1205            ]
1206            searchresult.playlists = [*searchresult.playlists, *playlists]
1207            items_received += len(api_result["playlists"]["items"])
1208
1209        if "shows" in api_result:
1210            podcasts = []
1211            for item in api_result["shows"]["items"]:
1212                if not (item and item["id"]):
1213                    continue
1214                # Filter out audiobooks - they have a distinctive description format
1215                description = item.get("description", "")
1216                if description.startswith("Author(s):") and "Narrator(s):" in description:
1217                    continue
1218                podcasts.append(parse_podcast(item, self))
1219            searchresult.podcasts = [*searchresult.podcasts, *podcasts]
1220            items_received += len(api_result["shows"]["items"])
1221
1222        if "audiobooks" in api_result and self.audiobooks_supported:
1223            audiobooks = [
1224                parse_audiobook(item, self)
1225                for item in api_result["audiobooks"]["items"]
1226                if (item and item["id"])
1227            ]
1228            searchresult.audiobooks = [*searchresult.audiobooks, *audiobooks]
1229            items_received += len(api_result["audiobooks"]["items"])
1230
1231        return items_received
1232
1233    def _create_backend(self) -> SpotifyPlaybackBackend:
1234        """Return the playback backend selected by this instance's configuration."""
1235        if self.get_setup_value(CONF_PLAYBACK_BACKEND) == BACKEND_SOLOIST:
1236            return SoloistBackend(self)
1237        return LibrespotBackend(self)
1238
1239    def _remove_unused_playback_credentials(self) -> None:
1240        """Remove the login material the unselected playback backend left behind (blocking)."""
1241        if isinstance(self.backend, SoloistBackend):
1242            credentials_file = Path(self.cache_dir) / CREDENTIALS_FILE
1243            if credentials_file.is_file():
1244                self.logger.debug("Removing leftover librespot credential %s", credentials_file)
1245                credentials_file.unlink(missing_ok=True)
1246            return
1247        session_dir = self._instance_storage_dir / SOLOIST_DATA_DIR_NAME
1248        if session_dir.is_dir():
1249            self.logger.debug("Removing leftover soloist session at %s", session_dir)
1250            self._remove_tree(session_dir)
1251
1252    def _remove_login_material(self) -> None:
1253        """Remove everything this instance stored that could log in again (blocking)."""
1254        self._remove_tree(self._instance_storage_dir)
1255        self._remove_tree(Path(self.cache_dir))
1256
1257    def _remove_tree(self, path: Path) -> None:
1258        """
1259        Remove a directory tree holding login material (blocking).
1260
1261        A failure is logged rather than swallowed: what is left behind is a
1262        reusable Spotify login, so it should not disappear quietly.
1263        """
1264
1265        def _report(_func: object, failed: str, err: BaseException) -> None:
1266            if not isinstance(err, FileNotFoundError):
1267                self.logger.warning("Failed to remove %s: %s", failed, err)
1268
1269        shutil.rmtree(path, onexc=_report)
1270
1271    @property
1272    def _soloist_backend(self) -> SoloistBackend | None:
1273        """Return the playback backend when the soloist one is in use, else None."""
1274        backend = getattr(self, "backend", None)
1275        return backend if isinstance(backend, SoloistBackend) else None
1276
1277    @property
1278    def _instance_storage_dir(self) -> Path:
1279        """Return this instance's private storage directory."""
1280        return Path(self.mass.storage_path) / "spotify" / self.instance_id
1281
1282    async def _get_auth_info(self, use_global_session: bool = False) -> dict[str, Any]:
1283        """
1284        Get auth info for API requests, preferring dev session if available.
1285
1286        :param use_global_session: Force use of global session (for features not available on dev).
1287        """
1288        if use_global_session or not self.dev_session_active:
1289            return await self.login()
1290
1291        # Try dev session first
1292        try:
1293            return await self.login_dev()
1294        except LoginFailed:
1295            # Fall back to global session
1296            self.logger.debug("Falling back to global session after dev session failure")
1297            return await self.login()
1298
1299    def _get_liked_songs_playlist_id(self) -> str:
1300        return f"{LIKED_SONGS_FAKE_PLAYLIST_ID_PREFIX}-{self.instance_id}"
1301
1302    @use_cache(86400, allow_expired_cache=True)  # 24h; serve stale + refresh in background
1303    async def _get_new_releases(self) -> list[Album]:
1304        """Get Spotify's curated 'new releases' albums."""
1305        try:
1306            result = await self._get_data("browse/new-releases", limit=50)
1307        except MediaNotFoundError:
1308            return []
1309        return [
1310            parse_album(item, self)
1311            for item in result.get("albums", {}).get("items", [])
1312            if item and item.get("id")
1313        ]
1314
1315    @use_cache(86400 * 7, allow_expired_cache=True)  # 7d; serve stale + refresh in background
1316    async def _get_categories(self, locale: str) -> list[BrowseFolder]:
1317        """Get Spotify's curated browse categories (genres & moods) as browse folders."""
1318        try:
1319            result = await self._get_data("browse/categories", locale=locale, limit=50)
1320        except MediaNotFoundError:
1321            return []
1322        return [
1323            BrowseFolder(
1324                item_id=cat["id"],
1325                provider=self.instance_id,
1326                path=f"{self.instance_id}://categories/{cat['id']}",
1327                name=cat["name"],
1328                is_playable=False,
1329            )
1330            for cat in result.get("categories", {}).get("items", [])
1331            if cat and cat.get("id") and cat.get("name")
1332        ]
1333
1334    @use_cache(86400, allow_expired_cache=True)  # 24h; serve stale + refresh in background
1335    async def _get_category_playlists(self, category_id: str, locale: str) -> list[Playlist]:
1336        """Get the playlists for a single Spotify browse category."""
1337        try:
1338            result = await self._get_data(
1339                f"browse/categories/{category_id}/playlists",
1340                locale=locale,
1341                limit=50,
1342                use_global_session=True,
1343            )
1344        except MediaNotFoundError:
1345            return []
1346        return [
1347            parse_playlist(item, self)
1348            for item in result.get("playlists", {}).get("items", [])
1349            if item and item.get("id") and item.get("name")
1350        ]
1351
1352    async def _get_liked_songs_playlist(self) -> Playlist:
1353        if self._sp_user is None:
1354            raise LoginFailed("User info not available - not logged in")
1355
1356        liked_songs = Playlist(
1357            item_id=self._get_liked_songs_playlist_id(),
1358            provider=self.instance_id,
1359            name=f"Liked Songs {self._sp_user['display_name']}",
1360            translation_key="liked_songs",
1361            translation_params=[self._sp_user["display_name"]],
1362            owner=self._sp_user["display_name"],
1363            provider_mappings={
1364                ProviderMapping(
1365                    item_id=self._get_liked_songs_playlist_id(),
1366                    provider_domain=self.domain,
1367                    provider_instance=self.instance_id,
1368                    url="https://open.spotify.com/collection/tracks",
1369                    is_unique=True,  # liked songs is user-specific
1370                )
1371            },
1372        )
1373
1374        liked_songs.is_editable = False  # TODO Editing requires special endpoints
1375
1376        # Add image to the playlist metadata
1377        image = MediaItemImage(
1378            type=ImageType.THUMB,
1379            path="https://misc.scdn.co/liked-songs/liked-songs-64.png",
1380            provider=self.instance_id,
1381            remotely_accessible=True,
1382        )
1383        if liked_songs.metadata.images is None:
1384            liked_songs.metadata.images = UniqueList([image])
1385        else:
1386            liked_songs.metadata.add_image(image)
1387
1388        return liked_songs
1389
1390    async def _get_playlist_pagination_meta(
1391        self, endpoint: str, page: int, use_global_session: bool
1392    ) -> dict[str, Any]:
1393        """
1394        Return pagination metadata for a Spotify playlist traversal.
1395
1396        :param endpoint: Spotify API endpoint for the playlist items.
1397        :param page: Requested playlist page.
1398        :param use_global_session: Whether the global Spotify session is required.
1399        """
1400        state_key = (endpoint, use_global_session)
1401        if state := self._playlist_pagination_states.get(state_key):
1402            self._playlist_pagination_states.move_to_end(state_key)
1403        else:
1404            state = _PlaylistPaginationState(lock=asyncio.Lock())
1405            self._playlist_pagination_states[state_key] = state
1406            while len(self._playlist_pagination_states) > _PLAYLIST_PAGINATION_STATE_LIMIT:
1407                self._playlist_pagination_states.popitem(last=False)
1408
1409        observed_snapshot = state.snapshot
1410        async with state.lock:
1411            snapshot = state.snapshot
1412            # A concurrent page may have populated this snapshot while this call waited.
1413            if snapshot and (page > 0 or snapshot is not observed_snapshot):
1414                return snapshot
1415
1416            if page == 0:
1417                state.snapshot = None
1418            meta = await self._get_paginated_meta(
1419                endpoint,
1420                limit=1,
1421                offset=0,
1422                use_global_session=use_global_session,
1423            )
1424            state.snapshot = meta
1425            return meta
1426
1427    async def _playlist_requires_global_token(self, prov_playlist_id: str) -> bool:
1428        """
1429        Check if a playlist requires global token (cached).
1430
1431        :param prov_playlist_id: The Spotify playlist ID.
1432        :returns: True if the playlist requires global token.
1433        """
1434        cache_key = f"playlist_global_token_{prov_playlist_id}"
1435        return bool(await self.mass.cache.get(cache_key, provider=self.instance_id))
1436
1437    async def _set_playlist_requires_global_token(self, prov_playlist_id: str) -> None:
1438        """
1439        Mark a playlist as requiring global token in cache.
1440
1441        :param prov_playlist_id: The Spotify playlist ID.
1442        """
1443        cache_key = f"playlist_global_token_{prov_playlist_id}"
1444        # Cache for 90 days - playlist ownership doesn't change
1445        await self.mass.cache.set(cache_key, True, provider=self.instance_id, expiration=86400 * 90)
1446
1447    async def _add_audiobook_chapters(self, audiobook: Audiobook) -> None:
1448        """Add chapter metadata to an audiobook from Spotify API data."""
1449        try:
1450            chapters_data = await self._get_audiobook_chapters_data(audiobook.item_id)
1451            if chapters_data:
1452                chapters = []
1453                total_duration_seconds = 0.0
1454
1455                for idx, chapter in enumerate(chapters_data):
1456                    duration_ms = chapter.get("duration_ms", 0)
1457                    duration_seconds = duration_ms / 1000.0
1458
1459                    chapter_obj = MediaItemChapter(
1460                        position=idx + 1,
1461                        name=chapter.get("name", f"Chapter {idx + 1}"),
1462                        start=total_duration_seconds,
1463                        end=total_duration_seconds + duration_seconds,
1464                    )
1465                    chapters.append(chapter_obj)
1466                    total_duration_seconds += duration_seconds
1467
1468                audiobook.metadata.chapters = chapters
1469                audiobook.duration = int(total_duration_seconds)
1470
1471        except (MediaNotFoundError, ResourceTemporarilyUnavailable, ProviderUnavailableError) as e:
1472            self.logger.warning(f"Failed to get chapters for audiobook {audiobook.item_id}: {e}")
1473
1474    @use_cache(43200)  # 12 hours - balances freshness with performance
1475    async def _get_podcast_episodes_data(self, prov_podcast_id: str) -> list[dict[str, Any]]:
1476        """
1477        Get raw episode data from Spotify API (cached).
1478
1479        :param prov_podcast_id: Spotify podcast ID.
1480        """
1481        episodes_data: list[dict[str, Any]] = []
1482
1483        try:
1484            async for item in self._get_all_items(
1485                f"shows/{prov_podcast_id}/episodes", market="from_token"
1486            ):
1487                if item and item.get("id"):
1488                    episodes_data.append(item)
1489        except MediaNotFoundError:
1490            self.logger.warning("Podcast %s not found", prov_podcast_id)
1491            return []
1492        except ResourceTemporarilyUnavailable as err:
1493            self.logger.warning(
1494                "Temporary error fetching episodes for %s: %s", prov_podcast_id, err
1495            )
1496            raise
1497
1498        return episodes_data
1499
1500    @use_cache(7200)  # 2 hours - shorter cache for resume point data
1501    async def _get_audiobook_chapters_data(self, prov_audiobook_id: str) -> list[dict[str, Any]]:
1502        """
1503        Get raw chapter data from Spotify API (cached).
1504
1505        :param prov_audiobook_id: Spotify audiobook ID.
1506        """
1507        chapters_data: list[dict[str, Any]] = []
1508
1509        try:
1510            async for item in self._get_all_items(
1511                f"audiobooks/{prov_audiobook_id}/chapters", market="from_token"
1512            ):
1513                if item and item.get("id"):
1514                    chapters_data.append(item)
1515        except MediaNotFoundError:
1516            self.logger.warning("Audiobook %s not found", prov_audiobook_id)
1517            return []
1518        except ResourceTemporarilyUnavailable as err:
1519            self.logger.warning(
1520                "Temporary error fetching chapters for %s: %s", prov_audiobook_id, err
1521            )
1522            raise
1523
1524        return chapters_data
1525
1526    async def _get_all_items(
1527        self, endpoint: str, key: str = "items", limit: int = 50, **kwargs: Any
1528    ) -> AsyncGenerator[dict[str, Any]]:
1529        """Get all items from a paged list."""
1530        offset = 0
1531        # single request to fetch the etag (used as cache checksum) and total
1532        meta = await self._get_cached_paginated_meta(endpoint, limit=1, offset=0, **kwargs)
1533        cache_checksum = meta["etag"]
1534        total = meta["total"]
1535        while True:
1536            # Avoid requesting beyond the known end. Spotify can return 5xx
1537            # for offset >= total on some endpoints (e.g. algorithmic playlists).
1538            if total and offset >= total:
1539                break
1540            result = await self._get_data_with_caching(
1541                endpoint, cache_checksum=cache_checksum, limit=limit, offset=offset, **kwargs
1542            )
1543            offset += limit
1544            if not result or key not in result or not result[key]:
1545                break
1546            for item in result[key]:
1547                yield item
1548            if len(result[key]) < limit:
1549                break
1550
1551    async def _get_data_with_caching(
1552        self, endpoint: str, cache_checksum: str | None, **kwargs: Any
1553    ) -> dict[str, Any]:
1554        """Get data from api with caching."""
1555        cache_key_parts = [endpoint]
1556        for key in sorted(kwargs.keys()):
1557            cache_key_parts.append(f"{key}{kwargs[key]}")
1558        cache_key = ".".join(map(str, cache_key_parts))
1559        if cached := await self.mass.cache.get(
1560            cache_key, provider=self.instance_id, checksum=cache_checksum, allow_bypass=False
1561        ):
1562            return cast("dict[str, Any]", cached)
1563        result = await self._get_data(endpoint, **kwargs)
1564        await self.mass.cache.set(
1565            cache_key, result, provider=self.instance_id, checksum=cache_checksum
1566        )
1567        return result
1568
1569    @use_cache(120, allow_bypass=False)  # short cache: repeated traversals reuse metadata
1570    async def _get_cached_paginated_meta(self, endpoint: str, **kwargs: Any) -> dict[str, Any]:
1571        """Get cached pagination metadata for a paginated API endpoint."""
1572        return await self._get_paginated_meta(endpoint, **kwargs)
1573
1574    async def _get_paginated_meta(self, endpoint: str, **kwargs: Any) -> dict[str, Any]:
1575        """Get etag and total item count for a paginated api endpoint."""
1576        _res = await self._get_data(endpoint, **kwargs)
1577        return {"etag": _res.get("etag"), "total": _res.get("total", 0)}
1578
1579    @throttle_with_retries
1580    async def _get_data(self, endpoint: str, **kwargs: Any) -> dict[str, Any]:
1581        """
1582        Get data from api.
1583
1584        :param endpoint: API endpoint to call.
1585        :param use_global_session: Force use of global session (for features not available on dev).
1586        """
1587        url = f"https://api.spotify.com/v1/{endpoint}"
1588        kwargs["market"] = "from_token"
1589        kwargs["country"] = "from_token"
1590        use_global_session = kwargs.pop("use_global_session", False)
1591        if not (auth_info := kwargs.pop("auth_info", None)):
1592            auth_info = await self._get_auth_info(use_global_session=use_global_session)
1593        headers = {"Authorization": f"Bearer {auth_info['access_token']}"}
1594        locale = self.mass.metadata.locale.replace("_", "-")
1595        language = locale.split("-")[0]
1596        headers["Accept-Language"] = f"{locale}, {language};q=0.9, *;q=0.5"
1597        self.logger.debug("handling get data %s with kwargs %s", url, kwargs)
1598        async with (
1599            self.mass.http_session.get(
1600                url,
1601                headers=headers,
1602                params=kwargs,
1603                timeout=aiohttp.ClientTimeout(total=120),
1604            ) as response,
1605        ):
1606            # handle spotify rate limiter
1607            if response.status == 429:
1608                backoff_time = int(response.headers["Retry-After"])
1609                raise RateLimited("Spotify Rate Limiter", backoff_time=backoff_time)
1610            # handle temporary server error
1611            if response.status in (502, 503):
1612                raise ResourceTemporarilyUnavailable(backoff_time=30)
1613
1614            # handle token expired, raise ResourceTemporarilyUnavailable
1615            # so it will be retried (and the token refreshed)
1616            if response.status == 401:
1617                if use_global_session or not self.dev_session_active:
1618                    self._auth_info_global = None
1619                else:
1620                    self._auth_info_dev = None
1621                raise ResourceTemporarilyUnavailable("Token expired", backoff_time=1)
1622
1623            if response.status in (400, 403, 404):
1624                try:
1625                    error = await response.json(loads=json_loads)
1626                    message = error.get("error", {}).get("message") or response.reason
1627                except aiohttp.ContentTypeError, JSONDecodeError:
1628                    message = (await response.text()) or response.reason
1629
1630                self.logger.debug(
1631                    "Spotify API error: endpoint=%s, status=%s, reason=%s, message=%s",
1632                    endpoint,
1633                    response.status,
1634                    response.reason,
1635                    message,
1636                )
1637
1638                raise MediaNotFoundError(f"{endpoint} not found")
1639
1640            response.raise_for_status()
1641            result: dict[str, Any] = await response.json(loads=json_loads)
1642            if etag := response.headers.get("ETag"):
1643                result["etag"] = etag
1644            return result
1645
1646    @throttle_with_retries
1647    async def _delete_data(self, endpoint: str, data: Any = None, **kwargs: Any) -> None:
1648        """Delete data from api."""
1649        url = f"https://api.spotify.com/v1/{endpoint}"
1650        use_global_session = kwargs.pop("use_global_session", False)
1651        if not (auth_info := kwargs.pop("auth_info", None)):
1652            auth_info = await self._get_auth_info(use_global_session=use_global_session)
1653        headers = {"Authorization": f"Bearer {auth_info['access_token']}"}
1654        async with self.mass.http_session.delete(
1655            url, headers=headers, params=kwargs, json=data, ssl=True
1656        ) as response:
1657            # handle spotify rate limiter
1658            if response.status == 429:
1659                backoff_time = int(response.headers["Retry-After"])
1660                raise RateLimited("Spotify Rate Limiter", backoff_time=backoff_time)
1661            # handle token expired, raise ResourceTemporarilyUnavailable
1662            # so it will be retried (and the token refreshed)
1663            if response.status == 401:
1664                if use_global_session or not self.dev_session_active:
1665                    self._auth_info_global = None
1666                else:
1667                    self._auth_info_dev = None
1668                raise ResourceTemporarilyUnavailable("Token expired", backoff_time=1)
1669            # handle temporary server error
1670            if response.status in (502, 503):
1671                raise ResourceTemporarilyUnavailable(backoff_time=30)
1672            response.raise_for_status()
1673
1674    @throttle_with_retries
1675    async def _put_data(self, endpoint: str, data: Any = None, **kwargs: Any) -> None:
1676        """Put data on api."""
1677        url = f"https://api.spotify.com/v1/{endpoint}"
1678        use_global_session = kwargs.pop("use_global_session", False)
1679        if not (auth_info := kwargs.pop("auth_info", None)):
1680            auth_info = await self._get_auth_info(use_global_session=use_global_session)
1681        headers = {"Authorization": f"Bearer {auth_info['access_token']}"}
1682        async with self.mass.http_session.put(
1683            url, headers=headers, params=kwargs, json=data, ssl=True
1684        ) as response:
1685            # handle spotify rate limiter
1686            if response.status == 429:
1687                backoff_time = int(response.headers["Retry-After"])
1688                raise RateLimited("Spotify Rate Limiter", backoff_time=backoff_time)
1689            # handle token expired, raise ResourceTemporarilyUnavailable
1690            # so it will be retried (and the token refreshed)
1691            if response.status == 401:
1692                if use_global_session or not self.dev_session_active:
1693                    self._auth_info_global = None
1694                else:
1695                    self._auth_info_dev = None
1696                raise ResourceTemporarilyUnavailable("Token expired", backoff_time=1)
1697
1698            # handle temporary server error
1699            if response.status in (502, 503):
1700                raise ResourceTemporarilyUnavailable(backoff_time=30)
1701            response.raise_for_status()
1702
1703    @throttle_with_retries
1704    async def _post_data(
1705        self, endpoint: str, data: Any = None, want_result: bool = True, **kwargs: Any
1706    ) -> dict[str, Any]:
1707        """Post data on api."""
1708        url = f"https://api.spotify.com/v1/{endpoint}"
1709        use_global_session = kwargs.pop("use_global_session", False)
1710        if not (auth_info := kwargs.pop("auth_info", None)):
1711            auth_info = await self._get_auth_info(use_global_session=use_global_session)
1712        headers = {"Authorization": f"Bearer {auth_info['access_token']}"}
1713        async with self.mass.http_session.post(
1714            url, headers=headers, params=kwargs, json=data, ssl=True
1715        ) as response:
1716            # handle spotify rate limiter
1717            if response.status == 429:
1718                backoff_time = int(response.headers["Retry-After"])
1719                raise RateLimited("Spotify Rate Limiter", backoff_time=backoff_time)
1720            # handle token expired, raise ResourceTemporarilyUnavailable
1721            # so it will be retried (and the token refreshed)
1722            if response.status == 401:
1723                if use_global_session or not self.dev_session_active:
1724                    self._auth_info_global = None
1725                else:
1726                    self._auth_info_dev = None
1727                raise ResourceTemporarilyUnavailable("Token expired", backoff_time=1)
1728            # handle temporary server error
1729            if response.status in (502, 503):
1730                raise ResourceTemporarilyUnavailable(backoff_time=30)
1731            response.raise_for_status()
1732            if not want_result:
1733                return {}
1734            result: dict[str, Any] = await response.json(loads=json_loads)
1735            return result
1736
1737    def _fix_create_playlist_api_bug(self, playlist_obj: dict[str, Any]) -> None:
1738        """Fix spotify API bug where incorrect owner id is returned from Create Playlist."""
1739        if self._sp_user is None:
1740            raise LoginFailed("User info not available - not logged in")
1741
1742        if playlist_obj["owner"]["id"] != self._sp_user["id"]:
1743            playlist_obj["owner"]["id"] = self._sp_user["id"]
1744            playlist_obj["owner"]["display_name"] = self._sp_user["display_name"]
1745        else:
1746            self.logger.warning(
1747                "FIXME: Spotify have fixed their Create Playlist API, this fix can be removed."
1748            )
1749
1750    async def _test_audiobook_support(self) -> bool:
1751        """Test if audiobooks are supported in user's region."""
1752        try:
1753            await self._get_data("me/audiobooks", limit=1)
1754            return True
1755        except aiohttp.ClientResponseError as e:
1756            if e.status == 403:
1757                return False  # Not available
1758            raise  # Re-raise other HTTP errors
1759        except MediaNotFoundError, ProviderUnavailableError:
1760            return False
1761
1762    def _stored_refresh_token(self, key: str) -> str | None:
1763        """
1764        Return the currently persisted refresh token, or None if not set.
1765
1766        Reads through the live setup_data (kept in sync with a just-rotated token) so a
1767        refresh never uses a stale, revoked token from a lagging in-memory config copy.
1768
1769        :param key: Setup data key of the refresh token to read.
1770        """
1771        token = self.get_setup_value(key)
1772        return cast("str", token) if token else None
1773
1774    def _refresh_token_superseded(self, key: str, used_token: str) -> bool:
1775        """
1776        Return whether the stored refresh token differs from the one just used.
1777
1778        :param key: Config key of the refresh token to check.
1779        :param used_token: The refresh token value that was just used to refresh.
1780        """
1781        stored_token = self._stored_refresh_token(key)
1782        if not stored_token:
1783            return False
1784        return stored_token != used_token
1785