feat(worker): Last.fm artist.getSimilar as a second discovery SimilaritySource

Adds LastfmSource, resolving Last.fm's often-empty artist mbid via the
injected MB browser (name->mbid cached) so every emitted SimilarArtist.mbid
is a real MusicBrainz artist MBID. Registered in build_similarity_sources,
gated on lastfm.api_key so the ListenBrainz-only path is unchanged when
unset.
This commit is contained in:
Jonathan
2026-07-13 19:23:18 +02:00
parent 284872b307
commit 146bf32f23
4 changed files with 202 additions and 2 deletions
+69
View File
@@ -0,0 +1,69 @@
import requests
from lyra_worker.similarity.base import SimilarArtist
_BASE = "https://ws.audioscrobbler.com/2.0/"
class LastfmSource:
"""SimilaritySource backed by Last.fm's artist.getsimilar.
Last.fm rows often carry an empty mbid, so names are resolved to a real
MusicBrainz artist MBID via the injected browser. Unresolvable rows are
skipped -- every emitted SimilarArtist.mbid must be a real MB artist MBID,
since downstream keys on it.
"""
name = "lastfm"
def __init__(self, api_key, browser, base_url=_BASE, timeout: float = 15.0,
similar_limit: int = 50):
self._key = api_key
self._browser = browser # MbBrowser: .search_artist(name) -> list[ArtistHit]
self._base = base_url
self._timeout = timeout
self._limit = similar_limit
self._name_cache: dict[str, str | None] = {}
def _get(self, method: str, **params):
res = requests.get(
self._base,
params={"method": method, "api_key": self._key, "format": "json", **params},
timeout=self._timeout,
)
return res.json()
def health(self) -> bool:
try:
data = self._get("artist.getsimilar", artist="Radiohead", limit="1")
return not isinstance(data.get("error"), (int, float))
except requests.RequestException:
return False
def _resolve(self, name: str) -> str | None:
key = name.lower()
if key in self._name_cache:
return self._name_cache[key]
hits = self._browser.search_artist(name)
mbid = hits[0].mbid if hits else None
self._name_cache[key] = mbid
return mbid
def similar_artists(self, mbid: str) -> list[SimilarArtist]:
data = self._get("artist.getsimilar", mbid=mbid, limit=str(self._limit))
if isinstance(data.get("error"), (int, float)):
return []
rows = data.get("similarartists", {}).get("artist", [])
if isinstance(rows, dict):
rows = [rows]
out: list[SimilarArtist] = []
for row in rows:
name = row.get("name") or ""
cand = row.get("mbid") or None
if not cand:
cand = self._resolve(name)
if not cand or cand == mbid or not name:
continue
score = float(row.get("match") or 0)
out.append(SimilarArtist(mbid=cand, name=name, score=score))
return out