diff --git a/.claude/launch.json b/.claude/launch.json new file mode 100644 index 0000000..6387b67 --- /dev/null +++ b/.claude/launch.json @@ -0,0 +1,11 @@ +{ + "version": "0.0.1", + "configurations": [ + { + "name": "muse", + "runtimeExecutable": "bash", + "runtimeArgs": ["dev.sh"], + "port": 5041 + } + ] +} diff --git a/.gitignore b/.gitignore index f6c9e57..282e3de 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,8 @@ config/ .venv/ .DS_Store dev/ + +# Local diagnostics: contain the live instance's Soulseek username, +# server paths and peer names. Not for a public repo. +FINDINGS.md +SERVER_AUDIT.md diff --git a/README.md b/README.md index 2e2ca6c..da4e7c1 100644 --- a/README.md +++ b/README.md @@ -148,8 +148,9 @@ automatically. Every job shows what it *resolved to*, so a bad auto-pick is visible in the list rather than quietly landing the wrong song. Lines that match nothing are reported back as skipped instead of failing the batch. -Failed jobs get a **Retry** button. Most Soulseek failures are the peer going -offline, not a bad request, so retrying later usually just works. +Failed jobs get a **Retry** button, but far fewer jobs reach that state now: +a request that loses its peer falls through to the next candidate on its own. +See "When a peer says no" below. ### Everything goes through a bounded queue @@ -179,10 +180,83 @@ is needed to catch `Come As You Are (Live)`, but substring matching would also penalise `Deliver Me` and `Olive`. And a term only counts against a result when you didn't type it, so `oasis live forever` and `nirvana unplugged` still work. -**Picking the file (Soulseek).** Format preference first, then whether the peer -can actually serve you (free upload slot, short queue), then filename -plausibility, then bitrate. A marginally better file behind 200 queued users -loses to a good-enough one available now. +**Picking the file (Soulseek).** Whether the peer can actually serve you comes +first — a free upload slot and a short queue — then format preference, then +filename plausibility, then bitrate. A marginally better file behind 200 queued +users loses to a good-enough one available now. + +That order is deliberate and was once the other way round, which was a bug +worth naming: format outranked availability, so a FLAC behind a five-thousand +deep queue always beat an MP3 that was free to download immediately. Whether +the peer can serve you is the single most predictive signal for whether the +job ever finishes, so nothing outranks it. + +## Searching for a name the file doesn't use + +Two things go wrong between "what Deezer calls this track" and "what a stranger +named the file", and both used to end the request. + +**The Soulseek server filters some artist names.** This is measurable: on a +healthy, logged-in client, `Michael Jackson Billie Jean` and bare +`Michael Jackson` both return **zero** responses, while `Billie Jean` returns +over a thousand files. The same is true of The Weeknd, Kanye West and Frank +Ocean. Sending one query of `artist title` — which is all muse used to do — +means every request for a filtered artist fails reporting "searched 0 files", +indistinguishable from a track nobody shares. + +So muse tries a **ladder** of queries, loosest last, and stops at the first +rung that yields enough candidates: + +``` +Michael Jackson Billie Jean 0 responses +Billie Jean 251 responses, 1083 files +``` + +The rungs drop, in order: a parenthetical suffix (`Novacane (Explicit Version)`), +a `feat.` credit, a leading article, everything after a `/` in a split title, +and finally the artist itself. Since most requests are answered by the first +query, the extra searches cost nothing in the common case — `MUSE_MAX_QUERIES` +caps how far it will go. + +Dropping the artist from the *query* does not mean ignoring it. The artist is +then matched against the whole **folder path**, which is where a well-organised +share puts it anyway: + +``` +anjya music/Michael Jackson/1982 - Thriller [flac]/06 - Billie Jean.flac + ^^^^^^^^^^^^^^^^ artist found here +``` + +**Names differ in punctuation and accents.** Normalisation folds accents away +before comparing, so `Beyoncé` matches a file called `Beyonce`, `Björk` matches +`Bjork`, and `Don't Stop` matches `Dont Stop`. Apostrophes are removed rather +than turned into a space; `&` still separates, so `R&B` stays two tokens +instead of collapsing into the unrelated word `rb`. Track numbers and release +tags (`02 - `, `[320kbps]`) are stripped before the comparison, and a +reordered filename (`Creep - Radiohead`) scores the same as `Radiohead - Creep`. + +The title has to stand on its own: a matching artist cannot carry a filename +whose title is wrong. Without that floor, a strong artist match let a +*different song by the right artist* through. + +## When a peer says no + +`Transfer rejected: File not shared.` is the commonest failure on Soulseek — +the peer's share index is stale and the file it advertised is gone. It is a +condition of one *file*, not of the request, and the right response is to take +the next candidate. + +muse keeps the whole ranked list (`MUSE_CANDIDATES`, default 8) on the job row +rather than only the winner, so when a transfer is rejected, times out, errors, +or slskd cannot open a connection to the peer at all, the next candidate is +enqueued instead of the request failing. The list is ordered to put a different +peer next, so one offline user cannot burn three attempts in a row. Only when +the list is exhausted — or `MUSE_MAX_ATTEMPTS` is reached — does the job fail, +and it then reports every peer it tried and what each one said. + +A download that sits with no bytes for `MUSE_STALL_MINUTES` is abandoned the +same way, as is one slskd has forgotten about. Both used to leave the job in +`downloading` forever, with a cheerful pulsing dot and no log line. ## Review: when muse isn't sure @@ -470,7 +544,14 @@ stat -c "%h" "/data/media/music// ()/ - .flac" | `MUSE_LIBRARY_DIR` | `/data/media/music` | where the library lives | | `MUSE_DB` | `/config/muse.db` | job history | | `MUSE_FORMATS` | `flac,mp3` | ordered preference; anything else is rejected | -| `MUSE_MIN_MATCH` | `0.45` | filename-to-title similarity floor | +| `MUSE_MIN_MATCH` | `0.45` | filename-to-track similarity floor | +| `MUSE_TITLE_FLOOR` | `0.5` | the title alone must clear this before the artist counts | +| `MUSE_CANDIDATES` | `8` | ranked candidates kept per request, for fallback | +| `MUSE_MAX_ATTEMPTS` | `5` | peers tried before a request is given up on | +| `MUSE_STALL_MINUTES` | `20` | a transfer moving no bytes for this long is abandoned | +| `MUSE_MAX_QUERIES` | `4` | rungs of the search ladder to try | +| `MUSE_MIN_CANDIDATES` | `3` | below this, escalate to the next query | +| `MUSE_SEARCH_RETRIES` | `3` | attempts at issuing a search before failing | | `MUSE_AUTO_CONFIDENCE` | `0.95` | at or above this, import without asking | | `MUSE_REVIEW_CONFIDENCE` | `0.70` | below this, held but probably wrong | | `MUSE_WORKERS` | `3` | concurrent Soulseek searches | diff --git a/app.py b/app.py index 846f2fc..9e8b646 100644 --- a/app.py +++ b/app.py @@ -26,13 +26,16 @@ file on disk, two names — the same trick the *arrs use. import asyncio import base64 +import collections import difflib import hashlib +import json import logging import os import re import secrets import sqlite3 +import unicodedata import uuid from contextlib import asynccontextmanager from datetime import datetime, timedelta, timezone @@ -42,7 +45,7 @@ from urllib.parse import urlencode import httpx from fastapi import FastAPI, HTTPException, Query, Request from fastapi.responses import (FileResponse, HTMLResponse, JSONResponse, - RedirectResponse) + RedirectResponse, Response) from fastapi.staticfiles import StaticFiles from pydantic import BaseModel @@ -68,6 +71,35 @@ 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")) +# The title alone must clear this before the artist is even consulted. Without +# a floor of its own, a file by the right artist scored well enough on the +# combined ratio to be fetched no matter which of their songs it actually was. +TITLE_FLOOR = float(os.getenv("MUSE_TITLE_FLOOR", "0.5")) + +# How many ranked candidates to keep per job. When a transfer fails, the next +# one is tried instead of failing the request -- "File not shared." is a +# per-file condition, so candidate two usually just works. +CANDIDATES = int(os.getenv("MUSE_CANDIDATES", "8")) +MAX_ATTEMPTS = int(os.getenv("MUSE_MAX_ATTEMPTS", "5")) + +# A download that has moved no bytes for this long is not going to. The peer +# queued it and vanished, or slskd lost the transfer; either way the job used +# to sit in `downloading` forever behind a cheerful pulsing dot. +STALL_MINUTES = float(os.getenv("MUSE_STALL_MINUTES", "20")) + +# Query variants to try before giving up. Each costs a real Soulseek search, +# so this escalates only when a search underdelivers -- see search_queries(). +MAX_QUERIES = int(os.getenv("MUSE_MAX_QUERIES", "4")) +MIN_CANDIDATES = int(os.getenv("MUSE_MIN_CANDIDATES", "3")) + +# Issuing a search is a POST that occasionally blips. Nine production failures +# were one slskd hiccup during a bulk import, every queued job dying inside the +# same second, each one permanently. +SEARCH_RETRIES = int(os.getenv("MUSE_SEARCH_RETRIES", "3")) + +# Soulseek results trickle in: what you have at t=5s is much worse than t=15s. +SEARCH_POLLS = int(os.getenv("MUSE_SEARCH_POLLS", "20")) + # How many Soulseek searches may run at once. # # This is a hard cap, not a suggestion. Each search takes 10-40s of polling @@ -177,7 +209,8 @@ 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"} +PUBLIC_PATHS = {"/healthz", "/auth/login", "/auth/callback", "/auth/logout", + "/login.css"} ROLES = ("admin", "user", "guest") @@ -334,6 +367,28 @@ SAFE = re.compile(r'[<>:"/\\|?*\x00-\x1f]') NOISE = re.compile(r"[^a-z0-9]+") YEAR = re.compile(r"^\d{4}$") +# Apostrophes vanish rather than becoming a separator, so "Don't" normalises +# onto a filename's "Dont". Everything else non-alphanumeric becomes a space, +# so "R&B" stays two tokens instead of collapsing to the unrelated word "rb". +QUOTES = re.compile("['\u2018\u2019\u02bc\u0060\"\u201c\u201d]") + +# Letters NFKD leaves alone because they are atomic rather than base + accent. +FOLD = str.maketrans({"\u00f8": "o", "\u00e6": "ae", "\u0153": "oe", + "\u00f0": "d", "\u00fe": "th", "\u0142": "l", + "\u00df": "ss", "\u0111": "d", "\u0131": "i"}) + +# A leading track number: "02 - ", "02.", "A1 ". Anchored and bounded so it +# cannot eat a title that merely opens with a digit ("1979", "99 Problems"). +TRACKNO = re.compile(r"^\s*(?:[a-d]?\d{1,3})\s*(?:[-._)\]]\s*)+", re.I) + +# Technical release tags. Deliberately NOT "live", "remix" or "acoustic": +# confidence() reads those to decide whether this is the recording that was +# asked for, and stripping them here would blind it. +JUNK = re.compile( + r"[\[({]\s*(?:\d{2,4}\s*k(?:bps)?|v\d|web|cd\s*\d*|vinyl|flac|mp3|" + r"lossless|hi-?res|\d{2,3}\s*bit|\d{2}\s*khz)[^\])}]*[\])}]" + r"|\b(?:\d{3}\s*kbps|web-?dl|cdrip|vbr|cbr)\b", re.I) + # Most filesystems cap a path component at 255 bytes. Track titles can be far # longer than that once a stranger has appended their own notes, and the # resulting ENAMETOOLONG would fail the import after the download had finished. @@ -365,7 +420,71 @@ def safe_year(year) -> str | None: def norm(s: str) -> str: - return NOISE.sub(" ", s.lower()).strip() + """Lowercase, fold accents away, and reduce to alphanumeric tokens. + + The accent fold is load-bearing. Deezer returns "Beyonc\u00e9", "Bj\u00f6rk" and + "Sigur R\u00f3s"; the strangers who named the files on Soulseek wrote + "Beyonce", "Bjork", "Sigur Ros". Without it the accented character is not + in [a-z0-9] and became a *space*, so "beyonce" was compared against + "beyonc " -- a containment miss and a mediocre ratio, on a track that was + sitting right there in the results. + """ + s = unicodedata.normalize("NFKD", str(s).lower().translate(FOLD)) + s = "".join(c for c in s if not unicodedata.combining(c)) + return NOISE.sub(" ", QUOTES.sub("", s)).strip() + + +def bare_name(remote_path: str) -> str: + """A remote filename with extension, track number and release tags gone. + + All of that is decoration a peer added that says nothing about which + recording the file holds. Leaving it in drags the similarity ratio down on + files that are in fact exactly right. + """ + base = remote_path.replace("\\", "/").rsplit("/", 1)[-1] + if "." in base: + base = base.rsplit(".", 1)[0] + return JUNK.sub(" ", TRACKNO.sub("", base)).strip() + + +def similarity(want: str, got: str) -> float: + """Best of a sequence ratio and a token-coverage ratio, 0..1. + + difflib alone is order-sensitive: "creep radiohead" against "radiohead + creep" scores far below what a human would call a match, and Soulseek + filenames put the fields in whatever order the ripper felt like. The token + measure is order-free, so taking the better of the two absorbs reordering + and interleaved junk without having to loosen MIN_MATCH. + """ + if not want or not got: + return 0.0 + seq = difflib.SequenceMatcher(None, want, got).ratio() + a, b = set(want.split()), set(got.split()) + return max(seq, len(a & b) / len(a) if a else 0.0) + + +def match_ratio(remote_path: str, artist: str, title: str) -> float: + """How well a remote path matches the track asked for, 0..1. + + Title and artist are measured separately and the title has to stand on its + own. A single ratio over the blob "artist title" let a strong artist match + carry a weak title -- the same failure the Deezer ranking keeps a dedicated + term for, where `Oasis - Live Forever` once resolved to `Cloudburst` + because the artist agreed and nothing else had to. + + The artist is looked for in the whole path rather than the filename: a file + at `Radiohead/Pablo Honey/02 Creep.flac` names its artist in the folder, + which is the commonest well-organised case there is. + """ + stem = norm(bare_name(remote_path)) + path = norm(remote_path.replace("\\", "/")) + nt, na = norm(title), norm(artist) + + t = 1.0 if (nt and nt in stem) else similarity(nt, stem) + if t < TITLE_FLOOR: + return 0.0 # wrong song; a matching artist must not rescue it + a = 1.0 if (na and na in path) else similarity(na, path) + return 0.75 * t + 0.25 * a # ─────────────────────────── storage ─────────────────────────── @@ -382,7 +501,7 @@ def init_db(): c.execute(""" CREATE TABLE IF NOT EXISTS jobs ( id TEXT PRIMARY KEY, - state TEXT NOT NULL, -- searching|queued|downloading|imported|failed + state TEXT NOT NULL, -- queued|searching|downloading|review|imported|failed detail TEXT, artist TEXT, title TEXT, album TEXT, year TEXT, track INTEGER, @@ -426,7 +545,15 @@ def init_db(): ("duration", "INTEGER"), ("confidence", "REAL"), ("match_why", "TEXT"), - ("reviewed_by", "TEXT")): + ("reviewed_by", "TEXT"), + # The candidate ladder. Keeping the runners-up on the + # row is what lets a rejected transfer fall through to + # the next peer minutes later, in a different task, + # after the search result set is long gone. + ("candidates", "TEXT"), + ("tried", "TEXT"), + ("attempts", "INTEGER NOT NULL DEFAULT 0"), + ("started_at", "TEXT")): 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()} @@ -521,15 +648,14 @@ def confidence(filename: str, artist: str, title: str, Returns the score and the human-readable reasons behind it, because a number alone gives the reviewer nothing to act on. """ - base = filename.replace("\\", "/").rsplit("/", 1)[-1] - stem = norm(base.rsplit(".", 1)[0]) + stem = norm(bare_name(filename)) path = norm(filename.replace("\\", "/")) na, nt = norm(artist), norm(title) why: list[str] = [] # Title is the load-bearing signal: a wrong title is a wrong song, whereas # a missing artist is routine in files organised into per-artist folders. - t_ratio = difflib.SequenceMatcher(None, nt, stem).ratio() + t_ratio = similarity(nt, stem) if nt and nt in stem: t_score, t_ratio = 1.0, max(t_ratio, 1.0) why.append("title matches") @@ -587,104 +713,310 @@ AUTO_CONFIDENCE = float(os.getenv("MUSE_AUTO_CONFIDENCE", "0.95")) REVIEW_CONFIDENCE = float(os.getenv("MUSE_REVIEW_CONFIDENCE", "0.70")) -def score(candidate: dict, peer: dict, want: str) -> float | None: +def score(candidate: dict, peer: dict, artist: str, title: 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. + Ordering intent: actually-gettable first, then correct format, then + filename plausibility, then bitrate. + + Availability leads deliberately. It used to sit *below* format -- 100 + points per format rank against 60 for a free upload slot -- which meant a + FLAC behind two hundred queued people always beat an MP3 available right + now. Whether the peer can serve you is the single most predictive signal + for whether the job ever finishes, so it outranks the preference for a + better file. Format is the tiebreak within an availability tier. """ 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) + ratio = match_ratio(candidate["filename"], artist, title) 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 += 220 if peer.get("hasFreeUploadSlot") else 0 + # Unsaturated: the old `max(0, 40 - queueLength)` scored a 40-deep queue + # and a 5000-deep one identically, which is the difference between waiting + # a minute and never being served at all. + s += 120 / (1 + peer.get("queueLength", 0) / 8) + s += min(peer.get("uploadSpeed", 0) / 50_000, 20) + s += (len(FORMATS) - FORMATS.index(ext)) * 60 # format preference + s += ratio * 90 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, - album: str | None = None): - """Search slskd for one track and queue the best candidate.""" - want = norm(f"{artist} {title}") +PAREN = re.compile(r"[\[(\{][^\])}]*[\])}]") +FEAT = re.compile(r"\b(?:feat|ft|featuring)\b\.?.*$", re.I) +ARTICLE = re.compile(r"^(?:the|a|an|los|las|les)\s+", re.I) + + +class SearchUnavailable(Exception): + """slskd could not accept a search. Transient, not a bad request.""" + + +def search_queries(artist: str, title: str) -> list[str]: + """Query variants to try, loosest last. + + The Soulseek server filters searches containing certain artist names. That + is measured, not theorised: on this network "Michael Jackson Billie Jean" + and bare "Michael Jackson" both return zero responses, while "Billie Jean" + returns 1083 files. Sending only `artist title` -- all this used to do -- + means every request for a filtered artist dies reporting "searched 0 + files", and that is exactly the roster of names in the failed jobs: + Michael Jackson, The Weeknd, Kanye West, Frank Ocean. + + So the ladder ends at the title alone, and match_ratio() then recognises + the artist from the folder path instead of from the query. The middle + rungs handle the ordinary case where Deezer's metadata is simply more + decorated than the filename a stranger typed. + """ + out: list[str] = [] + seen: set[str] = set() + + def add(*parts: str) -> None: + q = " ".join(p for p in parts if p).strip() + # Dedupe on the normalised form, so "Creep (Remastered)" and "Creep" + # are recognised as one query once the punctuation is gone. + k = norm(q) + if q and k and k not in seen: + seen.add(k) + out.append(q) + + plain_t = PAREN.sub(" ", title).strip() or title + lean_t = FEAT.sub(" ", plain_t).strip() or plain_t + head_t = lean_t.split("/")[0].strip() or lean_t # "A / B" -> "A" + lean_a = ARTICLE.sub("", FEAT.sub(" ", artist).strip()).strip() + + add(artist, title) # what was sent before, and still best when it works + add(artist, lean_t) # decoration and featuring credits gone + add(lean_a, head_t) # leading article gone, split title + add(head_t) # title alone -- this is the rung that beats the filter + add(lean_t) + return out[:MAX_QUERIES] + + +async def run_search(sc: httpx.AsyncClient, text: str) -> dict: + """Issue one Soulseek search and poll it to completion. + + The search is deleted only once it has actually finished. Deleting one + still InProgress makes slskd throw an optimistic-concurrency error inside + its own SearchService and lose the record; the old code deleted + unconditionally after its 40s budget, so every slow search corrupted a row. + """ search_id = str(uuid.uuid4()) - async with slskd_client() as sc: + for attempt in range(SEARCH_RETRIES): try: r = await sc.post("/api/v0/searches", - json={"id": search_id, "searchText": f"{artist} {title}"}) + json={"id": search_id, "searchText": text}) r.raise_for_status() + break except Exception as e: - set_state(job_id, "failed", f"slskd search failed: {e}") - return + if attempt == SEARCH_RETRIES - 1: + raise SearchUnavailable(errstr(e)) from e + await asyncio.sleep(2 * (attempt + 1)) - # 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 + results: dict = {} + finished = False + for _ in range(SEARCH_POLLS): + 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 as e: + # Swallowed silently before, which made a broken slskd look + # identical to a track nobody happens to share. + log.debug("poll %r: %s", text, errstr(e)) + continue + if str(results.get("state", "")).startswith("Completed"): + finished = True + 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. + if finished: 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 + log.debug("cleanup %r: %s", text, errstr(e)) + else: + log.info("search %r unfinished after %ds, leaving it for slskd to close", + text, SEARCH_POLLS * 2) + return results - # Confidence is recorded now, while the candidate list is still in - # hand, but it gates the IMPORT rather than the download. Holding the - # transfer instead would mean the peer is often gone by the time - # someone reviews it; this way the file is already on disk, approving - # is instant, and rejecting is a delete. - conf, why = confidence(f["filename"], artist, title, album) + +def spread_by_peer(ranked: list[tuple[float, dict, dict]]) -> list[dict]: + """Order candidates so consecutive attempts land on different peers. + + Straight score order routinely stacks three files from one well-stocked + user at the top; if that user is offline all three attempts burn on the + same dead connection. Peers arrive best-first, so round-robining them keeps + the strongest candidate first and still guarantees variety behind it. + """ + by_peer: dict[str, list[dict]] = {} + for s, peer, f in ranked: + by_peer.setdefault(peer["username"], []).append({ + "username": peer["username"], + "filename": f["filename"], + "size": f.get("size") or 0, + "score": round(s, 1), + }) + queues = list(by_peer.values()) + out: list[dict] = [] + while queues: + for q in list(queues): + out.append(q.pop(0)) + if not q: + queues.remove(q) + return out + + +def no_candidate_detail(seen: int, stats: "collections.Counter", + queries: list[str]) -> str: + """Say which filter emptied the results, not merely that they are empty. + + One message used to cover four unrelated situations, and 28 of 29 real + occurrences turned out to be the search itself coming back empty -- which + is now stated outright instead of being reported as a matching failure. + """ + if seen == 0: + tried = ", ".join(repr(q) for q in queries) + word = "y" if len(queries) == 1 else "ies" + return f"Soulseek returned nothing for {len(queries)} quer{word}: {tried}" + bits = [] + if stats["format"]: + bits.append(f"{stats['format']} not {'/'.join(FORMATS)}") + if stats["weak"]: + bits.append(f"{stats['weak']} did not match the title") + if stats["locked"]: + bits.append(f"{stats['locked']} locked") + return f"{seen} files found, none usable: " + ", ".join(bits) + + +def exhausted_detail(tried: list[dict], attempts: int) -> str: + """Summarise the whole ladder instead of overwriting it with the last error.""" + if not tried: + return "no candidate could be enqueued" + reasons = collections.Counter(t["reason"].split(":")[0] for t in tried) + plural = "" if attempts == 1 else "s" + joined = ", ".join(f"{n}x {r}" for r, n in reasons.most_common(3)) + return f"tried {attempts} peer{plural}: {joined}" + + +async def try_next_candidate(job_id: str, artist: str, title: str, + album: str | None = None) -> bool: + """Enqueue the next-best candidate. False once the ladder is exhausted. + + This is the difference between a request that survives one bad peer and one + that does not. "Transfer rejected: File not shared." is a *per-file* + condition -- the peer's share index is stale -- so candidate two almost + always works, and slskd's 500 on an unreachable peer is the same shape. + """ + with db() as c: + row = c.execute( + "SELECT candidates, tried, attempts FROM jobs WHERE id=?", + (job_id,)).fetchone() + if not row: + return False + cands = json.loads(row["candidates"] or "[]") + tried = json.loads(row["tried"] or "[]") + attempts = row["attempts"] or 0 + + while cands and attempts < MAX_ATTEMPTS: + cand = cands.pop(0) + attempts += 1 + try: + async with slskd_client() as sc: + er = await sc.post( + f"/api/v0/transfers/downloads/{cand['username']}", + json=[{"filename": cand["filename"], + "size": cand["size"]}]) + er.raise_for_status() + except Exception as e: + tried.append({"username": cand["username"], + "filename": cand["filename"], + "reason": f"could not reach peer: {errstr(e)}"}) + continue + + conf, why = confidence(cand["filename"], artist, title, album) + with db() as c: + c.execute( + "UPDATE jobs SET candidates=?, tried=?, attempts=? WHERE id=?", + (json.dumps(cands), json.dumps(tried), attempts, job_id)) + extra = f", attempt {attempts}" if attempts > 1 else "" set_state(job_id, "downloading", - f"from {peer['username']} (score {best_score:.0f})", - username=peer["username"], remote_file=f["filename"], - size=f["size"], confidence=conf, match_why="; ".join(why)) + f"from {cand['username']} (score {cand['score']:.0f}{extra})", + username=cand["username"], remote_file=cand["filename"], + size=cand["size"], confidence=conf, match_why="; ".join(why), + started_at=datetime.now(timezone.utc).isoformat()) + return True + + with db() as c: + c.execute("UPDATE jobs SET candidates=?, tried=?, attempts=? WHERE id=?", + (json.dumps(cands), json.dumps(tried), attempts, job_id)) + set_state(job_id, "failed", exhausted_detail(tried, attempts)) + return False + + +async def pick_and_enqueue(job_id: str, artist: str, title: str, + album: str | None = None): + """Search Soulseek for one track and queue the best candidate. + + Builds a ranked candidate *list* rather than a single winner, so a peer + that rejects, times out or cannot be reached costs one attempt instead of + the entire request. + """ + queries = search_queries(artist, title) + pool: dict[tuple[str, str], tuple[float, dict, dict]] = {} + seen = 0 + stats: "collections.Counter" = collections.Counter() + + async with slskd_client() as sc: + for i, q in enumerate(queries): + try: + results = await run_search(sc, q) + except SearchUnavailable as e: + set_state(job_id, "failed", + f"slskd would not accept a search: {e}") + return + + for peer in results.get("responses", []): + for f in peer.get("files", []): + seen += 1 + if f.get("isLocked"): + stats["locked"] += 1 + continue + if ext_of(f["filename"]) not in FORMATS: + stats["format"] += 1 + continue + sc_ = score(f, peer, artist, title) + if sc_ is None: + stats["weak"] += 1 + continue + key = (peer["username"], f["filename"]) + if key not in pool or pool[key][0] < sc_: + pool[key] = (sc_, peer, f) + + log.info("job %s q%d/%d %r -> %d responses, %d files, %d candidates", + job_id[:8], i + 1, len(queries), q, + len(results.get("responses", [])), + results.get("fileCount", 0), len(pool)) + # Escalate only on underdelivery: the common case stays one search, + # which is what keeps a 50-track paste civil to slskd. + if len(pool) >= MIN_CANDIDATES: + break + + if not pool: + set_state(job_id, "failed", no_candidate_detail(seen, stats, queries)) + return + + ranked = sorted(pool.values(), key=lambda r: -r[0]) + cands = spread_by_peer(ranked)[:CANDIDATES] + with db() as c: + c.execute("UPDATE jobs SET candidates=?, tried=?, attempts=0 WHERE id=?", + (json.dumps(cands), "[]", job_id)) + await try_next_candidate(job_id, artist, title, album) # ─────────────────────────── email ─────────────────────────── @@ -867,14 +1199,23 @@ def tag_and_file(job: sqlite3.Row, src: Path) -> Path: 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 +def downloaded_index() -> dict[str, Path]: + """Every completed file, by basename. + + slskd nests a download under the remote's parent directory name, so the + basename is the only reliable key. Built once per poll cycle and shared by + every job in it: the previous version walked the whole tree separately for + each in-flight job, every ten seconds, which on a large downloads + directory is a lot of stat() calls to answer the same question. + """ + out: dict[str, Path] = {} + try: + for p in DOWNLOAD_DIR.rglob("*"): + if p.is_file(): + out.setdefault(p.name, p) + except OSError as e: + log.warning("cannot walk %s: %s", DOWNLOAD_DIR, errstr(e)) + return out # ─────────────────────────── search queue ─────────────────────────── @@ -903,8 +1244,64 @@ async def search_worker(n: int): work_q.task_done() +def transfer_index(payload: list) -> dict[str, dict]: + """Flatten slskd's user -> directories -> files nesting. + + Keeps the whole file object, not just its state. `exception` is where the + literal "Transfer rejected: File not shared." lives, and discarding it was + why a failed job could only ever say "transfer Completed, Rejected" with no + reason a person could act on. + """ + out: dict[str, dict] = {} + for user in payload: + for d in user.get("directories", []): + for f in d.get("files", []): + out[f.get("filename", "")] = f + return out + + +def stalled_for(job: sqlite3.Row) -> float: + """Minutes since this transfer last had anything to show for itself.""" + started = job["started_at"] if "started_at" in job.keys() else None + if not started: + return 0.0 + try: + t = datetime.fromisoformat(started) + except ValueError: + return 0.0 + if t.tzinfo is None: + t = t.replace(tzinfo=timezone.utc) + return (datetime.now(timezone.utc) - t).total_seconds() / 60.0 + + +async def abandon(job: sqlite3.Row, reason: str) -> None: + """Give up on the current peer and fall through to the next candidate.""" + with db() as c: + row = c.execute("SELECT tried FROM jobs WHERE id=?", + (job["id"],)).fetchone() + tried = json.loads((row["tried"] if row else None) or "[]") + tried.append({"username": job["username"], "filename": job["remote_file"], + "reason": reason}) + with db() as c: + c.execute("UPDATE jobs SET tried=? WHERE id=?", + (json.dumps(tried), job["id"])) + log.info("job %s giving up on %s: %s", job["id"][:8], job["username"], reason) + await try_next_candidate(job["id"], job["artist"], job["title"], job["album"]) + + +async def cancel_transfer(username: str, filename: str) -> None: + """Ask slskd to drop a transfer we have stopped waiting for.""" + try: + async with slskd_client(timeout=PROBE_TIMEOUT) as sc: + await sc.delete(f"/api/v0/transfers/downloads/{username}", + params={"filename": filename}) + except Exception as e: + log.debug("cancel %s/%s: %s", username, filename, errstr(e)) + + async def import_worker(): """Poll slskd for in-flight downloads and import the finished ones.""" + missing: collections.Counter = collections.Counter() while True: await asyncio.sleep(10) try: @@ -912,28 +1309,36 @@ async def import_worker(): jobs = c.execute( "SELECT * FROM jobs WHERE state='downloading'").fetchall() if not jobs: + missing.clear() continue async with slskd_client() as sc: r = await sc.get("/api/v0/transfers/downloads") r.raise_for_status() - payload = r.json() + index = transfer_index(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", "")) + # One directory walk per cycle, shared by every job. This used to be + # a full recursive rglob per stuck job, every ten seconds, forever. + done_files: dict[str, Path] | None = None for job in jobs: - st = states.get(job["remote_file"], "") + t = index.get(job["remote_file"]) + st = str((t or {}).get("state", "")) + if "Completed, Succeeded" in st: - src = find_downloaded(job["remote_file"]) + if done_files is None: + done_files = downloaded_index() + src = done_files.get( + job["remote_file"].replace("\\", "/").rsplit("/", 1)[-1]) if not src: - continue # still being moved out of incomplete - # Record the real size, not the size the peer advertised — - # storage quotas are computed from this column. + # Normally just still being moved out of incomplete, but + # unbounded before: a misconfigured DOWNLOAD_DIR meant + # polling until the heat death of the process. + if stalled_for(job) > STALL_MINUTES: + set_state(job["id"], "failed", + "downloaded but the file never appeared in " + f"{DOWNLOAD_DIR} -- check MUSE_DOWNLOAD_DIR") + continue try: actual = src.stat().st_size except OSError: @@ -941,11 +1346,11 @@ async def import_worker(): conf = job["confidence"] if conf is not None and conf < AUTO_CONFIDENCE: - # Hold it. The file stays in downloads, untagged and - # unlinked, until someone approves or rejects it. + band = ("almost certainly wrong" + if conf < REVIEW_CONFIDENCE else "worth a look") set_state(job["id"], "review", - f"{conf:.0%} confident — {job['match_why'] or ''}", - size=actual) + f"{conf:.0%} confident, {band} -- " + f"{job['match_why'] or ''}", size=actual) await notify_review(job) continue @@ -953,10 +1358,36 @@ async def import_worker(): set_state(job["id"], "imported", str(dest), imported_path=str(dest), size=actual) await notify_imported(job, dest) + elif "Completed" in st and "Succeeded" not in st: - set_state(job["id"], "failed", f"transfer {st}") + # Every terminal non-success used to end the request here. + # "File not shared." is per-file and per-peer, so the only + # sane response is to try the next candidate. + why = (t or {}).get("exception") or st + await abandon(job, str(why)[:120]) + + elif t is None: + # Absent from slskd entirely: it restarted, or the transfer + # was cleared. Tolerate a couple of polls in case it is a + # race, then stop waiting for something nobody is doing. + missing[job["id"]] += 1 + if missing[job["id"]] >= 3: + del missing[job["id"]] + await abandon(job, "slskd forgot the transfer") + + else: + missing.pop(job["id"], None) + # Queued behind a wall, or connected and sending nothing. + # There was no clock on this at all, which is how a job sat + # in `downloading` for days behind a cheerful pulsing dot. + if not (t or {}).get("bytesTransferred") \ + and stalled_for(job) > STALL_MINUTES: + place = (t or {}).get("placeInQueue") + where = f" at #{place} in the queue" if place else "" + await cancel_transfer(job["username"], job["remote_file"]) + await abandon(job, f"no bytes in {STALL_MINUTES:.0f}m{where}") except Exception as e: - log.warning("worker: %s", e) + log.warning("import worker: %s", errstr(e)) # ─────────────────────────── api ─────────────────────────── @@ -1094,20 +1525,91 @@ async def healthz(): 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 - -
-
muse
-
Sign in to continue
-{buttons}
""" +# The sign-in page is the first screen anyone sees, and for a while it was the +# only one built out of hardcoded hex with no relation to the rest of the app. +# It cannot use the Next bundle -- /_next/* sits behind the session on purpose, +# so an unauthenticated visitor never receives it -- so it gets one small +# stylesheet of its own, served from a public path. `style-src 'self'` in the +# CSP permits exactly that; an inline