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.
101 lines
2.9 KiB
Python
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
|