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