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