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        return 2
272
273    @property
274    def audiobooks_supported(self) -> bool:
275        """Check if audiobooks are supported for this user/region."""
276        return self._audiobooks_supported
277
278    @property
279    def audiobook_progress_sync_enabled(self) -> bool:
280        """Check if audiobook progress sync is enabled."""
281        return bool(self.config.get_value(CONF_SYNC_AUDIOBOOK_PROGRESS, False))
282
283    @property
284    def podcast_progress_sync_enabled(self) -> bool:
285        """Check if played status sync is enabled."""
286        value = self.config.get_value(CONF_SYNC_PODCAST_PROGRESS, True)
287        return bool(value) if value is not None else True
288
289    @property
290    def supported_features(self) -> set[ProviderFeature]:
291        """Return the features supported by this Provider."""
292        features = self._supported_features.copy()
293        # Add audiobook features if enabled
294        if self.audiobooks_supported:
295            features.add(ProviderFeature.LIBRARY_AUDIOBOOKS)
296            features.add(ProviderFeature.LIBRARY_AUDIOBOOKS_EDIT)
297        return features
298
299    @property
300    def account_id(self) -> str | None:
301        """Return the Spotify user id of the logged-in account, if known."""
302        return str(self._sp_user["id"]) if self._sp_user else None
303
304    @property
305    def instance_name_postfix(self) -> str | None:
306        """Return a (default) instance name postfix for this provider instance."""
307        if self._sp_user:
308            return str(self._sp_user["display_name"])
309        return None
310
311    async def get_diagnostics(self) -> dict[str, SerializableType]:
312        """Return diagnostics info for this provider to include in diagnostics reports."""
313        return {
314            "logged_in": self._sp_user is not None,
315            "token_expires_in_sec": (
316                round(self._auth_info_global["expires_at"] - time.time())
317                if self._auth_info_global
318                else None
319            ),
320            "dev_session_active": self.dev_session_active,
321            "playback_backend": str(self.get_setup_value(CONF_PLAYBACK_BACKEND) or "librespot"),
322            "audiobooks_supported": self._audiobooks_supported,
323            **(await self.backend.get_diagnostics() if hasattr(self, "backend") else {}),
324        }
325
326    ## Library retrieval methods (generators)
327    async def get_library_artists(self) -> AsyncGenerator[Artist]:
328        """Retrieve library artists from spotify."""
329        endpoint = "me/following"
330        while True:
331            spotify_artists = await self._get_data(
332                endpoint,
333                type="artist",
334                limit=50,
335            )
336            for item in spotify_artists["artists"]["items"]:
337                if item and item["id"]:
338                    yield parse_artist(item, self)
339            if spotify_artists["artists"]["next"]:
340                endpoint = spotify_artists["artists"]["next"]
341                endpoint = endpoint.replace("https://api.spotify.com/v1/", "")
342            else:
343                break
344
345    async def get_library_albums(self) -> AsyncGenerator[Album]:
346        """Retrieve library albums from the provider."""
347        async for item in self._get_all_items("me/albums"):
348            if item["album"] and item["album"]["id"]:
349                yield parse_album(item["album"], self)
350
351    async def get_library_tracks(self) -> AsyncGenerator[Track]:
352        """Retrieve library tracks from the provider."""
353        async for item in self._get_all_items("me/tracks"):
354            if item and item["track"] and item["track"]["id"]:
355                yield parse_track(item["track"], self)
356
357    async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
358        """Retrieve library podcasts from spotify."""
359        async for item in self._get_all_items("me/shows"):
360            if item["show"] and item["show"]["id"]:
361                show_obj = item["show"]
362                # Filter out audiobooks - they have a distinctive description format
363                description = show_obj.get("description", "")
364                if description.startswith("Author(s):") and "Narrator(s):" in description:
365                    continue
366                yield parse_podcast(show_obj, self)
367
368    async def get_library_audiobooks(self) -> AsyncGenerator[Audiobook]:
369        """Retrieve library audiobooks from spotify."""
370        if not self.audiobooks_supported:
371            return
372        async for item in self._get_all_items("me/audiobooks"):
373            if item and item["id"]:
374                # Parse the basic audiobook
375                audiobook = parse_audiobook(item, self)
376                # Add chapters from Spotify API data
377                await self._add_audiobook_chapters(audiobook)
378                yield audiobook
379
380    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
381        """
382        Retrieve playlists from the provider.
383
384        Note: We use the global session here because playlists like "Daily Mix"
385        are only returned when using the non-dev (global) token.
386        """
387        yield await self._get_liked_songs_playlist()
388        async for item in self._get_all_items("me/playlists", use_global_session=True):
389            if item and item["id"]:
390                yield parse_playlist(item, self)
391
392    async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
393        """
394        Browse Spotify items, including curated sections (new releases, genres & moods).
395
396        :param path: The path to browse (e.g. provider_id:// or provider_id://new-releases).
397        """
398        path_parts = path.split("://")[1].split("/") if "://" in path else []
399        subpath = path_parts[0] if path_parts else None
400        sub_subpath = path_parts[1] if len(path_parts) > 1 else None
401        locale = self.mass.metadata.locale
402
403        if subpath == "new-releases":
404            return await self._get_new_releases()
405
406        if subpath == "categories" and sub_subpath:
407            return await self._get_category_playlists(sub_subpath, locale)
408
409        if subpath == "categories":
410            return await self._get_categories(locale)
411
412        # For root path, add curated folders on top of standard library folders.
413        # At the root the path always ends in "://", so curated paths can be appended directly.
414        if not subpath:
415            curated: list[BrowseFolder] = [
416                BrowseFolder(
417                    item_id="new-releases",
418                    provider=self.instance_id,
419                    path=f"{path}new-releases",
420                    name="New Releases",
421                    translation_key="new_releases",
422                    is_playable=True,
423                ),
424                BrowseFolder(
425                    item_id="categories",
426                    provider=self.instance_id,
427                    path=f"{path}categories",
428                    name="Genres & Moods",
429                    translation_key="genres_and_moods",
430                    is_playable=False,
431                ),
432            ]
433            standard = await super().browse(path)
434            return [*curated, *standard]
435
436        return await super().browse(path)
437
438    @use_cache()
439    async def search(
440        self, search_query: str, media_types: list[MediaType] | None = None, limit: int = 5
441    ) -> SearchResults:
442        """
443        Perform search on musicprovider.
444
445        :param search_query: Search query.
446        :param media_types: A list of media_types to include.
447        :param limit: Number of items to return in the search (per type).
448        """
449        searchresult = SearchResults()
450        if media_types is None:
451            return searchresult
452
453        searchtype = self._build_search_types(media_types)
454        if not searchtype:
455            return searchresult
456
457        search_query = search_query.replace("'", "")
458        offset = 0
459        page_limit = min(limit, 10)
460
461        while True:
462            api_result = await self._get_data(
463                "search", q=search_query, type=searchtype, limit=page_limit, offset=offset
464            )
465            items_received = self._process_search_results(api_result, searchresult)
466
467            offset += page_limit
468            if offset >= limit or items_received < page_limit:
469                break
470
471        return searchresult
472
473    @use_cache()
474    async def get_artist(self, prov_artist_id: str) -> Artist:
475        """Get full artist details by id."""
476        artist_obj = await self._get_data(f"artists/{prov_artist_id}")
477        return parse_artist(artist_obj, self)
478
479    @use_cache()
480    async def get_album(self, prov_album_id: str) -> Album:
481        """Get full album details by id."""
482        album_obj = await self._get_data(f"albums/{prov_album_id}")
483        return parse_album(album_obj, self)
484
485    @use_cache()
486    async def get_track(self, prov_track_id: str) -> Track:
487        """Get full track details by id."""
488        track_obj = await self._get_data(f"tracks/{prov_track_id}")
489        return parse_track(track_obj, self)
490
491    @use_cache()
492    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
493        """Get full playlist details by id."""
494        if prov_playlist_id == self._get_liked_songs_playlist_id():
495            return await self._get_liked_songs_playlist()
496
497        # Check cache to see if this playlist requires global token
498        use_global = await self._playlist_requires_global_token(prov_playlist_id)
499        if use_global:
500            playlist_obj = await self._get_data(
501                f"playlists/{prov_playlist_id}", use_global_session=True
502            )
503            return parse_playlist(playlist_obj, self)
504
505        # Try with dev token first (if available), fallback to global on 400 error
506        # Some playlists like Spotify-owned (Daily Mix) or Liked Songs only work with global token
507        try:
508            playlist_obj = await self._get_data(f"playlists/{prov_playlist_id}")
509            return parse_playlist(playlist_obj, self)
510        except MediaNotFoundError:
511            if self.dev_session_active:
512                # Remember that this playlist requires global token
513                await self._set_playlist_requires_global_token(prov_playlist_id)
514                playlist_obj = await self._get_data(
515                    f"playlists/{prov_playlist_id}", use_global_session=True
516                )
517                return parse_playlist(playlist_obj, self)
518            raise
519
520    @use_cache()
521    async def get_podcast(self, prov_podcast_id: str) -> Podcast:
522        """Get full podcast details by id."""
523        podcast_obj = await self._get_data(f"shows/{prov_podcast_id}")
524        if not podcast_obj:
525            raise MediaNotFoundError(f"Podcast not found: {prov_podcast_id}")
526        return parse_podcast(podcast_obj, self)
527
528    @use_cache()
529    async def get_audiobook(self, prov_audiobook_id: str) -> Audiobook:
530        """Get full audiobook details by id."""
531        if not self.audiobooks_supported:
532            raise UnsupportedFeaturedException("Audiobooks are not supported with this account")
533
534        audiobook_obj = await self._get_data(f"audiobooks/{prov_audiobook_id}")
535        if not audiobook_obj:
536            raise MediaNotFoundError(f"Audiobook not found: {prov_audiobook_id}")
537
538        # Parse basic audiobook without chapters first
539        audiobook = parse_audiobook(audiobook_obj, self)
540
541        # Add chapters from Spotify API data
542        await self._add_audiobook_chapters(audiobook)
543
544        # Note: Resume position will be handled by MA's internal system
545        # which calls get_resume_position() when needed
546
547        return audiobook
548
549    async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
550        """Get all podcast episodes."""
551        podcast = await self.get_podcast(prov_podcast_id)
552
553        # Get (cached) episode data
554        episodes_data = await self._get_podcast_episodes_data(prov_podcast_id)
555
556        # API lists newest-first; number down so bigger position = newer
557        total = len(episodes_data)
558        for idx, episode_data in enumerate(episodes_data):
559            episode = parse_podcast_episode(episode_data, self, podcast)
560            episode.position = total - idx
561
562            # Set played status if sync is enabled and resume data exists
563            if self.podcast_progress_sync_enabled and "resume_point" in episode_data:
564                resume_point = episode_data["resume_point"]
565                fully_played = resume_point.get("fully_played", False)
566                position_ms = resume_point.get("resume_position_ms", 0)
567
568                episode.fully_played = fully_played or None
569                episode.resume_position_ms = position_ms if position_ms > 0 else None
570
571            yield episode
572
573    @use_cache(86400)  # 24 hours
574    async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
575        """Get full podcast episode details by id."""
576        episode_obj = await self._get_data(f"episodes/{prov_episode_id}", market="from_token")
577        if not episode_obj:
578            raise MediaNotFoundError(f"Episode not found: {prov_episode_id}")
579        return parse_podcast_episode(episode_obj, self)
580
581    async def get_resume_position(
582        self, item_id: str, media_type: MediaType
583    ) -> tuple[bool, int, datetime | None]:
584        """Get resume position for episode/audiobook from Spotify."""
585        if media_type == MediaType.PODCAST_EPISODE:
586            if not self.podcast_progress_sync_enabled:
587                raise NotImplementedError("Spotify podcast resume sync disabled in settings")
588
589            try:
590                episode_obj = await self._get_data(f"episodes/{item_id}", market="from_token")
591            except MediaNotFoundError:
592                raise NotImplementedError("Episode not found on Spotify")
593            except (ResourceTemporarilyUnavailable, aiohttp.ClientError) as e:
594                self.logger.debug(f"Error fetching episode {item_id}: {e}")
595                raise NotImplementedError("Unable to fetch episode data from Spotify")
596
597            if (
598                not episode_obj
599                or "resume_point" not in episode_obj
600                or not episode_obj["resume_point"]
601            ):
602                raise NotImplementedError("No resume point data from Spotify")
603
604            resume_point = episode_obj["resume_point"]
605            fully_played = resume_point.get("fully_played", False)
606            position_ms = resume_point.get("resume_position_ms", 0)
607            return fully_played, position_ms, None
608
609        if media_type == MediaType.AUDIOBOOK:
610            if not self.audiobooks_supported:
611                raise NotImplementedError("Audiobook support is disabled")
612            if not self.audiobook_progress_sync_enabled:
613                raise NotImplementedError("Spotify audiobook resume sync disabled in settings")
614
615            try:
616                chapters_data = await self._get_audiobook_chapters_data(item_id)
617                if not chapters_data:
618                    raise NotImplementedError("No chapters data available")
619
620                total_position_ms = 0
621                fully_played = True
622
623                for chapter in chapters_data:
624                    resume_point = chapter.get("resume_point", {})
625                    chapter_fully_played = resume_point.get("fully_played", False)
626                    chapter_position_ms = resume_point.get("resume_position_ms", 0)
627
628                    if chapter_fully_played:
629                        total_position_ms += chapter.get("duration_ms", 0)
630                    elif chapter_position_ms > 0:
631                        total_position_ms += chapter_position_ms
632                        fully_played = False
633                        break
634                    else:
635                        fully_played = False
636                        break
637
638                return fully_played, total_position_ms, None
639
640            except (MediaNotFoundError, ResourceTemporarilyUnavailable, aiohttp.ClientError) as e:
641                self.logger.debug(f"Failed to get audiobook resume position for {item_id}: {e}")
642                raise NotImplementedError("Unable to get audiobook resume position from Spotify")
643
644        else:
645            raise NotImplementedError(f"Resume position not supported for {media_type}")
646
647    async def on_played(
648        self,
649        media_type: MediaType,
650        prov_item_id: str,
651        fully_played: bool,
652        position: int,
653        media_item: MediaItemType,
654        is_playing: bool = False,
655    ) -> None:
656        """
657        Call when an episode/audiobook is played in MA.
658
659        MA automatically handles internal position tracking - this method is for
660        provider-specific actions like syncing to external services.
661        """
662        if media_type == MediaType.PODCAST_EPISODE:
663            if not isinstance(media_item, PodcastEpisode):
664                return
665
666            # Log the playback for monitoring/debugging
667            safe_position = position or 0
668            if media_item.duration > 0:
669                completion_percentage = (safe_position / media_item.duration) * 100
670            else:
671                completion_percentage = 0
672
673            self.logger.debug(
674                f"Episode played: {prov_item_id} at {safe_position}s "
675                f"({completion_percentage:.1f}%, fully_played: {fully_played})"
676            )
677
678            # Note: No API exists to sync playback position back to Spotify for episodes
679            # MA handles all internal position tracking automatically
680
681        elif media_type == MediaType.AUDIOBOOK:
682            if not isinstance(media_item, Audiobook):
683                return
684
685            # Log the playback for monitoring/debugging
686            safe_position = position or 0
687            if media_item.duration > 0:
688                completion_percentage = (safe_position / media_item.duration) * 100
689            else:
690                completion_percentage = 0
691
692            self.logger.debug(
693                f"Audiobook played: {prov_item_id} at {safe_position}s "
694                f"({completion_percentage:.1f}%, fully_played: {fully_played})"
695            )
696
697            # Note: No API exists to sync playback position back to Spotify for audiobooks
698            # MA handles all internal position tracking automatically
699
700            # The resume position will be automatically updated by MA's internal tracking
701            # and will be retrieved via get_audiobook() which combines MA + Spotify positions
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 audio stream from Spotify via librespot."""
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            # Convert back to seconds for librespot
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