"""Spotify Web API client -- just enough to ask a room's own Spotify Connect receiver to resume whatever the account was last playing. This is the opposite direction from how Spotify Connect normally works: instead of the phone pushing playback to a speaker, `resume()` calls the Web API's "Transfer Playback" endpoint to pull it there. It needs a Spotify Developer app and a one-time login -- see tools/spotify_auth.py and the README's Spotify section -- because Spotify has no way to grant that without a human authorizing it once. Credentials are never hardcoded here: the client takes them as arguments and config.py reads them from the environment, so nothing secret ends up committed alongside the rest of the config. """ import base64 import json import threading import time import urllib.error import urllib.request class SpotifyError(RuntimeError): """Spotify answered, but said no (or never answered at all).""" class SpotifyClient: """One access token, refreshed on demand, guarded by a lock the same way HeosClient guards its socket.""" ACCOUNTS_URL = "https://accounts.spotify.com" API_URL = "https://api.spotify.com" def __init__(self, client_id, client_secret, refresh_token, timeout=8.0, accounts_url=None, api_url=None): self.client_id = client_id self.client_secret = client_secret self.refresh_token = refresh_token self.timeout = timeout self.accounts_url = accounts_url or self.ACCOUNTS_URL self.api_url = api_url or self.API_URL self._access_token = None self._expires_at = 0 self._lock = threading.Lock() # -- auth ------------------------------------------------------------ def _refresh_token(self) -> str: """Exchange the long-lived refresh token for a fresh access token. Spotify's access tokens last about an hour; refresh a bit early rather than racing the clock on every call.""" credentials = base64.b64encode( f"{self.client_id}:{self.client_secret}".encode()).decode() body = f"grant_type=refresh_token&refresh_token={self.refresh_token}".encode() request = urllib.request.Request( f"{self.accounts_url}/api/token", data=body, method="POST", headers={ "Authorization": f"Basic {credentials}", "Content-Type": "application/x-www-form-urlencoded", }, ) try: with urllib.request.urlopen(request, timeout=self.timeout) as response: payload = json.loads(response.read()) except urllib.error.HTTPError as exc: raise SpotifyError(f"Spotify login refresh failed: {_error_detail(exc)}") from exc except urllib.error.URLError as exc: raise SpotifyError(f"Could not reach Spotify: {exc.reason}") from exc self._access_token = payload["access_token"] self._expires_at = time.time() + payload.get("expires_in", 3600) return self._access_token def _token(self) -> str: with self._lock: if self._access_token and time.time() < self._expires_at - 30: return self._access_token return self._refresh_token() # -- transport --------------------------------------------------------- def _call(self, method: str, path: str, body: dict = None, retrying: bool = False): headers = {"Authorization": f"Bearer {self._token()}"} data = None if body is not None: data = json.dumps(body).encode() headers["Content-Type"] = "application/json" request = urllib.request.Request( f"{self.api_url}{path}", data=data, method=method, headers=headers) try: with urllib.request.urlopen(request, timeout=self.timeout) as response: raw = response.read() # A player command (seek, say) can answer 200 with a body that is # not JSON at all -- it worked, there is just nothing to read. try: return json.loads(raw) if raw else {} except ValueError: return {} except urllib.error.HTTPError as exc: if exc.code == 401 and not retrying: # The access token can go stale between calls even inside # its nominal lifetime -- one retry after a forced refresh # covers that without hiding a genuinely bad refresh token. exc.close() with self._lock: self._access_token = None return self._call(method, path, body, retrying=True) if exc.code == 404 and method == "PUT" and path == "/v1/me/player": raise SpotifyError("Nothing is queued to resume on that Spotify account") from exc raise SpotifyError(f"Spotify said no: {_error_detail(exc)}") from exc except urllib.error.URLError as exc: raise SpotifyError(f"Could not reach Spotify: {exc.reason}") from exc # -- the calls this panel needs ----------------------------------------- def devices(self) -> list: """Every Spotify Connect receiver visible to this account right now -- including a HEOS room nobody has ever connected to from the Spotify app, since Connect devices announce themselves on the LAN whether or not anything is currently playing.""" return self._call("GET", "/v1/me/player/devices").get("devices", []) def playback(self) -> dict: """The account's current playback -- which device, and whether it is playing -- or {} when the account has no playback session at all.""" return self._call("GET", "/v1/me/player") def resume(self, device_name: str) -> dict: """Transfer the account's current (usually paused) playback to the named device and resume it -- the reverse of tapping the device in the Spotify app's Connect picker.""" devices = self.devices() matches = [d for d in devices if d.get("name") == device_name] if not matches: visible = ", ".join(d.get("name", "?") for d in devices) or "none" raise SpotifyError( f"No Spotify Connect device named '{device_name}' is visible right now " f"(Spotify sees: {visible}) -- is the room powered on, or does its " "spotify_name in config.py need setting?") device = matches[0] self._call("PUT", "/v1/me/player", body={"device_ids": [device["id"]], "play": True}) return device def seek(self, position_ms: int): """Jump to position_ms in whatever the account is playing, on whichever device is playing it.""" self._call("PUT", f"/v1/me/player/seek?position_ms={int(position_ms)}") def _error_detail(exc: urllib.error.HTTPError) -> str: try: with exc: payload = json.loads(exc.read()) return payload.get("error_description") or payload.get("error", {}).get("message") or exc.reason except (ValueError, AttributeError, KeyError): return exc.reason