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