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>
95 lines
3.7 KiB
Python
95 lines
3.7 KiB
Python
"""Playwish – Verbindung Player → Server (Player-API mit Bearer-Token, nur ausgehend).
|
||
|
||
Alle Aufrufe fangen Fehler selbst ab: Ist der Server nicht erreichbar, liefern sie
|
||
None/False, und der Controller spielt einfach die Grund-Playlist weiter. Im
|
||
Offline-Fall wird nur alle OFFLINE_RETRY_S Sekunden ein neuer Versuch gemacht.
|
||
"""
|
||
|
||
import logging
|
||
import time
|
||
from dataclasses import dataclass
|
||
|
||
import httpx
|
||
|
||
log = logging.getLogger("playwish.server")
|
||
|
||
OFFLINE_RETRY_S = 15
|
||
TIMEOUT_S = 5
|
||
|
||
|
||
@dataclass
|
||
class Wunsch:
|
||
id: int
|
||
uri: str
|
||
title: str = ""
|
||
artist: str = ""
|
||
priority: bool = False # "Als Nächstes" des Besitzers
|
||
|
||
|
||
class ServerVerbindung:
|
||
def __init__(self, basis_url: str, token: str):
|
||
self._http = httpx.Client(base_url=basis_url.rstrip("/") + "/api/player", timeout=TIMEOUT_S,
|
||
headers={"Authorization": f"Bearer {token}"})
|
||
self.online: bool | None = None
|
||
self._naechster_versuch = 0.0
|
||
self._unbestaetigt: set[int] = set() # gespielte Wünsche, deren Meldung noch fehlt
|
||
|
||
# --- intern ----------------------------------------------------------------
|
||
|
||
def _anfrage(self, methode: str, pfad: str, **kwargs):
|
||
"""JSON der Antwort oder None bei jedem Fehler (Netz, Timeout, HTTP-Fehler)."""
|
||
if self.online is False and time.monotonic() < self._naechster_versuch:
|
||
return None
|
||
try:
|
||
r = self._http.request(methode, pfad, **kwargs)
|
||
if r.status_code == 401:
|
||
self._offline("Token abgelehnt (401) – PLAYER_API_TOKEN prüfen")
|
||
return None
|
||
r.raise_for_status()
|
||
if self.online is not True:
|
||
log.info("Server erreichbar")
|
||
self.online = True
|
||
return r.json() if r.content else {}
|
||
except (httpx.HTTPError, ValueError) as e:
|
||
self._offline(f"{type(e).__name__}")
|
||
return None
|
||
|
||
def _offline(self, grund: str) -> None:
|
||
if self.online is not False:
|
||
log.warning("Server nicht erreichbar (%s) – spiele Grund-Playlist weiter, neuer Versuch alle %d s",
|
||
grund, OFFLINE_RETRY_S)
|
||
self.online = False
|
||
self._naechster_versuch = time.monotonic() + OFFLINE_RETRY_S
|
||
|
||
# --- Player-API ----------------------------------------------------------------
|
||
|
||
def konfiguration(self) -> dict | None:
|
||
return self._anfrage("GET", "/config")
|
||
|
||
def naechster_wunsch(self) -> Wunsch | None:
|
||
self._bestaetigungen_nachholen()
|
||
daten = self._anfrage("GET", "/next-wish")
|
||
w = (daten or {}).get("wish")
|
||
if not w or w["id"] in self._unbestaetigt:
|
||
return None # noch nicht als gespielt bestätigt → nicht doppelt spielen
|
||
return Wunsch(id=w["id"], uri=w["track_uri"], title=w.get("title", ""), artist=w.get("artist", ""),
|
||
priority=bool(w.get("priority")))
|
||
|
||
def wunsch_gespielt(self, wunsch: Wunsch) -> None:
|
||
self._unbestaetigt.add(wunsch.id)
|
||
self._bestaetigungen_nachholen()
|
||
|
||
def _bestaetigungen_nachholen(self) -> None:
|
||
for wid in list(self._unbestaetigt):
|
||
if self._anfrage("POST", f"/wish/{wid}/played") is not None:
|
||
self._unbestaetigt.discard(wid)
|
||
|
||
def melden(self, zustand: dict) -> bool:
|
||
return self._anfrage("POST", "/now-playing", json=zustand) is not None
|
||
|
||
def bluetooth_melden(self, zustand: dict) -> bool:
|
||
return self._anfrage("POST", "/bluetooth", json=zustand) is not None
|
||
|
||
def quittieren(self, befehl_id: int, ok: bool, meldung: str | None = None) -> None:
|
||
self._anfrage("POST", f"/commands/{befehl_id}/ack", json={"ok": ok, "message": meldung})
|