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