music-assistant-server

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