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

Import a shared playlist link and build it in Navidrome

Browse Source 
Paste a Spotify, Apple Music or Deezer playlist URL; muse resolves it, matches
every track through the same Deezer path a pasted line takes, and writes an
.m3u the library scanner picks up.

Written per import rather than at the end, so the playlist appears in Navidrome
with the first track that lands and grows as the rest arrive -- twenty minutes
of nothing, then everything, is the worse shape. A track rejected in review
drops out on the next rewrite, and one that never resolves is reported instead
of silently vanishing.

Resolve is split from queue on purpose: a shared link is opaque, you cannot
tell twelve tracks from four hundred by looking at it, and four hundred is a
quota decision.

Two of the three sources are scrapes and the code says so plainly. Deezer has a
real API. Spotify and Apple Music have no free way to read a public playlist --
Apple's MusicKit needs a paid developer account -- so those read the public page
and parse the JSON it already embeds for its own player. No key, no OAuth, no
account, and no guarantee: they break whenever either company reshuffles their
frontend. So the resolvers fail loudly and name themselves, because an empty
playlist and a changed page must not produce the same message.

The .m3u uses MUSE_PLAYLIST_PREFIX (/music), not muse's own library path. Both
containers mount the same directory at different points and a playlist written
with muse's view resolves to nothing on Navidrome's side.

Also reorders the hooks in search-bar so the URL-seeding effect is declared
after the state it sets, rather than closing over a binding declared below it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Aryan SinghandClaude Opus 5 committed 2026-09-02 15:40:29 -05:00
1 parent e9fba3bb1d
commit 4aaf3e0998
7 files changed
+815 -19
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

app.py
+347 -5
View File
@@ -596,9 +596,29 @@ def init_db():
("candidates", "TEXT"),
("tried", "TEXT"),
("attempts", "INTEGER NOT NULL DEFAULT 0"),
("started_at", "TEXT")):
("started_at", "TEXT"),
# Which shared playlist this track came from, and
# where it sits in it -- an m3u in shuffled order is
# not the playlist anyone sent you.
("playlist_id", "TEXT"),
("playlist_seq", "INTEGER")):
if col not in cols:
c.execute(f"ALTER TABLE jobs ADD COLUMN {col} {decl}")
c.execute("""
CREATE TABLE IF NOT EXISTS playlists (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
source TEXT NOT NULL, -- deezer|spotify|apple
source_url TEXT,
requested_by TEXT,
tracks_total INTEGER NOT NULL DEFAULT 0,
tracks_written INTEGER NOT NULL DEFAULT 0,
m3u_path TEXT,
created_at TEXT DEFAULT (datetime('now'))
)
""")
c.execute("CREATE INDEX IF NOT EXISTS jobs_by_playlist "
"ON jobs (playlist_id)")
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"),
@@ -1534,6 +1554,10 @@ async def import_worker():
dest = tag_and_file(job, src)
set_state(job["id"], "imported", str(dest),
imported_path=str(dest), size=actual)
# Rewritten per track, so the playlist appears in Navidrome
# with the first arrival rather than the last.
if playlist_of(job):
write_playlist(job["playlist_id"])
await notify_imported(job, dest)
elif "Completed" in st and "Succeeded" not in st:
@@ -2300,7 +2324,8 @@ async def artist_detail(artist_id: int):
async def enqueue_job(artist: str, title: str, album=None, year=None,
track=None, deezer_id=None, requested_by=None,
cover=None, genre=None, duration=None) -> str:
cover=None, genre=None, duration=None,
playlist_id=None, playlist_seq=None) -> str:
"""Create a job and hand it to the search queue."""
job_id = str(uuid.uuid4())
# The single choke point for all three request paths, so normalise here.
@@ -2313,10 +2338,12 @@ async def enqueue_job(artist: str, title: str, album=None, year=None,
with db() as c:
c.execute(
"INSERT INTO jobs (id,state,detail,artist,title,album,year,track,"
"recording_mbid,release_mbid,requested_by,cover,genre,duration) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
"recording_mbid,release_mbid,requested_by,cover,genre,duration,"
"playlist_id,playlist_seq) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
(job_id, "queued", "", artist, title, album, year, track,
str(deezer_id or ""), None, requested_by, cover, genre, duration))
str(deezer_id or ""), None, requested_by, cover, genre, duration,
playlist_id, playlist_seq))
await work_q.put(job_id)
return job_id
@@ -2398,6 +2425,218 @@ MAX_IMPORT_LINES = 500
MAX_IMPORT_TRACKS = 1000
# ─────────────────────────── playlists ───────────────────────────
#
# Paste a playlist link from wherever it was shared and get the tracks.
#
# Deezer is a real API. Spotify and Apple Music are NOT: neither offers a free
# way to read a public playlist -- Apple's MusicKit needs a paid developer
# account -- so those two read the public web page and pull the JSON blob the
# page already embeds for its own client. That works, and it needs no key, no
# OAuth and no account. It is also a scrape, and it will break the day either
# company reshuffles their frontend.
#
# So the failure is made loud and specific rather than silent. A resolver that
# finds the page but not the tracks says the format changed, because "0 tracks"
# and "this playlist is empty" must not look the same to whoever has to fix it.
SPOTIFY_URL = re.compile(
r"open\.spotify\.com/(?:intl-[a-z-]+/)?(playlist|album)/([A-Za-z0-9]+)")
APPLE_URL = re.compile(
r"music\.apple\.com/[a-z]{2}/(playlist|album)/[^/]+/(pl\.[A-Za-z0-9.-]+|\d+)")
# Playlists are capped harder than a paste: one link can be thousands of tracks
# and the quota check cannot run until the whole thing has been resolved.
MAX_PLAYLIST_TRACKS = int(os.getenv("MUSE_MAX_PLAYLIST_TRACKS", "500"))
BROWSERISH = {
# These are public pages; a default httpx agent gets a different response
# from a real one on both hosts.
"User-Agent": ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/122.0 Safari/537.36"),
"Accept-Language": "en-US,en;q=0.9",
}
class PlaylistUnavailable(Exception):
"""The link could not be turned into a track list."""
def _walk(node, want: tuple[str, ...], out: list):
"""Collect every dict carrying all of `want`, at any depth."""
if isinstance(node, dict):
if all(k in node for k in want):
out.append(node)
for v in node.values():
_walk(v, want, out)
elif isinstance(node, list):
for v in node:
_walk(v, want, out)
async def _fetch(url: str) -> str:
async with httpx.AsyncClient(timeout=25, follow_redirects=True,
headers=BROWSERISH) as c:
r = await c.get(url)
if r.status_code == 404:
raise PlaylistUnavailable("that playlist does not exist, or is private")
r.raise_for_status()
return r.text
async def _deezer_playlist(kind: str, ident: str) -> tuple[str, list[tuple[str, str]]]:
async with httpx.AsyncClient(timeout=25) as c:
r = await c.get(f"{DEEZER}/{kind}/{ident}")
if r.status_code != 200:
raise PlaylistUnavailable(f"Deezer returned {r.status_code}")
d = r.json()
if d.get("error"):
raise PlaylistUnavailable(str(d["error"])[:120])
tracks = [((t.get("artist") or {}).get("name") or "", t.get("title") or "")
for t in (d.get("tracks") or {}).get("data", [])]
return d.get("title") or f"Deezer {kind}", [t for t in tracks if t[1]]
async def _spotify_playlist(kind: str, ident: str) -> tuple[str, list[tuple[str, str]]]:
# The embed player is server-rendered with the whole track list inline,
# which is why this needs no credentials.
html = await _fetch(f"https://open.spotify.com/embed/{kind}/{ident}")
m = re.search(r'<script id="__NEXT_DATA__"[^>]*>(.*?)</script>', html, re.S)
if not m:
raise PlaylistUnavailable(
"Spotify's embed page no longer carries __NEXT_DATA__ — the "
"resolver in app.py needs updating")
try:
entity = (json.loads(m.group(1))["props"]["pageProps"]["state"]
["data"]["entity"])
except (KeyError, ValueError) as e:
raise PlaylistUnavailable(
f"Spotify's embed payload changed shape ({errstr(e)}) — the "
"resolver in app.py needs updating") from e
items = entity.get("trackList") or []
if not items:
raise PlaylistUnavailable("Spotify returned no tracks for that link")
# title is the song, subtitle is the artist.
return (entity.get("name") or f"Spotify {kind}",
[(t.get("subtitle") or "", t.get("title") or "") for t in items
if t.get("title")])
async def _apple_playlist(kind: str, ident: str, url: str) -> tuple[str, list[tuple[str, str]]]:
html = await _fetch(url)
m = re.search(r'id="serialized-server-data"[^>]*>(.*?)</script>', html, re.S)
if not m:
raise PlaylistUnavailable(
"Apple Music's page no longer carries serialized-server-data — "
"the resolver in app.py needs updating")
try:
data = json.loads(m.group(1))
except ValueError as e:
raise PlaylistUnavailable(
f"Apple Music's payload did not parse ({errstr(e)})") from e
rows: list = []
_walk(data, ("title", "artistName"), rows)
if not rows:
raise PlaylistUnavailable(
"Apple Music's payload changed shape — the resolver in app.py "
"needs updating")
name = ""
heads: list = []
_walk(data, ("headerTitle",), heads)
if heads:
name = heads[0].get("headerTitle") or ""
seen, tracks = set(), []
for t in rows:
key = (t.get("artistName"), t.get("title"))
if key in seen:
continue
seen.add(key)
tracks.append((t.get("artistName") or "", t.get("title") or ""))
return name or f"Apple Music {kind}", tracks
async def resolve_playlist(url: str) -> tuple[str, str, list[tuple[str, str]]]:
"""Turn a shared playlist link into (source, name, [(artist, title), ...])."""
url = url.strip()
if m := DEEZER_URL.search(url):
name, tracks = await _deezer_playlist(*m.groups())
source = "deezer"
elif m := SPOTIFY_URL.search(url):
name, tracks = await _spotify_playlist(*m.groups())
source = "spotify"
elif m := APPLE_URL.search(url):
name, tracks = await _apple_playlist(*m.groups(), url)
source = "apple"
else:
raise PlaylistUnavailable(
"that link is not a Deezer, Spotify or Apple Music playlist")
if not tracks:
raise PlaylistUnavailable("that playlist appears to be empty")
if len(tracks) > MAX_PLAYLIST_TRACKS:
raise PlaylistUnavailable(
f"that playlist has {len(tracks)} tracks; the limit is "
f"{MAX_PLAYLIST_TRACKS}")
return source, name, tracks
# Navidrome mounts the library at a different path than muse does -- /music
# against /data/media/music -- so an m3u written with muse's own paths resolves
# to nothing on the other side of the wall.
PLAYLIST_PREFIX = os.getenv("MUSE_PLAYLIST_PREFIX", "/music")
def write_playlist(playlist_id: str) -> Path | None:
"""Write the .m3u for a playlist, or remove it when nothing has landed yet.
Rewritten from scratch every time one of its tracks imports, which is both
simpler than patching a file in place and the reason the playlist shows up
in Navidrome as soon as the first track arrives instead of after the last.
"""
with db() as c:
pl = c.execute("SELECT * FROM playlists WHERE id=?",
(playlist_id,)).fetchone()
if not pl:
return None
rows = c.execute(
"SELECT artist,title,imported_path FROM jobs WHERE playlist_id=? "
"AND state='imported' AND imported_path IS NOT NULL "
"ORDER BY playlist_seq", (playlist_id,)).fetchall()
dest = LIBRARY_DIR / f"{safe(pl['name'])}.m3u"
root = LIBRARY_DIR.resolve()
if not Path(os.path.normpath(dest)).resolve().is_relative_to(root):
log.warning("playlist %s: refusing to write outside the library", playlist_id)
return None
lines = ["#EXTM3U", f"#PLAYLIST:{pl['name']}"]
for r in rows:
try:
rel = Path(r["imported_path"]).resolve().relative_to(root)
except ValueError:
continue # imported somewhere else entirely; not ours to list
lines.append(f"#EXTINF:-1,{r['artist']} - {r['title']}")
lines.append(f"{PLAYLIST_PREFIX}/{rel.as_posix()}")
if len(lines) == 2:
# Nothing imported yet. Leaving a two-line file behind would make
# Navidrome show an empty playlist, which reads as a failure.
if dest.exists():
dest.unlink()
return None
dest.write_text("\n".join(lines) + "\n", encoding="utf-8")
with db() as c:
c.execute("UPDATE playlists SET m3u_path=?, tracks_written=? WHERE id=?",
(str(dest), len(rows), playlist_id))
log.info("playlist %s -> %s (%d tracks)", playlist_id[:8], dest, len(rows))
return dest
def playlist_of(job) -> str | None:
keys = job.keys() if hasattr(job, "keys") else ()
return job["playlist_id"] if "playlist_id" in keys else None
async def expand_url(kind: str, ident: str) -> list[str]:
"""Turn a Deezer album/playlist URL into 'Artist - Title' lines."""
out = []
@@ -2475,6 +2714,101 @@ async def import_list(body: ImportBody, request: Request):
return {"queued": queued, "skipped": skipped, "pending": work_q.qsize()}
class PlaylistBody(BaseModel):
url: str
# Resolve and show what is in it without queueing anything. A shared link
# is opaque -- you cannot tell 12 tracks from 400 by looking at it -- and
# queueing 400 against a quota you cannot see first is a bad trade.
preview: bool = False
name: str | None = None
@app.post("/api/playlist")
async def playlist_import(body: PlaylistBody, request: Request):
"""Turn a shared playlist link into queued requests and an m3u.
Every track goes through the same Deezer resolution a pasted line does, so
the matching, ranking and review behaviour is identical; the only new thing
is that the results stay grouped and ordered.
"""
try:
source, name, tracks = await resolve_playlist(body.url)
except PlaylistUnavailable as e:
raise HTTPException(422, str(e))
except Exception as e:
raise HTTPException(502, f"could not read that playlist: {errstr(e)}")
name = (body.name or name).strip()[:MAX_COMPONENT] or "Playlist"
if body.preview:
return {"source": source, "name": name, "count": len(tracks),
"tracks": [{"artist": a, "title": t} for a, t in tracks[:100]]}
check_quota(request, adding=len(tracks))
playlist_id = str(uuid.uuid4())
with db() as c:
c.execute("INSERT INTO playlists (id,name,source,source_url,"
"requested_by,tracks_total) VALUES (?,?,?,?,?,?)",
(playlist_id, name, source, body.url[:500],
session_user(request), len(tracks)))
queued, skipped = [], []
for seq, (artist, title) in enumerate(tracks):
line = f"{artist} {title}".strip()
try:
hits = await deezer_search(SPLIT.sub(" ", line), limit=10)
except Exception as e:
skipped.append({"line": line, "why": errstr(e)})
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"], requested_by=session_user(request),
cover=h.get("cover"), duration=h.get("length"),
playlist_id=playlist_id, playlist_seq=seq)
queued.append({"id": jid, "line": line, "cover": h.get("cover"),
"matched": f"{h['artist']} — {h['title']}"})
log.info("playlist %s (%s) %r: %d queued, %d skipped",
playlist_id[:8], source, name, len(queued), len(skipped))
return {"playlist_id": playlist_id, "name": name, "source": source,
"queued": queued, "skipped": skipped, "pending": work_q.qsize()}
@app.get("/api/playlists")
async def playlists(request: Request):
"""Playlists and how far along each one is.
Shared on purpose: an m3u on disk is visible to every Navidrome user
regardless, so pretending otherwise here would only be a fiction.
"""
with db() as c:
rows = c.execute(
"SELECT * FROM playlists ORDER BY created_at DESC LIMIT 100"
).fetchall()
counts = {}
for r in c.execute(
"SELECT playlist_id, state, COUNT(*) n FROM jobs "
"WHERE playlist_id IS NOT NULL GROUP BY playlist_id, state"):
counts.setdefault(r["playlist_id"], {})[r["state"]] = r["n"]
out = []
for r in rows:
c_ = counts.get(r["id"], {})
done = c_.get("imported", 0)
out.append({**dict(r), "counts": c_, "done": done,
"pending": sum(c_.get(s, 0) for s in
("queued", "searching", "downloading")),
"failed": c_.get("failed", 0),
"review": c_.get("review", 0)})
return {"playlists": out}
class ReviewBody(BaseModel):
action: str # approve | reject
@@ -2532,6 +2866,9 @@ async def review_decide(job_id: str, body: ReviewBody, request: Request):
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)
# It may already be listed from an earlier pass; the rewrite drops it.
if playlist_of(job):
write_playlist(job["playlist_id"])
return {"ok": True, "state": "failed"}
if not src:
@@ -2539,6 +2876,11 @@ async def review_decide(job_id: str, body: ReviewBody, request: Request):
dest = tag_and_file(job, src)
set_state(job_id, "imported", str(dest), imported_path=str(dest),
reviewed_by=me)
# After set_state, not before: write_playlist reads imported_path back out
# of the row, so running it first would write a playlist missing the very
# track that was just approved.
if playlist_of(job):
write_playlist(job["playlist_id"])
log.info("review: %s approved %s", me, job_id[:8])
return {"ok": True, "state": "imported", "path": str(dest)}