x3
c3f7d87e67
- Playlists: Alle werden jetzt geladen (Pagination) - Player: Fehlermeldung wenn kein Geraet aktiv - next/previous/repeat/shuffle zeigen Fehler statt stillschweigend zuversagen
253 lines
8.3 KiB
Python
253 lines
8.3 KiB
Python
import requests
|
|
from typing import Optional
|
|
|
|
BASE_URL = "https://api.spotify.com/v1"
|
|
|
|
|
|
class SpotifyClient:
|
|
def __init__(self, token: str):
|
|
self.token = token
|
|
self.headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
def _get(self, endpoint, params=None):
|
|
resp = requests.get(f"{BASE_URL}{endpoint}", headers=self.headers, params=params)
|
|
if resp.status_code == 401:
|
|
raise TokenExpired()
|
|
if resp.status_code == 429:
|
|
import time
|
|
retry = int(resp.headers.get("Retry-After", 1))
|
|
time.sleep(retry)
|
|
return self._get(endpoint, params)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
def _put(self, endpoint, data=None):
|
|
resp = requests.put(f"{BASE_URL}{endpoint}", headers=self.headers, json=data)
|
|
if resp.status_code == 401:
|
|
raise TokenExpired()
|
|
if resp.status_code == 429:
|
|
import time
|
|
retry = int(resp.headers.get("Retry-After", 1))
|
|
time.sleep(retry)
|
|
return self._put(endpoint, data)
|
|
if resp.status_code == 204:
|
|
return None
|
|
resp.raise_for_status()
|
|
return resp.json() if resp.text else None
|
|
|
|
def _post(self, endpoint, data=None):
|
|
resp = requests.post(f"{BASE_URL}{endpoint}", headers=self.headers, json=data)
|
|
if resp.status_code == 401:
|
|
raise TokenExpired()
|
|
if resp.status_code == 429:
|
|
import time
|
|
retry = int(resp.headers.get("Retry-After", 1))
|
|
time.sleep(retry)
|
|
return self._post(endpoint, data)
|
|
if resp.status_code == 204:
|
|
return None
|
|
resp.raise_for_status()
|
|
return resp.json() if resp.text else None
|
|
|
|
def _delete(self, endpoint, data=None):
|
|
resp = requests.delete(f"{BASE_URL}{endpoint}", headers=self.headers, json=data)
|
|
if resp.status_code == 401:
|
|
raise TokenExpired()
|
|
if resp.status_code == 429:
|
|
import time
|
|
retry = int(resp.headers.get("Retry-After", 1))
|
|
time.sleep(retry)
|
|
return self._delete(endpoint, data)
|
|
if resp.status_code == 204:
|
|
return None
|
|
resp.raise_for_status()
|
|
return resp.json() if resp.text else None
|
|
|
|
# --- Playback ---
|
|
def get_playback(self):
|
|
return self._get("/me/player")
|
|
|
|
def get_current_track(self):
|
|
return self._get("/me/player/currently-playing")
|
|
|
|
def play(self, context_uri=None, uris=None, offset=None, position_ms=None):
|
|
data = {}
|
|
if context_uri:
|
|
data["context_uri"] = context_uri
|
|
if uris:
|
|
data["uris"] = uris
|
|
if offset is not None:
|
|
data["offset"] = offset
|
|
if position_ms is not None:
|
|
data["position_ms"] = position_ms
|
|
return self._put("/me/player/play", data)
|
|
|
|
def pause(self):
|
|
return self._put("/me/player/pause")
|
|
|
|
def next_track(self):
|
|
return self._post("/me/player/next")
|
|
|
|
def previous_track(self):
|
|
return self._post("/me/player/previous")
|
|
|
|
def seek(self, position_ms):
|
|
return self._put(f"/me/player/seek?position_ms={position_ms}")
|
|
|
|
def set_volume(self, volume):
|
|
return self._put(f"/me/player/volume?volume={volume}")
|
|
|
|
def set_repeat(self, state):
|
|
return self._put(f"/me/player/repeat?state={state}")
|
|
|
|
def set_shuffle(self, state):
|
|
return self._put(f"/me/player/shuffle?state={state}")
|
|
|
|
def transfer_playback(self, device_id, force_play=True):
|
|
return self._put("/me/player", {
|
|
"device_ids": [device_id],
|
|
"play": force_play,
|
|
})
|
|
|
|
def get_devices(self):
|
|
return self._get("/me/player/devices")
|
|
|
|
# --- Player Info ---
|
|
def get_recently_played(self, limit=20):
|
|
return self._get("/me/player/recently-played", {"limit": limit})
|
|
|
|
def get_queue(self):
|
|
return self._get("/me/player/queue")
|
|
|
|
def add_to_queue(self, uri):
|
|
return self._post(f"/me/player/queue?uri={uri}")
|
|
|
|
# --- Library ---
|
|
def get_saved_tracks(self, limit=50, offset=0):
|
|
return self._get("/me/tracks", {"limit": limit, "offset": offset})
|
|
|
|
def get_saved_albums(self, limit=50, offset=0):
|
|
return self._get("/me/albums", {"limit": limit, "offset": offset})
|
|
|
|
def get_saved_artists(self, limit=50, offset=0):
|
|
return self._get("/me/following", {"type": "artist", "limit": limit, "offset": offset})
|
|
|
|
def save_track(self, track_ids):
|
|
return self._put("/me/tracks", {"ids": track_ids})
|
|
|
|
def remove_saved_track(self, track_ids):
|
|
return self._delete("/me/tracks", {"ids": track_ids})
|
|
|
|
def check_saved_track(self, track_ids):
|
|
return self._get(f"/me/tracks/contains?ids={','.join(track_ids)}")
|
|
|
|
# --- Playlists ---
|
|
def get_playlists(self, limit=50, offset=0):
|
|
return self._get("/me/playlists", {"limit": limit, "offset": offset})
|
|
|
|
def get_all_playlists(self):
|
|
all_items = []
|
|
offset = 0
|
|
limit = 50
|
|
while True:
|
|
data = self.get_playlists(limit=limit, offset=offset)
|
|
items = data.get("items", [])
|
|
all_items.extend(items)
|
|
if len(items) < limit:
|
|
break
|
|
offset += limit
|
|
return all_items
|
|
|
|
def get_playlist(self, playlist_id):
|
|
return self._get(f"/playlists/{playlist_id}")
|
|
|
|
def get_playlist_tracks(self, playlist_id, limit=100, offset=0):
|
|
return self._get(f"/playlists/{playlist_id}/tracks", {"limit": limit, "offset": offset})
|
|
|
|
def create_playlist(self, name, description="", public=False):
|
|
user = self.get_current_user()
|
|
return self._post(f"/users/{user['id']}/playlists", {
|
|
"name": name,
|
|
"description": description,
|
|
"public": public,
|
|
})
|
|
|
|
def add_to_playlist(self, playlist_id, uris):
|
|
return self._post(f"/playlists/{playlist_id}/tracks", {"uris": uris})
|
|
|
|
def remove_from_playlist(self, playlist_id, track_ids):
|
|
return self._delete(f"/playlists/{playlist_id}/tracks", {
|
|
"tracks": [{"id": tid} for tid in track_ids],
|
|
})
|
|
|
|
# --- Albums ---
|
|
def get_album(self, album_id):
|
|
return self._get(f"/albums/{album_id}")
|
|
|
|
def get_album_tracks(self, album_id, limit=50, offset=0):
|
|
return self._get(f"/albums/{album_id}/tracks", {"limit": limit, "offset": offset})
|
|
|
|
# --- Artists ---
|
|
def get_artist(self, artist_id):
|
|
return self._get(f"/artists/{artist_id}")
|
|
|
|
def get_artist_top_tracks(self, artist_id, country="US"):
|
|
return self._get(f"/artists/{artist_id}/top-tracks", {"country": country})
|
|
|
|
def get_artist_albums(self, artist_id, limit=50, offset=0):
|
|
return self._get(f"/artists/{artist_id}/albums", {
|
|
"limit": limit, "offset": offset, "include_groups": "album,single"
|
|
})
|
|
|
|
# --- Search ---
|
|
def search(self, query, types="track,album,artist,playlist", limit=20, offset=0):
|
|
return self._get("/search", {
|
|
"q": query, "type": types, "limit": limit, "offset": offset
|
|
})
|
|
|
|
# --- User ---
|
|
def get_current_user(self):
|
|
return self._get("/me")
|
|
|
|
|
|
class LyricsClient:
|
|
LRCLIB_URL = "https://lrclib.net/api"
|
|
|
|
def get_lyrics(self, track_name, artist_name, album_name=None, duration_ms=None):
|
|
import requests as req
|
|
|
|
params = {
|
|
"track_name": track_name,
|
|
"artist_name": artist_name,
|
|
}
|
|
if album_name:
|
|
params["album_name"] = album_name
|
|
if duration_ms:
|
|
params["duration"] = duration_ms // 1000
|
|
|
|
try:
|
|
resp = req.get(f"{self.LRCLIB_URL}/get", params=params, timeout=5)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
return data.get("syncedLyrics") or data.get("plainLyrics")
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
resp = req.get(f"{self.LRCLIB_URL}/search", params={
|
|
"q": f"{track_name} {artist_name}"
|
|
}, timeout=5)
|
|
if resp.status_code == 200:
|
|
results = resp.json()
|
|
if results:
|
|
data = results[0]
|
|
return data.get("syncedLyrics") or data.get("plainLyrics")
|
|
except Exception:
|
|
pass
|
|
|
|
return None
|
|
|
|
|
|
class TokenExpired(Exception):
|
|
pass
|