Jonathan
f72d193ef7
A 4xx/5xx (rate limit / auth) now raises instead of parsing an error body; callers already handle it (discovery per-source try/except; album fetch returns []). Closes the last small deferred cleanup. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
55 lines
2.0 KiB
Python
55 lines
2.0 KiB
Python
import requests
|
|
|
|
from lyra_worker.apicache import cached_api
|
|
|
|
_BASE = "https://ws.audioscrobbler.com/2.0/"
|
|
|
|
|
|
def _parse_top_albums(data) -> list[list]:
|
|
"""Parse a Last.fm artist.gettopalbums response into [[name, mbid|None, playcount], …]
|
|
ordered by playcount desc. A single-album body arrives as a dict; an error body → []."""
|
|
if isinstance(data.get("error"), (int, float)):
|
|
return []
|
|
rows = data.get("topalbums", {}).get("album", [])
|
|
if isinstance(rows, dict):
|
|
rows = [rows]
|
|
out: list[list] = []
|
|
for r in rows:
|
|
name = r.get("name") or ""
|
|
if not name:
|
|
continue
|
|
out.append([name, r.get("mbid") or None, int(r.get("playcount") or 0)])
|
|
out.sort(key=lambda a: a[2], reverse=True)
|
|
return out
|
|
|
|
|
|
class LastfmAlbumPopularity:
|
|
"""Ranks an artist's albums by Last.fm playcount, so discovery can surface each similar
|
|
artist's most-played album (alongside their newest). Read-through cached in ApiCache; any
|
|
failure yields [] so the sweep just falls back to newest-only."""
|
|
|
|
def __init__(self, api_key, base_url=_BASE, timeout: float = 15.0, limit: int = 25):
|
|
self._key = api_key
|
|
self._base = base_url
|
|
self._timeout = timeout
|
|
self._limit = limit
|
|
|
|
def _fetch(self, artist_name: str) -> list[list]:
|
|
try:
|
|
res = requests.get(self._base, params={
|
|
"method": "artist.gettopalbums", "artist": artist_name,
|
|
"api_key": self._key, "format": "json", "limit": str(self._limit),
|
|
"autocorrect": "1",
|
|
}, timeout=self._timeout)
|
|
res.raise_for_status()
|
|
return _parse_top_albums(res.json())
|
|
except Exception: # a Last.fm outage must never abort the sweep
|
|
return []
|
|
|
|
def top_albums(self, conn, artist_name: str) -> list[list]:
|
|
norm = (artist_name or "").strip().lower()
|
|
if not norm:
|
|
return []
|
|
return cached_api(conn, f"lfm-artist-top-albums:{norm}", 43200,
|
|
lambda: self._fetch(artist_name))
|