Aryan SinghandClaude Opus 5
6ef580780d
/api/admin/system probed slskd with the shared client, whose 30s timeout is sized for searches and transfers. So the panel sat on skeletons for the full 30s exactly when slskd was unreachable -- the case you open it to diagnose -- and the 5s UI poll stacked requests behind it. Probes now get 4s: above a LAN round-trip, under the poll interval. httpx.ConnectTimeout also stringifies to "", so the banner rendered "slskd unreachable -- " with nothing after the dash. Fall back to the exception class name. Add dev.sh, because the documented dev commands could not work: MUSE_DB defaults to /config/muse.db which a Mac cannot write, and the auth preflight refuses to boot without a client ID. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1829 lines
77 KiB
Python
1829 lines
77 KiB
Python
"""
|
||
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/<Artist>/<Album> (<Year>)/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 base64
|
||
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
|
||
from urllib.parse import urlencode
|
||
|
||
import httpx
|
||
from fastapi import FastAPI, HTTPException, Request
|
||
from fastapi.responses import (FileResponse, HTMLResponse, 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 ───────────────────────────
|
||
#
|
||
# OAuth against GitHub and/or Google, 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.
|
||
#
|
||
# WHO MAY LOG IN is two rules, checked in this order:
|
||
# 1. MUSE_ALLOWED_USERS — the bootstrap allowlist. Always admitted. This is
|
||
# how the very first admin gets in when the users table is empty, and it
|
||
# is the recovery path if someone disables themselves.
|
||
# 2. A row in `users` that is not disabled — how everyone invited later gets
|
||
# in, without an env change or a redeploy.
|
||
# Anyone matching neither is rejected and the attempt is logged.
|
||
AUTH_MODE = os.getenv("MUSE_AUTH", "github").strip().lower()
|
||
ENABLED_PROVIDERS = [p for p in (x.strip() for x in AUTH_MODE.split(","))
|
||
if p and p != "off"]
|
||
|
||
# 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("/")
|
||
ALLOWED_USERS = {u.strip().lower()
|
||
for u in os.getenv("MUSE_ALLOWED_USERS", "").split(",") if u.strip()}
|
||
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"
|
||
|
||
# One callback URL for every provider — the provider is carried in the state
|
||
# COOKIE, not the query string, so adding Google needs no change to the GitHub
|
||
# OAuth app that is already registered against /auth/callback.
|
||
PROVIDERS: dict[str, dict] = {
|
||
"github": {
|
||
"label": "GitHub",
|
||
"authorize": "https://github.com/login/oauth/authorize",
|
||
"token": "https://github.com/login/oauth/access_token",
|
||
"scope": "read:user",
|
||
"client_id": os.getenv("MUSE_GITHUB_CLIENT_ID", "").strip(),
|
||
"client_secret": os.getenv("MUSE_GITHUB_CLIENT_SECRET", "").strip(),
|
||
"id_env": "MUSE_GITHUB_CLIENT_ID",
|
||
"secret_env": "MUSE_GITHUB_CLIENT_SECRET",
|
||
},
|
||
"google": {
|
||
"label": "Google",
|
||
"authorize": "https://accounts.google.com/o/oauth2/v2/auth",
|
||
"token": "https://oauth2.googleapis.com/token",
|
||
"scope": "openid email profile",
|
||
"client_id": os.getenv("MUSE_GOOGLE_CLIENT_ID", "").strip(),
|
||
"client_secret": os.getenv("MUSE_GOOGLE_CLIENT_SECRET", "").strip(),
|
||
"id_env": "MUSE_GOOGLE_CLIENT_ID",
|
||
"secret_env": "MUSE_GOOGLE_CLIENT_SECRET",
|
||
},
|
||
}
|
||
|
||
GH_API = "https://api.github.com/user"
|
||
GOOGLE_API = "https://openidconnect.googleapis.com/v1/userinfo"
|
||
|
||
# Paths reachable without a session. Deliberately tiny: /healthz reports a bare
|
||
# ok so the container healthcheck works without leaking configuration.
|
||
PUBLIC_PATHS = {"/healthz", "/auth/login", "/auth/callback", "/auth/logout"}
|
||
|
||
ROLES = ("admin", "user", "guest")
|
||
|
||
|
||
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 not ENABLED_PROVIDERS:
|
||
raise RuntimeError(f"MUSE_AUTH must list providers or be 'off', got {AUTH_MODE!r}")
|
||
|
||
unknown = [p for p in ENABLED_PROVIDERS if p not in PROVIDERS]
|
||
if unknown:
|
||
raise RuntimeError(
|
||
f"MUSE_AUTH names unknown provider(s) {unknown}. "
|
||
f"Valid: {', '.join(PROVIDERS)}, or 'off'.")
|
||
|
||
missing: list[str] = []
|
||
for p in ENABLED_PROVIDERS:
|
||
cfg = PROVIDERS[p]
|
||
if not cfg["client_id"]:
|
||
missing.append(cfg["id_env"])
|
||
if not cfg["client_secret"]:
|
||
missing.append(cfg["secret_env"])
|
||
if not BASE_URL:
|
||
missing.append("MUSE_BASE_URL")
|
||
# The bootstrap allowlist is only mandatory while nobody exists yet —
|
||
# otherwise a working multi-user install would be held hostage by an env
|
||
# var it has outgrown. But an EMPTY database and no allowlist means nobody
|
||
# can ever log in, which is a misconfiguration worth refusing to start on.
|
||
if not ALLOWED_USERS and not _any_users():
|
||
missing.append("MUSE_ALLOWED_USERS")
|
||
if missing:
|
||
raise RuntimeError(
|
||
"Auth is enabled but " + ", ".join(sorted(set(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.")
|
||
log.info("auth: providers=%s bootstrap_allowlist=%d",
|
||
",".join(ENABLED_PROVIDERS), len(ALLOWED_USERS))
|
||
|
||
|
||
def _any_users() -> bool:
|
||
try:
|
||
with db() as c:
|
||
return c.execute("SELECT 1 FROM users LIMIT 1").fetchone() is not None
|
||
except sqlite3.Error:
|
||
return False
|
||
|
||
|
||
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 may_login(username: str, row: sqlite3.Row | None) -> bool:
|
||
"""The single authorization rule, used at login AND on every request.
|
||
|
||
Bootstrap allowlist first, then a live, non-disabled user row. Deliberately
|
||
one function: an authorization check that runs only at login is not a
|
||
revocation mechanism, it is a delay — disabling someone would leave their
|
||
session valid for up to MUSE_SESSION_DAYS.
|
||
"""
|
||
if username.lower() in ALLOWED_USERS:
|
||
return True
|
||
return bool(row) and not row["disabled"]
|
||
|
||
|
||
def current_user(request: Request) -> sqlite3.Row | None:
|
||
"""The signed-in user's row, or None. Re-validated on every request."""
|
||
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
|
||
u = c.execute("SELECT * FROM users WHERE username=?",
|
||
(row["username"],)).fetchone()
|
||
if not may_login(row["username"], u):
|
||
return None
|
||
if u is None:
|
||
# Allowlisted but no row — an admin deleted the bootstrap account while
|
||
# its session was still live. The allowlist is exactly the recovery
|
||
# path for that, so put the row back rather than half-authenticating
|
||
# them into a state where every role check fails.
|
||
u = upsert_user(row["username"])
|
||
log.info("auth: restored deleted allowlisted user %s", row["username"])
|
||
return u
|
||
|
||
|
||
def session_user(request: Request) -> str | None:
|
||
u = current_user(request)
|
||
return u["username"] if u else None
|
||
|
||
|
||
def user_role(request: Request) -> str:
|
||
u = current_user(request)
|
||
if u:
|
||
return u["role"]
|
||
# With auth off there is nobody to be, so everyone is an admin — otherwise
|
||
# the admin panel would be unreachable on a LAN-only install.
|
||
return "admin" if AUTH_MODE == "off" else "guest"
|
||
|
||
|
||
def require_admin(request: Request) -> str:
|
||
if AUTH_MODE == "off":
|
||
return "local"
|
||
u = current_user(request)
|
||
if not u:
|
||
raise HTTPException(401, "unauthenticated")
|
||
if u["role"] != "admin":
|
||
log.warning("admin: DENIED %s (role=%s) -> %s",
|
||
u["username"], u["role"], request.url.path)
|
||
raise HTTPException(403, "admin only")
|
||
return u["username"]
|
||
|
||
|
||
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
|
||
)
|
||
""")
|
||
# Users exist as rows even while the allowlist is a single name in an
|
||
# env var. Roles and quotas are read from here from the start, so
|
||
# opening this up to other people later is an INSERT and a config
|
||
# change rather than a schema migration against live job history.
|
||
c.execute("""
|
||
CREATE TABLE IF NOT EXISTS users (
|
||
username TEXT PRIMARY KEY COLLATE NOCASE,
|
||
provider TEXT NOT NULL DEFAULT 'github',
|
||
role TEXT NOT NULL DEFAULT 'user', -- admin|user|guest
|
||
display TEXT,
|
||
avatar TEXT,
|
||
quota INTEGER NOT NULL DEFAULT 0, -- requests/month, 0 = unlimited
|
||
disabled INTEGER NOT NULL DEFAULT 0,
|
||
created_at TEXT DEFAULT (datetime('now')),
|
||
last_seen TEXT
|
||
)
|
||
""")
|
||
# Added after the first release, so existing databases need it grafted
|
||
# on rather than recreated -- job history is the whole point of the file.
|
||
cols = {r["name"] for r in c.execute("PRAGMA table_info(jobs)").fetchall()}
|
||
for col, decl in (("requested_by", "TEXT"),
|
||
("cover", "TEXT"),
|
||
("genre", "TEXT"),
|
||
("duration", "INTEGER")):
|
||
if col not in cols:
|
||
c.execute(f"ALTER TABLE jobs ADD COLUMN {col} {decl}")
|
||
ucols = {r["name"] for r in c.execute("PRAGMA table_info(users)").fetchall()}
|
||
for col, decl in (("storage_quota", "INTEGER NOT NULL DEFAULT 0"),
|
||
("invited_by", "TEXT")):
|
||
if col not in ucols:
|
||
c.execute(f"ALTER TABLE users ADD COLUMN {col} {decl}")
|
||
c.execute("CREATE INDEX IF NOT EXISTS jobs_by_user ON jobs (requested_by)")
|
||
c.execute("CREATE INDEX IF NOT EXISTS jobs_by_state ON jobs (state)")
|
||
c.execute("CREATE INDEX IF NOT EXISTS jobs_by_created ON jobs (created_at)")
|
||
|
||
|
||
def upsert_user(username: str, avatar: str = "", display: str = "",
|
||
provider: str = "github") -> sqlite3.Row:
|
||
"""Record the sign-in.
|
||
|
||
The first user to ever log in becomes admin — no bootstrap ceremony, no
|
||
default password to forget. Everyone after that is a plain user until an
|
||
admin says otherwise; role is never touched on subsequent logins, or every
|
||
sign-in would silently undo a demotion.
|
||
"""
|
||
with db() as c:
|
||
first = c.execute("SELECT COUNT(*) n FROM users").fetchone()["n"] == 0
|
||
c.execute(
|
||
"INSERT INTO users (username,provider,role,display,avatar,last_seen) "
|
||
"VALUES (?,?,?,?,?,datetime('now')) "
|
||
"ON CONFLICT(username) DO UPDATE SET "
|
||
" last_seen=datetime('now'), provider=excluded.provider, "
|
||
" avatar=COALESCE(NULLIF(excluded.avatar,''), users.avatar), "
|
||
" display=COALESCE(NULLIF(excluded.display,''), users.display)",
|
||
(username, provider, "admin" if first else "user", display, avatar))
|
||
return c.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
|
||
|
||
|
||
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(timeout: float = 30.0) -> httpx.AsyncClient:
|
||
"""30s suits searches and transfer polling, which are legitimately slow.
|
||
|
||
Status probes pass something much shorter: they run on a UI poll, and a
|
||
probe that outlasts its own poll interval stacks requests and leaves the
|
||
admin panel on skeletons for the whole timeout — precisely when slskd is
|
||
down and you opened the panel to find out why.
|
||
"""
|
||
return httpx.AsyncClient(
|
||
base_url=SLSKD_URL,
|
||
headers={"X-API-Key": SLSKD_API_KEY},
|
||
timeout=httpx.Timeout(timeout),
|
||
)
|
||
|
||
|
||
# slskd is on the LAN, so a reachable instance answers in milliseconds. Long
|
||
# enough to absorb a slow moment, short enough to stay under the 5s UI poll.
|
||
PROBE_TIMEOUT = 4.0
|
||
|
||
|
||
def why(e: Exception) -> str:
|
||
"""httpx's timeout and connection errors often str() to an empty string,
|
||
which would render as 'slskd unreachable — ' with nothing after it."""
|
||
return (str(e) or type(e).__name__)[:120]
|
||
|
||
|
||
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)
|
||
# Record the real size, not the size the peer advertised —
|
||
# storage quotas are computed from this column.
|
||
try:
|
||
actual = src.stat().st_size
|
||
except OSError:
|
||
actual = job["size"]
|
||
set_state(job["id"], "imported", str(dest),
|
||
imported_path=str(dest), size=actual)
|
||
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):
|
||
# init_db first: the preflight consults the users table to decide whether
|
||
# the bootstrap allowlist is still mandatory.
|
||
init_db()
|
||
auth_preflight() # raises and stops the container if misconfigured
|
||
|
||
# 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()
|
||
|
||
|
||
# ─────────────────────── static front end / CSP ───────────────────────
|
||
#
|
||
# The UI is a Next.js static export: `next build` writes plain files into
|
||
# web/out and this process serves them, so the deployment stays one container,
|
||
# one process, one healthcheck, with no Node runtime in the image.
|
||
#
|
||
# That export contains inline <script> blocks (React's hydration payload),
|
||
# which `script-src 'self'` blocks outright. The usual fix is a per-request
|
||
# nonce, but a STATIC export has a better option: the scripts never change
|
||
# between requests, so their SHA-256 can go straight into the policy. Hashing
|
||
# them at startup means the CSP needs no per-request work, no HTML rewriting,
|
||
# and no maintenance when the bundle changes -- rebuild and the hashes follow.
|
||
WEB = Path(__file__).parent / "web" / "out"
|
||
|
||
_INLINE_SCRIPT = re.compile(rb"<script(?![^>]*\bsrc=)[^>]*>(.*?)</script>", re.S)
|
||
|
||
|
||
def script_hashes() -> list[str]:
|
||
out: set[str] = set()
|
||
for page in ("index.html", "404.html"):
|
||
f = WEB / page
|
||
if not f.exists():
|
||
continue
|
||
for body in _INLINE_SCRIPT.findall(f.read_bytes()):
|
||
digest = base64.b64encode(hashlib.sha256(body).digest()).decode()
|
||
out.add(f"'sha256-{digest}'")
|
||
return sorted(out)
|
||
|
||
|
||
SCRIPT_HASHES = " ".join(script_hashes())
|
||
|
||
# Framer Motion writes its pre-hydration `initial` state into style="" on the
|
||
# server-rendered markup, so those attributes have to be allowed or every
|
||
# animated element flashes in at full opacity before hydration corrects it.
|
||
#
|
||
# style-src-attr is the narrow lever for that: it permits style ATTRIBUTES
|
||
# while style-src 'self' still blocks injected <style> blocks and remote
|
||
# stylesheets. It is not a free pass either -- CSS-based exfiltration goes
|
||
# through url(), and img-src below already limits where those can point.
|
||
CSP = (
|
||
"default-src 'self'; "
|
||
f"script-src 'self' {SCRIPT_HASHES}; "
|
||
"style-src 'self'; style-src-attr 'unsafe-inline'; "
|
||
"img-src 'self' data: https://*.dzcdn.net https://*.githubusercontent.com; "
|
||
"media-src 'self' https://*.dzcdn.net; "
|
||
"connect-src 'self'; font-src 'self'; object-src 'none'; "
|
||
"frame-ancestors 'none'; base-uri 'none'; form-action 'self'"
|
||
)
|
||
|
||
|
||
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."""
|
||
# /_next/* is NOT public. Unlike the old hand-written page there is no
|
||
# local login screen to style — an unauthenticated visitor is redirected
|
||
# straight to GitHub and never receives our HTML — so the bundle can sit
|
||
# behind the session too, and not hand out a map of the app to anyone who
|
||
# finds the URL. It loads fine after the callback, cookie in hand.
|
||
path = request.url.path
|
||
if AUTH_MODE == "off" or path in PUBLIC_PATHS:
|
||
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)
|
||
|
||
# See CSP above. img-src and media-src name Deezer's CDN explicitly rather
|
||
# than allowing https: wholesale — under a blanket https: any injected
|
||
# markup could beacon to an arbitrary host by requesting an image from it,
|
||
# which is a working exfiltration channel even with connect-src locked to
|
||
# 'self'. Artwork comes from dzcdn.net and avatars from githubusercontent.
|
||
resp.headers["Content-Security-Policy"] = CSP
|
||
resp.headers["X-Content-Type-Options"] = "nosniff"
|
||
resp.headers["X-Frame-Options"] = "DENY"
|
||
resp.headers["Referrer-Policy"] = "no-referrer"
|
||
resp.headers["Permissions-Policy"] = "geolocation=(), microphone=(), camera=()"
|
||
if COOKIE_SECURE:
|
||
resp.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
|
||
return resp
|
||
|
||
|
||
def _set_cookie(resp, name, value, max_age):
|
||
resp.set_cookie(name, value, max_age=max_age, httponly=True,
|
||
secure=COOKIE_SECURE, samesite="lax", path="/")
|
||
|
||
|
||
@app.get("/healthz")
|
||
async def healthz():
|
||
"""Unauthenticated on purpose, for the container healthcheck. Returns
|
||
nothing but a literal ok — no versions, no config, no counts."""
|
||
return {"ok": True}
|
||
|
||
|
||
CHOOSER = """<!doctype html><html lang=en><head><meta charset=utf-8>
|
||
<meta name=viewport content="width=device-width,initial-scale=1"><title>muse</title>
|
||
</head><body style="margin:0;min-height:100vh;display:grid;place-items:center;
|
||
background:#08090d;color:#f2f4f9;font:15px/1.5 -apple-system,BlinkMacSystemFont,
|
||
ui-sans-serif,system-ui,sans-serif">
|
||
<div style="text-align:center;padding:32px">
|
||
<div style="font-size:26px;font-weight:700;letter-spacing:-.5px">muse</div>
|
||
<div style="color:#9aa3b6;font-size:13px;margin:6px 0 26px">Sign in to continue</div>
|
||
{buttons}</div></body></html>"""
|
||
|
||
_BTN = ('<a href="/auth/login?provider={p}" style="display:block;margin:10px auto;'
|
||
'padding:12px 26px;min-width:230px;border-radius:12px;text-decoration:none;'
|
||
'background:rgb(255 255 255/.06);border:1px solid rgb(255 255 255/.14);'
|
||
'color:#f2f4f9;font-weight:600">Continue with {label}</a>')
|
||
|
||
|
||
@app.get("/auth/login")
|
||
async def auth_login(provider: str = ""):
|
||
"""Start OAuth. With several providers enabled and none named, offer a
|
||
choice — a tiny self-contained page that uses only style attributes, which
|
||
the CSP already permits, so it needs no stylesheet and no extra hash."""
|
||
if AUTH_MODE == "off":
|
||
return RedirectResponse("/", status_code=302)
|
||
|
||
if not provider:
|
||
if len(ENABLED_PROVIDERS) == 1:
|
||
provider = ENABLED_PROVIDERS[0]
|
||
else:
|
||
buttons = "".join(
|
||
_BTN.format(p=p, label=PROVIDERS[p]["label"]) for p in ENABLED_PROVIDERS)
|
||
return HTMLResponse(CHOOSER.replace("{buttons}", buttons))
|
||
|
||
if provider not in ENABLED_PROVIDERS:
|
||
raise HTTPException(404, "unknown provider")
|
||
|
||
cfg = PROVIDERS[provider]
|
||
nonce = secrets.token_urlsafe(24)
|
||
params = {
|
||
"client_id": cfg["client_id"],
|
||
"redirect_uri": f"{BASE_URL}/auth/callback",
|
||
"scope": cfg["scope"],
|
||
"state": nonce,
|
||
"response_type": "code",
|
||
}
|
||
if provider == "github":
|
||
params["allow_signup"] = "false"
|
||
if provider == "google":
|
||
# Force the account chooser; otherwise Google silently reuses whichever
|
||
# account the browser last used, which on a shared machine signs you in
|
||
# as someone else with no visible prompt.
|
||
params["prompt"] = "select_account"
|
||
|
||
resp = RedirectResponse(f"{cfg['authorize']}?{urlencode(params)}", status_code=302)
|
||
# CSRF defence: the nonce is echoed back by the provider and must match
|
||
# this cookie, so a callback the user did not initiate is rejected. The
|
||
# provider name rides along in the COOKIE, never the query string, so the
|
||
# callback cannot be steered to a different provider by the redirect.
|
||
_set_cookie(resp, STATE_COOKIE, f"{provider}|{nonce}", 600)
|
||
return resp
|
||
|
||
|
||
async def fetch_identity(provider: str, token: str) -> tuple[str, str, str]:
|
||
"""(username, display, avatar) for a provider access token."""
|
||
async with httpx.AsyncClient(timeout=20) as c:
|
||
if provider == "github":
|
||
r = await c.get(GH_API, headers={"Authorization": f"Bearer {token}",
|
||
"Accept": "application/vnd.github+json"})
|
||
if r.status_code != 200:
|
||
raise HTTPException(401, "github user lookup failed")
|
||
p = r.json() or {}
|
||
return p.get("login") or "", p.get("name") or "", p.get("avatar_url") or ""
|
||
|
||
r = await c.get(GOOGLE_API, headers={"Authorization": f"Bearer {token}"})
|
||
if r.status_code != 200:
|
||
raise HTTPException(401, "google user lookup failed")
|
||
p = r.json() or {}
|
||
# Unverified Google emails must never become an identity: anyone can
|
||
# put someone else's address on an account they control, and the
|
||
# allowlist is keyed on exactly this string.
|
||
if not p.get("email") or not p.get("email_verified"):
|
||
raise HTTPException(403, "google account has no verified email")
|
||
return p["email"], p.get("name") or "", p.get("picture") or ""
|
||
|
||
|
||
@app.get("/auth/callback")
|
||
async def auth_callback(request: Request, code: str = "", state: str = ""):
|
||
if AUTH_MODE == "off":
|
||
return RedirectResponse("/", status_code=302)
|
||
|
||
cookie = request.cookies.get(STATE_COOKIE) or ""
|
||
provider, _, nonce = cookie.partition("|")
|
||
if (not code or not state or not nonce
|
||
or provider not in ENABLED_PROVIDERS
|
||
or not secrets.compare_digest(state, nonce)):
|
||
log.warning("auth: rejected callback with bad or missing state")
|
||
return JSONResponse({"error": "invalid state"}, status_code=400)
|
||
|
||
cfg = PROVIDERS[provider]
|
||
async with httpx.AsyncClient(timeout=20) as c:
|
||
t = await c.post(cfg["token"], headers={"Accept": "application/json"},
|
||
data={"client_id": cfg["client_id"],
|
||
"client_secret": cfg["client_secret"],
|
||
"code": code,
|
||
"grant_type": "authorization_code",
|
||
"redirect_uri": f"{BASE_URL}/auth/callback"})
|
||
token = (t.json() or {}).get("access_token") if t.status_code == 200 else None
|
||
if not token:
|
||
log.warning("auth: %s token exchange failed (%s)", provider, t.status_code)
|
||
return JSONResponse({"error": "token exchange failed"}, status_code=401)
|
||
|
||
username, display, avatar = await fetch_identity(provider, token)
|
||
if not username:
|
||
return JSONResponse({"error": "no username from provider"}, status_code=401)
|
||
|
||
with db() as c:
|
||
existing = c.execute("SELECT * FROM users WHERE username=?",
|
||
(username,)).fetchone()
|
||
if not may_login(username, existing):
|
||
# Log the attempt: on an internet-facing box this is the signal that
|
||
# someone found the login page.
|
||
why = "disabled" if existing else "not invited"
|
||
log.warning("auth: DENIED %s user %r (%s)", provider, username, why)
|
||
return JSONResponse({"error": "not authorised"}, status_code=403)
|
||
|
||
row = upsert_user(username, avatar, display, provider)
|
||
log.info("auth: signed in %s via %s (%s)", username, provider, row["role"])
|
||
resp = RedirectResponse("/", status_code=302)
|
||
_set_cookie(resp, SESSION_COOKIE, new_session(username), SESSION_DAYS * 86400)
|
||
resp.delete_cookie(STATE_COOKIE, path="/")
|
||
return resp
|
||
|
||
|
||
@app.get("/auth/logout")
|
||
async def auth_logout(request: Request):
|
||
tok = request.cookies.get(SESSION_COOKIE)
|
||
if tok:
|
||
with db() as c:
|
||
c.execute("DELETE FROM sessions WHERE token_hash=?", (_hash(tok),))
|
||
resp = RedirectResponse("/auth/login", status_code=302)
|
||
resp.delete_cookie(SESSION_COOKIE, path="/")
|
||
return resp
|
||
|
||
|
||
# ─────────────────────────── quotas ───────────────────────────
|
||
#
|
||
# Two limits, both per user, both 0 = unlimited:
|
||
# quota requests per CALENDAR month (resets on the 1st)
|
||
# storage_quota bytes of imported audio attributable to this user
|
||
#
|
||
# Calendar month rather than a rolling 30-day window on purpose: "you have 12
|
||
# left until the 1st" is something a person can reason about, where a rolling
|
||
# window silently drips capacity back at times nobody can predict.
|
||
|
||
def month_start() -> str:
|
||
return datetime.now(timezone.utc).strftime("%Y-%m-01 00:00:00")
|
||
|
||
|
||
def usage_for(username: str | None) -> dict:
|
||
"""Requests this month and bytes imported, for one user."""
|
||
if not username:
|
||
return {"requests": 0, "storage": 0}
|
||
with db() as c:
|
||
n = c.execute(
|
||
"SELECT COUNT(*) n FROM jobs WHERE requested_by=? AND created_at >= ?",
|
||
(username, month_start())).fetchone()["n"]
|
||
b = c.execute(
|
||
"SELECT COALESCE(SUM(size),0) b FROM jobs "
|
||
"WHERE requested_by=? AND state='imported'", (username,)).fetchone()["b"]
|
||
return {"requests": n, "storage": b}
|
||
|
||
|
||
def check_quota(request: Request, adding: int = 1):
|
||
"""Raise 429 if this request would exceed the caller's monthly allowance.
|
||
|
||
Admins are exempt — someone has to be able to fix the library when a quota
|
||
is what is in the way.
|
||
"""
|
||
u = current_user(request)
|
||
if not u or u["role"] == "admin":
|
||
return
|
||
used = usage_for(u["username"])
|
||
if u["quota"] and used["requests"] + adding > u["quota"]:
|
||
raise HTTPException(429, (
|
||
f"Monthly request quota reached ({used['requests']}/{u['quota']}). "
|
||
f"Resets on the 1st."))
|
||
if u["storage_quota"] and used["storage"] >= u["storage_quota"]:
|
||
raise HTTPException(429, (
|
||
f"Storage quota reached ({used['storage'] // 2**30} GB of "
|
||
f"{u['storage_quota'] // 2**30} GB)."))
|
||
|
||
|
||
@app.get("/api/me")
|
||
async def me(request: Request):
|
||
u = current_user(request)
|
||
if not u:
|
||
return {"user": None, "auth": AUTH_MODE,
|
||
"role": "admin" if AUTH_MODE == "off" else None,
|
||
"providers": ENABLED_PROVIDERS}
|
||
return {
|
||
"user": u["username"], "auth": AUTH_MODE, "role": u["role"],
|
||
"display": u["display"], "avatar": u["avatar"], "provider": u["provider"],
|
||
"quota": u["quota"], "storage_quota": u["storage_quota"],
|
||
"usage": usage_for(u["username"]),
|
||
"providers": ENABLED_PROVIDERS,
|
||
}
|
||
|
||
|
||
class RequestBody(BaseModel):
|
||
artist: str
|
||
title: str
|
||
album: str | None = None
|
||
year: str | None = None
|
||
track: int | None = None
|
||
deezer_id: int | None = None
|
||
cover: str | None = None
|
||
genre: str | None = None
|
||
length: int | None = None
|
||
|
||
|
||
# Album metadata (release year, genre, label) is not in the track search
|
||
# payload, so it costs one extra call per distinct album. Cached forever on
|
||
# purpose: a release date and a genre are immutable, so there is nothing to
|
||
# invalidate, and repeat searches for the same artist stop hitting Deezer at
|
||
# all. Trimmed only to stop it growing without bound on a long-lived process.
|
||
_album_cache: dict[int, dict] = {}
|
||
_album_lock = asyncio.Semaphore(6) # Deezer rate-limits ~50 req/5s per IP
|
||
|
||
|
||
async def album_meta(album_id) -> dict:
|
||
if not album_id:
|
||
return {}
|
||
key = int(album_id)
|
||
if key in _album_cache:
|
||
return _album_cache[key]
|
||
async with _album_lock:
|
||
if key in _album_cache: # filled while we waited for the slot
|
||
return _album_cache[key]
|
||
try:
|
||
async with httpx.AsyncClient(timeout=15) as c:
|
||
r = await c.get(f"{DEEZER}/album/{key}")
|
||
d = r.json() if r.status_code == 200 else {}
|
||
except Exception as e:
|
||
log.debug("album %s lookup failed: %s", key, e)
|
||
return {}
|
||
genres = [g.get("name") for g in (d.get("genres") or {}).get("data", []) if g.get("name")]
|
||
meta = {
|
||
"year": (d.get("release_date") or "")[:4] or None,
|
||
"genre": genres[0] if genres else None,
|
||
"genres": genres,
|
||
"label": d.get("label"),
|
||
"nb_tracks": d.get("nb_tracks"),
|
||
"record_type": d.get("record_type"),
|
||
}
|
||
if len(_album_cache) > 5000:
|
||
for k in list(_album_cache)[:1000]:
|
||
_album_cache.pop(k, None)
|
||
_album_cache[key] = meta
|
||
return meta
|
||
|
||
|
||
def track_row(t: dict) -> dict:
|
||
"""One Deezer track object -> the shape the UI and the job table use."""
|
||
artist = t.get("artist") or {}
|
||
album = t.get("album") or {}
|
||
return {
|
||
"deezer_id": t.get("id"),
|
||
"title": t.get("title") or "",
|
||
"artist": artist.get("name") or "",
|
||
"artist_id": artist.get("id"),
|
||
"artist_pic": artist.get("picture_medium"),
|
||
"album": album.get("title") or "",
|
||
"album_id": album.get("id"),
|
||
"cover": album.get("cover_medium"),
|
||
"cover_big": album.get("cover_big") or album.get("cover_medium"),
|
||
"length": (t.get("duration") or 0) * 1000,
|
||
# 30-second clip from Deezer's CDN. This is what lets you confirm you
|
||
# picked the right recording BEFORE committing a Soulseek search to it
|
||
# -- the single most useful thing in the payload for avoiding a wrong
|
||
# import, and it costs nothing extra.
|
||
"preview": t.get("preview") or None,
|
||
"explicit": bool(t.get("explicit_lyrics")),
|
||
"isrc": t.get("isrc"),
|
||
"rank": t.get("rank") or 0,
|
||
"year": None, "genre": None, "track": None,
|
||
}
|
||
|
||
|
||
async def add_album_meta(rows: list[dict]) -> list[dict]:
|
||
"""Fill year and genre on a result list, concurrently and de-duplicated."""
|
||
ids = {r["album_id"] for r in rows if r.get("album_id")}
|
||
if not ids:
|
||
return rows
|
||
metas = dict(zip(ids, await asyncio.gather(*(album_meta(i) for i in ids))))
|
||
for r in rows:
|
||
m = metas.get(r.get("album_id")) or {}
|
||
r["year"] = r["year"] or m.get("year")
|
||
r["genre"] = r["genre"] or m.get("genre")
|
||
return rows
|
||
|
||
|
||
async def deezer_search(q: str, limit: int = 25) -> list[dict]:
|
||
"""Ranked Deezer track search. Shared by /api/search and /api/import."""
|
||
if not q.strip():
|
||
return []
|
||
async with httpx.AsyncClient(timeout=20) as c:
|
||
r = await c.get(f"{DEEZER}/search", params={"q": q, "limit": limit * 2})
|
||
if r.status_code != 200:
|
||
raise HTTPException(502, f"deezer returned {r.status_code}")
|
||
data = r.json()
|
||
|
||
qn = norm(q)
|
||
scored = []
|
||
for t in data.get("data", []):
|
||
title = t.get("title") or ""
|
||
artist = (t.get("artist") or {}).get("name") or ""
|
||
|
||
s = float(t.get("rank") or 0) / 10_000 # Deezer popularity, 0-~100
|
||
|
||
# An artist match separates the real track from a covers compilation.
|
||
at = set(norm(artist).split())
|
||
if norm(artist) and norm(artist) in qn:
|
||
s += 300
|
||
|
||
# Title relevance. This has to dominate everything else: without it a
|
||
# completely different song by the right artist competes on equal
|
||
# footing, and "Oasis - Live Forever" resolved to "Cloudburst" because
|
||
# a variant penalty was enough to tip between two unrelated tracks.
|
||
# Whatever query words the artist name did not explain should show up
|
||
# in the title.
|
||
rest = set(qn.split()) - at
|
||
if rest:
|
||
s += 400 * len(rest & set(norm(title).split())) / len(rest)
|
||
for w in set(DERIVATIVE_RE.findall(norm(title))):
|
||
if not re.search(rf"\b{re.escape(w)}\b", qn):
|
||
s -= 150
|
||
# Catch-all for variants the keyword list will never cover in full
|
||
# ("Devonshire Mix", "Radio Edit", "2011 Version"): a plain title beats
|
||
# a parenthesised one. Deliberately small -- it should break ties
|
||
# between versions of the SAME song, never outweigh title relevance
|
||
# above and pick a different song.
|
||
if ("(" in title or "[" in title) and "(" not in q:
|
||
s -= 40
|
||
scored.append((s, track_row(t)))
|
||
|
||
seen, out = set(), []
|
||
for _, item in sorted(scored, key=lambda x: -x[0]):
|
||
key = (norm(item["artist"]), norm(item["title"]))
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append(item)
|
||
return out[:limit]
|
||
|
||
|
||
@app.get("/api/search")
|
||
async def search(q: str, limit: int = 25):
|
||
"""Track search. Returns individual songs, not albums — the whole point.
|
||
|
||
Year and genre are grafted on afterwards because Deezer's track search
|
||
omits both. Enrichment happens only here, not in deezer_search itself, so
|
||
the bulk-import path stays one HTTP call per line.
|
||
"""
|
||
return {"results": await add_album_meta(await deezer_search(q, limit))}
|
||
|
||
|
||
@app.get("/api/suggest")
|
||
async def suggest(q: str):
|
||
"""Autocomplete. Deliberately does NOT enrich or re-rank: this fires on
|
||
every few keystrokes, so it has to be one upstream call and nothing else.
|
||
Deezer's own autocomplete ordering is good enough for a dropdown, and the
|
||
real ranking still happens when the search is actually submitted."""
|
||
q = q.strip()
|
||
if len(q) < 2:
|
||
return {"tracks": [], "albums": [], "artists": []}
|
||
try:
|
||
async with httpx.AsyncClient(timeout=8) as c:
|
||
r = await c.get(f"{DEEZER}/search/autocomplete",
|
||
params={"q": q, "limit": 5})
|
||
d = r.json() if r.status_code == 200 else {}
|
||
except Exception:
|
||
return {"tracks": [], "albums": [], "artists": []}
|
||
|
||
def rows(key):
|
||
return (d.get(key) or {}).get("data", []) or []
|
||
|
||
return {
|
||
"tracks": [{"deezer_id": t.get("id"), "title": t.get("title"),
|
||
"artist": (t.get("artist") or {}).get("name"),
|
||
"cover": (t.get("album") or {}).get("cover_small"),
|
||
"preview": t.get("preview")}
|
||
for t in rows("tracks")[:5]],
|
||
"albums": [{"album_id": a.get("id"), "title": a.get("title"),
|
||
"artist": (a.get("artist") or {}).get("name"),
|
||
"cover": a.get("cover_small"),
|
||
"nb_tracks": a.get("nb_tracks")}
|
||
for a in rows("albums")[:3]],
|
||
"artists": [{"artist_id": a.get("id"), "name": a.get("name"),
|
||
"cover": a.get("picture_small"),
|
||
"nb_album": a.get("nb_album")}
|
||
for a in rows("artists")[:3]],
|
||
}
|
||
|
||
|
||
@app.get("/api/album/{album_id}")
|
||
async def album_detail(album_id: int):
|
||
"""Full track list for an album, so 'Get all' can queue it in one go."""
|
||
# Two calls, not one. /album/{id} embeds a track list, but those nested
|
||
# objects have no track_position -- and taking the list index instead would
|
||
# misnumber every multi-disc release, writing "13 - ..." for disc 2 track 1.
|
||
# /album/{id}/tracks is the only source that carries track_position and
|
||
# disk_number, and those numbers end up in the filename and the tags.
|
||
try:
|
||
async with httpx.AsyncClient(timeout=20) as c:
|
||
r, tr = await asyncio.gather(
|
||
c.get(f"{DEEZER}/album/{album_id}"),
|
||
c.get(f"{DEEZER}/album/{album_id}/tracks", params={"limit": 300}))
|
||
except Exception as e:
|
||
raise HTTPException(502, f"deezer: {e}")
|
||
if r.status_code != 200:
|
||
raise HTTPException(404, "album not found")
|
||
d = r.json()
|
||
|
||
artist = (d.get("artist") or {}).get("name") or ""
|
||
year = (d.get("release_date") or "")[:4] or None
|
||
genres = [g.get("name") for g in (d.get("genres") or {}).get("data", []) if g.get("name")]
|
||
raw = (tr.json().get("data", []) if tr.status_code == 200
|
||
else (d.get("tracks") or {}).get("data", []))
|
||
tracks = []
|
||
for t in raw:
|
||
row = track_row(t)
|
||
# These nested track objects carry no album of their own, so inherit
|
||
# what the parent already told us rather than looking it up again.
|
||
row.update(album=d.get("title") or "", album_id=d.get("id"),
|
||
cover=d.get("cover_medium"), cover_big=d.get("cover_big"),
|
||
artist=row["artist"] or artist, year=year,
|
||
genre=genres[0] if genres else None,
|
||
track=t.get("track_position"),
|
||
disc=t.get("disk_number"))
|
||
tracks.append(row)
|
||
|
||
return {"album": {"album_id": d.get("id"), "title": d.get("title"),
|
||
"artist": artist, "artist_id": (d.get("artist") or {}).get("id"),
|
||
"cover_big": d.get("cover_big"), "cover": d.get("cover_medium"),
|
||
"year": year, "genres": genres, "label": d.get("label"),
|
||
"nb_tracks": d.get("nb_tracks"),
|
||
"record_type": d.get("record_type"),
|
||
"duration": d.get("duration")},
|
||
"tracks": tracks}
|
||
|
||
|
||
@app.get("/api/artist/{artist_id}")
|
||
async def artist_detail(artist_id: int):
|
||
"""Artist header plus their top tracks and albums."""
|
||
try:
|
||
async with httpx.AsyncClient(timeout=20) as c:
|
||
info, top, albums = await asyncio.gather(
|
||
c.get(f"{DEEZER}/artist/{artist_id}"),
|
||
c.get(f"{DEEZER}/artist/{artist_id}/top", params={"limit": 15}),
|
||
c.get(f"{DEEZER}/artist/{artist_id}/albums", params={"limit": 24}))
|
||
except Exception as e:
|
||
raise HTTPException(502, f"deezer: {e}")
|
||
if info.status_code != 200:
|
||
raise HTTPException(404, "artist not found")
|
||
a = info.json()
|
||
tracks = [track_row(t) for t in (top.json().get("data", []) if top.status_code == 200 else [])]
|
||
await add_album_meta(tracks)
|
||
return {
|
||
"artist": {"artist_id": a.get("id"), "name": a.get("name"),
|
||
"picture": a.get("picture_big"), "nb_fan": a.get("nb_fan"),
|
||
"nb_album": a.get("nb_album")},
|
||
"tracks": tracks,
|
||
"albums": [{"album_id": x.get("id"), "title": x.get("title"),
|
||
"cover": x.get("cover_medium"), "cover_big": x.get("cover_big"),
|
||
"year": (x.get("release_date") or "")[:4] or None,
|
||
"record_type": x.get("record_type"),
|
||
"nb_tracks": x.get("nb_tracks")}
|
||
for x in (albums.json().get("data", []) if albums.status_code == 200 else [])],
|
||
}
|
||
|
||
|
||
async def enqueue_job(artist: str, title: str, album=None, year=None,
|
||
track=None, deezer_id=None, requested_by=None,
|
||
cover=None, genre=None, duration=None) -> str:
|
||
"""Create a job and hand it to the search queue."""
|
||
job_id = str(uuid.uuid4())
|
||
with db() as c:
|
||
c.execute(
|
||
"INSERT INTO jobs (id,state,detail,artist,title,album,year,track,"
|
||
"recording_mbid,release_mbid,requested_by,cover,genre,duration) "
|
||
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
|
||
(job_id, "queued", "", artist, title, album, year, track,
|
||
str(deezer_id or ""), None, requested_by, cover, genre, duration))
|
||
await work_q.put(job_id)
|
||
return job_id
|
||
|
||
|
||
async def enrich(deezer_id) -> dict:
|
||
"""Deezer's search payload has no track number or release date, and both
|
||
go into the filename and tags. One lookup per request, not per result."""
|
||
if not deezer_id:
|
||
return {}
|
||
try:
|
||
async with httpx.AsyncClient(timeout=15) as c:
|
||
r = await c.get(f"{DEEZER}/track/{deezer_id}")
|
||
if r.status_code != 200:
|
||
return {}
|
||
d = r.json()
|
||
rd = d.get("release_date") or (d.get("album") or {}).get("release_date")
|
||
return {
|
||
"track": d.get("track_position"),
|
||
"year": (rd or "")[:4] or None,
|
||
"album": (d.get("album") or {}).get("title"),
|
||
"artist": (d.get("artist") or {}).get("name"),
|
||
}
|
||
except Exception as e:
|
||
log.warning("deezer track lookup failed: %s", e)
|
||
return {}
|
||
|
||
|
||
@app.post("/api/request")
|
||
async def request_track(body: RequestBody, request: Request):
|
||
check_quota(request)
|
||
e = await enrich(body.deezer_id)
|
||
return {"id": await enqueue_job(
|
||
e.get("artist") or body.artist, body.title,
|
||
e.get("album") or body.album, e.get("year") or body.year,
|
||
e.get("track") or body.track, body.deezer_id,
|
||
requested_by=session_user(request), cover=body.cover,
|
||
genre=body.genre, duration=body.length)}
|
||
|
||
|
||
class BatchBody(BaseModel):
|
||
tracks: list[RequestBody]
|
||
|
||
|
||
@app.post("/api/request/batch")
|
||
async def request_batch(body: BatchBody, request: Request):
|
||
"""Queue a whole album at once.
|
||
|
||
Unlike /api/request this skips the per-track /track/{id} enrichment: the
|
||
caller got these rows from /api/album/{id}, which already carries the
|
||
album, year and track position for every one of them. Re-fetching would be
|
||
nb_tracks extra calls to learn what we were just told.
|
||
"""
|
||
# Checked against the whole batch, not one track: queueing a 30-track album
|
||
# when 5 requests remain should be refused up front, not accepted and then
|
||
# blocked halfway through.
|
||
check_quota(request, adding=len(body.tracks))
|
||
who = session_user(request)
|
||
ids = [await enqueue_job(t.artist, t.title, t.album, t.year, t.track,
|
||
t.deezer_id, requested_by=who, cover=t.cover,
|
||
genre=t.genre, duration=t.length)
|
||
for t in body.tracks]
|
||
log.info("batch: queued %d track(s) for %s", len(ids), who or "anon")
|
||
return {"queued": len(ids), "ids": ids, "pending": work_q.qsize()}
|
||
|
||
|
||
class ImportBody(BaseModel):
|
||
text: str
|
||
|
||
|
||
# "Artist - Title", "Artist – Title" (en dash), "Title by Artist". Anything
|
||
# that doesn't split just goes to Deezer whole — its search copes fine with
|
||
# free text, and guessing wrong is worse than not guessing.
|
||
SPLIT = re.compile(r"\s+[-–—|]\s+|\s+by\s+", re.I)
|
||
DEEZER_URL = re.compile(r"deezer\.com/(?:[a-z]{2}/)?(album|playlist)/(\d+)")
|
||
|
||
|
||
async def expand_url(kind: str, ident: str) -> list[str]:
|
||
"""Turn a Deezer album/playlist URL into 'Artist - Title' lines."""
|
||
out = []
|
||
try:
|
||
async with httpx.AsyncClient(timeout=25) as c:
|
||
r = await c.get(f"{DEEZER}/{kind}/{ident}")
|
||
if r.status_code != 200:
|
||
return []
|
||
for t in (r.json().get("tracks") or {}).get("data", []):
|
||
a = (t.get("artist") or {}).get("name") or ""
|
||
out.append(f"{a} - {t.get('title')}" if a else t.get("title", ""))
|
||
except Exception as e:
|
||
log.warning("expanding %s/%s failed: %s", kind, ident, e)
|
||
return [x for x in out if x]
|
||
|
||
|
||
@app.post("/api/import")
|
||
async def import_list(body: ImportBody, request: Request):
|
||
"""Paste a list, get it all queued.
|
||
|
||
Each line is resolved against Deezer and the top-ranked hit is taken
|
||
automatically — there is no picker in bulk mode. Resolved artist/album are
|
||
recorded on every job so a wrong auto-pick is visible in the list rather
|
||
than silently landing the wrong song in the library.
|
||
"""
|
||
lines, queued, skipped = [], [], []
|
||
for raw in body.text.splitlines():
|
||
line = raw.strip().lstrip("-•*0123456789. )").strip()
|
||
if not line:
|
||
continue
|
||
m = DEEZER_URL.search(line)
|
||
lines.extend(await expand_url(*m.groups()) if m else [line])
|
||
|
||
# After expansion — a single playlist URL can be 200 tracks, and the point
|
||
# of the check is the real number, not the number of lines pasted.
|
||
check_quota(request, adding=len(lines))
|
||
|
||
for line in lines:
|
||
# Search the line as typed; the ranker already weights an artist-name
|
||
# match heavily, so "Artist - Title" needs no special handling.
|
||
try:
|
||
# Ask for a real candidate pool, not one row. deezer_search ranks
|
||
# what it is given, so limit=1 fetched two results and handed back
|
||
# the better of a demo and a rehearsal take while the studio
|
||
# version -- 6x the popularity -- sat further down the list unseen.
|
||
hits = await deezer_search(SPLIT.sub(" ", line), limit=10)
|
||
except Exception as e:
|
||
skipped.append({"line": line, "why": str(e)[:120]})
|
||
continue
|
||
if not hits:
|
||
skipped.append({"line": line, "why": "no match on Deezer"})
|
||
continue
|
||
h = hits[0]
|
||
e = await enrich(h["deezer_id"])
|
||
jid = await enqueue_job(
|
||
e.get("artist") or h["artist"], h["title"],
|
||
e.get("album") or h["album"], e.get("year"), e.get("track"),
|
||
h["deezer_id"], requested_by=session_user(request),
|
||
cover=h.get("cover"), duration=h.get("length"))
|
||
queued.append({"id": jid, "line": line, "cover": h.get("cover"),
|
||
"matched": f"{h['artist']} — {h['title']}"})
|
||
|
||
log.info("import: %d queued, %d skipped", len(queued), len(skipped))
|
||
return {"queued": queued, "skipped": skipped, "pending": work_q.qsize()}
|
||
|
||
|
||
@app.get("/api/jobs")
|
||
async def jobs(request: Request, limit: int = 200, mine: bool = False):
|
||
"""Own requests by default; admins see everyone's.
|
||
|
||
Scoped rather than global because on a shared instance one person's
|
||
request history is a list of what they listen to, which is nobody else's
|
||
business by default.
|
||
"""
|
||
u = current_user(request)
|
||
everyone = (AUTH_MODE == "off") or (u is not None and u["role"] == "admin" and not mine)
|
||
who = None if everyone else (u["username"] if u else None)
|
||
|
||
where, args = ("", []) if everyone else ("WHERE requested_by IS ?", [who])
|
||
with db() as c:
|
||
rows = c.execute(
|
||
f"SELECT * FROM jobs {where} ORDER BY created_at DESC, rowid DESC LIMIT ?",
|
||
[*args, limit]).fetchall()
|
||
counts = {r["state"]: r["n"] for r in c.execute(
|
||
f"SELECT state, COUNT(*) n FROM jobs {where} GROUP BY state",
|
||
args).fetchall()}
|
||
return {"jobs": [dict(r) for r in rows], "counts": counts,
|
||
"pending": work_q.qsize(), "scope": "all" if everyone else "mine"}
|
||
|
||
|
||
@app.post("/api/retry/{job_id}")
|
||
async def retry(job_id: str, request: Request):
|
||
"""Put a failed job back on the queue. Soulseek failures are usually the
|
||
peer going offline, not the request being wrong, so retrying later often
|
||
just works."""
|
||
u = current_user(request)
|
||
with db() as c:
|
||
row = c.execute("SELECT id, requested_by FROM jobs WHERE id=?",
|
||
(job_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "no such job")
|
||
# Same 404 rather than 403 for someone else's job: a distinct error would
|
||
# confirm the id exists, which is a small oracle over other people's
|
||
# requests.
|
||
if AUTH_MODE != "off" and u and u["role"] != "admin" \
|
||
and row["requested_by"] != u["username"]:
|
||
raise HTTPException(404, "no such job")
|
||
set_state(job_id, "queued", "")
|
||
await work_q.put(job_id)
|
||
return {"ok": True}
|
||
|
||
|
||
# ─────────────────────────── admin ───────────────────────────
|
||
|
||
class NewUser(BaseModel):
|
||
username: str
|
||
provider: str = "github"
|
||
role: str = "user"
|
||
quota: int = 0
|
||
storage_quota_gb: float = 0
|
||
|
||
|
||
class PatchUser(BaseModel):
|
||
role: str | None = None
|
||
quota: int | None = None
|
||
storage_quota_gb: float | None = None
|
||
disabled: bool | None = None
|
||
|
||
|
||
def user_dict(r: sqlite3.Row) -> dict:
|
||
u = usage_for(r["username"])
|
||
return {
|
||
"username": r["username"], "provider": r["provider"], "role": r["role"],
|
||
"display": r["display"], "avatar": r["avatar"],
|
||
"quota": r["quota"], "storage_quota": r["storage_quota"],
|
||
"disabled": bool(r["disabled"]), "created_at": r["created_at"],
|
||
"last_seen": r["last_seen"], "invited_by": r["invited_by"],
|
||
"usage": u,
|
||
# Surfaced so the UI doesn't have to know the allowlist rule: these are
|
||
# the accounts that cannot actually be locked out by disabling them.
|
||
"bootstrap": r["username"].lower() in ALLOWED_USERS,
|
||
}
|
||
|
||
|
||
@app.get("/api/admin/users")
|
||
async def admin_users(request: Request):
|
||
require_admin(request)
|
||
with db() as c:
|
||
rows = c.execute("SELECT * FROM users ORDER BY role='admin' DESC, username").fetchall()
|
||
return {"users": [user_dict(r) for r in rows], "roles": list(ROLES)}
|
||
|
||
|
||
@app.post("/api/admin/users")
|
||
async def admin_add_user(body: NewUser, request: Request):
|
||
"""Invite someone. There is no email or token — the row IS the invitation:
|
||
once it exists, that identity can sign in with the matching provider."""
|
||
who = require_admin(request)
|
||
name = body.username.strip()
|
||
if not name:
|
||
raise HTTPException(400, "username required")
|
||
if body.role not in ROLES:
|
||
raise HTTPException(400, f"role must be one of {', '.join(ROLES)}")
|
||
if body.provider not in PROVIDERS:
|
||
raise HTTPException(400, f"provider must be one of {', '.join(PROVIDERS)}")
|
||
with db() as c:
|
||
if c.execute("SELECT 1 FROM users WHERE username=?", (name,)).fetchone():
|
||
raise HTTPException(409, "that user already exists")
|
||
c.execute(
|
||
"INSERT INTO users (username,provider,role,quota,storage_quota,invited_by) "
|
||
"VALUES (?,?,?,?,?,?)",
|
||
(name, body.provider, body.role, max(0, body.quota),
|
||
int(max(0.0, body.storage_quota_gb) * 2**30), who))
|
||
r = c.execute("SELECT * FROM users WHERE username=?", (name,)).fetchone()
|
||
log.info("admin: %s invited %s as %s", who, name, body.role)
|
||
return user_dict(r)
|
||
|
||
|
||
@app.patch("/api/admin/users/{username}")
|
||
async def admin_patch_user(username: str, body: PatchUser, request: Request):
|
||
who = require_admin(request)
|
||
with db() as c:
|
||
r = c.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
|
||
if not r:
|
||
raise HTTPException(404, "no such user")
|
||
|
||
# An install with no admin cannot be administered, and nothing in the
|
||
# UI can undo it. Refuse to remove the last one — whether by demotion
|
||
# or by disabling them.
|
||
losing_admin = r["role"] == "admin" and (
|
||
(body.role is not None and body.role != "admin") or body.disabled is True)
|
||
if losing_admin:
|
||
others = c.execute(
|
||
"SELECT COUNT(*) n FROM users "
|
||
"WHERE role='admin' AND disabled=0 AND username<>?",
|
||
(username,)).fetchone()["n"]
|
||
if others == 0:
|
||
raise HTTPException(
|
||
409, "this is the only active admin — promote someone else first")
|
||
|
||
sets, vals = [], []
|
||
if body.role is not None:
|
||
if body.role not in ROLES:
|
||
raise HTTPException(400, f"role must be one of {', '.join(ROLES)}")
|
||
sets.append("role=?"); vals.append(body.role)
|
||
if body.quota is not None:
|
||
sets.append("quota=?"); vals.append(max(0, body.quota))
|
||
if body.storage_quota_gb is not None:
|
||
sets.append("storage_quota=?")
|
||
vals.append(int(max(0.0, body.storage_quota_gb) * 2**30))
|
||
if body.disabled is not None:
|
||
sets.append("disabled=?"); vals.append(1 if body.disabled else 0)
|
||
if not sets:
|
||
raise HTTPException(400, "nothing to change")
|
||
|
||
c.execute(f"UPDATE users SET {', '.join(sets)} WHERE username=?",
|
||
[*vals, username])
|
||
# Disabling must take effect now. session_user re-checks on every
|
||
# request anyway, but dropping the rows makes it unambiguous.
|
||
if body.disabled:
|
||
c.execute("DELETE FROM sessions WHERE username=?", (username,))
|
||
r = c.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
|
||
log.info("admin: %s updated %s (%s)", who, username, ", ".join(sets))
|
||
return user_dict(r)
|
||
|
||
|
||
@app.delete("/api/admin/users/{username}")
|
||
async def admin_delete_user(username: str, request: Request):
|
||
who = require_admin(request)
|
||
with db() as c:
|
||
r = c.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
|
||
if not r:
|
||
raise HTTPException(404, "no such user")
|
||
if r["role"] == "admin":
|
||
others = c.execute(
|
||
"SELECT COUNT(*) n FROM users "
|
||
"WHERE role='admin' AND disabled=0 AND username<>?",
|
||
(username,)).fetchone()["n"]
|
||
if others == 0:
|
||
raise HTTPException(409, "this is the only active admin")
|
||
c.execute("DELETE FROM sessions WHERE username=?", (username,))
|
||
c.execute("DELETE FROM users WHERE username=?", (username,))
|
||
# Their jobs are deliberately kept: the history is a record of what is in
|
||
# the library and why, and deleting a person should not rewrite it.
|
||
log.info("admin: %s deleted user %s (jobs retained)", who, username)
|
||
return {"ok": True, "jobs_retained": True}
|
||
|
||
|
||
@app.get("/api/admin/activity")
|
||
async def admin_activity(request: Request, days: int = 30):
|
||
require_admin(request)
|
||
days = max(1, min(days, 365))
|
||
since = (datetime.now(timezone.utc) - timedelta(days=days)).strftime("%Y-%m-%d")
|
||
with db() as c:
|
||
daily = c.execute(
|
||
"SELECT substr(created_at,1,10) d, COUNT(*) n, "
|
||
" SUM(state='imported') imported, SUM(state='failed') failed "
|
||
"FROM jobs WHERE created_at >= ? GROUP BY d ORDER BY d", (since,)).fetchall()
|
||
per_user = c.execute(
|
||
"SELECT COALESCE(requested_by,'(unattributed)') u, COUNT(*) n, "
|
||
" COALESCE(SUM(CASE WHEN state='imported' THEN size END),0) bytes "
|
||
"FROM jobs WHERE created_at >= ? GROUP BY u ORDER BY n DESC",
|
||
(since,)).fetchall()
|
||
top = c.execute(
|
||
"SELECT artist, COUNT(*) n FROM jobs WHERE created_at >= ? "
|
||
"GROUP BY artist ORDER BY n DESC LIMIT 8", (since,)).fetchall()
|
||
|
||
# Zero-fill: a bar chart with gaps for quiet days misreads as missing data
|
||
# rather than as nothing having happened.
|
||
by_day = {r["d"]: r for r in daily}
|
||
today = datetime.now(timezone.utc).date()
|
||
series = []
|
||
for i in range(days - 1, -1, -1):
|
||
d = (today - timedelta(days=i)).isoformat()
|
||
r = by_day.get(d)
|
||
series.append({"date": d, "n": r["n"] if r else 0,
|
||
"imported": (r["imported"] or 0) if r else 0,
|
||
"failed": (r["failed"] or 0) if r else 0})
|
||
return {"days": days, "series": series,
|
||
"per_user": [dict(r) for r in per_user],
|
||
"top_artists": [dict(r) for r in top]}
|
||
|
||
|
||
def _proc_stat() -> tuple[int, int]:
|
||
"""(busy, total) jiffies from /proc/stat."""
|
||
with open("/proc/stat") as f:
|
||
parts = [int(x) for x in f.readline().split()[1:]]
|
||
idle = parts[3] + (parts[4] if len(parts) > 4 else 0)
|
||
return sum(parts) - idle, sum(parts)
|
||
|
||
|
||
_cpu_prev: tuple[int, int] | None = None
|
||
|
||
|
||
def system_stats() -> dict:
|
||
"""Host CPU/RAM plus disk for the paths that matter.
|
||
|
||
Reads /proc directly rather than adding psutil: it is ~30 lines, keeps the
|
||
image free of a compiled dependency, and /proc in a container reports the
|
||
HOST's CPU and memory — which is what you actually want to see for a
|
||
homelab box. Everything is individually guarded so a non-Linux dev machine
|
||
degrades to nulls instead of 500ing the whole panel.
|
||
"""
|
||
global _cpu_prev
|
||
out: dict = {}
|
||
|
||
try:
|
||
busy, total = _proc_stat()
|
||
if _cpu_prev:
|
||
db_, dt = busy - _cpu_prev[0], total - _cpu_prev[1]
|
||
out["cpu_pct"] = round(100 * db_ / dt, 1) if dt > 0 else None
|
||
else:
|
||
out["cpu_pct"] = None # first call has no baseline to diff
|
||
_cpu_prev = (busy, total)
|
||
except Exception:
|
||
out["cpu_pct"] = None
|
||
|
||
try:
|
||
mem = {}
|
||
with open("/proc/meminfo") as f:
|
||
for line in f:
|
||
k, _, v = line.partition(":")
|
||
mem[k] = int(v.split()[0]) * 1024
|
||
total_b = mem.get("MemTotal", 0)
|
||
avail = mem.get("MemAvailable", 0)
|
||
out["mem_total"] = total_b
|
||
out["mem_used"] = total_b - avail
|
||
out["mem_pct"] = round(100 * (total_b - avail) / total_b, 1) if total_b else None
|
||
except Exception:
|
||
out["mem_total"] = out["mem_used"] = out["mem_pct"] = None
|
||
|
||
try:
|
||
out["load"] = [round(x, 2) for x in os.getloadavg()]
|
||
out["cpus"] = os.cpu_count()
|
||
except Exception:
|
||
out["load"] = None
|
||
|
||
disks = {}
|
||
for label, p in (("library", LIBRARY_DIR), ("downloads", DOWNLOAD_DIR)):
|
||
try:
|
||
s = os.statvfs(p)
|
||
total_b = s.f_blocks * s.f_frsize
|
||
free = s.f_bavail * s.f_frsize
|
||
disks[label] = {"total": total_b, "used": total_b - free, "free": free,
|
||
"pct": round(100 * (total_b - free) / total_b, 1) if total_b else None}
|
||
except Exception:
|
||
disks[label] = None
|
||
out["disks"] = disks
|
||
return out
|
||
|
||
|
||
@app.get("/api/admin/system")
|
||
async def admin_system(request: Request):
|
||
require_admin(request)
|
||
out = {"system": system_stats(), "workers": WORKERS, "pending": work_q.qsize()}
|
||
|
||
with db() as c:
|
||
out["jobs"] = {r["state"]: r["n"] for r in c.execute(
|
||
"SELECT state, COUNT(*) n FROM jobs GROUP BY state").fetchall()}
|
||
out["library"] = {
|
||
"tracks": c.execute(
|
||
"SELECT COUNT(*) n FROM jobs WHERE state='imported'").fetchone()["n"],
|
||
"bytes": c.execute(
|
||
"SELECT COALESCE(SUM(size),0) b FROM jobs "
|
||
"WHERE state='imported'").fetchone()["b"],
|
||
}
|
||
out["users"] = c.execute("SELECT COUNT(*) n FROM users").fetchone()["n"]
|
||
out["sessions"] = c.execute(
|
||
"SELECT COUNT(*) n FROM sessions WHERE expires_at > ?",
|
||
(datetime.now(timezone.utc).isoformat(),)).fetchone()["n"]
|
||
|
||
slskd: dict = {"ok": False}
|
||
try:
|
||
async with slskd_client(PROBE_TIMEOUT) as sc:
|
||
r = await sc.get("/api/v0/application")
|
||
slskd["ok"] = r.status_code == 200
|
||
if r.status_code == 200:
|
||
d = r.json()
|
||
srv = d.get("server") or {}
|
||
slskd["connected"] = bool(srv.get("isConnected"))
|
||
slskd["logged_in"] = bool(srv.get("isLoggedIn"))
|
||
slskd["version"] = d.get("version")
|
||
shares = d.get("shares") or {}
|
||
slskd["shared_files"] = (shares.get("directories") or {}).get("fileCount") \
|
||
or shares.get("fileCount")
|
||
else:
|
||
slskd["error"] = f"HTTP {r.status_code}"
|
||
except Exception as e:
|
||
slskd["error"] = why(e)
|
||
out["slskd"] = slskd
|
||
return out
|
||
|
||
|
||
@app.get("/api/health")
|
||
async def health():
|
||
ok, detail = True, {}
|
||
async with slskd_client(PROBE_TIMEOUT) as sc:
|
||
try:
|
||
r = await sc.get("/api/v0/application")
|
||
detail["slskd"] = r.status_code
|
||
ok = ok and r.status_code == 200
|
||
except Exception as e:
|
||
detail["slskd"] = why(e)
|
||
ok = False
|
||
detail["library_writable"] = os.access(LIBRARY_DIR, os.W_OK)
|
||
detail["downloads_readable"] = os.access(DOWNLOAD_DIR, os.R_OK)
|
||
return {"ok": ok and detail["library_writable"], **detail}
|
||
|
||
|
||
@app.get("/")
|
||
async def index():
|
||
f = WEB / "index.html"
|
||
if not f.exists():
|
||
raise HTTPException(500, "UI not built — run `npm run build` in web/")
|
||
return FileResponse(f)
|
||
|
||
|
||
# Next.js emits its immutable, content-hashed chunks under /_next/static.
|
||
#
|
||
# Guarded rather than mounted unconditionally so that importing this module
|
||
# does not require build artifacts: CI typechecks the API before the UI has
|
||
# been built, and tests should not have to run npm first. A deploy that really
|
||
# is missing its UI says so at startup and again on the first request, rather
|
||
# than silently serving 404s.
|
||
if (WEB / "_next").is_dir():
|
||
app.mount("/_next", StaticFiles(directory=WEB / "_next"), name="next")
|
||
else:
|
||
log.warning("web/out/_next is missing — the UI was not built. "
|
||
"API endpoints work; the web page will not.")
|