music-assistant-server

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