feat(discovery): ListenBrainz similarity source

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jonathan
2026-07-12 00:17:58 +02:00
parent 0338d8e119
commit 6f504a7c04
2 changed files with 92 additions and 0 deletions
@@ -0,0 +1,49 @@
import requests
from lyra_worker.similarity.base import SimilarArtist
_DEFAULT_URL = "https://labs.api.listenbrainz.org"
# ListenBrainz Labs similar-artists dataset requires an algorithm parameter.
# NOTE: confirm this string + response field names against the live API during
# manual verification; parsing below is defensive if the shape shifts.
_DEFAULT_ALGORITHM = (
"session_based_days_7500_session_300_contribution_5_threshold_15_limit_50_filter_True_skip_30"
)
class ListenBrainzSource:
"""SimilaritySource backed by the (unauthenticated) ListenBrainz Labs API."""
name = "listenbrainz"
def __init__(self, base_url: str = _DEFAULT_URL, algorithm: str = _DEFAULT_ALGORITHM,
timeout: float = 15.0):
self._base = (base_url or _DEFAULT_URL).rstrip("/")
self._algorithm = algorithm
self._timeout = timeout
def health(self) -> bool:
try:
res = requests.get(self._base + "/", timeout=self._timeout)
return res.status_code < 500
except requests.RequestException:
return False
def similar_artists(self, mbid: str) -> list[SimilarArtist]:
res = requests.get(
self._base + "/similar-artists/json",
params={"artist_mbids": mbid, "algorithm": self._algorithm},
timeout=self._timeout,
)
res.raise_for_status()
out: list[SimilarArtist] = []
for row in res.json() or []:
ambid = row.get("artist_mbid")
if not ambid or ambid == mbid:
continue
out.append(SimilarArtist(
mbid=ambid,
name=row.get("name") or row.get("comment") or "",
score=float(row.get("score") or 0),
))
return out