reworked it all
This commit is contained in:
Executable
+815
@@ -0,0 +1,815 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Ainulindale discovery orchestrator.
|
||||
|
||||
Runs inside the Explo container (python 3.12, stdlib only) and drives the Explo
|
||||
binary instead of Explo's built-in cron. One idempotent "tick" does this:
|
||||
|
||||
1. Preflight make sure slskd is actually logged in to Soulseek, otherwise
|
||||
Explo would silently fall back to YouTube for the whole week.
|
||||
2. ListenBrainz import Weekly Exploration, but only if ListenBrainz has
|
||||
published a playlist we have not imported yet.
|
||||
3. Last.fm build a playlist from Last.fm recommendations and hand it to
|
||||
Explo as a "custom" playlist (always, or only as a fallback).
|
||||
4. Prune keep the newest N weeks per source, remove older playlists
|
||||
and their files; starred / playlisted tracks are rescued.
|
||||
|
||||
Because every step checks state first, the tick can run as often as you like
|
||||
and after any restart without ever producing duplicates.
|
||||
|
||||
Usage: discover.py run | status | prune [--dry-run] | lastfm-preview
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as dt
|
||||
import fcntl
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import random
|
||||
import re
|
||||
import secrets
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Configuration (all from environment, see .env.example)
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def env(name: str, default: str = "") -> str:
|
||||
return os.environ.get(name, default).strip()
|
||||
|
||||
|
||||
def env_bool(name: str, default: bool) -> bool:
|
||||
v = env(name)
|
||||
return default if v == "" else v.lower() in ("1", "true", "yes", "on")
|
||||
|
||||
|
||||
def env_int(name: str, default: int) -> int:
|
||||
try:
|
||||
return int(env(name, str(default)))
|
||||
except ValueError:
|
||||
return default
|
||||
|
||||
|
||||
CONFIG_DIR = env("WEB_DATA_PATH", "/opt/explo/config") # Explo reads its playlist cache from here
|
||||
DATA_DIR = env("DOWNLOAD_DIR", "/data")
|
||||
SLSKD_DIR = env("SLSKD_DIR", "/slskd")
|
||||
EXPLO_BIN = env("EXPLO_BIN", "/opt/explo/explo")
|
||||
EXPLO_ENV = env("WEB_ENV_PATH", "/opt/explo/.env")
|
||||
STATE_FILE = os.path.join(CONFIG_DIR, "ainulindale-state.json")
|
||||
LOCK_FILE = os.path.join(CONFIG_DIR, "ainulindale.lock")
|
||||
KEEPERS_DIR = os.path.join(DATA_DIR, "Keepers")
|
||||
|
||||
LB_USER = env("LISTENBRAINZ_USER")
|
||||
LB_TOKEN = env("LISTENBRAINZ_USER_TOKEN")
|
||||
LB_ENABLED = env_bool("LISTENBRAINZ_ENABLED", True)
|
||||
LB_MAX_ATTEMPTS = env_int("LISTENBRAINZ_MAX_ATTEMPTS", 3)
|
||||
|
||||
LASTFM_USER = env("LASTFM_USER")
|
||||
LASTFM_API_KEY = env("LASTFM_API_KEY")
|
||||
LASTFM_MODE = env("LASTFM_MODE", "always").lower() # always | fallback | off
|
||||
LASTFM_STATIONS = [s for s in env("LASTFM_STATIONS", "recommended").split(",") if s.strip()]
|
||||
LASTFM_TRACKS = env_int("LASTFM_TRACKS", 30)
|
||||
LASTFM_FALLBACK_DAY = env_int("LASTFM_FALLBACK_DAY", 3) # ISO weekday, 3 = Wednesday
|
||||
LASTFM_HISTORY = 1000 # remembered recommendations, so weeks do not repeat each other
|
||||
|
||||
NAVIDROME_URL = env("SYSTEM_URL").rstrip("/")
|
||||
NAVIDROME_USER = env("SYSTEM_USERNAME")
|
||||
NAVIDROME_PASS = env("SYSTEM_PASSWORD")
|
||||
|
||||
SLSKD_URL = env("SLSKD_URL").rstrip("/")
|
||||
SLSKD_API_KEY = env("SLSKD_API_KEY")
|
||||
USES_SLSKD = "slskd" in env("DOWNLOAD_SERVICES", "slskd,youtube").split(",")
|
||||
REQUIRE_SLSKD = env_bool("REQUIRE_SLSKD", True)
|
||||
SLSKD_WAIT_MIN = env_int("SLSKD_WAIT_MINUTES", 10)
|
||||
|
||||
KEEP_WEEKS = max(1, env_int("KEEP_WEEKS", 2))
|
||||
PRUNE_FILES = env_bool("PRUNE_FILES", True)
|
||||
KEEP_FAVOURITES = env_bool("KEEP_FAVOURITES", True)
|
||||
PRUNE_LEGACY_JAMS = env_bool("PRUNE_LEGACY_JAMS", False)
|
||||
EXPLO_TIMEOUT_MIN = env_int("EXPLO_TIMEOUT_MINUTES", 480)
|
||||
|
||||
USER_AGENT = "Mozilla/5.0 (compatible; Ainulindale/2.0; +https://github.com/Quinta0/Ainulindale)"
|
||||
AUDIO_EXT = {".flac", ".wav", ".mp3", ".m4a", ".ogg", ".opus", ".aac", ".wma", ".aiff", ".ape"}
|
||||
|
||||
|
||||
def log(msg: str) -> None:
|
||||
print(f"{dt.datetime.now():%Y-%m-%d %H:%M:%S} [ainulindale] {msg}", flush=True)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Small HTTP helper
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
class HttpError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def http_json(url, *, method="GET", headers=None, params=None, timeout=20, retries=2, allow_empty=False):
|
||||
if params:
|
||||
url = f"{url}{'&' if '?' in url else '?'}{urllib.parse.urlencode(params)}"
|
||||
hdrs = {"User-Agent": USER_AGENT, "Accept": "application/json"}
|
||||
hdrs.update(headers or {})
|
||||
last = None
|
||||
for attempt in range(retries + 1):
|
||||
try:
|
||||
data = b"" if method in ("PUT", "POST") else None
|
||||
req = urllib.request.Request(url, method=method, headers=hdrs, data=data)
|
||||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||||
body = resp.read()
|
||||
if not body.strip():
|
||||
if allow_empty:
|
||||
return None
|
||||
raise HttpError("empty response")
|
||||
return json.loads(body)
|
||||
except urllib.error.HTTPError as e:
|
||||
last = HttpError(f"HTTP {e.code} from {urllib.parse.urlsplit(url).netloc}")
|
||||
if e.code in (400, 401, 403, 404):
|
||||
break
|
||||
except (urllib.error.URLError, TimeoutError, OSError, ValueError) as e:
|
||||
last = HttpError(f"{type(e).__name__}: {e}")
|
||||
if attempt < retries:
|
||||
time.sleep(3 * (attempt + 1))
|
||||
raise last
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# State
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def load_state() -> dict:
|
||||
try:
|
||||
with open(STATE_FILE, encoding="utf-8") as f:
|
||||
state = json.load(f)
|
||||
except (OSError, ValueError):
|
||||
state = {}
|
||||
state.setdefault("listenbrainz", {})
|
||||
state.setdefault("lastfm", {})
|
||||
state["lastfm"].setdefault("history", [])
|
||||
return state
|
||||
|
||||
|
||||
def save_state(state: dict) -> None:
|
||||
os.makedirs(CONFIG_DIR, exist_ok=True)
|
||||
tmp = STATE_FILE + ".tmp"
|
||||
with open(tmp, "w", encoding="utf-8") as f:
|
||||
json.dump(state, f, indent=2, ensure_ascii=False)
|
||||
os.replace(tmp, STATE_FILE)
|
||||
|
||||
|
||||
def iso_week(d: dt.date | None = None) -> tuple[int, int]:
|
||||
y, w, _ = (d or dt.date.today()).isocalendar()
|
||||
return y, w
|
||||
|
||||
|
||||
def week_key(d: dt.date | None = None) -> str:
|
||||
y, w = iso_week(d)
|
||||
return f"{y}-W{w:02d}"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# slskd
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
class Slskd:
|
||||
def __init__(self, url: str, key: str):
|
||||
self.url, self.key = url, key
|
||||
|
||||
def server_state(self) -> dict:
|
||||
return http_json(f"{self.url}/api/v0/server", headers={"X-API-Key": self.key}, timeout=10, retries=0)
|
||||
|
||||
def connect(self) -> None:
|
||||
http_json(f"{self.url}/api/v0/server", method="PUT", headers={"X-API-Key": self.key},
|
||||
timeout=10, retries=0, allow_empty=True)
|
||||
|
||||
def wait_until_logged_in(self, minutes: int) -> bool:
|
||||
"""True once slskd reports Connected+LoggedIn. Nudges a reconnect while waiting."""
|
||||
deadline = time.monotonic() + minutes * 60
|
||||
nudged_at = None # not 0.0: monotonic() itself can be < 120 right after a host boot
|
||||
while True:
|
||||
try:
|
||||
st = self.server_state()
|
||||
if st.get("isConnected") and st.get("isLoggedIn"):
|
||||
return True
|
||||
desc = st.get("state", "unknown")
|
||||
transitioning = st.get("isConnecting") or st.get("isLoggingIn")
|
||||
if not transitioning and (nudged_at is None or time.monotonic() - nudged_at > 120):
|
||||
log(f"slskd is '{desc}', asking it to reconnect")
|
||||
try:
|
||||
self.connect()
|
||||
except HttpError as e:
|
||||
log(f"reconnect request failed: {e}")
|
||||
nudged_at = time.monotonic()
|
||||
except HttpError as e:
|
||||
log(f"slskd not reachable at {self.url}: {e}")
|
||||
if time.monotonic() >= deadline:
|
||||
return False
|
||||
time.sleep(20)
|
||||
|
||||
|
||||
def slskd_ready() -> bool:
|
||||
if not USES_SLSKD:
|
||||
return True
|
||||
if not (SLSKD_URL and SLSKD_API_KEY):
|
||||
log("SLSKD_URL / SLSKD_API_KEY not set, skipping slskd preflight")
|
||||
return True
|
||||
if Slskd(SLSKD_URL, SLSKD_API_KEY).wait_until_logged_in(SLSKD_WAIT_MIN):
|
||||
return True
|
||||
if REQUIRE_SLSKD:
|
||||
log(f"slskd did not log in to Soulseek within {SLSKD_WAIT_MIN} min. Skipping this run so the week "
|
||||
"is not filled with YouTube rips; the next tick will retry.")
|
||||
return False
|
||||
log("slskd is not logged in, continuing anyway because REQUIRE_SLSKD=false")
|
||||
return True
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# ListenBrainz
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
JSPF_PLAYLIST = "https://musicbrainz.org/doc/jspf#playlist"
|
||||
|
||||
|
||||
def lb_latest_playlist(user: str, patch: str = "weekly-exploration") -> dict | None:
|
||||
"""Newest 'created for you' playlist of the given type: {'mbid', 'date', 'title'} or None."""
|
||||
headers = {"Authorization": f"Token {LB_TOKEN}"} if LB_TOKEN else {}
|
||||
best = None
|
||||
offset = 0
|
||||
while True:
|
||||
page = http_json(
|
||||
f"https://api.listenbrainz.org/1/user/{urllib.parse.quote(user)}/playlists/createdfor",
|
||||
params={"count": 50, "offset": offset}, headers=headers, timeout=30)
|
||||
items = page.get("playlists", [])
|
||||
for item in items:
|
||||
pl = item.get("playlist", {})
|
||||
meta = pl.get("extension", {}).get(JSPF_PLAYLIST, {}).get("additional_metadata", {})
|
||||
if meta.get("algorithm_metadata", {}).get("source_patch") != patch:
|
||||
continue
|
||||
date = pl.get("date", "")
|
||||
if best is None or date > best["date"]:
|
||||
best = {"mbid": pl.get("identifier", "").rstrip("/").rsplit("/", 1)[-1],
|
||||
"date": date, "title": pl.get("title", "")}
|
||||
offset += len(items)
|
||||
if not items or offset >= page.get("playlist_count", 0):
|
||||
return best
|
||||
|
||||
|
||||
def lb_is_fresh(state: dict) -> bool:
|
||||
"""Has a ListenBrainz playlist published in the last 7 days been imported?"""
|
||||
date = state["listenbrainz"].get("imported_date", "")
|
||||
try:
|
||||
published = dt.datetime.fromisoformat(date.replace("Z", "+00:00"))
|
||||
except ValueError:
|
||||
return False
|
||||
return dt.datetime.now(dt.timezone.utc) - published < dt.timedelta(days=7)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Last.fm
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def track_key(artist: str, title: str) -> str:
|
||||
return re.sub(r"[^\w]+", "", f"{artist}|{title}".lower(), flags=re.UNICODE)
|
||||
|
||||
|
||||
def lastfm_station(user: str, station: str, want: int, seen: set[str]) -> list[dict]:
|
||||
"""
|
||||
Last.fm's own recommendation engine, via the JSON the website player uses.
|
||||
Undocumented but public and stable for years; every call returns a fresh
|
||||
batch, so we poll until we have enough unseen tracks.
|
||||
"""
|
||||
url = f"https://www.last.fm/player/station/user/{urllib.parse.quote(user)}/{station}"
|
||||
out: list[dict] = []
|
||||
dry = 0
|
||||
for _ in range(25):
|
||||
if len(out) >= want or dry >= 3:
|
||||
break
|
||||
batch = http_json(url, timeout=30).get("playlist", [])
|
||||
if not batch:
|
||||
break
|
||||
added = 0
|
||||
for t in batch:
|
||||
title = (t.get("name") or t.get("_name") or "").strip()
|
||||
artists = [a.get("name") or a.get("_name") or "" for a in t.get("artists", [])]
|
||||
artists = [a.strip() for a in artists if a.strip()]
|
||||
if not title or not artists:
|
||||
continue
|
||||
key = track_key(artists[0], title)
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
out.append({"title": title, "artist": ", ".join(artists), "mainArtist": artists[0], "release": ""})
|
||||
added += 1
|
||||
dry = 0 if added else dry + 1
|
||||
time.sleep(1)
|
||||
return out[:want]
|
||||
|
||||
|
||||
def lastfm_api(method: str, **params) -> dict:
|
||||
params.update({"method": method, "api_key": LASTFM_API_KEY, "format": "json"})
|
||||
data = http_json("https://ws.audioscrobbler.com/2.0/", params=params, timeout=20)
|
||||
if "error" in data:
|
||||
raise HttpError(f"Last.fm API error {data['error']}: {data.get('message')}")
|
||||
time.sleep(0.25) # stay well under the 5 req/s limit
|
||||
return data
|
||||
|
||||
|
||||
def lastfm_similar(user: str, want: int, seen: set[str]) -> list[dict]:
|
||||
"""
|
||||
Fallback built only on the official API: artists similar to what you played
|
||||
in the last 3 months, minus every artist you already know, one top track each.
|
||||
"""
|
||||
known = set()
|
||||
for page in (1, 2):
|
||||
top = lastfm_api("user.gettopartists", user=user, period="overall", limit=1000, page=page)
|
||||
artists = top.get("topartists", {}).get("artist", [])
|
||||
known.update(a["name"].lower() for a in artists)
|
||||
if len(artists) < 1000:
|
||||
break
|
||||
recent = lastfm_api("user.gettopartists", user=user, period="3month", limit=30)
|
||||
scores: dict[str, float] = {}
|
||||
for seed in recent.get("topartists", {}).get("artist", []):
|
||||
try:
|
||||
sim = lastfm_api("artist.getsimilar", artist=seed["name"], limit=15, autocorrect=1)
|
||||
except HttpError:
|
||||
continue
|
||||
for a in sim.get("similarartists", {}).get("artist", []):
|
||||
if a["name"].lower() not in known:
|
||||
scores[a["name"]] = scores.get(a["name"], 0.0) + float(a.get("match") or 0)
|
||||
ranked = sorted(scores, key=scores.get, reverse=True)
|
||||
out: list[dict] = []
|
||||
for artist in ranked:
|
||||
if len(out) >= want:
|
||||
break
|
||||
try:
|
||||
top = lastfm_api("artist.gettoptracks", artist=artist, limit=5, autocorrect=1)
|
||||
except HttpError:
|
||||
continue
|
||||
candidates = [t["name"] for t in top.get("toptracks", {}).get("track", []) if t.get("name")]
|
||||
random.shuffle(candidates)
|
||||
for title in candidates:
|
||||
key = track_key(artist, title)
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
out.append({"title": title, "artist": artist, "mainArtist": artist, "release": ""})
|
||||
break
|
||||
return out
|
||||
|
||||
|
||||
def lastfm_fill_albums(tracks: list[dict]) -> None:
|
||||
"""Optional nicety when an API key is present: album names make Explo's matching stricter."""
|
||||
for t in tracks:
|
||||
try:
|
||||
info = lastfm_api("track.getinfo", artist=t["mainArtist"], track=t["title"], autocorrect=1)
|
||||
t["release"] = info.get("track", {}).get("album", {}).get("title", "") or ""
|
||||
except HttpError:
|
||||
pass
|
||||
|
||||
|
||||
def lastfm_recommendations(state: dict, want: int) -> list[dict]:
|
||||
seen = set(state["lastfm"]["history"])
|
||||
tracks: list[dict] = []
|
||||
for station in LASTFM_STATIONS:
|
||||
if len(tracks) >= want:
|
||||
break
|
||||
try:
|
||||
got = lastfm_station(LASTFM_USER, station.strip(), want - len(tracks), seen)
|
||||
log(f"Last.fm station '{station}': {len(got)} new tracks")
|
||||
tracks += got
|
||||
except HttpError as e:
|
||||
log(f"Last.fm station '{station}' failed: {e}")
|
||||
if len(tracks) < want and LASTFM_API_KEY:
|
||||
try:
|
||||
got = lastfm_similar(LASTFM_USER, want - len(tracks), seen)
|
||||
log(f"Last.fm similar-artists (API): {len(got)} new tracks")
|
||||
tracks += got
|
||||
except HttpError as e:
|
||||
log(f"Last.fm API fallback failed: {e}")
|
||||
if tracks and LASTFM_API_KEY:
|
||||
lastfm_fill_albums(tracks)
|
||||
return tracks
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Explo bridge
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def lb_playlist_name(d: dt.date | None = None) -> str:
|
||||
y, w = iso_week(d)
|
||||
return f"Weekly-Exploration-{y}-Week{w}" # exactly how Explo names it with --replace-playlist=false
|
||||
|
||||
|
||||
def lastfm_ids(d: dt.date | None = None) -> tuple[str, str]:
|
||||
y, w = iso_week(d)
|
||||
# digits only after the prefix: Explo title-cases the id to build the download folder name
|
||||
return f"custom-lastfm-{y}-{w}", f"Lastfm-Recommended-{y}-Week{w}"
|
||||
|
||||
|
||||
def register_custom_playlist(pid: str, name: str, tracks: list[dict]) -> None:
|
||||
"""Write the two files `explo --playlist=custom-*` reads (Explo v1.2 cache format)."""
|
||||
cache_dir = os.path.join(CONFIG_DIR, "cache")
|
||||
os.makedirs(cache_dir, exist_ok=True)
|
||||
with open(os.path.join(cache_dir, f"{pid}.json"), "w", encoding="utf-8") as f:
|
||||
json.dump({"tracks": tracks}, f, ensure_ascii=False)
|
||||
meta_path = os.path.join(CONFIG_DIR, "custom-playlists.json")
|
||||
try:
|
||||
with open(meta_path, encoding="utf-8") as f:
|
||||
meta = json.load(f)
|
||||
if not isinstance(meta, list):
|
||||
meta = []
|
||||
except (OSError, ValueError):
|
||||
meta = []
|
||||
meta = [m for m in meta if m.get("id") != pid]
|
||||
meta.append({"id": pid, "name": name, "source": "lastfm", "refresh_days": 0,
|
||||
"last_fetched": dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")})
|
||||
with open(meta_path, "w", encoding="utf-8") as f:
|
||||
json.dump(meta, f, indent=2, ensure_ascii=False)
|
||||
|
||||
|
||||
def unregister_custom_playlist(pid: str) -> None:
|
||||
try:
|
||||
os.remove(os.path.join(CONFIG_DIR, "cache", f"{pid}.json"))
|
||||
except OSError:
|
||||
pass
|
||||
meta_path = os.path.join(CONFIG_DIR, "custom-playlists.json")
|
||||
try:
|
||||
with open(meta_path, encoding="utf-8") as f:
|
||||
meta = [m for m in json.load(f) if m.get("id") != pid]
|
||||
with open(meta_path, "w", encoding="utf-8") as f:
|
||||
json.dump(meta, f, indent=2, ensure_ascii=False)
|
||||
except (OSError, ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def run_explo(*flags: str) -> int:
|
||||
cmd = [EXPLO_BIN, "--config", EXPLO_ENV, *flags]
|
||||
log("running: " + " ".join(cmd))
|
||||
try:
|
||||
return subprocess.run(cmd, cwd=os.path.dirname(EXPLO_BIN) or ".", timeout=EXPLO_TIMEOUT_MIN * 60).returncode
|
||||
except subprocess.TimeoutExpired:
|
||||
log(f"Explo exceeded {EXPLO_TIMEOUT_MIN} min and was stopped")
|
||||
return 124
|
||||
except OSError as e:
|
||||
log(f"could not start Explo: {e}")
|
||||
return 127
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Navidrome (Subsonic API)
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
class Navidrome:
|
||||
def __init__(self, url: str, user: str, password: str):
|
||||
self.url, self.user, self.password = url, user, password
|
||||
|
||||
def call(self, endpoint: str, **params) -> dict:
|
||||
salt = secrets.token_hex(8)
|
||||
params.update({"u": self.user, "s": salt, "t": hashlib.md5((self.password + salt).encode()).hexdigest(),
|
||||
"v": "1.16.1", "c": "ainulindale", "f": "json"})
|
||||
resp = http_json(f"{self.url}/rest/{endpoint}", params=params, timeout=30).get("subsonic-response", {})
|
||||
if resp.get("status") != "ok":
|
||||
raise HttpError(f"Navidrome {endpoint}: {resp.get('error', {}).get('message', 'failed')}")
|
||||
return resp
|
||||
|
||||
def playlists(self) -> list[dict]:
|
||||
return self.call("getPlaylists").get("playlists", {}).get("playlist", []) or []
|
||||
|
||||
def playlist_entries(self, pid: str) -> list[dict]:
|
||||
return self.call("getPlaylist", id=pid).get("playlist", {}).get("entry", []) or []
|
||||
|
||||
def starred(self) -> list[dict]:
|
||||
return self.call("getStarred2").get("starred2", {}).get("song", []) or []
|
||||
|
||||
def delete_playlist(self, pid: str) -> None:
|
||||
self.call("deletePlaylist", id=pid)
|
||||
|
||||
def scan(self) -> None:
|
||||
self.call("startScan")
|
||||
|
||||
|
||||
def navidrome() -> Navidrome | None:
|
||||
if NAVIDROME_URL and NAVIDROME_USER:
|
||||
return Navidrome(NAVIDROME_URL, NAVIDROME_USER, NAVIDROME_PASS)
|
||||
return None
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Retention
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
# (label, playlist-name regex, download-folder regex, how many to keep)
|
||||
def families() -> list[tuple[str, re.Pattern, re.Pattern, int]]:
|
||||
fams = [
|
||||
("ListenBrainz weekly exploration",
|
||||
re.compile(r"^Weekly-Exploration-(\d{4})-Week(\d{1,2})$"),
|
||||
re.compile(r"^Weekly-Exploration-(\d{4})-Week(\d{1,2})$"), KEEP_WEEKS),
|
||||
("Last.fm recommendations",
|
||||
re.compile(r"^Lastfm-Recommended-(\d{4})-Week(\d{1,2})$"),
|
||||
re.compile(r"^custom-lastfm-(\d{4})-(\d{1,2})$", re.I), KEEP_WEEKS),
|
||||
]
|
||||
if PRUNE_LEGACY_JAMS:
|
||||
jams = re.compile(r"^(?:Weekly|Daily)-Jams(?:-(\d{4})-(?:Week|Day)(\d{1,3}))?$")
|
||||
fams.append(("legacy jams", jams, jams, 0))
|
||||
return fams
|
||||
|
||||
|
||||
def split_keep(names: list[str], rx: re.Pattern, keep: int) -> tuple[list[str], list[str]]:
|
||||
def order(n: str):
|
||||
g = rx.match(n).groups()
|
||||
return int(g[0] or 0), int(g[1] or 0)
|
||||
# a set: the same name can exist twice in Navidrome (duplicates left by older setups)
|
||||
# and must only ever use up one of the "keep" slots
|
||||
ranked = sorted({n for n in names if rx.match(n)}, key=order, reverse=True)
|
||||
return ranked[:keep], ranked[keep:]
|
||||
|
||||
|
||||
def rescue_or_delete(folder: str, protected: set[tuple[int, str]], dry: bool) -> tuple[int, int]:
|
||||
"""Delete an expired download folder; files you starred or playlisted move to Keepers/ instead."""
|
||||
rescued = removed = 0
|
||||
for root, _dirs, files in os.walk(folder):
|
||||
for fn in files:
|
||||
path = os.path.join(root, fn)
|
||||
ext = os.path.splitext(fn)[1].lower()
|
||||
try:
|
||||
size = os.path.getsize(path)
|
||||
except OSError:
|
||||
continue
|
||||
if ext in AUDIO_EXT and (size, ext.lstrip(".")) in protected:
|
||||
rescued += 1
|
||||
if not dry:
|
||||
os.makedirs(KEEPERS_DIR, exist_ok=True)
|
||||
dest = os.path.join(KEEPERS_DIR, fn)
|
||||
if os.path.exists(dest):
|
||||
stem, e = os.path.splitext(fn)
|
||||
dest = os.path.join(KEEPERS_DIR, f"{stem}.{int(time.time())}{e}")
|
||||
shutil.move(path, dest)
|
||||
else:
|
||||
removed += 1
|
||||
if not dry:
|
||||
shutil.rmtree(folder, ignore_errors=True)
|
||||
if os.path.isdir(folder):
|
||||
log(f" could not fully remove {folder} (permissions? files from an older root-run setup?)")
|
||||
return rescued, removed
|
||||
|
||||
|
||||
def sweep_empty_slskd_dirs(dry: bool) -> None:
|
||||
"""Explo moves finished files out of slskd's folder; drop empty shells older than a day."""
|
||||
if not os.path.isdir(SLSKD_DIR) or os.path.isdir(os.path.join(SLSKD_DIR, "explo")):
|
||||
return # an explo/ folder in here means slskd downloads into the library root: hands off
|
||||
cutoff = time.time() - 86400
|
||||
for name in os.listdir(SLSKD_DIR):
|
||||
path = os.path.join(SLSKD_DIR, name)
|
||||
if name == "incomplete" or not os.path.isdir(path):
|
||||
continue
|
||||
try:
|
||||
if not os.listdir(path) and os.path.getmtime(path) < cutoff:
|
||||
log(f" removing empty slskd folder {name}")
|
||||
if not dry:
|
||||
os.rmdir(path)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def prune(dry: bool = False) -> None:
|
||||
tag = "[dry-run] " if dry else ""
|
||||
nd = navidrome()
|
||||
doomed_playlists: list[dict] = []
|
||||
all_playlists: list[dict] = []
|
||||
if nd:
|
||||
try:
|
||||
all_playlists = nd.playlists()
|
||||
except HttpError as e:
|
||||
log(f"prune: Navidrome unreachable ({e}); nothing pruned this time")
|
||||
return
|
||||
try:
|
||||
folders = [d for d in os.listdir(DATA_DIR) if os.path.isdir(os.path.join(DATA_DIR, d))]
|
||||
except OSError:
|
||||
folders = []
|
||||
|
||||
doomed_folders: list[str] = []
|
||||
for label, pl_rx, dir_rx, keep in families():
|
||||
kept, old = split_keep([p["name"] for p in all_playlists], pl_rx, keep)
|
||||
doomed_playlists += [p for p in all_playlists if p["name"] in old]
|
||||
for name in kept: # duplicates of a week that stays: keep the fullest copy only
|
||||
twins = sorted((p for p in all_playlists if p["name"] == name),
|
||||
key=lambda p: int(p.get("songCount") or 0), reverse=True)
|
||||
if len(twins) > 1:
|
||||
log(f"{tag}prune {label}: {len(twins) - 1} duplicate(s) of {name}")
|
||||
doomed_playlists += twins[1:]
|
||||
_, old_dirs = split_keep(folders, dir_rx, keep)
|
||||
doomed_folders += old_dirs
|
||||
if old or old_dirs:
|
||||
log(f"{tag}prune {label}: {len(old)} playlist(s), {len(old_dirs)} folder(s) beyond the newest {keep}")
|
||||
|
||||
if not doomed_playlists and not doomed_folders:
|
||||
sweep_empty_slskd_dirs(dry)
|
||||
return
|
||||
|
||||
# Anything starred, or sitting in a playlist that survives, must not be deleted.
|
||||
# Navidrome hides real paths by default, so files are matched by exact byte size + extension.
|
||||
protected: set[tuple[int, str]] = set()
|
||||
if nd and PRUNE_FILES and KEEP_FAVOURITES and doomed_folders:
|
||||
try:
|
||||
doomed_ids = {p["id"] for p in doomed_playlists}
|
||||
songs = nd.starred()
|
||||
for p in all_playlists:
|
||||
if p["id"] not in doomed_ids:
|
||||
songs += nd.playlist_entries(p["id"])
|
||||
protected = {(int(s["size"]), str(s.get("suffix", "")).lower()) for s in songs if s.get("size")}
|
||||
except HttpError as e:
|
||||
log(f"prune: could not read favourites ({e}); keeping files this time to be safe")
|
||||
doomed_folders = []
|
||||
|
||||
for p in doomed_playlists:
|
||||
log(f"{tag} delete playlist {p['name']}")
|
||||
if not dry:
|
||||
try:
|
||||
nd.delete_playlist(p["id"])
|
||||
except HttpError as e:
|
||||
log(f" failed: {e}")
|
||||
m = re.match(r"^Lastfm-Recommended-(\d{4})-Week(\d{1,2})$", p["name"])
|
||||
if m: # forget the matching entry in Explo's custom playlist registry
|
||||
unregister_custom_playlist(f"custom-lastfm-{m.group(1)}-{int(m.group(2))}")
|
||||
|
||||
if PRUNE_FILES:
|
||||
for d in doomed_folders:
|
||||
rescued, removed = rescue_or_delete(os.path.join(DATA_DIR, d), protected, dry)
|
||||
log(f"{tag} delete folder {d}: {removed} file(s) removed, {rescued} kept in Keepers/")
|
||||
if doomed_folders and nd and not dry:
|
||||
try:
|
||||
nd.scan()
|
||||
except HttpError:
|
||||
pass # needs an admin user; Navidrome's own scheduled scan will catch up
|
||||
|
||||
sweep_empty_slskd_dirs(dry)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# The tick
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def delete_playlist_named(name: str) -> None:
|
||||
nd = navidrome()
|
||||
if not nd:
|
||||
return
|
||||
try:
|
||||
for p in nd.playlists():
|
||||
if p["name"] == name:
|
||||
log(f"removing earlier partial playlist {name}")
|
||||
nd.delete_playlist(p["id"])
|
||||
except HttpError as e:
|
||||
log(f"could not check for an existing '{name}' playlist: {e}")
|
||||
|
||||
|
||||
def step_listenbrainz(state: dict) -> str:
|
||||
"""Returns: imported | current | waiting | down | failed | disabled"""
|
||||
if not (LB_ENABLED and LB_USER):
|
||||
return "disabled"
|
||||
lb = state["listenbrainz"]
|
||||
try:
|
||||
latest = lb_latest_playlist(LB_USER)
|
||||
except HttpError as e:
|
||||
log(f"ListenBrainz is not answering: {e}")
|
||||
return "down"
|
||||
if not latest or not latest["mbid"]:
|
||||
log("ListenBrainz has no Weekly Exploration playlist for this user yet")
|
||||
return "waiting"
|
||||
if latest["mbid"] == lb.get("imported_mbid"):
|
||||
log(f"ListenBrainz: newest playlist ({latest['date'][:10]}) is already imported")
|
||||
return "current" if lb_is_fresh(state) else "waiting"
|
||||
if lb.get("attempt_mbid") == latest["mbid"] and lb.get("attempts", 0) >= LB_MAX_ATTEMPTS:
|
||||
log(f"ListenBrainz: giving up on playlist {latest['mbid']} after {LB_MAX_ATTEMPTS} failed attempts")
|
||||
return "failed"
|
||||
|
||||
log(f"ListenBrainz: new playlist '{latest['title']}' ({latest['date'][:10]})")
|
||||
if not slskd_ready():
|
||||
return "failed"
|
||||
lb["attempts"] = lb.get("attempts", 0) + 1 if lb.get("attempt_mbid") == latest["mbid"] else 1
|
||||
lb["attempt_mbid"] = latest["mbid"]
|
||||
save_state(state)
|
||||
|
||||
delete_playlist_named(lb_playlist_name())
|
||||
rc = run_explo("--playlist=weekly-exploration", "--replace-playlist=false")
|
||||
if rc != 0:
|
||||
log(f"Explo exited with code {rc} on the ListenBrainz import (attempt {lb['attempts']}/{LB_MAX_ATTEMPTS})")
|
||||
return "failed"
|
||||
lb.update({"imported_mbid": latest["mbid"], "imported_date": latest["date"],
|
||||
"imported_at": dt.datetime.now().isoformat(timespec="seconds"), "attempts": 0})
|
||||
save_state(state)
|
||||
return "imported"
|
||||
|
||||
|
||||
def step_lastfm(state: dict, lb_status: str) -> str:
|
||||
if LASTFM_MODE == "off" or not LASTFM_USER:
|
||||
return "disabled"
|
||||
lf = state["lastfm"]
|
||||
if lf.get("week") == week_key():
|
||||
log("Last.fm: this week's playlist already exists")
|
||||
return "current"
|
||||
if LASTFM_MODE == "fallback":
|
||||
if lb_status in ("imported", "current"):
|
||||
return "not needed"
|
||||
if dt.date.today().isoweekday() < LASTFM_FALLBACK_DAY:
|
||||
log("Last.fm fallback: giving ListenBrainz a little longer before stepping in")
|
||||
return "waiting"
|
||||
log(f"Last.fm fallback: ListenBrainz is '{lb_status}', stepping in")
|
||||
|
||||
tracks = lastfm_recommendations(state, LASTFM_TRACKS)
|
||||
if len(tracks) < min(5, LASTFM_TRACKS):
|
||||
log(f"Last.fm gave only {len(tracks)} usable tracks; will retry on the next tick")
|
||||
return "failed"
|
||||
if not slskd_ready():
|
||||
return "failed"
|
||||
|
||||
pid, name = lastfm_ids()
|
||||
register_custom_playlist(pid, name, tracks)
|
||||
rc = run_explo(f"--playlist={pid}") # replace-playlist defaults to true: a re-run overwrites, never duplicates
|
||||
if rc != 0:
|
||||
log(f"Explo exited with code {rc} on the Last.fm import")
|
||||
return "failed"
|
||||
lf["week"] = week_key()
|
||||
lf["history"] = (lf["history"] + [track_key(t["mainArtist"], t["title"]) for t in tracks])[-LASTFM_HISTORY:]
|
||||
save_state(state)
|
||||
return "imported"
|
||||
|
||||
|
||||
def tick() -> int:
|
||||
state = load_state()
|
||||
lb_status = step_listenbrainz(state)
|
||||
lf_status = step_lastfm(state, lb_status)
|
||||
try:
|
||||
prune()
|
||||
except Exception as e: # pruning must never take the run down with it
|
||||
log(f"prune failed: {type(e).__name__}: {e}")
|
||||
state = load_state()
|
||||
state["last_tick"] = {"at": dt.datetime.now().isoformat(timespec="seconds"),
|
||||
"listenbrainz": lb_status, "lastfm": lf_status}
|
||||
save_state(state)
|
||||
log(f"done: listenbrainz={lb_status} lastfm={lf_status}")
|
||||
return 75 if "failed" in (lb_status, lf_status) else 0
|
||||
|
||||
|
||||
def status() -> int:
|
||||
state = load_state()
|
||||
print(json.dumps({k: v for k, v in state.items() if k != "lastfm"}, indent=2))
|
||||
lf = dict(state["lastfm"])
|
||||
lf["history"] = f"{len(lf.get('history', []))} remembered tracks"
|
||||
print(json.dumps({"lastfm": lf}, indent=2))
|
||||
if USES_SLSKD and SLSKD_URL and SLSKD_API_KEY:
|
||||
try:
|
||||
print("slskd:", Slskd(SLSKD_URL, SLSKD_API_KEY).server_state().get("state"))
|
||||
except HttpError as e:
|
||||
print("slskd: unreachable -", e)
|
||||
if LB_USER:
|
||||
try:
|
||||
print("listenbrainz newest:", lb_latest_playlist(LB_USER))
|
||||
except HttpError as e:
|
||||
print("listenbrainz: down -", e)
|
||||
return 0
|
||||
|
||||
|
||||
def main() -> int:
|
||||
cmd = sys.argv[1] if len(sys.argv) > 1 else "run"
|
||||
if cmd == "status":
|
||||
return status()
|
||||
if cmd == "lastfm-preview":
|
||||
if not LASTFM_USER:
|
||||
print("LASTFM_USER is not set")
|
||||
return 2
|
||||
for t in lastfm_recommendations(load_state(), LASTFM_TRACKS):
|
||||
print(f"{t['artist']} - {t['title']}" + (f" [{t['release']}]" if t["release"] else ""))
|
||||
return 0
|
||||
if cmd not in ("run", "prune"):
|
||||
print(__doc__)
|
||||
return 2
|
||||
|
||||
os.makedirs(CONFIG_DIR, exist_ok=True)
|
||||
with open(LOCK_FILE, "w") as lock:
|
||||
try:
|
||||
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
except OSError:
|
||||
log("another run is still in progress, skipping")
|
||||
return 0
|
||||
if cmd == "prune":
|
||||
prune(dry="--dry-run" in sys.argv)
|
||||
return 0
|
||||
return tick()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user