Add bulk import, and a bounded queue to make it safe

Requesting songs one at a time does not scale past a handful, so this adds a
paste-a-list import. It accepts "Artist - Title", en dashes, "Title by
Artist", numbered lists, bare titles, and Deezer album/playlist URLs, which
get expanded into their tracks.

The queue is the load-bearing part. Every request previously spawned its own
asyncio task, so a 50-track paste would have launched 50 concurrent Soulseek
searches -- each 10-40s of polling and real load on slskd -- and drowned the
client. Work now goes through an asyncio.Queue drained by MUSE_WORKERS
consumers, default 3. Requests return immediately and the batch drains at a
civil rate. The queue is in memory, so jobs left 'queued' or 'searching' are
re-queued at startup rather than stranded by a restart.

Ranking needed two fixes found by testing against the real API:

Import was calling deezer_search(limit=1), which fetches two rows, so the
ranker was choosing between a demo and a rehearsal take while the studio
version -- six times the popularity -- sat unseen further down. It now pulls
a real candidate pool.

More seriously, the Deezer ranker had no title-relevance term at all: only
artist match, popularity and penalties. A different song by the right artist
therefore competed on equal footing, and "Oasis - Live Forever" resolved to
"Cloudburst" as soon as a variant penalty tipped the balance. Title overlap
now dominates, and the variant penalty is small enough only to break ties
between versions of the same song.

Derivative terms match on word boundaries rather than substrings, so a bare
"live" catches "Come As You Are (Live)" without penalising "Deliver Me" or
"Olive", and a term is ignored when the user typed it -- "oasis live forever"
still works.

Verified against live slskd: six tracks pasted, six resolved to the correct
song, five imported with correct album/year/track and link count 2, in about
two minutes. The sixth returned an empty Soulseek search and is retryable
from the UI.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

app.py

@@ -62,6 +62,16 @@ FORMATS = [f.strip() for f in os.getenv("MUSE_FORMATS", "flac,mp3").split(",") i
 # 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.
 #
@@ -79,8 +89,18 @@ MIN_MATCH = float(os.getenv("MUSE_MIN_MATCH", "0.45"))
 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")
+# 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")
 
 SAFE = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
 NOISE = re.compile(r"[^a-z0-9]+")
@@ -305,7 +325,32 @@ def find_downloaded(remote_file: str) -> Path | None:
     return None
 
 
-async def worker():
+# ─────────────────────────── 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)
@@ -347,11 +392,25 @@ async def worker():
 @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))
+
+    # 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
-    task.cancel()
+    for t in tasks:
+        t.cancel()
 
 
 app = FastAPI(title="muse", lifespan=lifespan)
@@ -366,11 +425,10 @@ class RequestBody(BaseModel):
     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."""
+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 {"results": []}
+        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:
@@ -385,13 +443,31 @@ async def search(q: str, limit: int = 25):
         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.
+
+        # 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
-        for w in DERIVATIVE:
-            if w in norm(title) and w not in qn:
+
+        # 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, {
             "deezer_id": t.get("id"),
             "title": title,
@@ -410,48 +486,157 @@ async def search(q: str, limit: int = 25):
             continue
         seen.add(key)
         out.append(item)
-    return {"results": out[:limit]}
+    return 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
+@app.get("/api/search")
+async def search(q: str, limit: int = 25):
+    """Track search. Returns individual songs, not albums — the whole point."""
+    return {"results": await deezer_search(q, limit)}
 
-    # 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)
 
+async def enqueue_job(artist: str, title: str, album=None, year=None,
+                      track=None, deezer_id=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) 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}
+            (job_id, "queued", "", artist, title, album, year, track,
+             str(deezer_id or ""), None))
+    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):
+    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)}
+
+
+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):
+    """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])
+
+    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"])
+        queued.append({"id": jid, "line": line,
+                       "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(limit: int = 50):
+async def jobs(limit: int = 200):
     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]}
+        counts = {r["state"]: r["n"] for r in c.execute(
+            "SELECT state, COUNT(*) n FROM jobs GROUP BY state").fetchall()}
+    return {"jobs": [dict(r) for r in rows], "counts": counts,
+            "pending": work_q.qsize()}
+
+
+@app.post("/api/retry/{job_id}")
+async def retry(job_id: str):
+    """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."""
+    with db() as c:
+        row = c.execute("SELECT id FROM jobs WHERE id=?", (job_id,)).fetchone()
+    if not row:
+        raise HTTPException(404, "no such job")
+    set_state(job_id, "queued", "")
+    await work_q.put(job_id)
+    return {"ok": True}
 
 
 @app.get("/api/health")