heos/spotify.py

"""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()
                return json.loads(raw) if raw else {}
        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 _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