Pierre-Olivier Mercier
d486558883
Give the outgoing HTTP clients (Navidrome, MusicBrainz, ListenBrainz) a transport-level retry budget (RADIEO_HTTP_RETRIES, default 2) so a brief connection blip on a remote endpoint no longer drops a fetch or a lookup. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
132 lines
4.3 KiB
Python
132 lines
4.3 KiB
Python
"""Minimal OpenSubsonic client (enough for Navidrome playback).
|
|
|
|
Uses salted-token authentication (``t = md5(password + salt)``), the scheme
|
|
recommended by the Subsonic API since 1.13.0 and supported by Navidrome.
|
|
"""
|
|
|
|
import hashlib
|
|
import logging
|
|
import secrets
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
from . import config
|
|
|
|
log = logging.getLogger("radieo.subsonic")
|
|
|
|
# Advertised API version and client name.
|
|
_API_VERSION = "1.16.1"
|
|
_CLIENT = "radieo"
|
|
|
|
# Content-Type -> file extension, used to name downloaded files.
|
|
_CTYPE_EXT = {
|
|
"audio/mpeg": ".mp3",
|
|
"audio/mp3": ".mp3",
|
|
"audio/flac": ".flac",
|
|
"audio/x-flac": ".flac",
|
|
"audio/ogg": ".ogg",
|
|
"application/ogg": ".ogg",
|
|
"audio/opus": ".opus",
|
|
"audio/mp4": ".m4a",
|
|
"audio/x-m4a": ".m4a",
|
|
"audio/aac": ".aac",
|
|
"audio/wav": ".wav",
|
|
"audio/x-wav": ".wav",
|
|
}
|
|
|
|
|
|
class SubsonicError(Exception):
|
|
pass
|
|
|
|
|
|
class SubsonicClient:
|
|
def __init__(self, base_url: str, user: str, password: str):
|
|
self._base = base_url.rstrip("/")
|
|
self._user = user
|
|
self._password = password
|
|
self._http = httpx.Client(
|
|
timeout=30.0,
|
|
follow_redirects=True,
|
|
transport=httpx.HTTPTransport(retries=config.HTTP_RETRIES),
|
|
)
|
|
|
|
def _auth_params(self) -> dict[str, str]:
|
|
salt = secrets.token_hex(8)
|
|
token = hashlib.md5((self._password + salt).encode()).hexdigest()
|
|
return {
|
|
"u": self._user,
|
|
"t": token,
|
|
"s": salt,
|
|
"v": _API_VERSION,
|
|
"c": _CLIENT,
|
|
"f": "json",
|
|
}
|
|
|
|
def _get_json(self, view: str, **params) -> dict:
|
|
url = f"{self._base}/rest/{view}"
|
|
resp = self._http.get(url, params={**self._auth_params(), **params})
|
|
resp.raise_for_status()
|
|
body = resp.json()["subsonic-response"]
|
|
if body.get("status") != "ok":
|
|
err = body.get("error", {})
|
|
raise SubsonicError(
|
|
f"{view}: {err.get('code')} {err.get('message')}"
|
|
)
|
|
return body
|
|
|
|
def ping(self) -> None:
|
|
self._get_json("ping")
|
|
|
|
def resolve_playlist_id(self, name_or_id: str) -> str:
|
|
"""Accept either a playlist id or a playlist name."""
|
|
body = self._get_json("getPlaylists")
|
|
playlists = body.get("playlists", {}).get("playlist", [])
|
|
for pl in playlists:
|
|
if pl.get("id") == name_or_id or pl.get("name") == name_or_id:
|
|
return pl["id"]
|
|
raise SubsonicError(f"playlist not found: {name_or_id!r}")
|
|
|
|
def get_playlist_songs(self, playlist_id: str) -> list[dict]:
|
|
body = self._get_json("getPlaylist", id=playlist_id)
|
|
return body.get("playlist", {}).get("entry", [])
|
|
|
|
def search_songs(self, query: str, count: int = 25) -> list[dict]:
|
|
"""Full-text song search (``search3``). Songs only, no artists/albums."""
|
|
body = self._get_json(
|
|
"search3",
|
|
query=query,
|
|
songCount=count,
|
|
artistCount=0,
|
|
albumCount=0,
|
|
)
|
|
return body.get("searchResult3", {}).get("song", [])
|
|
|
|
def download(self, song_id: str, dest: Path, hint_ext: str | None = None) -> str:
|
|
"""Download a song to ``dest``; return the file extension used.
|
|
|
|
``format=raw`` asks Navidrome for the original file (no transcoding),
|
|
keeping quality and letting Liquidsoap decode it.
|
|
"""
|
|
params = {**self._auth_params(), "id": song_id, "format": "raw"}
|
|
with self._http.stream(
|
|
"GET", f"{self._base}/rest/stream", params=params
|
|
) as resp:
|
|
resp.raise_for_status()
|
|
ctype = resp.headers.get("content-type", "").split(";")[0].strip()
|
|
if ctype.startswith(("application/json", "text/xml")):
|
|
raise SubsonicError(
|
|
f"stream {song_id}: error response {resp.read()[:200]!r}"
|
|
)
|
|
ext = _CTYPE_EXT.get(ctype)
|
|
if ext is None and hint_ext:
|
|
ext = "." + hint_ext.lstrip(".")
|
|
if ext is None:
|
|
ext = ".mp3"
|
|
with open(dest, "wb") as fh:
|
|
for chunk in resp.iter_bytes(chunk_size=65536):
|
|
fh.write(chunk)
|
|
return ext
|
|
|
|
def close(self) -> None:
|
|
self._http.close()
|