/
/
/
1"""Helpers/utils for the Spotify musicprovider."""
2
3from __future__ import annotations
4
5import asyncio
6import os
7import platform
8import time
9from typing import TYPE_CHECKING, Any
10from urllib.parse import urlencode
11
12import pkce
13from music_assistant_models.errors import LoginFailed
14
15from music_assistant.helpers.auth import AuthenticationHelper
16from music_assistant.helpers.process import check_output
17
18from .constants import CALLBACK_REDIRECT_URL, SCOPE
19
20if TYPE_CHECKING:
21 import aiohttp
22
23 from music_assistant import MusicAssistant
24
25
26async def get_librespot_binary() -> str:
27 """Find the correct librespot binary belonging to the platform."""
28
29 async def check_librespot(librespot_path: str) -> str | None:
30 try:
31 returncode, output = await check_output(librespot_path, "--version")
32 if returncode == 0 and b"librespot" in output:
33 return librespot_path
34 return None
35 except OSError:
36 return None
37
38 base_path = os.path.join(os.path.dirname(__file__), "bin")
39 system = platform.system().lower().replace("darwin", "macos")
40 architecture = platform.machine().lower()
41
42 if librespot_binary := await check_librespot(
43 os.path.join(base_path, f"librespot-{system}-{architecture}")
44 ):
45 return librespot_binary
46
47 msg = f"Unable to locate Librespot for {system}/{architecture}"
48 raise RuntimeError(msg)
49
50
51async def get_spotify_token(
52 http_session: aiohttp.ClientSession,
53 client_id: str,
54 refresh_token: str,
55 session_name: str = "spotify",
56) -> dict[str, Any]:
57 """Refresh Spotify access token using refresh token.
58
59 :param http_session: aiohttp client session.
60 :param client_id: Spotify client ID.
61 :param refresh_token: Spotify refresh token.
62 :param session_name: Name for logging purposes.
63 :return: Auth info dict with access_token, refresh_token, expires_at.
64 :raises LoginFailed: If token refresh fails.
65 """
66 params = {
67 "grant_type": "refresh_token",
68 "refresh_token": refresh_token,
69 "client_id": client_id,
70 }
71 err = "Unknown error"
72 for _ in range(2):
73 async with http_session.post(
74 "https://accounts.spotify.com/api/token", data=params
75 ) as response:
76 if response.status != 200:
77 err = await response.text()
78 if "revoked" in err:
79 raise LoginFailed(f"Token revoked for {session_name}: {err}")
80 # the token failed to refresh, we allow one retry
81 await asyncio.sleep(2)
82 continue
83 # if we reached this point, the token has been successfully refreshed
84 auth_info: dict[str, Any] = await response.json()
85 auth_info["expires_at"] = int(auth_info["expires_in"] + time.time())
86 return auth_info
87
88 raise LoginFailed(f"Failed to refresh {session_name} access token: {err}")
89
90
91async def pkce_auth_flow(
92 mass: MusicAssistant,
93 session_id: str,
94 client_id: str,
95) -> str:
96 """Perform Spotify PKCE auth flow and return refresh token.
97
98 :param mass: MusicAssistant instance.
99 :param session_id: Session ID for the authentication helper.
100 :param client_id: The client ID to use for authentication.
101 :return: Refresh token string.
102 """
103 # spotify PKCE auth flow
104 # https://developer.spotify.com/documentation/web-api/tutorials/code-pkce-flow
105 code_verifier, code_challenge = pkce.generate_pkce_pair()
106 async with AuthenticationHelper(mass, session_id) as auth_helper:
107 params = {
108 "response_type": "code",
109 "client_id": client_id,
110 "scope": " ".join(SCOPE),
111 "code_challenge_method": "S256",
112 "code_challenge": code_challenge,
113 "redirect_uri": CALLBACK_REDIRECT_URL,
114 "state": auth_helper.callback_url,
115 }
116 query_string = urlencode(params)
117 url = f"https://accounts.spotify.com/authorize?{query_string}"
118 result = await auth_helper.authenticate(url)
119 authorization_code = result["code"]
120
121 # now get the access token
122 token_params = {
123 "grant_type": "authorization_code",
124 "code": authorization_code,
125 "redirect_uri": CALLBACK_REDIRECT_URL,
126 "client_id": client_id,
127 "code_verifier": code_verifier,
128 }
129 async with mass.http_session.post(
130 "https://accounts.spotify.com/api/token", data=token_params
131 ) as response:
132 if response.status != 200:
133 error_text = await response.text()
134 raise LoginFailed(f"Failed to get access token: {error_text}")
135 token_result = await response.json()
136
137 return str(token_result["refresh_token"])
138