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