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

Rebuild the UI on Next.js, and give search real metadata

Browse Source 
The old page was hand-written HTML/CSS/JS. That was fine for a search box and
a list, but the roadmap here is multi-user, an admin dashboard with usage
graphs, and an import-review queue — all of which get painful in vanilla DOM
code. Moves to Next.js 16 / React 19 / Tailwind 4 / Framer Motion.

Built as a STATIC EXPORT served by FastAPI. Running `next start` alongside
uvicorn would add a Node runtime, a second process to supervise, and a second
thing that can be down while the other is up. There is no use for SSR — every
view is behind auth and renders from live JSON — so the deployment stays one
container, one process, one healthcheck. Image is 60MB; Node only exists in
the build stage.

Search now carries what a music UI needs:

- year and genre, grafted on from /album/{id} concurrently, deduplicated by
  album and cached forever since a release date is immutable
- 30-second previews, so you can hear a track before committing a Soulseek
  search to it. Not a library player — that is Navidrome's job.
- autocomplete over songs/albums/artists, debounced and aborted per keystroke
  so a slow response to "rad" cannot land after "radiohead"
- album and artist pages, with `Get all` to queue a release in one request
- the accent colour is sampled from the artwork, so the palette follows the
  music. Deezer's CDN sends ACAO:*, which is what makes canvas sampling work
  with no image proxy.

/api/album/{id} deliberately calls Deezer twice: the track list embedded in
the album response has no track_position, and using the list index instead
misnumbers every multi-disc release — verified against the White Album, where
disc 2 track 1 would have been filed as 18.

Schema is multi-user ready ahead of the feature: users table with roles and
quotas, requested_by on jobs, and the first login becomes admin. Auth is still
the single-user GitHub allowlist.

CSP survives all of this intact. Next emits inline hydration scripts, which
script-src 'self' blocks; rather than per-request nonces, their SHA-256 is
computed at startup from the built page — a static export never varies between
requests, so hashes need no maintenance and no HTML rewriting. The one
concession is style-src-attr 'unsafe-inline', because Framer Motion writes its
pre-hydration state into style="" and without it every element flashes in.
That permits style attributes only. img-src/media-src now name Deezer's CDN
explicitly instead of allowing https: wholesale, closing an exfiltration
channel that survived connect-src 'self'. /_next/* sits behind the session
too — there is no local login page to style, so the bundle need not be public.

Verified: unauthenticated /→302 and every /api/*→401 including the four new
endpoints, forged state→400, forged cookie→401, misconfigured auth refuses to
start, and the production image hydrates with zero CSP violations.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Aryan SinghandClaude Opus 5 committed 2026-08-04 14:26:31 -05:00
1 parent bc46fd80b3
commit 3b4b68daaf
34 files changed
+10655 -281
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

app.py
+389 -41
View File
@@ -25,6 +25,7 @@ file on disk, two names — the same trick the *arrs use.
"""
import asyncio
import base64
import difflib
import hashlib
import logging
@@ -201,8 +202,15 @@ def session_user(request: Request) -> str | None:
return None
# Re-check the allowlist on every request, so removing someone from
# MUSE_ALLOWED_USERS locks them out immediately instead of whenever their
# session happens to expire.
return row["username"] if row["username"].lower() in ALLOWED_USERS else None
# session happens to expire. Same for the per-user disable flag: a
# revocation that only took effect on expiry would be a revocation in name
# only.
if row["username"].lower() not in ALLOWED_USERS:
return None
with db() as c:
u = c.execute("SELECT disabled FROM users WHERE username=?",
(row["username"],)).fetchone()
return None if u and u["disabled"] else row["username"]
SAFE = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
@@ -250,6 +258,49 @@ def init_db():
expires_at TEXT NOT NULL
)
""")
# Users exist as rows even while the allowlist is a single name in an
# env var. Roles and quotas are read from here from the start, so
# opening this up to other people later is an INSERT and a config
# change rather than a schema migration against live job history.
c.execute("""
CREATE TABLE IF NOT EXISTS users (
username TEXT PRIMARY KEY COLLATE NOCASE,
provider TEXT NOT NULL DEFAULT 'github',
role TEXT NOT NULL DEFAULT 'user', -- admin|user|guest
display TEXT,
avatar TEXT,
quota INTEGER NOT NULL DEFAULT 0, -- requests/month, 0 = unlimited
disabled INTEGER NOT NULL DEFAULT 0,
created_at TEXT DEFAULT (datetime('now')),
last_seen TEXT
)
""")
# Added after the first release, so existing databases need it grafted
# on rather than recreated -- job history is the whole point of the file.
cols = {r["name"] for r in c.execute("PRAGMA table_info(jobs)").fetchall()}
for col, decl in (("requested_by", "TEXT"),
("cover", "TEXT"),
("genre", "TEXT"),
("duration", "INTEGER")):
if col not in cols:
c.execute(f"ALTER TABLE jobs ADD COLUMN {col} {decl}")
c.execute("CREATE INDEX IF NOT EXISTS jobs_by_user ON jobs (requested_by)")
c.execute("CREATE INDEX IF NOT EXISTS jobs_by_state ON jobs (state)")
def upsert_user(username: str, avatar: str = "", display: str = "") -> sqlite3.Row:
"""Record the sign-in. The first user to ever log in becomes admin --
there is no bootstrap ceremony and no default password to forget."""
with db() as c:
first = c.execute("SELECT COUNT(*) n FROM users").fetchone()["n"] == 0
c.execute(
"INSERT INTO users (username,provider,role,display,avatar,last_seen) "
"VALUES (?,?,?,?,?,datetime('now')) "
"ON CONFLICT(username) DO UPDATE SET "
" last_seen=datetime('now'), avatar=excluded.avatar, "
" display=COALESCE(NULLIF(excluded.display,''), users.display)",
(username, "github", "admin" if first else "user", display, avatar))
return c.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
def set_state(job_id: str, state: str, detail: str = "", **cols):
@@ -525,6 +576,56 @@ async def lifespan(_: FastAPI):
t.cancel()
# ─────────────────────── static front end / CSP ───────────────────────
#
# The UI is a Next.js static export: `next build` writes plain files into
# web/out and this process serves them, so the deployment stays one container,
# one process, one healthcheck, with no Node runtime in the image.
#
# That export contains inline <script> blocks (React's hydration payload),
# which `script-src 'self'` blocks outright. The usual fix is a per-request
# nonce, but a STATIC export has a better option: the scripts never change
# between requests, so their SHA-256 can go straight into the policy. Hashing
# them at startup means the CSP needs no per-request work, no HTML rewriting,
# and no maintenance when the bundle changes -- rebuild and the hashes follow.
WEB = Path(__file__).parent / "web" / "out"
_INLINE_SCRIPT = re.compile(rb"<script(?![^>]*\bsrc=)[^>]*>(.*?)</script>", re.S)
def script_hashes() -> list[str]:
out: set[str] = set()
for page in ("index.html", "404.html"):
f = WEB / page
if not f.exists():
continue
for body in _INLINE_SCRIPT.findall(f.read_bytes()):
digest = base64.b64encode(hashlib.sha256(body).digest()).decode()
out.add(f"'sha256-{digest}'")
return sorted(out)
SCRIPT_HASHES = " ".join(script_hashes())
# Framer Motion writes its pre-hydration `initial` state into style="" on the
# server-rendered markup, so those attributes have to be allowed or every
# animated element flashes in at full opacity before hydration corrects it.
#
# style-src-attr is the narrow lever for that: it permits style ATTRIBUTES
# while style-src 'self' still blocks injected <style> blocks and remote
# stylesheets. It is not a free pass either -- CSS-based exfiltration goes
# through url(), and img-src below already limits where those can point.
CSP = (
"default-src 'self'; "
f"script-src 'self' {SCRIPT_HASHES}; "
"style-src 'self'; style-src-attr 'unsafe-inline'; "
"img-src 'self' data: https://*.dzcdn.net https://*.githubusercontent.com; "
"media-src 'self' https://*.dzcdn.net; "
"connect-src 'self'; font-src 'self'; object-src 'none'; "
"frame-ancestors 'none'; base-uri 'none'; form-action 'self'"
)
app = FastAPI(
title="muse",
lifespan=lifespan,
@@ -538,8 +639,13 @@ app = FastAPI(
async def guard(request: Request, call_next):
"""Require a session for everything except PUBLIC_PATHS, and stamp
security headers on every response."""
# /_next/* is NOT public. Unlike the old hand-written page there is no
# local login screen to style — an unauthenticated visitor is redirected
# straight to GitHub and never receives our HTML — so the bundle can sit
# behind the session too, and not hand out a map of the app to anyone who
# finds the URL. It loads fine after the callback, cookie in hand.
path = request.url.path
if AUTH_MODE == "off" or path in PUBLIC_PATHS or path.startswith("/static/"):
if AUTH_MODE == "off" or path in PUBLIC_PATHS:
resp = await call_next(request)
elif session_user(request):
resp = await call_next(request)
@@ -549,15 +655,12 @@ async def guard(request: Request, call_next):
else:
resp = RedirectResponse("/auth/login", status_code=302)
# Strict CSP is why the stylesheet and script live in separate files rather
# than inline in index.html: with 'self' only and no 'unsafe-inline', an
# injected <script> cannot execute. img-src allows Deezer's cover CDN over
# https; connect-src is self only, so nothing can exfiltrate to a third
# party from this page.
resp.headers["Content-Security-Policy"] = (
"default-src 'self'; img-src 'self' https: data:; "
"style-src 'self'; script-src 'self'; connect-src 'self'; "
"frame-ancestors 'none'; base-uri 'none'; form-action 'self'")
# See CSP above. img-src and media-src name Deezer's CDN explicitly rather
# than allowing https: wholesale — under a blanket https: any injected
# markup could beacon to an arbitrary host by requesting an image from it,
# which is a working exfiltration channel even with connect-src locked to
# 'self'. Artwork comes from dzcdn.net and avatars from githubusercontent.
resp.headers["Content-Security-Policy"] = CSP
resp.headers["X-Content-Type-Options"] = "nosniff"
resp.headers["X-Frame-Options"] = "DENY"
resp.headers["Referrer-Policy"] = "no-referrer"
@@ -619,7 +722,8 @@ async def auth_callback(request: Request, code: str = "", state: str = ""):
"Accept": "application/vnd.github+json"})
if u.status_code != 200:
return JSONResponse({"error": "github user lookup failed"}, status_code=401)
login = (u.json() or {}).get("login") or ""
profile = u.json() or {}
login = profile.get("login") or ""
if login.lower() not in ALLOWED_USERS:
# Log the attempt: on an internet-facing box this is the signal that
@@ -627,7 +731,13 @@ async def auth_callback(request: Request, code: str = "", state: str = ""):
log.warning("auth: DENIED github user %r (not in MUSE_ALLOWED_USERS)", login)
return JSONResponse({"error": "not authorised"}, status_code=403)
log.info("auth: signed in %s", login)
row = upsert_user(login, profile.get("avatar_url") or "",
profile.get("name") or "")
if row and row["disabled"]:
log.warning("auth: DENIED %r (account disabled)", login)
return JSONResponse({"error": "account disabled"}, status_code=403)
log.info("auth: signed in %s (%s)", login, row["role"] if row else "user")
resp = RedirectResponse("/", status_code=302)
_set_cookie(resp, SESSION_COOKIE, new_session(login), SESSION_DAYS * 86400)
resp.delete_cookie(STATE_COOKIE, path="/")
@@ -647,7 +757,13 @@ async def auth_logout(request: Request):
@app.get("/api/me")
async def me(request: Request):
return {"user": session_user(request), "auth": AUTH_MODE}
who = session_user(request)
if not who:
return {"user": None, "auth": AUTH_MODE}
with db() as c:
u = c.execute("SELECT username,role,display,avatar,quota FROM users "
"WHERE username=?", (who,)).fetchone()
return {"user": who, "auth": AUTH_MODE, **(dict(u) if u else {"role": "user"})}
class RequestBody(BaseModel):
@@ -657,6 +773,90 @@ class RequestBody(BaseModel):
year: str | None = None
track: int | None = None
deezer_id: int | None = None
cover: str | None = None
genre: str | None = None
length: int | None = None
# Album metadata (release year, genre, label) is not in the track search
# payload, so it costs one extra call per distinct album. Cached forever on
# purpose: a release date and a genre are immutable, so there is nothing to
# invalidate, and repeat searches for the same artist stop hitting Deezer at
# all. Trimmed only to stop it growing without bound on a long-lived process.
_album_cache: dict[int, dict] = {}
_album_lock = asyncio.Semaphore(6) # Deezer rate-limits ~50 req/5s per IP
async def album_meta(album_id) -> dict:
if not album_id:
return {}
key = int(album_id)
if key in _album_cache:
return _album_cache[key]
async with _album_lock:
if key in _album_cache: # filled while we waited for the slot
return _album_cache[key]
try:
async with httpx.AsyncClient(timeout=15) as c:
r = await c.get(f"{DEEZER}/album/{key}")
d = r.json() if r.status_code == 200 else {}
except Exception as e:
log.debug("album %s lookup failed: %s", key, e)
return {}
genres = [g.get("name") for g in (d.get("genres") or {}).get("data", []) if g.get("name")]
meta = {
"year": (d.get("release_date") or "")[:4] or None,
"genre": genres[0] if genres else None,
"genres": genres,
"label": d.get("label"),
"nb_tracks": d.get("nb_tracks"),
"record_type": d.get("record_type"),
}
if len(_album_cache) > 5000:
for k in list(_album_cache)[:1000]:
_album_cache.pop(k, None)
_album_cache[key] = meta
return meta
def track_row(t: dict) -> dict:
"""One Deezer track object -> the shape the UI and the job table use."""
artist = t.get("artist") or {}
album = t.get("album") or {}
return {
"deezer_id": t.get("id"),
"title": t.get("title") or "",
"artist": artist.get("name") or "",
"artist_id": artist.get("id"),
"artist_pic": artist.get("picture_medium"),
"album": album.get("title") or "",
"album_id": album.get("id"),
"cover": album.get("cover_medium"),
"cover_big": album.get("cover_big") or album.get("cover_medium"),
"length": (t.get("duration") or 0) * 1000,
# 30-second clip from Deezer's CDN. This is what lets you confirm you
# picked the right recording BEFORE committing a Soulseek search to it
# -- the single most useful thing in the payload for avoiding a wrong
# import, and it costs nothing extra.
"preview": t.get("preview") or None,
"explicit": bool(t.get("explicit_lyrics")),
"isrc": t.get("isrc"),
"rank": t.get("rank") or 0,
"year": None, "genre": None, "track": None,
}
async def add_album_meta(rows: list[dict]) -> list[dict]:
"""Fill year and genre on a result list, concurrently and de-duplicated."""
ids = {r["album_id"] for r in rows if r.get("album_id")}
if not ids:
return rows
metas = dict(zip(ids, await asyncio.gather(*(album_meta(i) for i in ids))))
for r in rows:
m = metas.get(r.get("album_id")) or {}
r["year"] = r["year"] or m.get("year")
r["genre"] = r["genre"] or m.get("genre")
return rows
async def deezer_search(q: str, limit: int = 25) -> list[dict]:
@@ -674,7 +874,6 @@ async def deezer_search(q: str, limit: int = 25) -> list[dict]:
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
@@ -702,16 +901,7 @@ async def deezer_search(q: str, limit: int = 25) -> list[dict]:
# 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,
"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,
}))
scored.append((s, track_row(t)))
seen, out = set(), []
for _, item in sorted(scored, key=lambda x: -x[0]):
@@ -725,20 +915,142 @@ async def deezer_search(q: str, limit: int = 25) -> list[dict]:
@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)}
"""Track search. Returns individual songs, not albums — the whole point.
Year and genre are grafted on afterwards because Deezer's track search
omits both. Enrichment happens only here, not in deezer_search itself, so
the bulk-import path stays one HTTP call per line.
"""
return {"results": await add_album_meta(await deezer_search(q, limit))}
@app.get("/api/suggest")
async def suggest(q: str):
"""Autocomplete. Deliberately does NOT enrich or re-rank: this fires on
every few keystrokes, so it has to be one upstream call and nothing else.
Deezer's own autocomplete ordering is good enough for a dropdown, and the
real ranking still happens when the search is actually submitted."""
q = q.strip()
if len(q) < 2:
return {"tracks": [], "albums": [], "artists": []}
try:
async with httpx.AsyncClient(timeout=8) as c:
r = await c.get(f"{DEEZER}/search/autocomplete",
params={"q": q, "limit": 5})
d = r.json() if r.status_code == 200 else {}
except Exception:
return {"tracks": [], "albums": [], "artists": []}
def rows(key):
return (d.get(key) or {}).get("data", []) or []
return {
"tracks": [{"deezer_id": t.get("id"), "title": t.get("title"),
"artist": (t.get("artist") or {}).get("name"),
"cover": (t.get("album") or {}).get("cover_small"),
"preview": t.get("preview")}
for t in rows("tracks")[:5]],
"albums": [{"album_id": a.get("id"), "title": a.get("title"),
"artist": (a.get("artist") or {}).get("name"),
"cover": a.get("cover_small"),
"nb_tracks": a.get("nb_tracks")}
for a in rows("albums")[:3]],
"artists": [{"artist_id": a.get("id"), "name": a.get("name"),
"cover": a.get("picture_small"),
"nb_album": a.get("nb_album")}
for a in rows("artists")[:3]],
}
@app.get("/api/album/{album_id}")
async def album_detail(album_id: int):
"""Full track list for an album, so 'Get all' can queue it in one go."""
# Two calls, not one. /album/{id} embeds a track list, but those nested
# objects have no track_position -- and taking the list index instead would
# misnumber every multi-disc release, writing "13 - ..." for disc 2 track 1.
# /album/{id}/tracks is the only source that carries track_position and
# disk_number, and those numbers end up in the filename and the tags.
try:
async with httpx.AsyncClient(timeout=20) as c:
r, tr = await asyncio.gather(
c.get(f"{DEEZER}/album/{album_id}"),
c.get(f"{DEEZER}/album/{album_id}/tracks", params={"limit": 300}))
except Exception as e:
raise HTTPException(502, f"deezer: {e}")
if r.status_code != 200:
raise HTTPException(404, "album not found")
d = r.json()
artist = (d.get("artist") or {}).get("name") or ""
year = (d.get("release_date") or "")[:4] or None
genres = [g.get("name") for g in (d.get("genres") or {}).get("data", []) if g.get("name")]
raw = (tr.json().get("data", []) if tr.status_code == 200
else (d.get("tracks") or {}).get("data", []))
tracks = []
for t in raw:
row = track_row(t)
# These nested track objects carry no album of their own, so inherit
# what the parent already told us rather than looking it up again.
row.update(album=d.get("title") or "", album_id=d.get("id"),
cover=d.get("cover_medium"), cover_big=d.get("cover_big"),
artist=row["artist"] or artist, year=year,
genre=genres[0] if genres else None,
track=t.get("track_position"),
disc=t.get("disk_number"))
tracks.append(row)
return {"album": {"album_id": d.get("id"), "title": d.get("title"),
"artist": artist, "artist_id": (d.get("artist") or {}).get("id"),
"cover_big": d.get("cover_big"), "cover": d.get("cover_medium"),
"year": year, "genres": genres, "label": d.get("label"),
"nb_tracks": d.get("nb_tracks"),
"record_type": d.get("record_type"),
"duration": d.get("duration")},
"tracks": tracks}
@app.get("/api/artist/{artist_id}")
async def artist_detail(artist_id: int):
"""Artist header plus their top tracks and albums."""
try:
async with httpx.AsyncClient(timeout=20) as c:
info, top, albums = await asyncio.gather(
c.get(f"{DEEZER}/artist/{artist_id}"),
c.get(f"{DEEZER}/artist/{artist_id}/top", params={"limit": 15}),
c.get(f"{DEEZER}/artist/{artist_id}/albums", params={"limit": 24}))
except Exception as e:
raise HTTPException(502, f"deezer: {e}")
if info.status_code != 200:
raise HTTPException(404, "artist not found")
a = info.json()
tracks = [track_row(t) for t in (top.json().get("data", []) if top.status_code == 200 else [])]
await add_album_meta(tracks)
return {
"artist": {"artist_id": a.get("id"), "name": a.get("name"),
"picture": a.get("picture_big"), "nb_fan": a.get("nb_fan"),
"nb_album": a.get("nb_album")},
"tracks": tracks,
"albums": [{"album_id": x.get("id"), "title": x.get("title"),
"cover": x.get("cover_medium"), "cover_big": x.get("cover_big"),
"year": (x.get("release_date") or "")[:4] or None,
"record_type": x.get("record_type"),
"nb_tracks": x.get("nb_tracks")}
for x in (albums.json().get("data", []) if albums.status_code == 200 else [])],
}
async def enqueue_job(artist: str, title: str, album=None, year=None,
track=None, deezer_id=None) -> str:
track=None, deezer_id=None, requested_by=None,
cover=None, genre=None, duration=None) -> str:
"""Create a job and hand it to the search queue."""
job_id = str(uuid.uuid4())
with db() as c:
c.execute(
"INSERT INTO jobs (id,state,detail,artist,title,album,year,track,"
"recording_mbid,release_mbid) VALUES (?,?,?,?,?,?,?,?,?,?)",
"recording_mbid,release_mbid,requested_by,cover,genre,duration) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
(job_id, "queued", "", artist, title, album, year, track,
str(deezer_id or ""), None))
str(deezer_id or ""), None, requested_by, cover, genre, duration))
await work_q.put(job_id)
return job_id
@@ -767,12 +1079,36 @@ async def enrich(deezer_id) -> dict:
@app.post("/api/request")
async def request_track(body: RequestBody):
async def request_track(body: RequestBody, request: Request):
e = await enrich(body.deezer_id)
return {"id": await enqueue_job(
e.get("artist") or body.artist, body.title,
e.get("album") or body.album, e.get("year") or body.year,
e.get("track") or body.track, body.deezer_id)}
e.get("track") or body.track, body.deezer_id,
requested_by=session_user(request), cover=body.cover,
genre=body.genre, duration=body.length)}
class BatchBody(BaseModel):
tracks: list[RequestBody]
@app.post("/api/request/batch")
async def request_batch(body: BatchBody, request: Request):
"""Queue a whole album at once.
Unlike /api/request this skips the per-track /track/{id} enrichment: the
caller got these rows from /api/album/{id}, which already carries the
album, year and track position for every one of them. Re-fetching would be
nb_tracks extra calls to learn what we were just told.
"""
who = session_user(request)
ids = [await enqueue_job(t.artist, t.title, t.album, t.year, t.track,
t.deezer_id, requested_by=who, cover=t.cover,
genre=t.genre, duration=t.length)
for t in body.tracks]
log.info("batch: queued %d track(s) for %s", len(ids), who or "anon")
return {"queued": len(ids), "ids": ids, "pending": work_q.qsize()}
class ImportBody(BaseModel):
@@ -803,7 +1139,7 @@ async def expand_url(kind: str, ident: str) -> list[str]:
@app.post("/api/import")
async def import_list(body: ImportBody):
async def import_list(body: ImportBody, request: Request):
"""Paste a list, get it all queued.
Each line is resolved against Deezer and the top-ranked hit is taken
@@ -839,8 +1175,9 @@ async def import_list(body: ImportBody):
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,
h["deezer_id"], requested_by=session_user(request),
cover=h.get("cover"), duration=h.get("length"))
queued.append({"id": jid, "line": line, "cover": h.get("cover"),
"matched": f"{h['artist']} — {h['title']}"})
log.info("import: %d queued, %d skipped", len(queued), len(skipped))
@@ -889,12 +1226,23 @@ async def health():
return {"ok": ok and detail["library_writable"], **detail}
STATIC = Path(__file__).parent / "static"
@app.get("/")
async def index():
return FileResponse(STATIC / "index.html")
f = WEB / "index.html"
if not f.exists():
raise HTTPException(500, "UI not built — run `npm run build` in web/")
return FileResponse(f)
app.mount("/static", StaticFiles(directory=STATIC), name="static")
# Next.js emits its immutable, content-hashed chunks under /_next/static.
#
# Guarded rather than mounted unconditionally so that importing this module
# does not require build artifacts: CI typechecks the API before the UI has
# been built, and tests should not have to run npm first. A deploy that really
# is missing its UI says so at startup and again on the first request, rather
# than silently serving 404s.
if (WEB / "_next").is_dir():
app.mount("/_next", StaticFiles(directory=WEB / "_next"), name="next")
else:
log.warning("web/out/_next is missing — the UI was not built. "
"API endpoints work; the web page will not.")