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