1  """Native Spotify tools for Hermes (registered via plugins/spotify)."""
  2  
  3  from __future__ import annotations
  4  
  5  from typing import Any, Dict, List
  6  
  7  from hermes_cli.auth import get_auth_status
  8  from plugins.spotify.client import (
  9      SpotifyAPIError,
 10      SpotifyAuthRequiredError,
 11      SpotifyClient,
 12      SpotifyError,
 13      normalize_spotify_id,
 14      normalize_spotify_uri,
 15      normalize_spotify_uris,
 16  )
 17  from tools.registry import tool_error, tool_result
 18  
 19  
 20  def _check_spotify_available() -> bool:
 21      try:
 22          return bool(get_auth_status("spotify").get("logged_in"))
 23      except Exception:
 24          return False
 25  
 26  
 27  def _spotify_client() -> SpotifyClient:
 28      return SpotifyClient()
 29  
 30  
 31  def _spotify_tool_error(exc: Exception) -> str:
 32      if isinstance(exc, (SpotifyError, SpotifyAuthRequiredError)):
 33          return tool_error(str(exc))
 34      if isinstance(exc, SpotifyAPIError):
 35          return tool_error(str(exc), status_code=exc.status_code)
 36      return tool_error(f"Spotify tool failed: {type(exc).__name__}: {exc}")
 37  
 38  
 39  def _coerce_limit(raw: Any, *, default: int = 20, minimum: int = 1, maximum: int = 50) -> int:
 40      try:
 41          value = int(raw)
 42      except Exception:
 43          value = default
 44      return max(minimum, min(maximum, value))
 45  
 46  
 47  def _coerce_bool(raw: Any, default: bool = False) -> bool:
 48      if isinstance(raw, bool):
 49          return raw
 50      if isinstance(raw, str):
 51          cleaned = raw.strip().lower()
 52          if cleaned in {"1", "true", "yes", "on"}:
 53              return True
 54          if cleaned in {"0", "false", "no", "off"}:
 55              return False
 56      return default
 57  
 58  
 59  def _as_list(raw: Any) -> List[str]:
 60      if raw is None:
 61          return []
 62      if isinstance(raw, list):
 63          return [str(item).strip() for item in raw if str(item).strip()]
 64      return [str(raw).strip()] if str(raw).strip() else []
 65  
 66  
 67  def _describe_empty_playback(payload: Any, *, action: str) -> dict | None:
 68      if not isinstance(payload, dict) or not payload.get("empty"):
 69          return None
 70      if action == "get_currently_playing":
 71          return {
 72              "success": True,
 73              "action": action,
 74              "is_playing": False,
 75              "status_code": payload.get("status_code", 204),
 76              "message": payload.get("message") or "Spotify is not currently playing anything.",
 77          }
 78      if action == "get_state":
 79          return {
 80              "success": True,
 81              "action": action,
 82              "has_active_device": False,
 83              "status_code": payload.get("status_code", 204),
 84              "message": payload.get("message") or "No active Spotify playback session was found.",
 85          }
 86      return None
 87  
 88  
 89  def _handle_spotify_playback(args: dict, **kw) -> str:
 90      action = str(args.get("action") or "get_state").strip().lower()
 91      client = _spotify_client()
 92      try:
 93          if action == "get_state":
 94              payload = client.get_playback_state(market=args.get("market"))
 95              empty_result = _describe_empty_playback(payload, action=action)
 96              return tool_result(empty_result or payload)
 97          if action == "get_currently_playing":
 98              payload = client.get_currently_playing(market=args.get("market"))
 99              empty_result = _describe_empty_playback(payload, action=action)
100              return tool_result(empty_result or payload)
101          if action == "play":
102              offset = args.get("offset")
103              if isinstance(offset, dict):
104                  payload_offset = {k: v for k, v in offset.items() if v is not None}
105              else:
106                  payload_offset = None
107              uris = normalize_spotify_uris(_as_list(args.get("uris")), "track") if args.get("uris") else None
108              context_uri = None
109              if args.get("context_uri"):
110                  raw_context = str(args.get("context_uri"))
111                  context_type = None
112                  if raw_context.startswith("spotify:album:") or "/album/" in raw_context:
113                      context_type = "album"
114                  elif raw_context.startswith("spotify:playlist:") or "/playlist/" in raw_context:
115                      context_type = "playlist"
116                  elif raw_context.startswith("spotify:artist:") or "/artist/" in raw_context:
117                      context_type = "artist"
118                  context_uri = normalize_spotify_uri(raw_context, context_type)
119              result = client.start_playback(
120                  device_id=args.get("device_id"),
121                  context_uri=context_uri,
122                  uris=uris,
123                  offset=payload_offset,
124                  position_ms=args.get("position_ms"),
125              )
126              return tool_result({"success": True, "action": action, "result": result})
127          if action == "pause":
128              result = client.pause_playback(device_id=args.get("device_id"))
129              return tool_result({"success": True, "action": action, "result": result})
130          if action == "next":
131              result = client.skip_next(device_id=args.get("device_id"))
132              return tool_result({"success": True, "action": action, "result": result})
133          if action == "previous":
134              result = client.skip_previous(device_id=args.get("device_id"))
135              return tool_result({"success": True, "action": action, "result": result})
136          if action == "seek":
137              if args.get("position_ms") is None:
138                  return tool_error("position_ms is required for action='seek'")
139              result = client.seek(position_ms=int(args["position_ms"]), device_id=args.get("device_id"))
140              return tool_result({"success": True, "action": action, "result": result})
141          if action == "set_repeat":
142              state = str(args.get("state") or "").strip().lower()
143              if state not in {"track", "context", "off"}:
144                  return tool_error("state must be one of: track, context, off")
145              result = client.set_repeat(state=state, device_id=args.get("device_id"))
146              return tool_result({"success": True, "action": action, "result": result})
147          if action == "set_shuffle":
148              result = client.set_shuffle(state=_coerce_bool(args.get("state")), device_id=args.get("device_id"))
149              return tool_result({"success": True, "action": action, "result": result})
150          if action == "set_volume":
151              if args.get("volume_percent") is None:
152                  return tool_error("volume_percent is required for action='set_volume'")
153              result = client.set_volume(volume_percent=max(0, min(100, int(args["volume_percent"]))), device_id=args.get("device_id"))
154              return tool_result({"success": True, "action": action, "result": result})
155          if action == "recently_played":
156              after = args.get("after")
157              before = args.get("before")
158              if after and before:
159                  return tool_error("Provide only one of 'after' or 'before'")
160              return tool_result(client.get_recently_played(
161                  limit=_coerce_limit(args.get("limit"), default=20),
162                  after=int(after) if after is not None else None,
163                  before=int(before) if before is not None else None,
164              ))
165          return tool_error(f"Unknown spotify_playback action: {action}")
166      except Exception as exc:
167          return _spotify_tool_error(exc)
168  
169  
170  def _handle_spotify_devices(args: dict, **kw) -> str:
171      action = str(args.get("action") or "list").strip().lower()
172      client = _spotify_client()
173      try:
174          if action == "list":
175              return tool_result(client.get_devices())
176          if action == "transfer":
177              device_id = str(args.get("device_id") or "").strip()
178              if not device_id:
179                  return tool_error("device_id is required for action='transfer'")
180              result = client.transfer_playback(device_id=device_id, play=_coerce_bool(args.get("play")))
181              return tool_result({"success": True, "action": action, "result": result})
182          return tool_error(f"Unknown spotify_devices action: {action}")
183      except Exception as exc:
184          return _spotify_tool_error(exc)
185  
186  
187  def _handle_spotify_queue(args: dict, **kw) -> str:
188      action = str(args.get("action") or "get").strip().lower()
189      client = _spotify_client()
190      try:
191          if action == "get":
192              return tool_result(client.get_queue())
193          if action == "add":
194              uri = normalize_spotify_uri(str(args.get("uri") or ""), None)
195              result = client.add_to_queue(uri=uri, device_id=args.get("device_id"))
196              return tool_result({"success": True, "action": action, "uri": uri, "result": result})
197          return tool_error(f"Unknown spotify_queue action: {action}")
198      except Exception as exc:
199          return _spotify_tool_error(exc)
200  
201  
202  def _handle_spotify_search(args: dict, **kw) -> str:
203      client = _spotify_client()
204      query = str(args.get("query") or "").strip()
205      if not query:
206          return tool_error("query is required")
207      raw_types = _as_list(args.get("types") or args.get("type") or ["track"])
208      search_types = [value.lower() for value in raw_types if value.lower() in {"album", "artist", "playlist", "track", "show", "episode", "audiobook"}]
209      if not search_types:
210          return tool_error("types must contain one or more of: album, artist, playlist, track, show, episode, audiobook")
211      try:
212          return tool_result(client.search(
213              query=query,
214              search_types=search_types,
215              limit=_coerce_limit(args.get("limit"), default=10),
216              offset=max(0, int(args.get("offset") or 0)),
217              market=args.get("market"),
218              include_external=args.get("include_external"),
219          ))
220      except Exception as exc:
221          return _spotify_tool_error(exc)
222  
223  
224  def _handle_spotify_playlists(args: dict, **kw) -> str:
225      action = str(args.get("action") or "list").strip().lower()
226      client = _spotify_client()
227      try:
228          if action == "list":
229              return tool_result(client.get_my_playlists(
230                  limit=_coerce_limit(args.get("limit"), default=20),
231                  offset=max(0, int(args.get("offset") or 0)),
232              ))
233          if action == "get":
234              playlist_id = normalize_spotify_id(str(args.get("playlist_id") or ""), "playlist")
235              return tool_result(client.get_playlist(playlist_id=playlist_id, market=args.get("market")))
236          if action == "create":
237              name = str(args.get("name") or "").strip()
238              if not name:
239                  return tool_error("name is required for action='create'")
240              return tool_result(client.create_playlist(
241                  name=name,
242                  public=_coerce_bool(args.get("public")),
243                  collaborative=_coerce_bool(args.get("collaborative")),
244                  description=args.get("description"),
245              ))
246          if action == "add_items":
247              playlist_id = normalize_spotify_id(str(args.get("playlist_id") or ""), "playlist")
248              uris = normalize_spotify_uris(_as_list(args.get("uris")))
249              return tool_result(client.add_playlist_items(
250                  playlist_id=playlist_id,
251                  uris=uris,
252                  position=args.get("position"),
253              ))
254          if action == "remove_items":
255              playlist_id = normalize_spotify_id(str(args.get("playlist_id") or ""), "playlist")
256              uris = normalize_spotify_uris(_as_list(args.get("uris")))
257              return tool_result(client.remove_playlist_items(
258                  playlist_id=playlist_id,
259                  uris=uris,
260                  snapshot_id=args.get("snapshot_id"),
261              ))
262          if action == "update_details":
263              playlist_id = normalize_spotify_id(str(args.get("playlist_id") or ""), "playlist")
264              return tool_result(client.update_playlist_details(
265                  playlist_id=playlist_id,
266                  name=args.get("name"),
267                  public=args.get("public"),
268                  collaborative=args.get("collaborative"),
269                  description=args.get("description"),
270              ))
271          return tool_error(f"Unknown spotify_playlists action: {action}")
272      except Exception as exc:
273          return _spotify_tool_error(exc)
274  
275  
276  def _handle_spotify_albums(args: dict, **kw) -> str:
277      action = str(args.get("action") or "get").strip().lower()
278      client = _spotify_client()
279      try:
280          album_id = normalize_spotify_id(str(args.get("album_id") or args.get("id") or ""), "album")
281          if action == "get":
282              return tool_result(client.get_album(album_id=album_id, market=args.get("market")))
283          if action == "tracks":
284              return tool_result(client.get_album_tracks(
285                  album_id=album_id,
286                  limit=_coerce_limit(args.get("limit"), default=20),
287                  offset=max(0, int(args.get("offset") or 0)),
288                  market=args.get("market"),
289              ))
290          return tool_error(f"Unknown spotify_albums action: {action}")
291      except Exception as exc:
292          return _spotify_tool_error(exc)
293  
294  
295  def _handle_spotify_library(args: dict, **kw) -> str:
296      """Unified handler for saved tracks + saved albums (formerly two tools)."""
297      kind = str(args.get("kind") or "").strip().lower()
298      if kind not in {"tracks", "albums"}:
299          return tool_error("kind must be one of: tracks, albums")
300      action = str(args.get("action") or "list").strip().lower()
301      item_type = "track" if kind == "tracks" else "album"
302      client = _spotify_client()
303      try:
304          if action == "list":
305              limit = _coerce_limit(args.get("limit"), default=20)
306              offset = max(0, int(args.get("offset") or 0))
307              market = args.get("market")
308              if kind == "tracks":
309                  return tool_result(client.get_saved_tracks(limit=limit, offset=offset, market=market))
310              return tool_result(client.get_saved_albums(limit=limit, offset=offset, market=market))
311          if action == "save":
312              uris = normalize_spotify_uris(_as_list(args.get("uris") or args.get("items")), item_type)
313              return tool_result(client.save_library_items(uris=uris))
314          if action == "remove":
315              ids = [normalize_spotify_id(item, item_type) for item in _as_list(args.get("ids") or args.get("items"))]
316              if not ids:
317                  return tool_error("ids/items is required for action='remove'")
318              if kind == "tracks":
319                  return tool_result(client.remove_saved_tracks(track_ids=ids))
320              return tool_result(client.remove_saved_albums(album_ids=ids))
321          return tool_error(f"Unknown spotify_library action: {action}")
322      except Exception as exc:
323          return _spotify_tool_error(exc)
324  
325  
326  COMMON_STRING = {"type": "string"}
327  
328  SPOTIFY_PLAYBACK_SCHEMA = {
329      "name": "spotify_playback",
330      "description": "Control Spotify playback, inspect the active playback state, or fetch recently played tracks.",
331      "parameters": {
332          "type": "object",
333          "properties": {
334              "action": {"type": "string", "enum": ["get_state", "get_currently_playing", "play", "pause", "next", "previous", "seek", "set_repeat", "set_shuffle", "set_volume", "recently_played"]},
335              "device_id": COMMON_STRING,
336              "market": COMMON_STRING,
337              "context_uri": COMMON_STRING,
338              "uris": {"type": "array", "items": COMMON_STRING},
339              "offset": {"type": "object"},
340              "position_ms": {"type": "integer"},
341              "state": {"description": "For set_repeat use track/context/off. For set_shuffle use boolean-like true/false.", "oneOf": [{"type": "string"}, {"type": "boolean"}]},
342              "volume_percent": {"type": "integer"},
343              "limit": {"type": "integer", "description": "For recently_played: number of tracks (max 50)"},
344              "after": {"type": "integer", "description": "For recently_played: Unix ms cursor (after this timestamp)"},
345              "before": {"type": "integer", "description": "For recently_played: Unix ms cursor (before this timestamp)"},
346          },
347          "required": ["action"],
348      },
349  }
350  
351  SPOTIFY_DEVICES_SCHEMA = {
352      "name": "spotify_devices",
353      "description": "List Spotify Connect devices or transfer playback to a different device.",
354      "parameters": {
355          "type": "object",
356          "properties": {
357              "action": {"type": "string", "enum": ["list", "transfer"]},
358              "device_id": COMMON_STRING,
359              "play": {"type": "boolean"},
360          },
361          "required": ["action"],
362      },
363  }
364  
365  SPOTIFY_QUEUE_SCHEMA = {
366      "name": "spotify_queue",
367      "description": "Inspect the user's Spotify queue or add an item to it.",
368      "parameters": {
369          "type": "object",
370          "properties": {
371              "action": {"type": "string", "enum": ["get", "add"]},
372              "uri": COMMON_STRING,
373              "device_id": COMMON_STRING,
374          },
375          "required": ["action"],
376      },
377  }
378  
379  SPOTIFY_SEARCH_SCHEMA = {
380      "name": "spotify_search",
381      "description": "Search the Spotify catalog for tracks, albums, artists, playlists, shows, or episodes.",
382      "parameters": {
383          "type": "object",
384          "properties": {
385              "query": COMMON_STRING,
386              "types": {"type": "array", "items": COMMON_STRING},
387              "type": COMMON_STRING,
388              "limit": {"type": "integer"},
389              "offset": {"type": "integer"},
390              "market": COMMON_STRING,
391              "include_external": COMMON_STRING,
392          },
393          "required": ["query"],
394      },
395  }
396  
397  SPOTIFY_PLAYLISTS_SCHEMA = {
398      "name": "spotify_playlists",
399      "description": "List, inspect, create, update, and modify Spotify playlists.",
400      "parameters": {
401          "type": "object",
402          "properties": {
403              "action": {"type": "string", "enum": ["list", "get", "create", "add_items", "remove_items", "update_details"]},
404              "playlist_id": COMMON_STRING,
405              "market": COMMON_STRING,
406              "limit": {"type": "integer"},
407              "offset": {"type": "integer"},
408              "name": COMMON_STRING,
409              "description": COMMON_STRING,
410              "public": {"type": "boolean"},
411              "collaborative": {"type": "boolean"},
412              "uris": {"type": "array", "items": COMMON_STRING},
413              "position": {"type": "integer"},
414              "snapshot_id": COMMON_STRING,
415          },
416          "required": ["action"],
417      },
418  }
419  
420  SPOTIFY_ALBUMS_SCHEMA = {
421      "name": "spotify_albums",
422      "description": "Fetch Spotify album metadata or album tracks.",
423      "parameters": {
424          "type": "object",
425          "properties": {
426              "action": {"type": "string", "enum": ["get", "tracks"]},
427              "album_id": COMMON_STRING,
428              "id": COMMON_STRING,
429              "market": COMMON_STRING,
430              "limit": {"type": "integer"},
431              "offset": {"type": "integer"},
432          },
433          "required": ["action"],
434      },
435  }
436  
437  SPOTIFY_LIBRARY_SCHEMA = {
438      "name": "spotify_library",
439      "description": "List, save, or remove the user's saved Spotify tracks or albums. Use `kind` to select which.",
440      "parameters": {
441          "type": "object",
442          "properties": {
443              "kind": {"type": "string", "enum": ["tracks", "albums"], "description": "Which library to operate on"},
444              "action": {"type": "string", "enum": ["list", "save", "remove"]},
445              "limit": {"type": "integer"},
446              "offset": {"type": "integer"},
447              "market": COMMON_STRING,
448              "uris": {"type": "array", "items": COMMON_STRING},
449              "ids": {"type": "array", "items": COMMON_STRING},
450              "items": {"type": "array", "items": COMMON_STRING},
451          },
452          "required": ["kind", "action"],
453      },
454  }