music-assistant-server

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
10
11from music_assistant_models.errors import LoginFailed
12
13from music_assistant.helpers.process import check_output
14
15if TYPE_CHECKING:
16    import aiohttp
17
18
19async def get_librespot_binary() -> str:
20    """Find the correct librespot binary belonging to the platform."""
21
22    async def check_librespot(librespot_path: str) -> str | None:
23        try:
24            returncode, output = await check_output(librespot_path, "--version")
25            if returncode == 0 and b"librespot" in output:
26                return librespot_path
27            return None
28        except OSError:
29            return None
30
31    base_path = os.path.join(os.path.dirname(__file__), "bin")
32    system = platform.system().lower().replace("darwin", "macos")
33    architecture = platform.machine().lower()
34
35    if librespot_binary := await check_librespot(
36        os.path.join(base_path, f"librespot-{system}-{architecture}")
37    ):
38        return librespot_binary
39
40    msg = f"Unable to locate Librespot for {system}/{architecture}"
41    raise RuntimeError(msg)
42
43
44async def get_spotify_token(
45    http_session: aiohttp.ClientSession,
46    client_id: str,
47    refresh_token: str,
48    session_name: str = "spotify",
49) -> dict[str, Any]:
50    """
51    Refresh Spotify access token using refresh token.
52
53    :param http_session: aiohttp client session.
54    :param client_id: Spotify client ID.
55    :param refresh_token: Spotify refresh token.
56    :param session_name: Name for logging purposes.
57    :return: Auth info dict with access_token, refresh_token, expires_at.
58    :raises LoginFailed: If token refresh fails.
59    """
60    params = {
61        "grant_type": "refresh_token",
62        "refresh_token": refresh_token,
63        "client_id": client_id,
64    }
65    err = "Unknown error"
66    for _ in range(2):
67        async with http_session.post(
68            "https://accounts.spotify.com/api/token", data=params
69        ) as response:
70            if response.status != 200:
71                err = await response.text()
72                # invalid_grant means the refresh token is revoked or expired (Spotify
73                # enforces a 6-month lifetime); retrying won't recover it, so fail now and
74                # let the caller clear the stored token and prompt re-authentication.
75                if "invalid_grant" in err or "revoked" in err:
76                    raise LoginFailed(
77                        f"Refresh token no longer valid for {session_name}: {err}",
78                        translation_key="refresh_token_invalid",
79                        translation_owner="provider.spotify",
80                    )
81                # the token failed to refresh, we allow one retry
82                await asyncio.sleep(2)
83                continue
84            # if we reached this point, the token has been successfully refreshed
85            auth_info: dict[str, Any] = await response.json()
86            auth_info["expires_at"] = int(auth_info["expires_in"] + time.time())
87            # Spotify only returns a refresh_token when it rotates one; when the response
88            # omits it, keep using the existing token (per Spotify's refresh-token docs).
89            auth_info.setdefault("refresh_token", refresh_token)
90            return auth_info
91
92    raise LoginFailed(f"Failed to refresh {session_name} access token: {err}")
93