Quinta/Ainulindale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
Ainulindale/discovery/discover.py
T
Quinta d7ef929160 reworked it all
2026-09-18 18:13:05 +02:00

816 lines
32 KiB
Python
Executable File
Raw Permalink Blame History

#!/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 View Git Blame Copy Permalink