c8e686c04e
# PR Description 1. `adjust_playback_position` - Adjust the playback position within the currently playing track 2. `skip_to_previous_track` - Skip to the previous track in the user's queue, if any 3. `skip_to_next_track` - Skip to the next track in the user's queue, if any 4. `pause_playback` - Pause the currently playing track, if any 5. `resume_playback` - Resume the currently playing track, if any 6. `start_tracks_playback_by_id` - Start playback of a list of tracks (songs) 7. `get_playback_state` - Get information about the user's current playback state, including track or episode, and active device 8. `get_currently_playing` - Get information about the user's currently playing track 9. `get_track_from_id` - Get information about a track 10. `get_recommendations` - Get track (song) recommendations based on seed artists, genres, and tracks, and multiple target audio stats 11. `get_tracks_audio_features` - Get audio features for a list of tracks (songs) ---------------------- My favorite feature of this toolkit is 1. Start playing my favorite song 2. Get the song that I'm currently playing 3. Get audio features of that song 4. Ask for recommended songs that are similar to it 5. Jam out ------------
97 lines
3.5 KiB
Python
97 lines
3.5 KiB
Python
import httpx
|
|
|
|
from arcade.core.schema import ToolContext
|
|
from arcade_spotify.tools.constants import ENDPOINTS, SPOTIFY_BASE_URL
|
|
from arcade_spotify.tools.models import PlaybackState
|
|
|
|
|
|
async def send_spotify_request(
|
|
context: ToolContext,
|
|
method: str,
|
|
url: str,
|
|
params: dict | None = None,
|
|
json_data: dict | None = None,
|
|
) -> httpx.Response:
|
|
"""
|
|
Send an asynchronous request to the Spotify API.
|
|
|
|
Args:
|
|
context: The tool context containing the authorization token.
|
|
method: The HTTP method (GET, POST, PUT, DELETE, etc.).
|
|
url: The full URL for the API endpoint.
|
|
params: Query parameters to include in the request.
|
|
json_data: JSON data to include in the request body.
|
|
|
|
Returns:
|
|
The response object from the API request.
|
|
|
|
Raises:
|
|
ToolExecutionError: If the request fails for any reason.
|
|
"""
|
|
headers = {"Authorization": f"Bearer {context.authorization.token}"}
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.request(method, url, headers=headers, params=params, json=json_data)
|
|
|
|
return response
|
|
|
|
|
|
def handle_404_playback_state(response, message, is_playing: bool) -> dict | None:
|
|
if response.status_code == 404:
|
|
return convert_to_playback_state({
|
|
"is_playing": is_playing,
|
|
"message": message,
|
|
}).to_dict()
|
|
|
|
|
|
def get_url(endpoint: str, **kwargs) -> str:
|
|
"""
|
|
Get the full Spotify URL for a given endpoint.
|
|
|
|
:param endpoint: The endpoint key from ENDPOINTS
|
|
:param kwargs: The parameters to format the URL with
|
|
:return: The full URL
|
|
"""
|
|
return f"{SPOTIFY_BASE_URL}{ENDPOINTS[endpoint].format(**kwargs)}"
|
|
|
|
|
|
def convert_to_playback_state(data: dict) -> PlaybackState:
|
|
"""
|
|
Convert the Spotify API endpoint "/me/player" response data to a PlaybackState object.
|
|
|
|
Args:
|
|
data: The response data from the Spotify API endpoint "/me/player".
|
|
|
|
Returns:
|
|
An instance of PlaybackState populated with the data.
|
|
"""
|
|
playback_state = PlaybackState(
|
|
device_name=data.get("device", {}).get("name"),
|
|
device_id=data.get("device", {}).get("id"),
|
|
currently_playing_type=data.get("currently_playing_type"),
|
|
is_playing=data.get("is_playing"),
|
|
progress_ms=data.get("progress_ms"),
|
|
message=data.get("message"),
|
|
)
|
|
|
|
if data.get("currently_playing_type") == "track":
|
|
item = data.get("item", {})
|
|
album = item.get("album", {})
|
|
playback_state.album_name = album.get("name")
|
|
playback_state.album_id = album.get("id")
|
|
playback_state.album_artists = [artist.get("name") for artist in album.get("artists", [])]
|
|
playback_state.album_spotify_url = album.get("external_urls", {}).get("spotify")
|
|
playback_state.track_name = item.get("name")
|
|
playback_state.track_id = item.get("id")
|
|
playback_state.track_artists = [artist.get("name") for artist in item.get("artists", [])]
|
|
elif data.get("currently_playing_type") == "episode":
|
|
item = data.get("item", {})
|
|
show = item.get("show", {})
|
|
playback_state.show_name = show.get("name")
|
|
playback_state.show_id = show.get("id")
|
|
playback_state.show_spotify_url = show.get("external_urls", {}).get("spotify")
|
|
playback_state.episode_name = item.get("name")
|
|
playback_state.episode_id = item.get("id")
|
|
playback_state.episode_spotify_url = item.get("external_urls", {}).get("spotify")
|
|
|
|
return playback_state
|