""" muse — request a song, get it in Navidrome. The gap this fills: Lidarr thinks in MusicBrainz *releases*, so the smallest thing it can want is an album. Soulseek trades individual files. Nothing in between existed, so requesting one song meant either hoovering a whole discography through Lidarr or hand-filing downloads out of slskd's UI. The trick that makes this simple: the user PICKS a specific track before anything is downloaded. That means we hold ground-truth metadata — artist, album, year, track number — before the file exists. Taggers like beets have to infer all of it afterwards from whatever a stranger named the file; we never guess. We tag what was asked for. Flow: /api/search Deezer track search -> the user picks one /api/request /track/{id} for exact metadata, then slskd search -> score candidates -> enqueue best worker poll slskd -> on completion: tag, then hardlink into /data/media/music// ()/NN - Title.ext Hardlink, not copy: the download stays in slskd's completed dir so it keeps being shared, while the library gets its own name for the same bytes. One file on disk, two names — the same trick the *arrs use. """ import asyncio import difflib import hashlib import logging import os import re import secrets import sqlite3 import uuid from contextlib import asynccontextmanager from datetime import datetime, timedelta, timezone from pathlib import Path import httpx from fastapi import FastAPI, HTTPException, Request from fastapi.responses import FileResponse, JSONResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel log = logging.getLogger("muse") logging.basicConfig( level=os.getenv("MUSE_LOG_LEVEL", "INFO"), format="[%(levelname)s|%(name)s] %(asctime)s: %(message)s", datefmt="%Y-%m-%dT%H:%M:%S%z", ) SLSKD_URL = os.getenv("SLSKD_URL", "http://192.168.1.87:5030").rstrip("/") SLSKD_API_KEY = os.getenv("SLSKD_API_KEY", "") DOWNLOAD_DIR = Path(os.getenv("MUSE_DOWNLOAD_DIR", "/data/torrents/soulseek/complete")) LIBRARY_DIR = Path(os.getenv("MUSE_LIBRARY_DIR", "/data/media/music")) DB_PATH = Path(os.getenv("MUSE_DB", "/config/muse.db")) # Ordered best-first. A candidate's extension must appear here or it is # rejected outright — this is what keeps .m4a rips and stray .cue files out. FORMATS = [f.strip() for f in os.getenv("MUSE_FORMATS", "flac,mp3").split(",") if f.strip()] # Below this, the remote filename doesn't plausibly contain the track title. # Soulseek filenames are chaotic ("08-radiohead-creep_(accoustic)-ent.mp3"), # so this compares against a normalised form, not the raw string. MIN_MATCH = float(os.getenv("MUSE_MIN_MATCH", "0.45")) # How many Soulseek searches may run at once. # # This is a hard cap, not a suggestion. Each search takes 10-40s of polling # and real load on slskd and the Soulseek server, so firing one task per # request — which is what an earlier version did — means pasting a 50-track # list launches 50 concurrent searches and drowns the client. Everything goes # through a bounded queue instead: requests return instantly, work drains at a # civil rate. WORKERS = int(os.getenv("MUSE_WORKERS", "3")) # Search uses Deezer's public API: no key, no OAuth, and vastly better # relevance than MusicBrainz for the way people actually type. # # MusicBrainz was tried first and rejected on measurements, not taste. It is a # superb database but a poor search engine: "radiohead creep" returned eight # covers and mashups before the Radiohead recording, and "bohemian rhapsody" # never surfaced Queen at all across 90 results — because its text index # matches "radiohead" inside a cover's *title* as readily as in the artist # field. Client-side re-ranking can only reorder what came back, so it could # not fix a result set that lacked the right row. Deezer returns # "Creep / Radiohead / Pablo Honey" first for the same query. # # Deezer search omits track number and release date, so those are fetched from # /track/{id} at request time — one extra call per request, not per result. DEEZER = "https://api.deezer.com" # Derivative versions to push down when the user did not ask for them. # Matched on WORD BOUNDARIES, not as substrings: a bare "live" is needed to # catch "Come As You Are (Live)", but substring matching would also penalise # "Deliver Me" and "Olive". Same trap for "demo" vs "Demolition". # # A term only counts against a result when the user did NOT type it, so asking # for "oasis live forever" or "nirvana unplugged demo" still works. DERIVATIVE = ("cover", "covers", "karaoke", "tribute", "remix", "remixed", "instrumental", "acapella", "originally performed", "made popular by", "in the style of", "demo", "rehearsal", "live", "unplugged", "re recorded", "rerecorded", "sped up", "slowed", "reverb", "8d", "nightcore") DERIVATIVE_RE = re.compile(r"\b(" + "|".join(DERIVATIVE) + r")\b") # ─────────────────────────── auth ─────────────────────────── # # Single-user GitHub OAuth, built so this can face the internet. # # FAILS CLOSED. Auth is on unless MUSE_AUTH is explicitly set to "off", and if # it is on with incomplete config the app refuses to start. The failure mode # worth engineering against is not a broken login page — it is a misconfigured # deploy quietly serving an unauthenticated download client to the internet. # You should have to *mean* it to run without auth. # # Sessions are server-side: the cookie carries a 256-bit random token and # nothing else, and the database stores only its SHA-256. There is no signed # payload to get wrong, a stolen database yields no usable cookies, and logout # and revocation actually revoke. AUTH_MODE = os.getenv("MUSE_AUTH", "github").strip().lower() GH_CLIENT_ID = os.getenv("MUSE_GITHUB_CLIENT_ID", "").strip() GH_CLIENT_SECRET = os.getenv("MUSE_GITHUB_CLIENT_SECRET", "").strip() ALLOWED_USERS = {u.strip().lower() for u in os.getenv("MUSE_ALLOWED_USERS", "").split(",") if u.strip()} # Absolute external URL, e.g. https://muse.aryan-singh.dev. Used to build the # OAuth callback. Taken from config rather than from X-Forwarded-* headers on # purpose: those are attacker-controllable and this is a redirect target. BASE_URL = os.getenv("MUSE_BASE_URL", "").rstrip("/") COOKIE_SECURE = os.getenv("MUSE_COOKIE_SECURE", "true").lower() != "false" SESSION_DAYS = int(os.getenv("MUSE_SESSION_DAYS", "7")) SESSION_COOKIE = "muse_session" STATE_COOKIE = "muse_oauth_state" GH_AUTH = "https://github.com/login/oauth/authorize" GH_TOKEN = "https://github.com/login/oauth/access_token" GH_API = "https://api.github.com/user" # Paths reachable without a session. Deliberately tiny: /healthz reports a bare # ok so the container healthcheck works without leaking configuration, and the # static assets must load on the login page itself. PUBLIC_PATHS = {"/healthz", "/auth/login", "/auth/callback", "/auth/logout"} def auth_preflight(): if AUTH_MODE == "off": log.warning("=" * 62) log.warning("MUSE_AUTH=off — every endpoint is UNAUTHENTICATED.") log.warning("Never expose this to the internet in this state.") log.warning("=" * 62) return if AUTH_MODE != "github": raise RuntimeError(f"MUSE_AUTH must be 'github' or 'off', got {AUTH_MODE!r}") missing = [n for n, v in ( ("MUSE_GITHUB_CLIENT_ID", GH_CLIENT_ID), ("MUSE_GITHUB_CLIENT_SECRET", GH_CLIENT_SECRET), ("MUSE_ALLOWED_USERS", ALLOWED_USERS), ("MUSE_BASE_URL", BASE_URL)) if not v] if missing: raise RuntimeError( "GitHub auth is enabled but " + ", ".join(missing) + " not set. " "Refusing to start: set them, or set MUSE_AUTH=off to run without " "authentication (LAN only).") if not BASE_URL.startswith("https://") and COOKIE_SECURE: log.warning("MUSE_BASE_URL is not https but cookies are Secure — the " "browser will refuse to send them. Set MUSE_COOKIE_SECURE=false " "only if you are testing over plain http on a LAN.") def _hash(tok: str) -> str: return hashlib.sha256(tok.encode()).hexdigest() def new_session(username: str) -> str: token = secrets.token_urlsafe(32) exp = datetime.now(timezone.utc) + timedelta(days=SESSION_DAYS) with db() as c: c.execute("INSERT INTO sessions (token_hash,username,expires_at) VALUES (?,?,?)", (_hash(token), username, exp.isoformat())) # Opportunistic sweep; no scheduler needed for a handful of rows. c.execute("DELETE FROM sessions WHERE expires_at < ?", (datetime.now(timezone.utc).isoformat(),)) return token def session_user(request: Request) -> str | None: tok = request.cookies.get(SESSION_COOKIE) if not tok: return None with db() as c: row = c.execute( "SELECT username, expires_at FROM sessions WHERE token_hash=?", (_hash(tok),)).fetchone() if not row: return None try: if datetime.fromisoformat(row["expires_at"]) < datetime.now(timezone.utc): return None except ValueError: return None # Re-check the allowlist on every request, so removing someone from # MUSE_ALLOWED_USERS locks them out immediately instead of whenever their # session happens to expire. return row["username"] if row["username"].lower() in ALLOWED_USERS else None SAFE = re.compile(r'[<>:"/\\|?*\x00-\x1f]') NOISE = re.compile(r"[^a-z0-9]+") def safe(name: str) -> str: """Filesystem-safe path component. Trailing dots/spaces break on some FS.""" return SAFE.sub("_", name).strip(". ") or "Unknown" def norm(s: str) -> str: return NOISE.sub(" ", s.lower()).strip() # ─────────────────────────── storage ─────────────────────────── def db(): c = sqlite3.connect(DB_PATH, timeout=30) c.row_factory = sqlite3.Row return c def init_db(): DB_PATH.parent.mkdir(parents=True, exist_ok=True) with db() as c: c.execute(""" CREATE TABLE IF NOT EXISTS jobs ( id TEXT PRIMARY KEY, state TEXT NOT NULL, -- searching|queued|downloading|imported|failed detail TEXT, artist TEXT, title TEXT, album TEXT, year TEXT, track INTEGER, recording_mbid TEXT, release_mbid TEXT, username TEXT, remote_file TEXT, size INTEGER, imported_path TEXT, created_at TEXT DEFAULT (datetime('now')) ) """) c.execute(""" CREATE TABLE IF NOT EXISTS sessions ( token_hash TEXT PRIMARY KEY, -- sha256 of the cookie value username TEXT NOT NULL, created_at TEXT DEFAULT (datetime('now')), expires_at TEXT NOT NULL ) """) def set_state(job_id: str, state: str, detail: str = "", **cols): sets = ", ".join(f"{k}=?" for k in cols) sql = f"UPDATE jobs SET state=?, detail=?{', ' + sets if sets else ''} WHERE id=?" with db() as c: c.execute(sql, [state, detail, *cols.values(), job_id]) log.info("job %s -> %s %s", job_id[:8], state, detail) # ─────────────────────────── slskd ─────────────────────────── def slskd_client() -> httpx.AsyncClient: return httpx.AsyncClient( base_url=SLSKD_URL, headers={"X-API-Key": SLSKD_API_KEY}, timeout=httpx.Timeout(30.0), ) def ext_of(remote_path: str) -> str: # Soulseek paths are Windows-style; the API's `extension` field comes back # empty in practice, so derive it ourselves. return remote_path.rsplit(".", 1)[-1].lower() if "." in remote_path else "" def score(candidate: dict, peer: dict, want: str) -> float | None: """Rank one remote file. None = reject. Ordering intent: correct format first, then actually-gettable (a peer with a free slot and a short queue beats a marginally better file behind 200 people), then filename plausibility, then bitrate. """ ext = ext_of(candidate["filename"]) if ext not in FORMATS: return None base = candidate["filename"].replace("\\", "/").rsplit("/", 1)[-1] ratio = difflib.SequenceMatcher(None, norm(base), want).ratio() # A short title inside a long scene filename scores badly on raw ratio, so # give credit for simple containment too. if want in norm(base): ratio = max(ratio, 0.9) if ratio < MIN_MATCH: return None s = 0.0 s += (len(FORMATS) - FORMATS.index(ext)) * 100 # format preference s += 60 if peer.get("hasFreeUploadSlot") else 0 s += max(0, 40 - peer.get("queueLength", 0)) # short queues s += min(peer.get("uploadSpeed", 0) / 50_000, 20) # capped speed bonus s += ratio * 50 s += min(candidate.get("bitRate") or 0, 320) / 32 # 0-10 return s async def pick_and_enqueue(job_id: str, artist: str, title: str): """Search slskd for one track and queue the best candidate.""" want = norm(f"{artist} {title}") search_id = str(uuid.uuid4()) async with slskd_client() as sc: try: r = await sc.post("/api/v0/searches", json={"id": search_id, "searchText": f"{artist} {title}"}) r.raise_for_status() except Exception as e: set_state(job_id, "failed", f"slskd search failed: {e}") return # Poll until slskd calls it done, or we run out of patience. Soulseek # searches trickle in; results at t=5s are much worse than at t=15s. results = {} for _ in range(20): await asyncio.sleep(2) try: rr = await sc.get(f"/api/v0/searches/{search_id}", params={"includeResponses": "true"}) rr.raise_for_status() results = rr.json() except Exception: continue if str(results.get("state", "")).startswith("Completed"): break best, best_score = None, -1.0 for peer in results.get("responses", []): for f in peer.get("files", []): if f.get("isLocked"): continue sc_ = score(f, peer, want) if sc_ is not None and sc_ > best_score: best, best_score = (peer, f), sc_ # Tidy up so slskd's search history doesn't grow without bound. try: await sc.delete(f"/api/v0/searches/{search_id}") except Exception: pass if not best: set_state(job_id, "failed", f"no {'/'.join(FORMATS)} candidate matched (searched {results.get('fileCount', 0)} files)") return peer, f = best try: er = await sc.post(f"/api/v0/transfers/downloads/{peer['username']}", json=[{"filename": f["filename"], "size": f["size"]}]) er.raise_for_status() except Exception as e: set_state(job_id, "failed", f"enqueue failed: {e}") return set_state(job_id, "downloading", f"from {peer['username']} (score {best_score:.0f})", username=peer["username"], remote_file=f["filename"], size=f["size"]) # ─────────────────────────── import ─────────────────────────── def tag_and_file(job: sqlite3.Row, src: Path) -> Path: """Write known-good tags, then hardlink into the library.""" from mutagen import File as MFile from mutagen.easyid3 import EasyID3 from mutagen.flac import FLAC from mutagen.mp3 import MP3 try: if src.suffix.lower() == ".flac": audio = FLAC(src) else: try: audio = EasyID3(src) except Exception: m = MP3(src) m.add_tags() m.save() audio = EasyID3(src) audio["title"] = job["title"] audio["artist"] = job["artist"] audio["albumartist"] = job["artist"] if job["album"]: audio["album"] = job["album"] if job["year"]: audio["date"] = str(job["year"]) if job["track"]: audio["tracknumber"] = str(job["track"]) # Deliberately NOT writing the Deezer id into musicbrainz_trackid -- # that field means a specific thing and a wrong value there would # mislead Picard, beets, or anything else that later reads this file. audio.save() except Exception as e: # A tagging failure must not strand the file — Navidrome can still # read whatever tags it shipped with. log.warning("tagging failed for %s: %s", src.name, e) album = f"{safe(job['album'] or 'Singles')}" if job["year"]: album += f" ({job['year']})" dest_dir = LIBRARY_DIR / safe(job["artist"]) / album dest_dir.mkdir(parents=True, exist_ok=True) num = f"{int(job['track']):02d} - " if job["track"] else "" dest = dest_dir / f"{num}{safe(job['title'])}{src.suffix.lower()}" if dest.exists(): return dest try: os.link(src, dest) # same filesystem -> free, and keeps sharing except OSError: import shutil shutil.copy2(src, dest) # cross-device fallback return dest def find_downloaded(remote_file: str) -> Path | None: """Locate the completed file. slskd nests it under the remote's parent directory name, so match on basename rather than guessing the path.""" base = remote_file.replace("\\", "/").rsplit("/", 1)[-1] for p in DOWNLOAD_DIR.rglob("*"): if p.is_file() and p.name == base: return p return None # ─────────────────────────── search queue ─────────────────────────── work_q: "asyncio.Queue[str]" = asyncio.Queue() async def search_worker(n: int): """Drain the search queue, one job at a time, WORKERS of these running.""" while True: job_id = await work_q.get() try: with db() as c: job = c.execute("SELECT * FROM jobs WHERE id=?", (job_id,)).fetchone() if job and job["state"] in ("queued", "searching"): set_state(job_id, "searching", "") await pick_and_enqueue(job_id, job["artist"], job["title"]) except Exception as e: log.warning("worker %d: %s", n, e) try: set_state(job_id, "failed", str(e)[:200]) except Exception: pass finally: work_q.task_done() async def import_worker(): """Poll slskd for in-flight downloads and import the finished ones.""" while True: await asyncio.sleep(10) try: with db() as c: jobs = c.execute( "SELECT * FROM jobs WHERE state='downloading'").fetchall() if not jobs: continue async with slskd_client() as sc: r = await sc.get("/api/v0/transfers/downloads") r.raise_for_status() payload = r.json() # Flatten slskd's user -> directories -> files nesting. states: dict[str, str] = {} for user in payload: for d in user.get("directories", []): for f in d.get("files", []): states[f.get("filename", "")] = str(f.get("state", "")) for job in jobs: st = states.get(job["remote_file"], "") if "Completed, Succeeded" in st: src = find_downloaded(job["remote_file"]) if not src: continue # still being moved out of incomplete dest = tag_and_file(job, src) set_state(job["id"], "imported", str(dest), imported_path=str(dest)) elif "Completed" in st and "Succeeded" not in st: set_state(job["id"], "failed", f"transfer {st}") except Exception as e: log.warning("worker: %s", e) # ─────────────────────────── api ─────────────────────────── @asynccontextmanager async def lifespan(_: FastAPI): auth_preflight() # raises and stops the container if misconfigured init_db() # The queue lives in memory, so a restart mid-import would otherwise # strand every unstarted job in 'queued' forever with nothing to pick it # up. Anything not yet handed to slskd goes back on the queue. with db() as c: orphans = [r["id"] for r in c.execute( "SELECT id FROM jobs WHERE state IN ('queued','searching')").fetchall()] for jid in orphans: work_q.put_nowait(jid) if orphans: log.info("requeued %d job(s) left over from the last run", len(orphans)) tasks = [asyncio.create_task(import_worker())] tasks += [asyncio.create_task(search_worker(i)) for i in range(WORKERS)] log.info("muse up — slskd=%s library=%s formats=%s workers=%d", SLSKD_URL, LIBRARY_DIR, ",".join(FORMATS), WORKERS) yield for t in tasks: t.cancel() app = FastAPI( title="muse", lifespan=lifespan, # No interactive docs on an internet-facing app: they enumerate every # endpoint and schema for anyone who reaches the login page. docs_url=None, redoc_url=None, openapi_url=None, ) @app.middleware("http") async def guard(request: Request, call_next): """Require a session for everything except PUBLIC_PATHS, and stamp security headers on every response.""" path = request.url.path if AUTH_MODE == "off" or path in PUBLIC_PATHS or path.startswith("/static/"): resp = await call_next(request) elif session_user(request): resp = await call_next(request) elif path.startswith("/api/"): # An API caller gets a status code, not an HTML redirect it can't use. resp = JSONResponse({"error": "unauthenticated"}, status_code=401) else: resp = RedirectResponse("/auth/login", status_code=302) # Strict CSP is why the stylesheet and script live in separate files rather # than inline in index.html: with 'self' only and no 'unsafe-inline', an # injected