Files
vynl/backend/app/services/lastfm.py
root 90945932ad Add Last.fm import support
Users can import their top tracks from Last.fm by entering their username.
No OAuth required - uses the public Last.fm API with user.getTopTracks and
user.getInfo endpoints. Includes a preview feature to verify the username
and see sample tracks before importing. Supports configurable time periods
(all time, 7 days, 1/3/6/12 months). Free tier playlist limit enforced.
2026-03-30 22:49:13 -05:00

101 lines
2.9 KiB
Python

import httpx
from app.core.config import settings
LASTFM_API_URL = "http://ws.audioscrobbler.com/2.0/"
async def get_user_info(username: str) -> dict | None:
"""Verify a Last.fm username exists and return basic info."""
params = {
"method": "user.getInfo",
"user": username,
"api_key": settings.LASTFM_API_KEY,
"format": "json",
}
async with httpx.AsyncClient(timeout=15) as client:
resp = await client.get(LASTFM_API_URL, params=params)
if resp.status_code != 200:
return None
data = resp.json()
if "error" in data:
return None
user = data.get("user", {})
image_url = None
images = user.get("image", [])
for img in images:
if img.get("size") == "large" and img.get("#text"):
image_url = img["#text"]
return {
"name": user.get("name", username),
"display_name": user.get("realname") or user.get("name", username),
"playcount": int(user.get("playcount", 0)),
"image_url": image_url,
"url": user.get("url", ""),
}
async def get_top_tracks(
username: str, period: str = "overall", limit: int = 50
) -> list[dict]:
"""Fetch a user's top tracks from Last.fm.
Args:
username: Last.fm username.
period: Time period - overall, 7day, 1month, 3month, 6month, 12month.
limit: Max number of tracks to return (max 1000).
Returns:
List of dicts with: title, artist, playcount, image_url.
"""
valid_periods = {"overall", "7day", "1month", "3month", "6month", "12month"}
if period not in valid_periods:
period = "overall"
params = {
"method": "user.getTopTracks",
"user": username,
"period": period,
"limit": min(limit, 1000),
"api_key": settings.LASTFM_API_KEY,
"format": "json",
}
async with httpx.AsyncClient(timeout=30) as client:
resp = await client.get(LASTFM_API_URL, params=params)
if resp.status_code != 200:
raise ValueError(f"Last.fm API returned status {resp.status_code}")
data = resp.json()
if "error" in data:
raise ValueError(data.get("message", "Last.fm API error"))
raw_tracks = data.get("toptracks", {}).get("track", [])
tracks = []
for t in raw_tracks:
image_url = None
images = t.get("image", [])
for img in images:
if img.get("size") == "large" and img.get("#text"):
image_url = img["#text"]
artist_name = ""
artist = t.get("artist")
if isinstance(artist, dict):
artist_name = artist.get("name", "")
elif isinstance(artist, str):
artist_name = artist
tracks.append({
"title": t.get("name", "Unknown"),
"artist": artist_name or "Unknown",
"playcount": int(t.get("playcount", 0)),
"image_url": image_url,
})
return tracks