aryansingh/muse
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

muse: request a single song, get it tagged in your library

Browse Source 
Lidarr's smallest unit is a MusicBrainz release, so asking for one track
means monitoring a whole discography. Soulseek trades individual files but
has no concept of a request. muse bridges the two: search, click, and the
song lands in the library correctly tagged.

The design point is that the user picks the exact track before anything
downloads, so muse holds artist, album, year and track number as ground
truth and tags deterministically rather than inferring them afterwards from
whatever a stranger named the file.

Search is Deezer rather than MusicBrainz, decided on measurement:
MusicBrainz put eight covers above the real recording for "radiohead creep"
and never surfaced Queen for "bohemian rhapsody" across 90 results, because
its index matches a word in a cover's title as readily as in the artist
field. Re-ranking cannot rescue a result set missing the right row.

Imports are os.link(), so downloads keep seeding from the Soulseek share
while the library gets a second name for the same bytes.
This commit is contained in:
aryansingh committed 2026-08-04 03:05:45 -05:00
commit 269f13524d
7 files changed
+874
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

app.py
+481
View File
@@ -0,0 +1,481 @@
"""
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 difflib
import logging
import os
import re
import sqlite3
import uuid
from contextlib import asynccontextmanager
from pathlib import Path
import httpx
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse
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"))
# 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.
DERIVATIVE = ("cover", "karaoke", "tribute", "remix", "instrumental",
"originally performed", "made popular by", "in the style of")
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'))
)
""")
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
async def 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):
init_db()
task = asyncio.create_task(worker())
log.info("muse up — slskd=%s library=%s formats=%s",
SLSKD_URL, LIBRARY_DIR, ",".join(FORMATS))
yield
task.cancel()
app = FastAPI(title="muse", lifespan=lifespan)
class RequestBody(BaseModel):
artist: str
title: str
album: str | None = None
year: str | None = None
track: int | None = None
deezer_id: int | None = None
@app.get("/api/search")
async def search(q: str, limit: int = 25):
"""Track search. Returns individual songs, not albums — the whole point."""
if not q.strip():
return {"results": []}
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 ""
album = (t.get("album") or {}).get("title") or ""
s = float(t.get("rank") or 0) / 10_000 # Deezer popularity, 0-~100
# An artist match is a far stronger signal than a title match: it is
# what separates the real track from a covers compilation.
if norm(artist) and norm(artist) in qn:
s += 300
for w in DERIVATIVE:
if w in norm(title) and w not in qn:
s -= 150
scored.append((s, {
"deezer_id": t.get("id"),
"title": title,
"artist": artist,
"album": album,
"length": (t.get("duration") or 0) * 1000,
"cover": (t.get("album") or {}).get("cover_medium"),
# filled in at request time from /track/{id}
"year": None, "track": None,
}))
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 {"results": out[:limit]}
@app.post("/api/request")
async def request_track(body: RequestBody):
artist, title = body.artist, body.title
album, year, track = body.album, body.year, body.track
# Deezer's search payload has no track number or release date, and both go
# into the filename and tags. One lookup here beats one per search result.
if body.deezer_id:
try:
async with httpx.AsyncClient(timeout=15) as c:
r = await c.get(f"{DEEZER}/track/{body.deezer_id}")
if r.status_code == 200:
d = r.json()
track = d.get("track_position") or track
rd = d.get("release_date") or (d.get("album") or {}).get("release_date")
year = (rd or "")[:4] or year
album = (d.get("album") or {}).get("title") or album
artist = (d.get("artist") or {}).get("name") or artist
except Exception as e:
log.warning("deezer track lookup failed: %s", e)
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) VALUES (?,?,?,?,?,?,?,?,?,?)",
(job_id, "searching", "", artist, title, album,
year, track, str(body.deezer_id or ""), None))
asyncio.create_task(pick_and_enqueue(job_id, artist, title))
return {"id": job_id}
@app.get("/api/jobs")
async def jobs(limit: int = 50):
with db() as c:
rows = c.execute(
"SELECT * FROM jobs ORDER BY created_at DESC, rowid DESC LIMIT ?",
(limit,)).fetchall()
return {"jobs": [dict(r) for r in rows]}
@app.get("/api/health")
async def health():
ok, detail = True, {}
async with slskd_client() 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"] = str(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}
STATIC = Path(__file__).parent / "static"
@app.get("/")
async def index():
return FileResponse(STATIC / "index.html")
app.mount("/static", StaticFiles(directory=STATIC), name="static")