trav
b7f8761bd1
Snapshot of the existing codebase before working through the TASKS.md backlog. Real library data (data/) and the iTunes import fixture (itunes-test-library/) are gitignored.
200 lines
7.2 KiB
Python
200 lines
7.2 KiB
Python
"""Last.fm scrobbling over the plain web API (no extra deps beyond requests).
|
|
|
|
Auth: auth.getMobileSession with username/password + the user's own API
|
|
key/secret (register at https://www.last.fm/api/account/create). Only the
|
|
returned session key is persisted, never the password.
|
|
|
|
Scrobbles happen when a track plays to the very end (the player's
|
|
track_finished signal). Failures are queued to <data_dir>/scrobble_queue.json
|
|
and retried on startup and after the next successful call. All network I/O
|
|
runs on background threads.
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from PyQt6.QtCore import QObject, pyqtSignal
|
|
|
|
|
|
API_URL = "https://ws.audioscrobbler.com/2.0/"
|
|
QUEUE_FILE = "scrobble_queue.json"
|
|
QUEUE_LIMIT = 500
|
|
|
|
|
|
def sign(params: dict, secret: str) -> str:
|
|
"""Last.fm api_sig: md5 of concatenated key+value pairs (sorted by key,
|
|
excluding 'format'/'callback') followed by the shared secret."""
|
|
pieces = [f"{k}{params[k]}" for k in sorted(params)
|
|
if k not in ("format", "callback")]
|
|
raw = "".join(pieces) + secret
|
|
return hashlib.md5(raw.encode("utf-8")).hexdigest()
|
|
|
|
|
|
class LastFm(QObject):
|
|
login_finished = pyqtSignal(bool, str) # success, message
|
|
status_message = pyqtSignal(str)
|
|
|
|
def __init__(self, prefs, data_dir: Path, parent=None):
|
|
super().__init__(parent)
|
|
self._prefs = prefs
|
|
self._queue_path = Path(data_dir) / QUEUE_FILE
|
|
# Retry anything left over from previous runs
|
|
if self.is_logged_in() and self._load_queue():
|
|
self._spawn(self._flush_queue)
|
|
|
|
# ---- state ----
|
|
|
|
def is_logged_in(self) -> bool:
|
|
lf = self._prefs.lastfm
|
|
return bool(lf.get("session_key") and lf.get("api_key")
|
|
and lf.get("api_secret"))
|
|
|
|
def scrobbling_enabled(self) -> bool:
|
|
return self.is_logged_in() and self._prefs.lastfm.get("scrobble_enabled")
|
|
|
|
# ---- auth ----
|
|
|
|
def login(self, api_key: str, api_secret: str, username: str, password: str):
|
|
def work():
|
|
try:
|
|
data = self._call(api_key, api_secret, "auth.getMobileSession",
|
|
{"username": username, "password": password},
|
|
session_key=None)
|
|
session = data["session"]
|
|
self._prefs.update_lastfm(
|
|
api_key=api_key, api_secret=api_secret,
|
|
username=session.get("name", username),
|
|
session_key=session["key"])
|
|
self.login_finished.emit(
|
|
True, f"Logged in as {session.get('name', username)}")
|
|
except Exception as e:
|
|
self.login_finished.emit(False, f"Login failed: {e}")
|
|
self._spawn(work)
|
|
|
|
def logout(self):
|
|
self._prefs.update_lastfm(session_key="", scrobble_enabled=False)
|
|
|
|
# ---- playback hooks ----
|
|
|
|
def handle_track_started(self, track):
|
|
if track is None or not self.scrobbling_enabled() or not track.artist:
|
|
return
|
|
params = {"artist": track.artist, "track": track.name}
|
|
if track.album:
|
|
params["album"] = track.album
|
|
if track.total_time:
|
|
params["duration"] = str(track.total_time // 1000)
|
|
self._spawn(lambda: self._try_call("track.updateNowPlaying", params))
|
|
|
|
def handle_track_finished(self, track):
|
|
if track is None or not self.scrobbling_enabled() or not track.artist:
|
|
return
|
|
duration_s = (track.total_time or 0) // 1000
|
|
entry = {
|
|
"artist": track.artist,
|
|
"track": track.name,
|
|
"album": track.album or "",
|
|
"timestamp": int(time.time()) - duration_s,
|
|
"duration": duration_s,
|
|
}
|
|
self._spawn(lambda: self._scrobble_entry(entry))
|
|
|
|
# ---- internals ----
|
|
|
|
def _scrobble_entry(self, entry: dict):
|
|
params = {"artist": entry["artist"], "track": entry["track"],
|
|
"timestamp": str(entry["timestamp"])}
|
|
if entry.get("album"):
|
|
params["album"] = entry["album"]
|
|
if entry.get("duration"):
|
|
params["duration"] = str(entry["duration"])
|
|
try:
|
|
self._call_with_session("track.scrobble", params)
|
|
self.status_message.emit(
|
|
f"Scrobbled: {entry['artist']} — {entry['track']}")
|
|
self._flush_queue()
|
|
except Exception as e:
|
|
self._enqueue(entry)
|
|
self.status_message.emit(f"Scrobble queued (offline?): {e}")
|
|
|
|
def _flush_queue(self):
|
|
queue = self._load_queue()
|
|
if not queue:
|
|
return
|
|
remaining = []
|
|
for entry in queue:
|
|
try:
|
|
params = {"artist": entry["artist"], "track": entry["track"],
|
|
"timestamp": str(entry["timestamp"])}
|
|
if entry.get("album"):
|
|
params["album"] = entry["album"]
|
|
self._call_with_session("track.scrobble", params)
|
|
except Exception:
|
|
remaining.append(entry)
|
|
self._save_queue(remaining)
|
|
flushed = len(queue) - len(remaining)
|
|
if flushed:
|
|
self.status_message.emit(f"Sent {flushed} queued scrobble(s)")
|
|
|
|
def _try_call(self, method: str, params: dict):
|
|
try:
|
|
self._call_with_session(method, params)
|
|
except Exception:
|
|
pass # now-playing updates are best-effort
|
|
|
|
def _call_with_session(self, method: str, params: dict) -> dict:
|
|
lf = self._prefs.lastfm
|
|
return self._call(lf["api_key"], lf["api_secret"], method, params,
|
|
session_key=lf["session_key"])
|
|
|
|
@staticmethod
|
|
def _call(api_key: str, secret: str, method: str, params: dict,
|
|
session_key: str | None) -> dict:
|
|
import requests
|
|
payload = dict(params)
|
|
payload["method"] = method
|
|
payload["api_key"] = api_key
|
|
if session_key:
|
|
payload["sk"] = session_key
|
|
payload["api_sig"] = sign(payload, secret)
|
|
payload["format"] = "json"
|
|
response = requests.post(API_URL, data=payload, timeout=15)
|
|
data = response.json()
|
|
if "error" in data:
|
|
raise RuntimeError(data.get("message", f"error {data['error']}"))
|
|
response.raise_for_status()
|
|
return data
|
|
|
|
# ---- offline queue ----
|
|
|
|
def _enqueue(self, entry: dict):
|
|
queue = self._load_queue()
|
|
queue.append(entry)
|
|
self._save_queue(queue[-QUEUE_LIMIT:])
|
|
|
|
def _load_queue(self) -> list:
|
|
if not self._queue_path.exists():
|
|
return []
|
|
try:
|
|
with open(self._queue_path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except (ValueError, OSError):
|
|
return []
|
|
|
|
def _save_queue(self, queue: list):
|
|
if not queue:
|
|
if self._queue_path.exists():
|
|
self._queue_path.unlink()
|
|
return
|
|
self._queue_path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = self._queue_path.with_suffix(".json.tmp")
|
|
with open(tmp, "w", encoding="utf-8") as f:
|
|
json.dump(queue, f, indent=2)
|
|
tmp.replace(self._queue_path)
|
|
|
|
@staticmethod
|
|
def _spawn(fn):
|
|
threading.Thread(target=fn, daemon=True).start()
|