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