Add Spotify player and user recommendation APIs with PUT and GET wrappers for playback control and user track recommendations

app/spotify_http.py

@@ -135,4 +135,43 @@ def spotify_post(
         time.sleep(sleep_s)
 
 
+def spotify_put(
+    url: str,
+    access_token: str,
+    *,
+    params: Optional[Dict[str, Any]] = None,
+    json: Optional[Dict[str, Any]] = None,
+    data: Optional[Dict[str, Any]] = None,
+    timeout: int = 20,
+) -> Dict[str, Any]:
+    """
+    PUT wrapper with basic retry/backoff for rate limits and transient upstream errors.
+    """
+    max_retries = 2
+    attempt = 0
+    while True:
+        resp = requests.put(
+            url,
+            headers={"Authorization": f"Bearer {access_token}"},
+            params=params,
+            json=json,
+            data=data,
+            timeout=timeout,
+        )
+        if resp.status_code < 400:
+            if resp.status_code == 204 or not resp.content:
+                return {}
+            return resp.json()
+
+        err = _parse_spotify_error(resp)
+        retryable = err.status_code in {429, 502, 503}
+        if not retryable or attempt >= max_retries:
+            raise err
+
+        attempt += 1
+        if err.status_code == 429 and err.retry_after is not None:
+            sleep_s = err.retry_after
+        else:
+            sleep_s = min(8.0, (2.0 ** (attempt - 1))) + random.uniform(0.0, 0.25)
+        time.sleep(sleep_s)
 

app/spotify_player_api.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+from .spotify_http import spotify_get, spotify_post, spotify_put
+
+
+def get_playback_state(
+    access_token: str,
+    *,
+    market: Optional[str] = None,
+    additional_types: Optional[str] = None,
+) -> Dict[str, Any]:
+    params: Dict[str, Any] = {}
+    if market:
+        params["market"] = market
+    if additional_types:
+        params["additional_types"] = additional_types
+    return spotify_get("https://api.spotify.com/v1/me/player", access_token, params=params)
+
+
+def get_available_devices(access_token: str) -> Dict[str, Any]:
+    return spotify_get("https://api.spotify.com/v1/me/player/devices", access_token)
+
+
+def get_currently_playing(
+    access_token: str,
+    *,
+    market: Optional[str] = None,
+    additional_types: Optional[str] = None,
+) -> Dict[str, Any]:
+    params: Dict[str, Any] = {}
+    if market:
+        params["market"] = market
+    if additional_types:
+        params["additional_types"] = additional_types
+    return spotify_get("https://api.spotify.com/v1/me/player/currently-playing", access_token, params=params)
+
+
+def start_or_resume_playback(
+    access_token: str,
+    *,
+    device_id: Optional[str] = None,
+    context_uri: Optional[str] = None,
+    uris: Optional[list[str]] = None,
+    offset: Optional[Dict[str, Any]] = None,
+    position_ms: Optional[int] = None,
+) -> Dict[str, Any]:
+    params: Dict[str, Any] = {}
+    if device_id:
+        params["device_id"] = device_id
+    body: Dict[str, Any] = {}
+    if context_uri:
+        body["context_uri"] = context_uri
+    if uris:
+        body["uris"] = uris
+    if offset is not None:
+        body["offset"] = offset
+    if position_ms is not None:
+        body["position_ms"] = position_ms
+    return spotify_put("https://api.spotify.com/v1/me/player/play", access_token, params=params, json=body)
+
+
+def pause_playback(access_token: str, *, device_id: Optional[str] = None) -> Dict[str, Any]:
+    params: Dict[str, Any] = {}
+    if device_id:
+        params["device_id"] = device_id
+    return spotify_put("https://api.spotify.com/v1/me/player/pause", access_token, params=params)
+
+
+def skip_next(access_token: str, *, device_id: Optional[str] = None) -> Dict[str, Any]:
+    params: Dict[str, Any] = {}
+    if device_id:
+        params["device_id"] = device_id
+    return spotify_post("https://api.spotify.com/v1/me/player/next", access_token, params=params)
+
+
+def skip_previous(access_token: str, *, device_id: Optional[str] = None) -> Dict[str, Any]:
+    params: Dict[str, Any] = {}
+    if device_id:
+        params["device_id"] = device_id
+    return spotify_post("https://api.spotify.com/v1/me/player/previous", access_token, params=params)
+
+
+def transfer_playback(
+    access_token: str,
+    *,
+    device_ids: list[str],
+    play: Optional[bool] = None,
+) -> Dict[str, Any]:
+    body: Dict[str, Any] = {"device_ids": device_ids}
+    if play is not None:
+        body["play"] = play
+    return spotify_put("https://api.spotify.com/v1/me/player", access_token, json=body)
+
+
+def add_to_queue(
+    access_token: str,
+    *,
+    uri: str,
+    device_id: Optional[str] = None,
+) -> Dict[str, Any]:
+    params: Dict[str, Any] = {"uri": uri}
+    if device_id:
+        params["device_id"] = device_id
+    return spotify_post("https://api.spotify.com/v1/me/player/queue", access_token, params=params)
+
+
+

app/spotify_user_reco.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+import random
+from typing import Any, Dict, Optional
+
+from .config import get_settings
+from .spotify_http import spotify_get
+from .spotify_client import recommend_track_for_emotion
+
+
+settings = get_settings()
+
+
+def _get_user_top_tracks(access_token: str, limit: int = 20) -> list[Dict[str, Any]]:
+  data = spotify_get(
+    "https://api.spotify.com/v1/me/top/tracks",
+    access_token,
+    params={"limit": limit, "time_range": "medium_term"},
+  )
+  return (data.get("items") or [])[:limit]
+
+
+def _get_recently_played(access_token: str, limit: int = 20) -> list[Dict[str, Any]]:
+  data = spotify_get(
+    "https://api.spotify.com/v1/me/player/recently-played",
+    access_token,
+    params={"limit": limit},
+  )
+  items = data.get("items") or []
+  out: list[Dict[str, Any]] = []
+  for it in items:
+    t = (it or {}).get("track") or {}
+    if not t:
+      continue
+    out.append(t)
+  return out[:limit]
+
+
+def recommend_user_track_for_emotion(
+  access_token: str,
+  emotion: str,
+  *,
+  market: Optional[str] = None,
+) -> Dict[str, Any]:
+  mk = market or settings.spotify_market or "NG"
+  emotion_norm = (emotion or "").lower()
+
+  try:
+    top_tracks = _get_user_top_tracks(access_token, limit=15)
+  except Exception:
+    top_tracks = []
+
+  try:
+    recent_tracks = _get_recently_played(access_token, limit=15)
+  except Exception:
+    recent_tracks = []
+
+  seeds = top_tracks + recent_tracks
+  if not seeds:
+    return recommend_track_for_emotion(emotion, source="user")
+
+  artist_names: list[str] = []
+  track_names: list[str] = []
+
+  for t in seeds:
+    for a in t.get("artists") or []:
+      name = a.get("name")
+      if isinstance(name, str):
+        artist_names.append(name)
+    name = t.get("name")
+    if isinstance(name, str):
+      track_names.append(name)
+
+  artist_names = list(dict.fromkeys(artist_names))[:10]
+  track_names = list(dict.fromkeys(track_names))[:10]
+
+  queries: list[str] = []
+
+  for name in artist_names[:5]:
+    queries.append(f"{emotion_norm} {name}")
+    queries.append(f"{name}")
+
+  for title in track_names[:5]:
+    queries.append(f"{emotion_norm} {title}")
+
+  queries.append(emotion_norm)
+
+  items: list[Dict[str, Any]] = []
+  for q in queries:
+    data = spotify_get(
+      "https://api.spotify.com/v1/search",
+      access_token,
+      params={"q": q, "type": "track", "limit": 10, "market": mk},
+    )
+    batch = (data.get("tracks") or {}).get("items") or []
+    if batch:
+      items.extend(batch)
+    if len(items) >= 30:
+      break
+
+  if not items:
+    return recommend_track_for_emotion(emotion, source="user")
+
+  track = random.choice(items)
+  return {
+    "id": track.get("id"),
+    "name": track.get("name"),
+    "artists": [a.get("name") for a in track.get("artists", [])],
+    "preview_url": track.get("preview_url"),
+    "external_url": (track.get("external_urls") or {}).get("spotify"),
+  }
+
+