Jonathan
146bf32f23
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.
70 lines
2.4 KiB
Python
70 lines
2.4 KiB
Python
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
|