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:
1 parent
e9fba3bb1d
commit
eeb0e096a7
7 files changed
+815
-19
No files matched your search
@@ -0,0 +1,176 @@
|
||||
"""Playlist import: link resolution, ordering, and the m3u Navidrome reads.
|
||||
|
||||
The resolvers are exercised against synthetic payloads rather than the live
|
||||
sites. Two reasons: a test that fails when Spotify has a bad afternoon is not
|
||||
testing this code, and the failure that actually matters here is the *shape*
|
||||
changing -- which is precisely what a fixture can assert and a live fetch
|
||||
cannot.
|
||||
|
||||
Run standalone: python tests/test_playlist.py
|
||||
"""
|
||||
import asyncio, json, os, sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
os.environ.setdefault("MUSE_AUTH", "off")
|
||||
os.environ.setdefault("MUSE_DB", "dev/playlist.db")
|
||||
os.environ.setdefault("MUSE_LIBRARY_DIR", "dev/pl-library")
|
||||
|
||||
for f in ("dev/playlist.db", "dev/playlist.db-wal", "dev/playlist.db-shm"):
|
||||
if os.path.exists(f):
|
||||
os.remove(f)
|
||||
|
||||
import app # noqa: E402
|
||||
|
||||
app.init_db()
|
||||
app.LIBRARY_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
ok = fail = 0
|
||||
|
||||
|
||||
def check(name, cond, detail=""):
|
||||
global ok, fail
|
||||
if cond:
|
||||
ok += 1
|
||||
print(f" PASS {name}")
|
||||
else:
|
||||
fail += 1
|
||||
print(f" FAIL {name} {detail}")
|
||||
|
||||
|
||||
SPOTIFY_PAGE = (
|
||||
'<html><script id="__NEXT_DATA__" type="application/json">'
|
||||
+ json.dumps({"props": {"pageProps": {"state": {"data": {"entity": {
|
||||
"name": "Late Night",
|
||||
"trackList": [
|
||||
{"title": "The Color Violet", "subtitle": "Tory Lanez"},
|
||||
{"title": "No Pole", "subtitle": "Don Toliver"},
|
||||
]}}}}}})
|
||||
+ "</script></html>")
|
||||
|
||||
APPLE_PAGE = (
|
||||
'<html><script id="serialized-server-data" type="application/json">'
|
||||
+ json.dumps([{"data": {"sections": [
|
||||
{"headerTitle": "Heavy Rotation"},
|
||||
{"items": [
|
||||
{"title": "YUKON", "artistName": "Justin Bieber"},
|
||||
{"title": "LET THE WORLD BURN", "artistName": "Chris Grey"},
|
||||
{"title": "YUKON", "artistName": "Justin Bieber"}, # dupe
|
||||
]}]}}])
|
||||
+ "</script></html>")
|
||||
|
||||
|
||||
def fake_fetch(body):
|
||||
async def _f(url):
|
||||
return body
|
||||
return _f
|
||||
|
||||
|
||||
def make_playlist(pid, name, rows):
|
||||
"""rows: (seq, artist, title, state, imported_path)"""
|
||||
with app.db() as c:
|
||||
c.execute("INSERT OR REPLACE INTO playlists (id,name,source,"
|
||||
"source_url,tracks_total) VALUES (?,?,?,?,?)",
|
||||
(pid, name, "spotify", "https://x", len(rows)))
|
||||
for seq, artist, title, state, path in rows:
|
||||
c.execute(
|
||||
"INSERT INTO jobs (id,state,detail,artist,title,"
|
||||
"imported_path,playlist_id,playlist_seq) "
|
||||
"VALUES (?,?,?,?,?,?,?,?)",
|
||||
(f"{pid}-{seq}", state, "", artist, title, path, pid, seq))
|
||||
|
||||
|
||||
def main() -> int:
|
||||
print("\nlink routing")
|
||||
for url, want in [
|
||||
("https://open.spotify.com/playlist/abc123", "spotify"),
|
||||
("https://open.spotify.com/intl-pt/playlist/abc123", "spotify"),
|
||||
("https://music.apple.com/us/playlist/hits/pl.abc-123", "apple"),
|
||||
("https://www.deezer.com/en/playlist/1313621735", "deezer"),
|
||||
]:
|
||||
matched = ("spotify" if app.SPOTIFY_URL.search(url)
|
||||
else "apple" if app.APPLE_URL.search(url)
|
||||
else "deezer" if app.DEEZER_URL.search(url) else "none")
|
||||
check(f"{want} link routes to {want}", matched == want, f"got {matched}")
|
||||
|
||||
print("\nresolvers, against a fixed payload shape")
|
||||
real = app._fetch
|
||||
try:
|
||||
app._fetch = fake_fetch(SPOTIFY_PAGE)
|
||||
name, tracks = asyncio.run(app._spotify_playlist("playlist", "x"))
|
||||
check("spotify name", name == "Late Night", name)
|
||||
check("spotify pulls artist from subtitle",
|
||||
tracks[0] == ("Tory Lanez", "The Color Violet"), tracks[:1])
|
||||
|
||||
app._fetch = fake_fetch(APPLE_PAGE)
|
||||
name, tracks = asyncio.run(app._apple_playlist("playlist", "pl.x", "u"))
|
||||
check("apple name", name == "Heavy Rotation", name)
|
||||
check("apple pulls artist from artistName",
|
||||
("Justin Bieber", "YUKON") in tracks, tracks)
|
||||
check("apple de-duplicates repeated entries", len(tracks) == 2, tracks)
|
||||
|
||||
# The failure that actually matters: they changed the page.
|
||||
app._fetch = fake_fetch("<html>nothing useful here</html>")
|
||||
for fn, label in ((app._spotify_playlist, "spotify"),
|
||||
(app._apple_playlist, "apple")):
|
||||
try:
|
||||
asyncio.run(fn("playlist", "x", "u") if label == "apple"
|
||||
else fn("playlist", "x"))
|
||||
check(f"{label} shape change is reported", False, "no raise")
|
||||
except app.PlaylistUnavailable as e:
|
||||
check(f"{label} shape change names the resolver",
|
||||
"needs updating" in str(e), str(e))
|
||||
finally:
|
||||
app._fetch = real
|
||||
|
||||
print("\nthe m3u Navidrome reads")
|
||||
lib = app.LIBRARY_DIR
|
||||
(lib / "Tory Lanez").mkdir(parents=True, exist_ok=True)
|
||||
(lib / "Don Toliver").mkdir(parents=True, exist_ok=True)
|
||||
a = lib / "Tory Lanez" / "The Color Violet.flac"; a.write_text("x")
|
||||
b = lib / "Don Toliver" / "No Pole.flac"; b.write_text("x")
|
||||
|
||||
make_playlist("p1", "Late Night", [
|
||||
(1, "Don Toliver", "No Pole", "imported", str(b)),
|
||||
(0, "Tory Lanez", "The Color Violet", "imported", str(a)),
|
||||
(2, "Nobody", "Missing", "failed", None),
|
||||
])
|
||||
dest = app.write_playlist("p1")
|
||||
text = dest.read_text()
|
||||
check("m3u is written", dest and dest.exists(), dest)
|
||||
check("declares the playlist name for Navidrome",
|
||||
"#PLAYLIST:Late Night" in text, text)
|
||||
check("uses Navidrome's mount, not muse's",
|
||||
"/music/Tory Lanez/The Color Violet.flac" in text
|
||||
and str(lib) not in text, text)
|
||||
check("keeps the playlist's own order, not insertion order",
|
||||
text.index("The Color Violet") < text.index("No Pole"), text)
|
||||
check("omits tracks that never imported", "Missing" not in text, text)
|
||||
|
||||
print("\nedge cases")
|
||||
make_playlist("p2", "Nothing Landed Yet", [
|
||||
(0, "A", "B", "searching", None)])
|
||||
check("no m3u while the playlist is empty",
|
||||
app.write_playlist("p2") is None)
|
||||
check("and no stray file left behind",
|
||||
not (lib / "Nothing Landed Yet.m3u").exists())
|
||||
|
||||
make_playlist("p3", "../../escape", [
|
||||
(0, "Tory Lanez", "The Color Violet", "imported", str(a))])
|
||||
out = app.write_playlist("p3")
|
||||
check("a traversing playlist name cannot escape the library",
|
||||
out is None or out.resolve().is_relative_to(lib.resolve()), out)
|
||||
|
||||
# A track approved and then rejected has to leave the file.
|
||||
with app.db() as c:
|
||||
c.execute("UPDATE jobs SET state='failed' WHERE id=?", ("p1-0",))
|
||||
app.write_playlist("p1")
|
||||
check("rejecting a track drops it from the m3u",
|
||||
"The Color Violet" not in dest.read_text(), dest.read_text())
|
||||
|
||||
print(f"\n{ok} passed, {fail} failed")
|
||||
return 1 if fail else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in new issue
Block a user