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