Add review, discover, email; rebuild the design system

Three features and a typographic pass over everything.

REVIEW. Picking a file and trusting it are different questions, so they now
use different numbers. The existing rank decides which candidate to fetch and
is right to weigh peer upload slots and queue length -- but none of that says
anything about whether the file is the correct recording, and a fast peer must
not be able to raise the confidence of a wrong match. So confidence() scores
the filename alone: title 0.75, artist 0.25, album +0.05, and a variant tag
you did not ask for multiplies by 0.8 AND caps below the auto threshold.

Confidence gates the IMPORT, not the download. The transfer completes either
way and the file waits in downloads, so approving is instant and rejecting is
a delete. Holding the transfer instead would mean the peer is usually gone by
the time anyone looks.

The two cases it exists for: "02 Creep.flac" in an unsorted folder scores 0.75
-- title exactly right, artist simply absent, the commonest honest case there
is -- and must be reviewed rather than binned. And "Creep (Live at
Glastonbury)" scores 0.80 with BOTH artist and title perfect; the cap is what
stops a live recording auto-filing over the studio version. Ask for "Creep
(Acoustic)" and it auto-imports at 0.97, because then the variant is the
request.

DISCOVER. Related artists seeded from what you have actually imported, plus a
recap of top artists, genres and decades. Built on muse's own request history,
never on listening data: Navidrome owns play counts and history, and a second
source of truth for those would disagree with the one you listen through.

EMAIL. Optional, off unless MUSE_SMTP_HOST is set. Sends on import and on
review-needed. smtplib is blocking so it runs off the event loop, and every
failure is logged and swallowed -- a notification is never worth failing an
import over. Google accounts need no configuration, since the identity is an
address. Each user owns their own setting under Settings, deliberately not an
admin route: an address you cannot change yourself is someone else's chore.

DESIGN. A real type scale (display/title/headline/body/callout/caption) whose
tracking tightens as size grows, which is the single biggest tell of typography
nobody set. Specular top edges on glass surfaces -- blur alone reads as a grey
rectangle. A two-part elevation scale, contact shadow plus soft light. Semantic
colours are now per-theme: the amber tuned for the dark background failed
contrast outright on white, which was exactly the text a reviewer most needs to
read. The tab underline became a sliding pill so the eye tracks one object, and
Review only appears when something is actually waiting.

Add tests/test_matching.py: 13 assertions on the confidence bands rather than
the exact scores. 90 tests pass across the three suites.

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

app.py

@@ -414,12 +414,17 @@ def init_db():
         for col, decl in (("requested_by", "TEXT"),
                           ("cover", "TEXT"),
                           ("genre", "TEXT"),
-                          ("duration", "INTEGER")):
+                          ("duration", "INTEGER"),
+                          ("confidence", "REAL"),
+                          ("match_why", "TEXT"),
+                          ("reviewed_by", "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()}
         for col, decl in (("storage_quota", "INTEGER NOT NULL DEFAULT 0"),
-                          ("invited_by", "TEXT")):
+                          ("invited_by", "TEXT"),
+                          ("email", "TEXT"),
+                          ("notify", "INTEGER NOT NULL DEFAULT 1")):
             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)")
@@ -491,6 +496,84 @@ def ext_of(remote_path: str) -> str:
     return remote_path.rsplit(".", 1)[-1].lower() if "." in remote_path else ""
 
 
+def confidence(filename: str, artist: str, title: str,
+               album: str | None = None) -> tuple[float, list[str]]:
+    """How sure are we this file is the song that was asked for, 0..1.
+
+    Deliberately separate from score(). score() decides *which* candidate to
+    fetch and is right to weigh peer speed and queue length; none of that says
+    anything about whether the file is the correct recording. Mixing the two
+    would let a fast peer raise the confidence of a wrong match.
+
+    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])
+    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()
+    if nt and nt in stem:
+        t_score, t_ratio = 1.0, max(t_ratio, 1.0)
+        why.append("title matches")
+    elif t_ratio >= 0.75:
+        t_score = t_ratio
+        why.append(f"title close ({t_ratio:.0%})")
+    else:
+        t_score = t_ratio
+        why.append(f"title differs ({t_ratio:.0%})")
+
+    # Artist may legitimately live in the path rather than the filename.
+    if na and na in stem:
+        a_score = 1.0
+        why.append("artist in filename")
+    elif na and na in path:
+        a_score = 0.9
+        why.append("artist in folder")
+    elif na and set(na.split()) & set(path.split()):
+        a_score = 0.5
+        why.append("artist partly matches")
+    else:
+        a_score = 0.0
+        why.append("artist not found in path")
+
+    # 0.75/0.25 rather than an even split. A file named exactly right but
+    # sitting in an unorganised folder is the commonest honest case there is,
+    # and it has to land in the review band, not below it — 0.65 would have
+    # binned it alongside genuinely wrong songs.
+    c = 0.75 * t_score + 0.25 * a_score
+
+    if album and norm(album) and norm(album) in path:
+        c = min(1.0, c + 0.05)
+        why.append("album in path")
+
+    # A live or remix version of the right song is still the wrong file when
+    # the request was for the studio recording, and that is exactly the case a
+    # reviewer needs to see rather than have silently imported. So the penalty
+    # caps confidence below AUTO instead of merely scaling it: "a human must
+    # decide" is the whole meaning of the flag, and a perfect artist+title
+    # match should not be able to auto-import past it.
+    variants = [w for w in ("live", "remix", "cover", "karaoke", "instrumental",
+                            "acoustic", "demo", "edit", "radio edit")
+                if re.search(rf"\b{w}\b", stem) and w not in nt]
+    if variants:
+        c = min(c * 0.8, AUTO_CONFIDENCE - 0.1)
+        why.append(f"looks like a {variants[0]} version, which was not requested")
+
+    return round(max(0.0, min(1.0, c)), 3), why
+
+
+# >= AUTO imports without asking. Below REVIEW the match is bad enough that a
+# human almost certainly rejects it, but it is still held rather than binned:
+# the file is already downloaded and only the reviewer knows if it is right.
+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:
     """Rank one remote file. None = reject.
 
@@ -521,7 +604,8 @@ def score(candidate: dict, peer: dict, want: str) -> float | None:
     return s
 
 
-async def pick_and_enqueue(job_id: str, artist: str, title: str):
+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}")
     search_id = str(uuid.uuid4())
@@ -578,9 +662,105 @@ async def pick_and_enqueue(job_id: str, artist: str, title: str):
             set_state(job_id, "failed", f"enqueue failed: {e}")
             return
 
+        # 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)
         set_state(job_id, "downloading",
                   f"from {peer['username']} (score {best_score:.0f})",
-                  username=peer["username"], remote_file=f["filename"], size=f["size"])
+                  username=peer["username"], remote_file=f["filename"],
+                  size=f["size"], confidence=conf, match_why="; ".join(why))
+
+
+# ─────────────────────────── email ───────────────────────────
+#
+# Optional and off unless SMTP_HOST is set. A homelab tool that cannot send
+# mail should still import music, so every failure here is logged and
+# swallowed — a notification is never worth failing an import over.
+
+SMTP_HOST = os.getenv("MUSE_SMTP_HOST", "").strip()
+SMTP_PORT = int(os.getenv("MUSE_SMTP_PORT", "587"))
+SMTP_USER = os.getenv("MUSE_SMTP_USER", "").strip()
+SMTP_PASS = os.getenv("MUSE_SMTP_PASS", "")
+SMTP_FROM = os.getenv("MUSE_SMTP_FROM", "").strip() or SMTP_USER
+SMTP_TLS = os.getenv("MUSE_SMTP_TLS", "starttls").strip().lower()
+MAIL_ON = bool(SMTP_HOST and SMTP_FROM)
+
+
+def _send_sync(to: str, subject: str, body: str):
+    import smtplib
+    from email.message import EmailMessage
+
+    msg = EmailMessage()
+    msg["From"], msg["To"], msg["Subject"] = SMTP_FROM, to, subject
+    msg.set_content(body)
+    if SMTP_TLS == "ssl":
+        s = smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, timeout=20)
+    else:
+        s = smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=20)
+    with s:
+        if SMTP_TLS == "starttls":
+            s.starttls()
+        if SMTP_USER:
+            s.login(SMTP_USER, SMTP_PASS)
+        s.send_message(msg)
+
+
+async def send_mail(to: str, subject: str, body: str):
+    """Fire-and-forget. smtplib is blocking, so it runs off the event loop —
+    otherwise a slow relay would stall the import worker for every job."""
+    if not (MAIL_ON and to):
+        return
+    try:
+        await asyncio.to_thread(_send_sync, to, subject, body)
+        log.info("mail: sent %r to %s", subject, to)
+    except Exception as e:
+        log.warning("mail: sending to %s failed: %s", to, errstr(e))
+
+
+def recipient(username: str | None) -> str | None:
+    """Where to write to, and whether they want to hear from us at all."""
+    if not username:
+        return None
+    with db() as c:
+        r = c.execute("SELECT email, notify, provider FROM users WHERE username=?",
+                      (username,)).fetchone()
+    if not r or not r["notify"]:
+        return None
+    # A Google identity IS an email address, so there is nothing to configure
+    # for those accounts unless they want mail somewhere else.
+    if r["email"]:
+        return r["email"]
+    return username if r["provider"] == "google" and "@" in username else None
+
+
+async def notify_imported(job, dest):
+    to = recipient(job["requested_by"])
+    if not to:
+        return
+    album = job["album"] or "Singles"
+    if job["year"]:
+        album += f" ({job['year']})"
+    await send_mail(
+        to, f"muse: {job['title']} is in your library",
+        f"{job['artist']} — {job['title']}\n{album}\n\n"
+        f"Filed as: {dest}\n\nIt should appear in Navidrome on the next scan.\n")
+
+
+async def notify_review(job):
+    to = recipient(job["requested_by"])
+    if not to:
+        return
+    await send_mail(
+        to, f"muse: {job['title']} needs a look",
+        f"{job['artist']} — {job['title']}\n\n"
+        f"The best match found was only {(job['confidence'] or 0):.0%} confident, "
+        f"so it has not been added to your library yet.\n\n"
+        f"Why: {job['match_why'] or 'unknown'}\n"
+        f"File: {job['remote_file']}\n\n"
+        f"Approve or reject it under Review in muse.\n")
 
 
 # ─────────────────────────── import ───────────────────────────
@@ -676,7 +856,8 @@ async def search_worker(n: int):
                 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"])
+                await pick_and_enqueue(job_id, job["artist"], job["title"],
+                                       job["album"])
         except Exception as e:
             log.warning("worker %d: %s", n, e)
             try:
@@ -716,15 +897,27 @@ async def import_worker():
                     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"]
+
+                    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.
+                        set_state(job["id"], "review",
+                                  f"{conf:.0%} confident — {job['match_why'] or ''}",
+                                  size=actual)
+                        await notify_review(job)
+                        continue
+
+                    dest = tag_and_file(job, src)
                     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}")
         except Exception as e:
@@ -1070,9 +1263,42 @@ async def me(request: Request):
         "quota": u["quota"], "storage_quota": u["storage_quota"],
         "usage": usage_for(u["username"]),
         "providers": ENABLED_PROVIDERS,
+        "email": u["email"], "notify": bool(u["notify"]),
+        "mail_configured": MAIL_ON,
     }
 
 
+class SettingsBody(BaseModel):
+    email: str | None = None
+    notify: bool | None = None
+
+
+EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s.]+\.[^@\s]+$")
+
+
+@app.patch("/api/me")
+async def update_me(body: SettingsBody, request: Request):
+    """Your own notification settings. Not an admin route on purpose — an
+    address you cannot change yourself is a support ticket."""
+    me = session_user(request)
+    if not me:
+        raise HTTPException(401, "unauthenticated")
+    sets, vals = [], []
+    if body.email is not None:
+        e = body.email.strip()
+        if e and not EMAIL_RE.match(e):
+            raise HTTPException(400, "that does not look like an email address")
+        sets.append("email=?"); vals.append(e or None)
+    if body.notify is not None:
+        sets.append("notify=?"); vals.append(1 if body.notify else 0)
+    if not sets:
+        raise HTTPException(400, "nothing to change")
+    with db() as c:
+        c.execute(f"UPDATE users SET {', '.join(sets)} WHERE username=?",
+                  [*vals, me])
+    return {"ok": True, "mail_configured": MAIL_ON}
+
+
 class RequestBody(BaseModel):
     artist: str
     title: str
@@ -1530,6 +1756,158 @@ async def import_list(body: ImportBody, request: Request):
     return {"queued": queued, "skipped": skipped, "pending": work_q.qsize()}
 
 
+class ReviewBody(BaseModel):
+    action: str          # approve | reject
+
+
+@app.get("/api/review")
+async def review_queue(request: Request):
+    """Downloads held back because the match was not convincing enough.
+
+    Scoped like /api/jobs: your own unless you are an admin, because a held
+    file is a record of what someone asked for.
+    """
+    me = session_user(request)
+    admin = user_role(request) == "admin"
+    with db() as c:
+        if admin:
+            rows = c.execute(
+                "SELECT * FROM jobs WHERE state='review' ORDER BY created_at DESC"
+            ).fetchall()
+        else:
+            rows = c.execute(
+                "SELECT * FROM jobs WHERE state='review' AND requested_by IS ? "
+                "ORDER BY created_at DESC", (me,)).fetchall()
+    return {"items": [{
+        "id": r["id"], "artist": r["artist"], "title": r["title"],
+        "album": r["album"], "year": r["year"], "cover": r["cover"],
+        "confidence": r["confidence"], "why": r["match_why"],
+        "file": (r["remote_file"] or "").replace("\\", "/").rsplit("/", 1)[-1],
+        "path": r["remote_file"], "size": r["size"],
+        "requested_by": r["requested_by"], "created_at": r["created_at"],
+    } for r in rows], "thresholds": {"auto": AUTO_CONFIDENCE,
+                                     "review": REVIEW_CONFIDENCE}}
+
+
+@app.post("/api/review/{job_id}")
+async def review_decide(job_id: str, body: ReviewBody, request: Request):
+    me = session_user(request)
+    admin = user_role(request) == "admin"
+    if body.action not in ("approve", "reject"):
+        raise HTTPException(400, "action must be approve or reject")
+
+    with db() as c:
+        job = c.execute("SELECT * FROM jobs WHERE id=? AND state='review'",
+                        (job_id,)).fetchone()
+    # 404 rather than 403 for someone else's job: a distinct error would
+    # confirm the id exists, same as /api/retry.
+    if not job or (not admin and job["requested_by"] != me):
+        raise HTTPException(404, "no such review item")
+
+    src = find_downloaded(job["remote_file"] or "")
+    if body.action == "reject":
+        if src:
+            try:
+                src.unlink()
+            except OSError as e:
+                log.warning("review: could not delete %s: %s", src, e)
+        set_state(job_id, "failed", "rejected in review", reviewed_by=me)
+        return {"ok": True, "state": "failed"}
+
+    if not src:
+        raise HTTPException(409, "the downloaded file is no longer on disk")
+    dest = tag_and_file(job, src)
+    set_state(job_id, "imported", str(dest), imported_path=str(dest),
+              reviewed_by=me)
+    log.info("review: %s approved %s", me, job_id[:8])
+    return {"ok": True, "state": "imported", "path": str(dest)}
+
+
+@app.get("/api/discover")
+async def discover(request: Request):
+    """What to listen to next, derived from what you have actually asked for.
+
+    Deliberately built on muse's own request history plus Deezer's related
+    artists, not on listening data — Navidrome owns play counts and history,
+    and duplicating them here would mean two disagreeing sources of truth.
+    """
+    me = session_user(request)
+    with db() as c:
+        top = c.execute(
+            "SELECT artist, COUNT(*) n FROM jobs "
+            "WHERE requested_by IS ? AND state='imported' "
+            "GROUP BY artist COLLATE NOCASE ORDER BY n DESC LIMIT 5",
+            (me,)).fetchall()
+    if not top:
+        return {"seeds": [], "artists": [], "tracks": []}
+
+    async def related(name: str) -> list[dict]:
+        try:
+            async with httpx.AsyncClient(timeout=15) as c:
+                s = await c.get(f"{DEEZER}/search/artist",
+                                params={"q": name, "limit": 1})
+                hits = (s.json() or {}).get("data") or []
+                if not hits:
+                    return []
+                r = await c.get(f"{DEEZER}/artist/{hits[0]['id']}/related",
+                                params={"limit": 6})
+                return (r.json() or {}).get("data") or []
+        except Exception as e:
+            log.info("discover: related for %r failed: %s", name, errstr(e))
+            return []
+
+    seeds = [r["artist"] for r in top]
+    groups = await asyncio.gather(*(related(s) for s in seeds))
+
+    have = {norm(r["artist"]) for r in top}
+    seen, artists = set(), []
+    for g in groups:
+        for a in g:
+            k = norm(a.get("name") or "")
+            if not k or k in have or k in seen:
+                continue
+            seen.add(k)
+            artists.append({"id": a.get("id"), "name": a.get("name"),
+                            "picture": a.get("picture_medium"),
+                            "fans": a.get("nb_fan")})
+    artists.sort(key=lambda a: -(a["fans"] or 0))
+    return {"seeds": seeds, "artists": artists[:18]}
+
+
+@app.get("/api/recap")
+async def recap(request: Request, days: int = Query(365, ge=7, le=3650)):
+    """Your year in muse — from request history, which muse does own."""
+    me = session_user(request)
+    since = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat(" ")[:19]
+    with db() as c:
+        base = ("FROM jobs WHERE requested_by IS ? AND created_at >= ? "
+                "AND state='imported'")
+        total = c.execute(f"SELECT COUNT(*) n, COALESCE(SUM(size),0) b {base}",
+                          (me, since)).fetchone()
+        artists = c.execute(
+            f"SELECT artist, COUNT(*) n {base} GROUP BY artist COLLATE NOCASE "
+            "ORDER BY n DESC LIMIT 10", (me, since)).fetchall()
+        genres = c.execute(
+            f"SELECT genre, COUNT(*) n {base} AND genre IS NOT NULL AND genre<>'' "
+            "GROUP BY genre ORDER BY n DESC LIMIT 8", (me, since)).fetchall()
+        decades = c.execute(
+            f"SELECT substr(year,1,3) d, COUNT(*) n {base} "
+            "AND year GLOB '[0-9][0-9][0-9][0-9]' GROUP BY d ORDER BY d",
+            (me, since)).fetchall()
+        busiest = c.execute(
+            f"SELECT substr(created_at,1,10) d, COUNT(*) n {base} "
+            "GROUP BY d ORDER BY n DESC LIMIT 1", (me, since)).fetchone()
+    return {
+        "days": days,
+        "tracks": total["n"], "bytes": total["b"],
+        "artists": [{"name": r["artist"], "n": r["n"]} for r in artists],
+        "genres": [{"name": r["genre"], "n": r["n"]} for r in genres],
+        "decades": [{"decade": f"{r['d']}0s", "n": r["n"]} for r in decades],
+        "busiest_day": ({"date": busiest["d"], "n": busiest["n"]}
+                        if busiest and busiest["n"] else None),
+    }
+
+
 @app.get("/api/jobs")
 async def jobs(request: Request, limit: int = Query(200, ge=1, le=1000),
                mine: bool = False):
@@ -1602,7 +1980,7 @@ def user_dict(r: sqlite3.Row) -> dict:
         "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,
+        "usage": u, "email": r["email"], "notify": bool(r["notify"]),
         # 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,