99admandClaude Opus 5.5
fc15f2b933
Name, Seitentitel, Dienste (playwish-web, playwish-controller), Dienst-Benutzer (playwish), Pfade (/opt/playwish, /usr/local/src/playwish), Vorlagen und Repository-Adresse. „Audiofy“ war bereits von mehreren Produkten belegt. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
117 lines
4.8 KiB
Python
117 lines
4.8 KiB
Python
"""Playwish-Web – Spotify-Zugriff für Suche und Titelinfos.
|
||
|
||
Eigener Refresh-Token des Servers (auth.py --profil server, keine Scopes nötig).
|
||
Im Development Mode liefert die Suche höchstens 10 Treffer.
|
||
"""
|
||
|
||
import base64
|
||
import logging
|
||
import threading
|
||
import time
|
||
from functools import lru_cache
|
||
|
||
import httpx
|
||
|
||
from .config import Konfig, env_wert_setzen, konfig
|
||
|
||
log = logging.getLogger("playwish.spotify")
|
||
|
||
API = "https://api.spotify.com/v1"
|
||
TOKEN_URL = "https://accounts.spotify.com/api/token"
|
||
CACHE_S = 300 # Suchergebnisse so lange zwischenspeichern
|
||
CACHE_MAX = 500
|
||
|
||
|
||
class SpotifyFehler(Exception):
|
||
def __init__(self, status: int, retry_after: int | None = None):
|
||
super().__init__(f"Spotify-Fehler {status}")
|
||
self.status = status
|
||
self.retry_after = retry_after
|
||
|
||
|
||
def titel_aus(t: dict) -> dict:
|
||
"""Spotify-Track-Objekt → schlankes Dict für Website und Datenbank."""
|
||
bilder = sorted((t.get("album") or {}).get("images") or [], key=lambda b: b.get("width") or 0)
|
||
# kleinstes Bild ab 200 px (gut für Listen), sonst das größte
|
||
cover = next((b["url"] for b in bilder if (b.get("width") or 0) >= 200), bilder[-1]["url"] if bilder else None)
|
||
return {
|
||
"track_uri": t["uri"],
|
||
"title": t.get("name") or "",
|
||
"artist": ", ".join(a.get("name", "") for a in t.get("artists") or []),
|
||
"artist_uris": [a["uri"] for a in t.get("artists") or [] if a.get("uri")],
|
||
"artists": [{"name": a.get("name", ""), "uri": a["uri"]} for a in t.get("artists") or [] if a.get("uri")],
|
||
"album": (t.get("album") or {}).get("name"),
|
||
"cover_url": cover,
|
||
"duration_ms": t.get("duration_ms"),
|
||
"explicit": bool(t.get("explicit")),
|
||
}
|
||
|
||
|
||
class Spotify:
|
||
def __init__(self, k: Konfig):
|
||
self._k = k
|
||
self._refresh_token = k.spotify_refresh_token
|
||
self._token: str | None = None
|
||
self._ablauf = 0.0
|
||
self._lock = threading.Lock()
|
||
self._http = httpx.Client(timeout=8)
|
||
self._cache: dict[str, tuple[float, list[dict]]] = {}
|
||
|
||
def verfuegbar(self) -> bool:
|
||
return bool(self._k.spotify_client_id and self._k.spotify_client_secret and self._refresh_token)
|
||
|
||
def _access_token(self) -> str:
|
||
with self._lock:
|
||
if self._token and time.monotonic() < self._ablauf - 60:
|
||
return self._token
|
||
basic = base64.b64encode(f"{self._k.spotify_client_id}:{self._k.spotify_client_secret}".encode()).decode()
|
||
r = self._http.post(TOKEN_URL, headers={"Authorization": f"Basic {basic}"},
|
||
data={"grant_type": "refresh_token", "refresh_token": self._refresh_token})
|
||
if r.status_code != 200:
|
||
log.error("Token-Erneuerung fehlgeschlagen (%s)", r.status_code)
|
||
raise SpotifyFehler(r.status_code)
|
||
daten = r.json()
|
||
self._token = daten["access_token"]
|
||
self._ablauf = time.monotonic() + daten.get("expires_in", 3600)
|
||
neu = daten.get("refresh_token")
|
||
if neu and neu != self._refresh_token:
|
||
self._refresh_token = neu
|
||
env_wert_setzen(self._k.env_pfad, "SPOTIFY_REFRESH_TOKEN", neu)
|
||
log.info("Neuer Refresh-Token von Spotify gespeichert")
|
||
return self._token
|
||
|
||
def _get(self, pfad: str, params: dict) -> dict:
|
||
for versuch in range(2):
|
||
r = self._http.get(API + pfad, params=params, headers={"Authorization": f"Bearer {self._access_token()}"})
|
||
if r.status_code == 401 and versuch == 0:
|
||
self._token = None
|
||
continue
|
||
if r.status_code == 429:
|
||
raise SpotifyFehler(429, int(r.headers.get("Retry-After", "5")))
|
||
if r.status_code >= 400:
|
||
raise SpotifyFehler(r.status_code)
|
||
return r.json()
|
||
raise SpotifyFehler(401)
|
||
|
||
def suche(self, q: str) -> list[dict]:
|
||
schluessel = " ".join(q.casefold().split())
|
||
treffer = self._cache.get(schluessel)
|
||
if treffer and time.monotonic() - treffer[0] < CACHE_S:
|
||
return treffer[1]
|
||
daten = self._get("/search", {"q": q, "type": "track", "limit": 10, "market": "from_token"})
|
||
ergebnis = [titel_aus(t) for t in (daten.get("tracks") or {}).get("items") or []
|
||
if t and t.get("uri", "").startswith("spotify:track:") and t.get("is_playable", True)]
|
||
if len(self._cache) >= CACHE_MAX:
|
||
self._cache.clear()
|
||
self._cache[schluessel] = (time.monotonic(), ergebnis)
|
||
return ergebnis
|
||
|
||
def titel(self, uri: str) -> dict:
|
||
return titel_aus(self._get(f"/tracks/{uri.rsplit(':', 1)[-1]}", {"market": "from_token"}))
|
||
|
||
|
||
@lru_cache
|
||
def spotify() -> Spotify:
|
||
"""FastAPI-Dependency (in Tests per dependency_overrides ersetzbar)."""
|
||
return Spotify(konfig())
|