414 lines
17 KiB
Python
414 lines
17 KiB
Python
"""Native Spotify tools for Hermes (registered via plugins/spotify).
|
|
|
|
Each tool routes ``args["action"]`` through a dict dispatch table; every entry
|
|
has the signature ``(client, args, action) -> str`` and issues its Spotify Web
|
|
API call through ``client.request`` (auth refresh + error mapping live there).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Callable, Dict, List, Optional
|
|
|
|
from hermes_cli.auth import get_auth_status
|
|
from plugins.spotify.client import (
|
|
SpotifyClient,
|
|
SpotifyError,
|
|
normalize_spotify_id,
|
|
normalize_spotify_uri,
|
|
normalize_spotify_uris,
|
|
)
|
|
from tools.registry import tool_error, tool_result
|
|
|
|
_Handler = Callable[[SpotifyClient, dict, str], str]
|
|
|
|
|
|
def _check_spotify_available() -> bool:
|
|
try:
|
|
return bool(get_auth_status("spotify").get("logged_in"))
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _spotify_tool_error(exc: Exception) -> str:
|
|
if isinstance(exc, SpotifyError): # includes SpotifyAPIError / SpotifyAuthRequiredError
|
|
return tool_error(str(exc))
|
|
return tool_error(f"Spotify tool failed: {type(exc).__name__}: {exc}")
|
|
|
|
|
|
def _coerce_bool(raw: Any, default: bool = False) -> bool:
|
|
if isinstance(raw, bool):
|
|
return raw
|
|
if isinstance(raw, str):
|
|
cleaned = raw.strip().lower()
|
|
if cleaned in {"1", "true", "yes", "on"}:
|
|
return True
|
|
if cleaned in {"0", "false", "no", "off"}:
|
|
return False
|
|
return default
|
|
|
|
|
|
def _as_list(raw: Any) -> List[str]:
|
|
if raw is None:
|
|
return []
|
|
items = raw if isinstance(raw, list) else [raw]
|
|
return [str(item).strip() for item in items if str(item).strip()]
|
|
|
|
|
|
def _offset(args: dict) -> int:
|
|
return max(0, int(args.get("offset") or 0))
|
|
|
|
|
|
def _limit(args: dict, default: int = 20) -> int:
|
|
"""Clamp ``limit`` to Spotify's 1..50 window; non-numeric input falls back to *default*."""
|
|
raw: Any = args.get("limit")
|
|
try:
|
|
value = int(raw)
|
|
except Exception:
|
|
value = default
|
|
return max(1, min(50, value))
|
|
|
|
|
|
def _ok(action: str, result: Any, **extra: Any) -> str:
|
|
return tool_result({"success": True, "action": action, **extra, "result": result})
|
|
|
|
|
|
def _dispatcher(tool_name: str, default: str, table: Dict[str, _Handler], prepare: Optional[Callable[[dict], dict]] = None):
|
|
"""Build a tool handler that routes ``args['action']`` through *table*.
|
|
|
|
The client is constructed outside the error boundary (auth failures propagate
|
|
as exceptions, as they always have). *prepare* runs inside the boundary before
|
|
the action lookup, so its validation errors surface even for unknown actions.
|
|
"""
|
|
def handle(args: dict, **kw) -> str:
|
|
action = str(args.get("action") or default).strip().lower()
|
|
client = SpotifyClient()
|
|
try:
|
|
if prepare is not None:
|
|
args = prepare(args)
|
|
handler = table.get(action)
|
|
if handler is None:
|
|
return tool_error(f"Unknown {tool_name} action: {action}")
|
|
return handler(client, args, action)
|
|
except Exception as exc:
|
|
return _spotify_tool_error(exc)
|
|
handle.__name__ = handle.__qualname__ = f"_handle_{tool_name}"
|
|
return handle
|
|
|
|
|
|
# -- spotify_playback ---------------------------------------------------------
|
|
|
|
# action -> (flag key reported False, fallback message) when Spotify returns 204/empty.
|
|
_EMPTY_PLAYBACK = {
|
|
"get_currently_playing": ("is_playing", "Spotify is not currently playing anything."),
|
|
"get_state": ("has_active_device", "No active Spotify playback session was found."),
|
|
}
|
|
|
|
|
|
def _pb_read(fetch: Callable[..., Any], args: dict, action: str) -> str:
|
|
payload = fetch(market=args.get("market"))
|
|
if isinstance(payload, dict) and payload.get("empty"):
|
|
flag, fallback = _EMPTY_PLAYBACK[action]
|
|
payload = {
|
|
"success": True,
|
|
"action": action,
|
|
flag: False,
|
|
"status_code": payload.get("status_code", 204),
|
|
"message": payload.get("message") or fallback,
|
|
}
|
|
return tool_result(payload)
|
|
|
|
|
|
_CONTEXT_TYPES = (("album", "spotify:album:", "/album/"), ("playlist", "spotify:playlist:", "/playlist/"), ("artist", "spotify:artist:", "/artist/"))
|
|
|
|
|
|
def _pb_play(client: SpotifyClient, args: dict, action: str) -> str:
|
|
offset = args.get("offset")
|
|
payload_offset = {k: v for k, v in offset.items() if v is not None} if isinstance(offset, dict) else None
|
|
uris = normalize_spotify_uris(_as_list(args.get("uris")), "track") if args.get("uris") else None
|
|
context_uri = None
|
|
if args.get("context_uri"):
|
|
raw = str(args.get("context_uri"))
|
|
# Infer the context type so mismatches raise; unknown kinds pass through unchecked.
|
|
context_type = next((t for t, prefix, frag in _CONTEXT_TYPES if raw.startswith(prefix) or frag in raw), None)
|
|
context_uri = normalize_spotify_uri(raw, context_type)
|
|
body = {"context_uri": context_uri, "uris": uris, "offset": payload_offset, "position_ms": args.get("position_ms")}
|
|
return _ok(action, client.request("PUT", "/me/player/play", params={"device_id": args.get("device_id")}, json_body=body))
|
|
|
|
|
|
def _pb_device_cmd(method: str, path: str, **extra: Any) -> _Handler:
|
|
"""Handler for a player command whose only free argument is ``device_id`` (*extra* = fixed params)."""
|
|
return lambda c, a, act: _ok(act, c.request(method, path, params={**extra, "device_id": a.get("device_id")}))
|
|
|
|
|
|
def _pb_seek(client: SpotifyClient, args: dict, action: str) -> str:
|
|
if args.get("position_ms") is None:
|
|
return tool_error("position_ms is required for action='seek'")
|
|
return _pb_device_cmd("PUT", "/me/player/seek", position_ms=int(args["position_ms"]))(client, args, action)
|
|
|
|
|
|
def _pb_set_repeat(client: SpotifyClient, args: dict, action: str) -> str:
|
|
state = str(args.get("state") or "").strip().lower()
|
|
if state not in {"track", "context", "off"}:
|
|
return tool_error("state must be one of: track, context, off")
|
|
return _pb_device_cmd("PUT", "/me/player/repeat", state=state)(client, args, action)
|
|
|
|
|
|
def _pb_set_shuffle(client: SpotifyClient, args: dict, action: str) -> str:
|
|
state = str(_coerce_bool(args.get("state"))).lower()
|
|
return _pb_device_cmd("PUT", "/me/player/shuffle", state=state)(client, args, action)
|
|
|
|
|
|
def _pb_set_volume(client: SpotifyClient, args: dict, action: str) -> str:
|
|
if args.get("volume_percent") is None:
|
|
return tool_error("volume_percent is required for action='set_volume'")
|
|
volume = max(0, min(100, int(args["volume_percent"])))
|
|
return _pb_device_cmd("PUT", "/me/player/volume", volume_percent=volume)(client, args, action)
|
|
|
|
|
|
def _pb_recently_played(client: SpotifyClient, args: dict, action: str) -> str:
|
|
after, before = args.get("after"), args.get("before")
|
|
if after and before:
|
|
return tool_error("Provide only one of 'after' or 'before'")
|
|
params = {
|
|
"limit": _limit(args),
|
|
"after": int(after) if after is not None else None,
|
|
"before": int(before) if before is not None else None,
|
|
}
|
|
return tool_result(client.request("GET", "/me/player/recently-played", params=params))
|
|
|
|
|
|
_handle_spotify_playback = _dispatcher("spotify_playback", "get_state", {
|
|
"get_state": lambda c, a, act: _pb_read(c.get_playback_state, a, act),
|
|
"get_currently_playing": lambda c, a, act: _pb_read(c.get_currently_playing, a, act),
|
|
"play": _pb_play,
|
|
"pause": _pb_device_cmd("PUT", "/me/player/pause"),
|
|
"next": _pb_device_cmd("POST", "/me/player/next"),
|
|
"previous": _pb_device_cmd("POST", "/me/player/previous"),
|
|
"seek": _pb_seek,
|
|
"set_repeat": _pb_set_repeat,
|
|
"set_shuffle": _pb_set_shuffle,
|
|
"set_volume": _pb_set_volume,
|
|
"recently_played": _pb_recently_played,
|
|
})
|
|
|
|
|
|
# -- spotify_devices / spotify_queue / spotify_search ---------------------------
|
|
|
|
def _dev_transfer(client: SpotifyClient, args: dict, action: str) -> str:
|
|
device_id = str(args.get("device_id") or "").strip()
|
|
if not device_id:
|
|
return tool_error("device_id is required for action='transfer'")
|
|
body = {"device_ids": [device_id], "play": _coerce_bool(args.get("play"))}
|
|
return _ok(action, client.request("PUT", "/me/player", json_body=body))
|
|
|
|
|
|
_handle_spotify_devices = _dispatcher("spotify_devices", "list", {
|
|
"list": lambda c, a, act: tool_result(c.request("GET", "/me/player/devices")),
|
|
"transfer": _dev_transfer,
|
|
})
|
|
|
|
|
|
def _queue_add(client: SpotifyClient, args: dict, action: str) -> str:
|
|
uri = normalize_spotify_uri(str(args.get("uri") or ""), None)
|
|
result = client.request("POST", "/me/player/queue", params={"uri": uri, "device_id": args.get("device_id")})
|
|
return _ok(action, result, uri=uri)
|
|
|
|
|
|
_handle_spotify_queue = _dispatcher("spotify_queue", "get", {
|
|
"get": lambda c, a, act: tool_result(c.request("GET", "/me/player/queue")),
|
|
"add": _queue_add,
|
|
})
|
|
|
|
_SEARCH_TYPES = {"album", "artist", "playlist", "track", "show", "episode", "audiobook"}
|
|
|
|
|
|
def _handle_spotify_search(args: dict, **kw) -> str:
|
|
client = SpotifyClient()
|
|
query = str(args.get("query") or "").strip()
|
|
if not query:
|
|
return tool_error("query is required")
|
|
raw_types = _as_list(args.get("types") or args.get("type") or ["track"])
|
|
search_types = [value.lower() for value in raw_types if value.lower() in _SEARCH_TYPES]
|
|
if not search_types:
|
|
return tool_error("types must contain one or more of: album, artist, playlist, track, show, episode, audiobook")
|
|
params = {
|
|
"q": query, "type": ",".join(search_types), "limit": _limit(args, 10), "offset": _offset(args),
|
|
"market": args.get("market"), "include_external": args.get("include_external"),
|
|
}
|
|
try:
|
|
return tool_result(client.request("GET", "/search", params=params))
|
|
except Exception as exc:
|
|
return _spotify_tool_error(exc)
|
|
|
|
|
|
# -- spotify_playlists ---------------------------------------------------------
|
|
|
|
def _playlist_path(args: dict, suffix: str = "") -> str:
|
|
return f"/playlists/{normalize_spotify_id(str(args.get('playlist_id') or ''), 'playlist')}{suffix}"
|
|
|
|
|
|
def _pl_create(client: SpotifyClient, args: dict, action: str) -> str:
|
|
name = str(args.get("name") or "").strip()
|
|
if not name:
|
|
return tool_error("name is required for action='create'")
|
|
body = {
|
|
"name": name, "public": _coerce_bool(args.get("public")),
|
|
"collaborative": _coerce_bool(args.get("collaborative")), "description": args.get("description"),
|
|
}
|
|
return tool_result(client.request("POST", "/me/playlists", json_body=body))
|
|
|
|
|
|
def _pl_remove_items(client: SpotifyClient, args: dict, action: str) -> str:
|
|
path = _playlist_path(args, "/items")
|
|
body = {"items": [{"uri": u} for u in normalize_spotify_uris(_as_list(args.get("uris")))], "snapshot_id": args.get("snapshot_id")}
|
|
return tool_result(client.request("DELETE", path, json_body=body))
|
|
|
|
|
|
_handle_spotify_playlists = _dispatcher("spotify_playlists", "list", {
|
|
"list": lambda c, a, act: tool_result(c.request("GET", "/me/playlists", params={"limit": _limit(a), "offset": _offset(a)})),
|
|
"get": lambda c, a, act: tool_result(c.request("GET", _playlist_path(a), params={"market": a.get("market")})),
|
|
"create": _pl_create,
|
|
"add_items": lambda c, a, act: tool_result(c.request("POST", _playlist_path(a, "/items"), json_body={
|
|
"uris": normalize_spotify_uris(_as_list(a.get("uris"))), "position": a.get("position"),
|
|
})),
|
|
"remove_items": _pl_remove_items,
|
|
"update_details": lambda c, a, act: tool_result(c.request("PUT", _playlist_path(a), json_body={
|
|
"name": a.get("name"), "public": a.get("public"), "collaborative": a.get("collaborative"), "description": a.get("description"),
|
|
})),
|
|
})
|
|
|
|
|
|
# -- spotify_albums ------------------------------------------------------------
|
|
|
|
def _page_params(args: dict) -> dict:
|
|
return {"limit": _limit(args), "offset": _offset(args), "market": args.get("market")}
|
|
|
|
|
|
def _prepare_album(args: dict) -> dict:
|
|
return {**args, "_path": f"/albums/{normalize_spotify_id(str(args.get('album_id') or args.get('id') or ''), 'album')}"}
|
|
|
|
|
|
_handle_spotify_albums = _dispatcher("spotify_albums", "get", {
|
|
"get": lambda c, a, act: tool_result(c.request("GET", a["_path"], params={"market": a.get("market")})),
|
|
"tracks": lambda c, a, act: tool_result(c.request("GET", a["_path"] + "/tracks", params=_page_params(a))),
|
|
}, prepare=_prepare_album)
|
|
|
|
|
|
# -- spotify_library — saved tracks + saved albums, selected by `kind` ---------
|
|
|
|
def _lib_list(client: SpotifyClient, args: dict, action: str) -> str:
|
|
return tool_result(client.request("GET", f"/me/{args['kind']}", params=_page_params(args)))
|
|
|
|
|
|
def _lib_save(client: SpotifyClient, args: dict, action: str) -> str:
|
|
uris = normalize_spotify_uris(_as_list(args.get("uris") or args.get("items")), args["_item_type"])
|
|
return tool_result(client.request("PUT", "/me/library", params={"uris": ",".join(uris)}))
|
|
|
|
|
|
def _lib_remove(client: SpotifyClient, args: dict, action: str) -> str:
|
|
item_type = args["_item_type"]
|
|
ids = [normalize_spotify_id(item, item_type) for item in _as_list(args.get("ids") or args.get("items"))]
|
|
if not ids:
|
|
return tool_error("ids/items is required for action='remove'")
|
|
uris = ",".join(f"spotify:{item_type}:{i}" for i in ids)
|
|
return tool_result(client.request("DELETE", "/me/library", params={"uris": uris}))
|
|
|
|
|
|
_dispatch_library = _dispatcher("spotify_library", "list", {"list": _lib_list, "save": _lib_save, "remove": _lib_remove})
|
|
|
|
|
|
def _handle_spotify_library(args: dict, **kw) -> str:
|
|
kind = str(args.get("kind") or "").strip().lower()
|
|
if kind not in {"tracks", "albums"}:
|
|
return tool_error("kind must be one of: tracks, albums")
|
|
return _dispatch_library({**args, "kind": kind, "_item_type": kind[:-1]}) # tracks->track, albums->album
|
|
|
|
|
|
# -- Schemas (sent to the model — byte-stable; property order matters) ----------
|
|
|
|
COMMON_STRING = {"type": "string"}
|
|
_INT = {"type": "integer"}
|
|
_BOOL = {"type": "boolean"}
|
|
_STR_ARRAY = {"type": "array", "items": COMMON_STRING}
|
|
|
|
|
|
def _strs(*names: str) -> dict:
|
|
return dict.fromkeys(names, COMMON_STRING)
|
|
|
|
|
|
def _enum(*values: str) -> dict:
|
|
return {"type": "string", "enum": list(values)}
|
|
|
|
|
|
def _schema(name: str, description: str, properties: dict, required: list) -> dict:
|
|
return {"name": name, "description": description, "parameters": {"type": "object", "properties": properties, "required": required}}
|
|
|
|
|
|
SPOTIFY_PLAYBACK_SCHEMA = _schema(
|
|
"spotify_playback",
|
|
"Control Spotify playback, inspect the active playback state, or fetch recently played tracks.",
|
|
{
|
|
"action": _enum("get_state", "get_currently_playing", "play", "pause", "next", "previous", "seek", "set_repeat", "set_shuffle", "set_volume", "recently_played"),
|
|
**_strs("device_id", "market", "context_uri"),
|
|
"uris": _STR_ARRAY,
|
|
"offset": {"type": "object"},
|
|
"position_ms": _INT,
|
|
"state": {"description": "For set_repeat use track/context/off. For set_shuffle use boolean-like true/false.", "oneOf": [{"type": "string"}, {"type": "boolean"}]},
|
|
"volume_percent": _INT,
|
|
"limit": {"type": "integer", "description": "For recently_played: number of tracks (max 50)"},
|
|
"after": {"type": "integer", "description": "For recently_played: Unix ms cursor (after this timestamp)"},
|
|
"before": {"type": "integer", "description": "For recently_played: Unix ms cursor (before this timestamp)"},
|
|
},
|
|
["action"],
|
|
)
|
|
|
|
SPOTIFY_DEVICES_SCHEMA = _schema(
|
|
"spotify_devices",
|
|
"List Spotify Connect devices or transfer playback to a different device.",
|
|
{"action": _enum("list", "transfer"), "device_id": COMMON_STRING, "play": _BOOL},
|
|
["action"],
|
|
)
|
|
|
|
SPOTIFY_QUEUE_SCHEMA = _schema(
|
|
"spotify_queue",
|
|
"Inspect the user's Spotify queue or add an item to it.",
|
|
{"action": _enum("get", "add"), **_strs("uri", "device_id")},
|
|
["action"],
|
|
)
|
|
|
|
SPOTIFY_SEARCH_SCHEMA = _schema(
|
|
"spotify_search",
|
|
"Search the Spotify catalog for tracks, albums, artists, playlists, shows, or episodes.",
|
|
{"query": COMMON_STRING, "types": _STR_ARRAY, "type": COMMON_STRING, "limit": _INT, "offset": _INT, **_strs("market", "include_external")},
|
|
["query"],
|
|
)
|
|
|
|
SPOTIFY_PLAYLISTS_SCHEMA = _schema(
|
|
"spotify_playlists",
|
|
"List, inspect, create, update, and modify Spotify playlists.",
|
|
{
|
|
"action": _enum("list", "get", "create", "add_items", "remove_items", "update_details"),
|
|
**_strs("playlist_id", "market"), "limit": _INT, "offset": _INT, **_strs("name", "description"),
|
|
"public": _BOOL, "collaborative": _BOOL, "uris": _STR_ARRAY, "position": _INT, "snapshot_id": COMMON_STRING,
|
|
},
|
|
["action"],
|
|
)
|
|
|
|
SPOTIFY_ALBUMS_SCHEMA = _schema(
|
|
"spotify_albums",
|
|
"Fetch Spotify album metadata or album tracks.",
|
|
{"action": _enum("get", "tracks"), **_strs("album_id", "id", "market"), "limit": _INT, "offset": _INT},
|
|
["action"],
|
|
)
|
|
|
|
SPOTIFY_LIBRARY_SCHEMA = _schema(
|
|
"spotify_library",
|
|
"List, save, or remove the user's saved Spotify tracks or albums. Use `kind` to select which.",
|
|
{
|
|
"kind": {"type": "string", "enum": ["tracks", "albums"], "description": "Which library to operate on"},
|
|
"action": _enum("list", "save", "remove"),
|
|
"limit": _INT, "offset": _INT, "market": COMMON_STRING, "uris": _STR_ARRAY, "ids": _STR_ARRAY, "items": _STR_ARRAY,
|
|
},
|
|
["kind", "action"],
|
|
)
|