This is used to link objects of different providers/qualities together.
"""
- raise NotImplementedError
+ self.logger.debug("Matching providers for playlists is not possible, ignoring request")
def _refresh_playlist_tracks(self, playlist: Playlist) -> None:
"""Refresh playlist tracks by forcing a cache refresh."""
from __future__ import annotations
from typing import TYPE_CHECKING, cast
-from urllib.parse import urlencode
-import pkce
from music_assistant_models.config_entries import ConfigEntry, ConfigValueType
from music_assistant_models.enums import ConfigEntryType, ProviderFeature
-from music_assistant_models.errors import InvalidDataError, SetupFailedError
+from music_assistant_models.errors import InvalidDataError, LoginFailed
from music_assistant.helpers.app_vars import app_var # type: ignore[attr-defined]
-from music_assistant.helpers.auth import AuthenticationHelper
from .constants import (
CALLBACK_REDIRECT_URL,
CONF_ACTION_AUTH,
+ CONF_ACTION_AUTH_DEV,
CONF_ACTION_CLEAR_AUTH,
+ CONF_ACTION_CLEAR_AUTH_DEV,
CONF_CLIENT_ID,
- CONF_REFRESH_TOKEN,
+ CONF_REFRESH_TOKEN_DEPRECATED,
+ CONF_REFRESH_TOKEN_DEV,
+ CONF_REFRESH_TOKEN_GLOBAL,
CONF_SYNC_AUDIOBOOK_PROGRESS,
CONF_SYNC_PODCAST_PROGRESS,
- SCOPE,
)
+from .helpers import pkce_auth_flow
from .provider import SpotifyProvider
if TYPE_CHECKING:
}
+async def _handle_auth_actions(
+ mass: MusicAssistant,
+ action: str | None,
+ values: dict[str, ConfigValueType] | None,
+) -> None:
+ """Handle authentication-related actions for config entries."""
+ if values is None:
+ return
+
+ if action == CONF_ACTION_AUTH:
+ refresh_token = await pkce_auth_flow(mass, cast("str", values["session_id"]), app_var(2))
+ values[CONF_REFRESH_TOKEN_GLOBAL] = refresh_token
+
+ elif action == CONF_ACTION_AUTH_DEV:
+ custom_client_id = values.get(CONF_CLIENT_ID)
+ if not custom_client_id:
+ raise InvalidDataError("Client ID is required for developer authentication")
+ refresh_token = await pkce_auth_flow(
+ mass, cast("str", values["session_id"]), cast("str", custom_client_id)
+ )
+ values[CONF_REFRESH_TOKEN_DEV] = refresh_token
+
+ elif action == CONF_ACTION_CLEAR_AUTH:
+ values[CONF_REFRESH_TOKEN_GLOBAL] = None
+
+ elif action == CONF_ACTION_CLEAR_AUTH_DEV:
+ values[CONF_REFRESH_TOKEN_DEV] = None
+ values[CONF_CLIENT_ID] = None
+
+
async def get_config_entries(
mass: MusicAssistant,
instance_id: str | None = None,
action: str | None = None,
values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """
- Return Config entries to setup this provider.
-
- instance_id: id of an existing provider instance (None if new instance setup).
- action: [optional] action key called from config entries UI.
- values: the (intermediate) raw values for config entries sent with the action.
- """
+ """Return Config entries to setup this provider."""
# Check if audiobooks are supported by existing provider instance
audiobooks_supported = (
instance_id
and getattr(prov_instance, "audiobooks_supported", False)
)
- if action == CONF_ACTION_AUTH:
- # spotify PKCE auth flow
- # https://developer.spotify.com/documentation/web-api/tutorials/code-pkce-flow
-
- if values is None:
- raise InvalidDataError("values cannot be None for authentication action")
-
- code_verifier, code_challenge = pkce.generate_pkce_pair()
- async with AuthenticationHelper(mass, cast("str", values["session_id"])) as auth_helper:
- params = {
- "response_type": "code",
- "client_id": values.get(CONF_CLIENT_ID) or app_var(2),
- "scope": " ".join(SCOPE),
- "code_challenge_method": "S256",
- "code_challenge": code_challenge,
- "redirect_uri": CALLBACK_REDIRECT_URL,
- "state": auth_helper.callback_url,
- }
- query_string = urlencode(params)
- url = f"https://accounts.spotify.com/authorize?{query_string}"
- result = await auth_helper.authenticate(url)
- authorization_code = result["code"]
- # now get the access token
- params = {
- "grant_type": "authorization_code",
- "code": authorization_code,
- "redirect_uri": CALLBACK_REDIRECT_URL,
- "client_id": values.get(CONF_CLIENT_ID) or app_var(2),
- "code_verifier": code_verifier,
- }
- async with mass.http_session.post(
- "https://accounts.spotify.com/api/token", data=params
- ) as response:
- result = await response.json()
- values[CONF_REFRESH_TOKEN] = result["refresh_token"]
-
- # handle action clear authentication
- if action == CONF_ACTION_CLEAR_AUTH:
- if values is None:
- raise InvalidDataError("values cannot be None for clear auth action")
- values[CONF_REFRESH_TOKEN] = None
-
- auth_required = (values or {}).get(CONF_REFRESH_TOKEN) in (None, "")
-
- if auth_required and values is not None:
- values[CONF_CLIENT_ID] = None
+ # Handle any authentication actions
+ await _handle_auth_actions(mass, action, values)
+
+ # Determine authentication states from current values
+ # Note: encrypted values are sent as placeholder text, which indicates value IS set
+ global_token = (values or {}).get(CONF_REFRESH_TOKEN_GLOBAL)
+ dev_token = (values or {}).get(CONF_REFRESH_TOKEN_DEV)
+ global_authenticated = global_token not in (None, "")
+ dev_authenticated = dev_token not in (None, "")
+
+ # Build label text based on state - these are dynamic based on current values
+ if not global_authenticated:
label_text = (
"You need to authenticate to Spotify. Click the authenticate button below "
"to start the authentication process which will open in a new (popup) window, "
"so make sure to disable any popup blockers.\n\n"
- "Also make sure to perform this action from your local network"
+ "Also make sure to perform this action from your local network."
)
elif action == CONF_ACTION_AUTH:
- label_text = "Authenticated to Spotify. Press save to complete setup."
+ label_text = "Authenticated to Spotify. Don't forget to save to complete setup."
else:
label_text = "Authenticated to Spotify. No further action required."
+ # Build dev label text
+ if action == CONF_ACTION_AUTH_DEV:
+ dev_label_text = "Developer session authenticated. Don't forget to save to complete setup."
+ elif dev_authenticated:
+ dev_label_text = (
+ "Developer API session authenticated. "
+ "This session will be used for most API requests to avoid rate limits."
+ )
+ else:
+ dev_label_text = (
+ "Optionally, enter your own Spotify Developer Client ID to speed up performance."
+ )
+
return (
+ # Global authentication section
ConfigEntry(
key="label_text",
type=ConfigEntryType.LABEL,
label=label_text,
),
ConfigEntry(
- key=CONF_REFRESH_TOKEN,
+ key=CONF_REFRESH_TOKEN_GLOBAL,
type=ConfigEntryType.SECURE_STRING,
- label=CONF_REFRESH_TOKEN,
+ label=CONF_REFRESH_TOKEN_GLOBAL,
hidden=True,
required=True,
- value=values.get(CONF_REFRESH_TOKEN) if values else None,
- ),
- ConfigEntry(
- key=CONF_CLIENT_ID,
- type=ConfigEntryType.SECURE_STRING,
- label="Client ID (optional)",
- description="By default, a generic client ID is used which is (heavy) rate limited. "
- "To speedup performance, it is advised that you create your own Spotify Developer "
- "account and use that client ID here, but this comes at the cost of some features "
- "due to Spotify policies. For example, Radio mode/recommendations and featured "
- "playlists will not work with a custom client ID. \n\n"
- f"Use {CALLBACK_REDIRECT_URL} as callback URL.",
- required=False,
- value=values.get(CONF_CLIENT_ID) if values else None,
- hidden=not auth_required,
+ default_value="",
+ value=values.get(CONF_REFRESH_TOKEN_GLOBAL, "") if values else "",
),
ConfigEntry(
key=CONF_ACTION_AUTH,
label="Authenticate with Spotify",
description="This button will redirect you to Spotify to authenticate.",
action=CONF_ACTION_AUTH,
- hidden=not auth_required,
+ # Show only when not authenticated
+ hidden=global_authenticated,
),
ConfigEntry(
key=CONF_ACTION_CLEAR_AUTH,
action=CONF_ACTION_CLEAR_AUTH,
action_label="Clear authentication",
required=False,
- hidden=auth_required,
+ # Show only when authenticated
+ hidden=not global_authenticated,
),
+ # Developer API section
+ ConfigEntry(
+ key="dev_label_text",
+ type=ConfigEntryType.LABEL,
+ label=dev_label_text,
+ category="Developer Token",
+ # Show only when global auth is complete
+ hidden=not global_authenticated,
+ ),
+ ConfigEntry(
+ key=CONF_CLIENT_ID,
+ type=ConfigEntryType.SECURE_STRING,
+ label="Client ID (optional)",
+ description="Enter your own Spotify Developer Client ID to speed up performance "
+ "by avoiding global rate limits. Some features like recommendations and similar "
+ "tracks will continue to use the global session due to Spotify API restrictions.\n\n"
+ f"Use {CALLBACK_REDIRECT_URL} as callback URL in your Spotify Developer app.",
+ required=False,
+ default_value="",
+ value=values.get(CONF_CLIENT_ID, "") if values else "",
+ category="Developer Token",
+ # Show only when global auth is complete
+ hidden=not global_authenticated or dev_authenticated,
+ ),
+ ConfigEntry(
+ key=CONF_REFRESH_TOKEN_DEV,
+ type=ConfigEntryType.SECURE_STRING,
+ label=CONF_REFRESH_TOKEN_DEV,
+ hidden=True,
+ required=False,
+ default_value="",
+ value=values.get(CONF_REFRESH_TOKEN_DEV, "") if values else "",
+ ),
+ ConfigEntry(
+ key=CONF_ACTION_AUTH_DEV,
+ type=ConfigEntryType.ACTION,
+ label="Authenticate Developer Session",
+ description="Authenticate with your custom Client ID.",
+ action=CONF_ACTION_AUTH_DEV,
+ category="Developer Token",
+ # Show only when global is authenticated and dev is NOT authenticated
+ # The client_id dependency is checked at action time, not visibility
+ hidden=not global_authenticated or dev_authenticated,
+ ),
+ ConfigEntry(
+ key=CONF_ACTION_CLEAR_AUTH_DEV,
+ type=ConfigEntryType.ACTION,
+ label="Clear Developer Authentication",
+ description="Clear the developer session authentication and client ID.",
+ action=CONF_ACTION_CLEAR_AUTH_DEV,
+ action_label="Clear developer authentication",
+ required=False,
+ category="Developer Token",
+ # Show when dev token is set
+ hidden=not global_authenticated or not dev_authenticated,
+ ),
+ # Sync options
ConfigEntry(
key=CONF_SYNC_PODCAST_PROGRESS,
type=ConfigEntryType.BOOLEAN,
default_value=False,
value=values.get(CONF_SYNC_PODCAST_PROGRESS, True) if values else True,
category="sync_options",
+ hidden=not global_authenticated,
),
ConfigEntry(
key=CONF_SYNC_AUDIOBOOK_PROGRESS,
default_value=False,
value=values.get(CONF_SYNC_AUDIOBOOK_PROGRESS, False) if values else False,
category="sync_options",
- hidden=not audiobooks_supported,
+ hidden=not global_authenticated or not audiobooks_supported,
),
)
mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
) -> ProviderInstanceType:
"""Initialize provider(instance) with given configuration."""
- if config.get_value(CONF_REFRESH_TOKEN) in (None, ""):
+ # Migration: handle legacy refresh_token
+ legacy_token = config.get_value(CONF_REFRESH_TOKEN_DEPRECATED)
+ global_token = config.get_value(CONF_REFRESH_TOKEN_GLOBAL)
+
+ if legacy_token and not global_token:
+ # Migrate legacy token to appropriate new key
+ if config.get_value(CONF_CLIENT_ID):
+ # Had custom client ID, migrate to dev token
+ mass.config.set_raw_provider_config_value(
+ config.instance_id, CONF_REFRESH_TOKEN_DEV, legacy_token, encrypted=True
+ )
+ else:
+ # No custom client ID, migrate to global token
+ mass.config.set_raw_provider_config_value(
+ config.instance_id, CONF_REFRESH_TOKEN_GLOBAL, legacy_token, encrypted=True
+ )
+ # Remove the deprecated legacy token from config
+ mass.config.set_raw_provider_config_value(
+ config.instance_id, CONF_REFRESH_TOKEN_DEPRECATED, None
+ )
+ # Re-fetch the updated config value
+ global_token = config.get_value(CONF_REFRESH_TOKEN_GLOBAL)
+
+ if global_token in (None, ""):
msg = "Re-Authentication required"
- raise SetupFailedError(msg)
+ raise LoginFailed(msg)
return SpotifyProvider(mass, manifest, config, SUPPORTED_FEATURES)
# Configuration Keys
CONF_CLIENT_ID = "client_id"
CONF_ACTION_AUTH = "auth"
-CONF_REFRESH_TOKEN = "refresh_token"
+CONF_ACTION_AUTH_DEV = "auth_dev"
+CONF_REFRESH_TOKEN_DEPRECATED = "refresh_token" # Legacy key for migration, will be removed
+CONF_REFRESH_TOKEN_GLOBAL = "refresh_token_global" # Token authenticated with MA's client ID
+CONF_REFRESH_TOKEN_DEV = "refresh_token_dev" # Token authenticated with user's custom client ID
CONF_ACTION_CLEAR_AUTH = "clear_auth"
+CONF_ACTION_CLEAR_AUTH_DEV = "clear_auth_dev"
CONF_SYNC_PODCAST_PROGRESS = "sync_podcast_progress"
CONF_SYNC_AUDIOBOOK_PROGRESS = "sync_audiobook_progress"
from __future__ import annotations
+import asyncio
import os
import platform
+import time
+from typing import TYPE_CHECKING, Any
+from urllib.parse import urlencode
+import pkce
+from music_assistant_models.errors import LoginFailed
+
+from music_assistant.helpers.auth import AuthenticationHelper
from music_assistant.helpers.process import check_output
+from .constants import CALLBACK_REDIRECT_URL, SCOPE
+
+if TYPE_CHECKING:
+ import aiohttp
+
+ from music_assistant import MusicAssistant
+
async def get_librespot_binary() -> str:
"""Find the correct librespot binary belonging to the platform."""
msg = f"Unable to locate Librespot for {system}/{architecture}"
raise RuntimeError(msg)
+
+
+async def get_spotify_token(
+ http_session: aiohttp.ClientSession,
+ client_id: str,
+ refresh_token: str,
+ session_name: str = "spotify",
+) -> dict[str, Any]:
+ """Refresh Spotify access token using refresh token.
+
+ :param http_session: aiohttp client session.
+ :param client_id: Spotify client ID.
+ :param refresh_token: Spotify refresh token.
+ :param session_name: Name for logging purposes.
+ :return: Auth info dict with access_token, refresh_token, expires_at.
+ :raises LoginFailed: If token refresh fails.
+ """
+ params = {
+ "grant_type": "refresh_token",
+ "refresh_token": refresh_token,
+ "client_id": client_id,
+ }
+ err = "Unknown error"
+ for _ in range(2):
+ async with http_session.post(
+ "https://accounts.spotify.com/api/token", data=params
+ ) as response:
+ if response.status != 200:
+ err = await response.text()
+ if "revoked" in err:
+ raise LoginFailed(f"Token revoked for {session_name}: {err}")
+ # the token failed to refresh, we allow one retry
+ await asyncio.sleep(2)
+ continue
+ # if we reached this point, the token has been successfully refreshed
+ auth_info: dict[str, Any] = await response.json()
+ auth_info["expires_at"] = int(auth_info["expires_in"] + time.time())
+ return auth_info
+
+ raise LoginFailed(f"Failed to refresh {session_name} access token: {err}")
+
+
+async def pkce_auth_flow(
+ mass: MusicAssistant,
+ session_id: str,
+ client_id: str,
+) -> str:
+ """Perform Spotify PKCE auth flow and return refresh token.
+
+ :param mass: MusicAssistant instance.
+ :param session_id: Session ID for the authentication helper.
+ :param client_id: The client ID to use for authentication.
+ :return: Refresh token string.
+ """
+ # spotify PKCE auth flow
+ # https://developer.spotify.com/documentation/web-api/tutorials/code-pkce-flow
+ code_verifier, code_challenge = pkce.generate_pkce_pair()
+ async with AuthenticationHelper(mass, session_id) as auth_helper:
+ params = {
+ "response_type": "code",
+ "client_id": client_id,
+ "scope": " ".join(SCOPE),
+ "code_challenge_method": "S256",
+ "code_challenge": code_challenge,
+ "redirect_uri": CALLBACK_REDIRECT_URL,
+ "state": auth_helper.callback_url,
+ }
+ query_string = urlencode(params)
+ url = f"https://accounts.spotify.com/authorize?{query_string}"
+ result = await auth_helper.authenticate(url)
+ authorization_code = result["code"]
+
+ # now get the access token
+ token_params = {
+ "grant_type": "authorization_code",
+ "code": authorization_code,
+ "redirect_uri": CALLBACK_REDIRECT_URL,
+ "client_id": client_id,
+ "code_verifier": code_verifier,
+ }
+ async with mass.http_session.post(
+ "https://accounts.spotify.com/api/token", data=token_params
+ ) as response:
+ if response.status != 200:
+ error_text = await response.text()
+ raise LoginFailed(f"Failed to get access token: {error_text}")
+ token_result = await response.json()
+
+ return str(token_result["refresh_token"])
from __future__ import annotations
-import asyncio
import os
import time
from collections.abc import AsyncGenerator
from .constants import (
CONF_CLIENT_ID,
- CONF_REFRESH_TOKEN,
+ CONF_REFRESH_TOKEN_DEV,
+ CONF_REFRESH_TOKEN_GLOBAL,
CONF_SYNC_AUDIOBOOK_PROGRESS,
CONF_SYNC_PODCAST_PROGRESS,
LIKED_SONGS_FAKE_PLAYLIST_ID_PREFIX,
)
-from .helpers import get_librespot_binary
+from .helpers import get_librespot_binary, get_spotify_token
from .parsers import (
parse_album,
parse_artist,
class SpotifyProvider(MusicProvider):
"""Implementation of a Spotify MusicProvider."""
- _auth_info: dict[str, Any] | None = None
+ # Global session (MA's client ID) - always present
+ _auth_info_global: dict[str, Any] | None = None
+ # Developer session (user's custom client ID) - optional
+ _auth_info_dev: dict[str, Any] | None = None
_sp_user: dict[str, Any] | None = None
_librespot_bin: str | None = None
_audiobooks_supported = False
- custom_client_id_active: bool = False
+ # True if user has configured a custom client ID with valid authentication
+ dev_session_active: bool = False
throttler: ThrottlerManager
async def handle_async_init(self) -> None:
"""Handle async initialization of the provider."""
self.cache_dir = os.path.join(self.mass.cache_path, self.instance_id)
+ # Default throttler for global session (heavy rate limited)
self.throttler = ThrottlerManager(rate_limit=1, period=2)
self.streamer = LibrespotStreamer(self)
- if self.config.get_value(CONF_CLIENT_ID):
- # loosen the throttler a bit when a custom client id is used
- self.throttler = ThrottlerManager(rate_limit=45, period=30)
- self.custom_client_id_active = True
+
# check if we have a librespot binary for this arch
self._librespot_bin = await get_librespot_binary()
- # try login which will raise if it fails
+ # try login which will raise if it fails (logs in global session)
await self.login()
+
+ # Check if user has a custom client ID with valid dev token
+ client_id = self.config.get_value(CONF_CLIENT_ID)
+ dev_token = self.config.get_value(CONF_REFRESH_TOKEN_DEV)
+
+ if client_id and dev_token and self._sp_user:
+ await self.login_dev()
+ # Verify user matches
+ userinfo = await self._get_data("me", use_global_session=False)
+ if userinfo["id"] != self._sp_user["id"]:
+ raise LoginFailed(
+ "Developer session must use the same Spotify account as the main session."
+ )
+ # loosen the throttler when a custom client id is used
+ self.throttler = ThrottlerManager(rate_limit=45, period=30)
+ self.dev_session_active = True
+ self.logger.info("Developer Spotify session active.")
+
self._audiobooks_supported = await self._test_audiobook_support()
if not self._audiobooks_supported:
self.logger.info(
if self.audiobooks_supported:
features.add(ProviderFeature.LIBRARY_AUDIOBOOKS)
features.add(ProviderFeature.LIBRARY_AUDIOBOOKS_EDIT)
- if not self.custom_client_id_active:
- # Spotify has killed the similar tracks api for developers
- # https://developer.spotify.com/blog/2024-11-27-changes-to-the-web-api
- return {*features, ProviderFeature.SIMILAR_TRACKS}
return features
@property
async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
"""Get playlist tracks."""
result: list[Track] = []
+ is_liked_songs = prov_playlist_id == self._get_liked_songs_playlist_id()
uri = (
"me/tracks"
if prov_playlist_id == self._get_liked_songs_playlist_id()
else f"playlists/{prov_playlist_id}/tracks"
)
# do single request to get the etag (which we use as checksum for caching)
- cache_checksum = await self._get_etag(uri, limit=1, offset=0)
+ cache_checksum = await self._get_etag(
+ uri, limit=1, offset=0, use_global_session=is_liked_songs
+ )
page_size = 50
offset = page * page_size
spotify_result = await self._get_data_with_caching(
- uri, cache_checksum, limit=page_size, offset=offset
+ uri, cache_checksum, limit=page_size, offset=offset, use_global_session=is_liked_songs
)
for index, item in enumerate(spotify_result["items"], 1):
if not (item and item["track"] and item["track"]["id"]):
@use_cache(86400 * 14) # 14 days
async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
"""Retrieve a dynamic list of tracks based on the provided item."""
+ # Recommendations endpoint is only available on global session (not developer API)
+ # https://developer.spotify.com/blog/2024-11-27-changes-to-the-web-api
endpoint = "recommendations"
- items = await self._get_data(endpoint, seed_tracks=prov_track_id, limit=limit)
+ items = await self._get_data(
+ endpoint, seed_tracks=prov_track_id, limit=limit, use_global_session=True
+ )
return [parse_track(item, self) for item in items["tracks"] if (item and item["id"])]
async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
@lock
async def login(self, force_refresh: bool = False) -> dict[str, Any]:
- """Log-in Spotify and return Auth/token info."""
+ """Log-in Spotify global session and return Auth/token info.
+
+ This uses MA's global client ID which has full API access but heavy rate limits.
+ """
# return existing token if we have one in memory
if (
not force_refresh
- and self._auth_info
- and (self._auth_info["expires_at"] > (time.time() - 600))
+ and self._auth_info_global
+ and (self._auth_info_global["expires_at"] > (time.time() - 600))
):
- return self._auth_info
+ return self._auth_info_global
# request new access token using the refresh token
- if not (refresh_token := self.config.get_value(CONF_REFRESH_TOKEN)):
+ if not (refresh_token := self.config.get_value(CONF_REFRESH_TOKEN_GLOBAL)):
raise LoginFailed("Authentication required")
- client_id = self.config.get_value(CONF_CLIENT_ID) or app_var(2)
- params = {
- "grant_type": "refresh_token",
- "refresh_token": refresh_token,
- "client_id": client_id,
- }
- err = "Unknown error"
- for _ in range(2):
- async with self.mass.http_session.post(
- "https://accounts.spotify.com/api/token", data=params
- ) as response:
- if response.status != 200:
- err = await response.text()
- if "revoked" in err:
- err_msg = f"Failed to refresh access token: {err}"
- # clear refresh token if it's invalid
- self.update_config_value(CONF_REFRESH_TOKEN, None)
- if self.available:
- # If we're already loaded, we need to unload and set an error
- self.unload_with_error(err_msg)
- raise LoginFailed(err_msg)
- # the token failed to refresh, we allow one retry
- await asyncio.sleep(2)
- continue
- # if we reached this point, the token has been successfully refreshed
- auth_info: dict[str, Any] = await response.json()
- auth_info["expires_at"] = int(auth_info["expires_in"] + time.time())
- self.logger.debug("Successfully refreshed access token")
- break
- else:
- if self.available:
+ try:
+ auth_info = await get_spotify_token(
+ self.mass.http_session,
+ app_var(2), # Always use MA's global client ID
+ cast("str", refresh_token),
+ "global",
+ )
+ self.logger.debug("Successfully refreshed global access token")
+ except LoginFailed as err:
+ if "revoked" in str(err):
+ # clear refresh token if it's invalid
+ self.update_config_value(CONF_REFRESH_TOKEN_GLOBAL, None)
+ if self.available:
+ self.unload_with_error(str(err))
+ elif self.available:
self.mass.create_task(
- self.mass.unload_provider_with_error(
- self.instance_id, f"Failed to refresh access token: {err}"
- )
+ self.mass.unload_provider_with_error(self.instance_id, str(err))
)
- raise LoginFailed(f"Failed to refresh access token: {err}")
+ raise
+
+ # make sure that our updated creds get stored in memory + config
+ self._auth_info_global = auth_info
+ self.update_config_value(
+ CONF_REFRESH_TOKEN_GLOBAL, auth_info["refresh_token"], encrypted=True
+ )
+
+ # Setup librespot with global token only if dev token is not configured
+ # (if dev token exists, librespot will be set up in login_dev instead)
+ if not self.config.get_value(CONF_REFRESH_TOKEN_DEV):
+ await self._setup_librespot_auth(auth_info["access_token"])
+
+ # get logged-in user info
+ if not self._sp_user:
+ self._sp_user = userinfo = await self._get_data(
+ "me", auth_info=auth_info, use_global_session=True
+ )
+ self.mass.metadata.set_default_preferred_language(userinfo["country"])
+ self.logger.info("Successfully logged in to Spotify as %s", userinfo["display_name"])
+ return auth_info
+
+ @lock
+ async def login_dev(self, force_refresh: bool = False) -> dict[str, Any]:
+ """Log-in Spotify developer session and return Auth/token info.
+
+ This uses the user's custom client ID which has less rate limits but limited API access.
+ """
+ # return existing token if we have one in memory
+ if (
+ not force_refresh
+ and self._auth_info_dev
+ and (self._auth_info_dev["expires_at"] > (time.time() - 600))
+ ):
+ return self._auth_info_dev
+ # request new access token using the refresh token
+ refresh_token = self.config.get_value(CONF_REFRESH_TOKEN_DEV)
+ client_id = self.config.get_value(CONF_CLIENT_ID)
+ if not refresh_token or not client_id:
+ raise LoginFailed("Developer authentication not configured")
+
+ try:
+ auth_info = await get_spotify_token(
+ self.mass.http_session,
+ cast("str", client_id),
+ cast("str", refresh_token),
+ "developer",
+ )
+ self.logger.debug("Successfully refreshed developer access token")
+ except LoginFailed as err:
+ if "revoked" in str(err):
+ # clear refresh token if it's invalid
+ self.update_config_value(CONF_REFRESH_TOKEN_DEV, None)
+ # Don't unload - we can still use the global session
+ self.dev_session_active = False
+ self.logger.warning(str(err))
+ raise
# make sure that our updated creds get stored in memory + config
- self._auth_info = auth_info
- self.update_config_value(CONF_REFRESH_TOKEN, auth_info["refresh_token"], encrypted=True)
- # check if librespot still has valid auth
+ self._auth_info_dev = auth_info
+ self.update_config_value(CONF_REFRESH_TOKEN_DEV, auth_info["refresh_token"], encrypted=True)
+
+ # Setup librespot with dev token (preferred over global token)
+ await self._setup_librespot_auth(auth_info["access_token"])
+
+ self.logger.info("Successfully logged in to Spotify developer session")
+ return auth_info
+
+ async def _setup_librespot_auth(self, access_token: str) -> None:
+ """Set up librespot authentication with the given access token.
+
+ :param access_token: Spotify access token to use for librespot authentication.
+ """
if self._librespot_bin is None:
raise LoginFailed("Librespot binary not available")
+
args = [
self._librespot_bin,
"--cache",
# librespot will then get its own token from spotify (somehow) and cache that.
args += [
"--access-token",
- auth_info["access_token"],
+ access_token,
]
ret_code, stdout = await check_output(*args)
if ret_code != 0:
# this should not happen, but guard it just in case
- err = stdout.decode("utf-8").strip()
- raise LoginFailed(f"Failed to verify credentials on Librespot: {err}")
+ err_str = stdout.decode("utf-8").strip()
+ raise LoginFailed(f"Failed to verify credentials on Librespot: {err_str}")
- # get logged-in user info
- if not self._sp_user:
- self._sp_user = userinfo = await self._get_data("me", auth_info=auth_info)
- self.mass.metadata.set_default_preferred_language(userinfo["country"])
- self.logger.info("Successfully logged in to Spotify as %s", userinfo["display_name"])
- return auth_info
+ async def _get_auth_info(self, use_global_session: bool = False) -> dict[str, Any]:
+ """Get auth info for API requests, preferring dev session if available.
+
+ :param use_global_session: Force use of global session (for features not available on dev).
+ """
+ if use_global_session or not self.dev_session_active:
+ return await self.login()
+
+ # Try dev session first
+ try:
+ return await self.login_dev()
+ except LoginFailed:
+ # Fall back to global session
+ self.logger.debug("Falling back to global session after dev session failure")
+ return await self.login()
def _get_liked_songs_playlist_id(self) -> str:
return f"{LIKED_SONGS_FAKE_PLAYLIST_ID_PREFIX}-{self.instance_id}"
@throttle_with_retries
async def _get_data(self, endpoint: str, **kwargs: Any) -> dict[str, Any]:
- """Get data from api."""
+ """Get data from api.
+
+ :param endpoint: API endpoint to call.
+ :param use_global_session: Force use of global session (for features not available on dev).
+ """
url = f"https://api.spotify.com/v1/{endpoint}"
kwargs["market"] = "from_token"
kwargs["country"] = "from_token"
+ use_global_session = kwargs.pop("use_global_session", False)
if not (auth_info := kwargs.pop("auth_info", None)):
- auth_info = await self.login()
+ auth_info = await self._get_auth_info(use_global_session=use_global_session)
headers = {"Authorization": f"Bearer {auth_info['access_token']}"}
locale = self.mass.metadata.locale.replace("_", "-")
language = locale.split("-")[0]
# handle token expired, raise ResourceTemporarilyUnavailable
# so it will be retried (and the token refreshed)
if response.status == 401:
- self._auth_info = None
+ if use_global_session or not self.dev_session_active:
+ self._auth_info_global = None
+ else:
+ self._auth_info_dev = None
raise ResourceTemporarilyUnavailable("Token expired", backoff_time=1)
# handle 404 not found, convert to MediaNotFoundError
async def _delete_data(self, endpoint: str, data: Any = None, **kwargs: Any) -> None:
"""Delete data from api."""
url = f"https://api.spotify.com/v1/{endpoint}"
- auth_info = kwargs.pop("auth_info", await self.login())
+ use_global_session = kwargs.pop("use_global_session", False)
+ if not (auth_info := kwargs.pop("auth_info", None)):
+ auth_info = await self._get_auth_info(use_global_session=use_global_session)
headers = {"Authorization": f"Bearer {auth_info['access_token']}"}
async with self.mass.http_session.delete(
url, headers=headers, params=kwargs, json=data, ssl=True
# handle token expired, raise ResourceTemporarilyUnavailable
# so it will be retried (and the token refreshed)
if response.status == 401:
- self._auth_info = None
+ if use_global_session or not self.dev_session_active:
+ self._auth_info_global = None
+ else:
+ self._auth_info_dev = None
raise ResourceTemporarilyUnavailable("Token expired", backoff_time=1)
# handle temporary server error
if response.status in (502, 503):
async def _put_data(self, endpoint: str, data: Any = None, **kwargs: Any) -> None:
"""Put data on api."""
url = f"https://api.spotify.com/v1/{endpoint}"
- auth_info = kwargs.pop("auth_info", await self.login())
+ use_global_session = kwargs.pop("use_global_session", False)
+ if not (auth_info := kwargs.pop("auth_info", None)):
+ auth_info = await self._get_auth_info(use_global_session=use_global_session)
headers = {"Authorization": f"Bearer {auth_info['access_token']}"}
async with self.mass.http_session.put(
url, headers=headers, params=kwargs, json=data, ssl=True
# handle token expired, raise ResourceTemporarilyUnavailable
# so it will be retried (and the token refreshed)
if response.status == 401:
- self._auth_info = None
+ if use_global_session or not self.dev_session_active:
+ self._auth_info_global = None
+ else:
+ self._auth_info_dev = None
raise ResourceTemporarilyUnavailable("Token expired", backoff_time=1)
# handle temporary server error
) -> dict[str, Any]:
"""Post data on api."""
url = f"https://api.spotify.com/v1/{endpoint}"
- auth_info = kwargs.pop("auth_info", await self.login())
+ use_global_session = kwargs.pop("use_global_session", False)
+ if not (auth_info := kwargs.pop("auth_info", None)):
+ auth_info = await self._get_auth_info(use_global_session=use_global_session)
headers = {"Authorization": f"Bearer {auth_info['access_token']}"}
async with self.mass.http_session.post(
url, headers=headers, params=kwargs, json=data, ssl=True
# handle token expired, raise ResourceTemporarilyUnavailable
# so it will be retried (and the token refreshed)
if response.status == 401:
- self._auth_info = None
+ if use_global_session or not self.dev_session_active:
+ self._auth_info_global = None
+ else:
+ self._auth_info_dev = None
raise ResourceTemporarilyUnavailable("Token expired", backoff_time=1)
# handle temporary server error
if response.status in (502, 503):
projects / music-assistant-server.git / commitdiff