travandClaude Opus 5.5
f899794068
No identify dialog pops up mid-download any more: songs are looked up quietly as they land and reviewed together once the last one is in, as rows in the app's columns. Songs that are one album get the album-wide fields once, unified, plus a cover from the album-art search. Ticked rows apply as one undo step; unticked ones get the ordinary dialog after. Popups wait for LinTunes to be the app in front. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
374 lines
14 KiB
Python
374 lines
14 KiB
Python
"""Round 76: Spotify links in File ▸ Import from URL.
|
|
|
|
trav's ``spotify-youtube`` script, brought inside: a Spotify track, album or
|
|
playlist link is read off Spotify's public embed page, each song is found on
|
|
YouTube (the result closest to Spotify's length wins), downloaded with the
|
|
`song` flags, tagged with what Spotify said, then imported and identified
|
|
like any other link. No network here: the page is canned and yt-dlp is fake.
|
|
"""
|
|
import json
|
|
import os
|
|
import stat
|
|
import sys
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from lintunes import spotify_link, tagging, url_import
|
|
from lintunes.gui.url_import_dialog import UrlImportDialog
|
|
from lintunes.library_manager import LibraryManager
|
|
from lintunes.models import Library
|
|
from lintunes.preferences import Preferences
|
|
from lintunes.spotify_link import (
|
|
SpotifyError, SpotifyTrack, parse_embed, rank_candidates, spotify_source,
|
|
)
|
|
|
|
ALBUM_URL = "https://open.spotify.com/album/57F44c0MTziVzHPEuJtH9A?si=xyz"
|
|
|
|
|
|
def _page(entity) -> str:
|
|
data = {"props": {"pageProps": {"state": {"data": {"entity": entity}}}}}
|
|
return ('<html><script id="__NEXT_DATA__" type="application/json">'
|
|
+ json.dumps(data) + "</script></html>")
|
|
|
|
|
|
ALBUM = _page({
|
|
"type": "album", "name": "Last Splash", "title": "Last Splash",
|
|
"subtitle": "The Breeders",
|
|
"trackList": [
|
|
{"uri": "spotify:track:t1", "title": "New Year",
|
|
"subtitle": "The Breeders", "duration": 116706},
|
|
{"uri": "spotify:track:t2", "title": "Cannonball",
|
|
"subtitle": "The Breeders", "duration": 213000},
|
|
{"uri": "spotify:track:t3", "title": "", # dropped
|
|
"subtitle": "The Breeders", "duration": 1000},
|
|
]})
|
|
|
|
PLAYLIST = _page({
|
|
"type": "playlist", "name": "Chill",
|
|
"trackList": [
|
|
{"uri": "spotify:track:p1", "title": "Song A", "subtitle": "X, Y",
|
|
"duration": 200000}]})
|
|
|
|
TRACK = _page({
|
|
"type": "track", "name": "Never Gonna Give You Up",
|
|
"title": "Never Gonna Give You Up", "id": "4uLU",
|
|
"uri": "spotify:track:4uLU",
|
|
"artists": [{"name": "Rick Astley"}], "duration": 213573})
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# A. reading a link
|
|
|
|
|
|
@pytest.mark.parametrize("text, source", [
|
|
(ALBUM_URL, ("album", "57F44c0MTziVzHPEuJtH9A")),
|
|
("https://open.spotify.com/intl-de/track/4uLU", ("track", "4uLU")),
|
|
("https://open.spotify.com/embed/playlist/37i9", ("playlist", "37i9")),
|
|
("spotify:playlist:37i9", ("playlist", "37i9")),
|
|
("https://open.spotify.com/artist/0gxy", None),
|
|
("https://open.spotify.com/episode/0gxy", None),
|
|
("https://www.youtube.com/watch?v=abc", None),
|
|
("", None),
|
|
])
|
|
def test_spotify_source(text, source):
|
|
assert spotify_source(text) == source
|
|
|
|
|
|
def test_link_kind_and_uris_count_as_links():
|
|
assert url_import.link_kind(ALBUM_URL) == "spotify"
|
|
assert url_import.link_kind("spotify:track:4uLU") == "spotify"
|
|
assert url_import.looks_like_url("spotify:track:4uLU")
|
|
assert url_import.link_kind("https://open.spotify.com/artist/x") \
|
|
== "unknown"
|
|
|
|
|
|
def test_an_album_page_numbers_its_tracks():
|
|
name, tracks = parse_embed(ALBUM)
|
|
assert name == "Last Splash"
|
|
assert [(t.id, t.title, t.album, t.track_number, t.track_count,
|
|
t.duration) for t in tracks] == [
|
|
("t1", "New Year", "Last Splash", 1, 2, 117),
|
|
("t2", "Cannonball", "Last Splash", 2, 2, 213)]
|
|
|
|
|
|
def test_a_playlist_page_has_no_album():
|
|
_, [track] = parse_embed(PLAYLIST)
|
|
assert (track.artist, track.album, track.track_number) == ("X, Y", "", 0)
|
|
assert spotify_link.tag_fields(track) == {"name": "Song A",
|
|
"artist": "X, Y"}
|
|
|
|
|
|
def test_a_track_page_is_one_song():
|
|
name, [track] = parse_embed(TRACK)
|
|
assert name == "Never Gonna Give You Up"
|
|
assert (track.id, track.artist, track.duration) == ("4uLU",
|
|
"Rick Astley", 214)
|
|
|
|
|
|
@pytest.mark.parametrize("html", [
|
|
"<html>no data</html>",
|
|
_page({"type": "playlist", "name": "Empty", "trackList": []}),
|
|
])
|
|
def test_a_changed_page_says_so(html):
|
|
with pytest.raises(SpotifyError):
|
|
parse_embed(html)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# B. choosing the YouTube result
|
|
|
|
|
|
def _e(vid, duration, title="x"):
|
|
return {"id": vid, "duration": duration, "title": title,
|
|
"url": f"https://www.youtube.com/watch?v={vid}"}
|
|
|
|
|
|
def test_the_closest_length_wins():
|
|
track = SpotifyTrack("t", "Cannonball", "The Breeders", duration=213)
|
|
ranked = rank_candidates([_e("video", 260), _e("audio", 214),
|
|
_e("lyric", 220)], track)
|
|
assert [e["id"] for e in ranked] == ["audio", "lyric", "video"]
|
|
|
|
|
|
def test_live_and_covers_lose_unless_the_song_is_one():
|
|
track = SpotifyTrack("t", "Cannonball", "The Breeders", duration=213)
|
|
ranked = rank_candidates([_e("live", 213, "Cannonball (Live 1994)"),
|
|
_e("studio", 216, "Cannonball")], track)
|
|
assert ranked[0]["id"] == "studio"
|
|
live = SpotifyTrack("t", "Cannonball - Live", "The Breeders",
|
|
duration=213)
|
|
assert rank_candidates([_e("live", 213, "Cannonball (Live 1994)"),
|
|
_e("studio", 216, "Cannonball")],
|
|
live)[0]["id"] == "live"
|
|
|
|
|
|
def test_without_lengths_youtube_order_stands():
|
|
track = SpotifyTrack("t", "Song", "A")
|
|
assert [e["id"] for e in rank_candidates(
|
|
[_e("a", None), _e("b", 100)], track)] == ["a", "b"]
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# C. the worker, against a fake yt-dlp
|
|
|
|
# Search: "ytsearch5:<query>" lists $FAKE_SEARCH[query] as (id, seconds,
|
|
# title). Download: writes "<id>.mp3", failing ids starting with "bad".
|
|
FAKE_YTDLP = '''#!/usr/bin/env python3
|
|
import json, os, sys
|
|
args = sys.argv[1:]
|
|
with open(os.environ["FAKE_YTDLP_LOG"], "a") as log:
|
|
log.write("\\x1f".join(args) + "\\n")
|
|
urls = args[args.index("--") + 1:]
|
|
if "--flat-playlist" in args:
|
|
query = urls[0].split(":", 1)[1]
|
|
for vid, secs, title in json.loads(os.environ["FAKE_SEARCH"]).get(query, []):
|
|
print(f"LTENTRY {vid}\\t{secs}\\t"
|
|
f"https://www.youtube.com/watch?v={vid}\\t{title}", flush=True)
|
|
sys.exit(0)
|
|
dest = args[args.index("-P") + 1]
|
|
for url in urls:
|
|
vid = url.rsplit("=", 1)[-1]
|
|
if vid.startswith("bad"):
|
|
print(f"ERROR: [youtube] {vid}: Video unavailable", file=sys.stderr,
|
|
flush=True)
|
|
continue
|
|
print(f"LTSTART 1\\t1\\t{vid}\\t{vid}", flush=True)
|
|
print("LTPROG 50.0%", flush=True)
|
|
path = os.path.join(dest, f"{vid} [{vid}].mp3")
|
|
with open(path, "wb") as f:
|
|
f.write(open(os.environ["FAKE_YTDLP_AUDIO"], "rb").read())
|
|
print(f"LTFILE {path}", flush=True)
|
|
'''
|
|
|
|
|
|
@pytest.fixture
|
|
def fake(tmp_path, monkeypatch, mp3_file):
|
|
bin_dir = tmp_path / "bin"
|
|
bin_dir.mkdir()
|
|
script = bin_dir / "yt-dlp"
|
|
script.write_text(FAKE_YTDLP.replace("/usr/bin/env python3",
|
|
sys.executable))
|
|
script.chmod(script.stat().st_mode | stat.S_IEXEC)
|
|
log = tmp_path / "argv.log"
|
|
monkeypatch.setenv("PATH", f"{bin_dir}{os.pathsep}{os.environ['PATH']}")
|
|
monkeypatch.setenv("FAKE_YTDLP_LOG", str(log))
|
|
monkeypatch.setenv("FAKE_YTDLP_AUDIO", str(mp3_file))
|
|
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config"))
|
|
|
|
def search(results):
|
|
monkeypatch.setenv("FAKE_SEARCH", json.dumps(results))
|
|
search({})
|
|
search.argv = lambda: ([line.split("\x1f") for line in
|
|
log.read_text().splitlines()]
|
|
if log.exists() else [])
|
|
return search
|
|
|
|
|
|
def _worker(tmp_path, tracks):
|
|
worker = url_import.SpotifyImportWorker(tracks)
|
|
worker.temp_dir = tmp_path / "dl"
|
|
worker.temp_dir.mkdir()
|
|
events = []
|
|
worker.item_started.connect(
|
|
lambda i, n, sid, t: events.append(("start", i, n, sid)))
|
|
worker.item_failed.connect(lambda sid, why: events.append(("bad", sid)))
|
|
worker.downloaded.connect(lambda p: events.append(("file", p)))
|
|
worker.finished.connect(lambda d: events.append(("done", d["downloaded"])))
|
|
worker.failed.connect(lambda m: events.append(("failed", m)))
|
|
return worker, events
|
|
|
|
|
|
def test_each_song_is_searched_picked_downloaded_and_tagged(fake, tmp_path):
|
|
_, tracks = parse_embed(ALBUM)
|
|
fake({"The Breeders - New Year": [["vid1", 300, "New Year (video)"],
|
|
["aud1", 117, "New Year"]],
|
|
"The Breeders - Cannonball": [["bad2", 213, "Cannonball"],
|
|
["aud2", 214, "Cannonball"]]})
|
|
worker, events = _worker(tmp_path, tracks)
|
|
worker._run()
|
|
|
|
files = [e[1] for e in events if e[0] == "file"]
|
|
assert [e for e in events if e[0] != "file"] == [
|
|
("start", 1, 2, "t1"), ("start", 2, 2, "t2"), ("done", 2)]
|
|
# The right length, then the next result when the best one fails.
|
|
assert [os.path.basename(f) for f in files] == ["aud1 [aud1].mp3",
|
|
"aud2 [aud2].mp3"]
|
|
tags = tagging.read_tags(files[1])
|
|
assert (tags["name"], tags["artist"], tags["album"],
|
|
tags["track_number"], tags["track_count"]) == (
|
|
"Cannonball", "The Breeders", "Last Splash", 2, 2)
|
|
downloads = [a for a in fake.argv() if "--flat-playlist" not in a]
|
|
assert all("--extract-audio" in a for a in downloads) # the `song` flags
|
|
|
|
|
|
def test_a_song_youtube_doesnt_have_is_named(fake, tmp_path):
|
|
_, tracks = parse_embed(ALBUM)
|
|
fake({"The Breeders - Cannonball": [["aud2", 214, "Cannonball"]]})
|
|
worker, events = _worker(tmp_path, tracks)
|
|
worker._run()
|
|
assert [e for e in events if e[0] != "file"] == [
|
|
("start", 1, 2, "t1"), ("bad", "t1"), ("start", 2, 2, "t2"),
|
|
("done", 1)]
|
|
|
|
|
|
def test_nothing_found_at_all_fails(fake, tmp_path):
|
|
_, tracks = parse_embed(PLAYLIST)
|
|
worker, events = _worker(tmp_path, tracks)
|
|
worker._run()
|
|
assert events[-1][0] == "failed"
|
|
assert not worker.busy()
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# D. the dialog and the window
|
|
|
|
|
|
def _pump(qapp, until, timeout=10.0):
|
|
deadline = time.monotonic() + timeout
|
|
while not until() and time.monotonic() < deadline:
|
|
qapp.processEvents()
|
|
time.sleep(0.01)
|
|
assert until(), "timed out waiting"
|
|
|
|
|
|
@pytest.fixture
|
|
def canned(monkeypatch):
|
|
pages = {"album": ALBUM, "track": TRACK, "playlist": PLAYLIST}
|
|
monkeypatch.setattr(spotify_link, "fetch",
|
|
lambda kind, _id: parse_embed(pages[kind]))
|
|
|
|
|
|
def _dialog(qapp, url):
|
|
qapp.clipboard().setText(url)
|
|
dialog = UrlImportDialog(None)
|
|
assert dialog.url() == url
|
|
return dialog
|
|
|
|
|
|
def test_an_album_is_a_checklist_of_spotify_songs(qapp, canned):
|
|
dialog = _dialog(qapp, ALBUM_URL)
|
|
dialog._on_import()
|
|
_pump(qapp, lambda: dialog._probe is None)
|
|
assert dialog._picking and dialog._list.count() == 2
|
|
assert dialog._ok.text() == "Import 2 Songs"
|
|
dialog._list.item(0).setCheckState(
|
|
dialog._list.item(0).checkState().Unchecked)
|
|
dialog._on_import()
|
|
assert [t.title for t in dialog.spotify_tracks()] == ["Cannonball"]
|
|
|
|
|
|
def test_a_spotify_track_goes_straight_through(qapp, canned):
|
|
dialog = _dialog(qapp, "https://open.spotify.com/track/4uLU")
|
|
dialog._on_import()
|
|
_pump(qapp, lambda: dialog.result() == dialog.DialogCode.Accepted)
|
|
assert [t.artist for t in dialog.spotify_tracks()] == ["Rick Astley"]
|
|
|
|
|
|
def test_a_youtube_link_has_no_spotify_tracks(qapp):
|
|
dialog = _dialog(qapp, "https://www.youtube.com/watch?v=abc")
|
|
dialog._on_import()
|
|
assert dialog.spotify_tracks() == []
|
|
|
|
|
|
@pytest.fixture
|
|
def window(qapp, tmp_path, fake, monkeypatch):
|
|
from lintunes.gui.main_window import MainWindow
|
|
library = Library(music_folder=str(tmp_path / "media"))
|
|
(tmp_path / "media").mkdir()
|
|
manager = LibraryManager(library, tmp_path / "data")
|
|
win = MainWindow(manager, Preferences(tmp_path / "data"))
|
|
identified = []
|
|
monkeypatch.setattr(win, "_enqueue_identify", identified.extend)
|
|
win.identified = identified
|
|
reviews = []
|
|
monkeypatch.setattr(win, "_show_batch_review", reviews.append)
|
|
win.reviews = reviews
|
|
yield win
|
|
if win._url_batch is not None: # lookups still out: never review them
|
|
win._url_batch.cancel()
|
|
win.close()
|
|
|
|
|
|
def test_spotify_songs_import_tagged_and_go_to_identify(window, qapp, fake,
|
|
monkeypatch):
|
|
from lintunes.gui import main_window as mw
|
|
_, tracks = parse_embed(ALBUM)
|
|
fake({"The Breeders - New Year": [["aud1", 117, "New Year"]],
|
|
"The Breeders - Cannonball": [["aud2", 214, "Cannonball"]]})
|
|
chosen = [{"id": t.id, "title": t.label(), "spotify": t}
|
|
for t in tracks]
|
|
|
|
class _Dialog:
|
|
cookies_used = None
|
|
def __init__(self, *a, **kw):
|
|
pass
|
|
def exec(self):
|
|
return True
|
|
def add_to_playlist(self):
|
|
return False
|
|
def chosen_entries(self):
|
|
return chosen
|
|
def spotify_tracks(self):
|
|
return tracks
|
|
def download_urls(self):
|
|
raise AssertionError("a Spotify link is never handed to yt-dlp")
|
|
monkeypatch.setattr(mw, "UrlImportDialog", _Dialog)
|
|
|
|
window._import_from_url()
|
|
worker = window._url_worker
|
|
assert isinstance(worker, url_import.SpotifyImportWorker)
|
|
_pump(qapp, lambda: worker.temp_dir is None)
|
|
assert window._url_progress.status("t1") == "Imported"
|
|
assert window._url_progress.status("t2") == "Imported"
|
|
lib = window._manager.library.tracks.values()
|
|
assert sorted((t.artist, t.album, t.name) for t in lib) == [
|
|
("The Breeders", "Last Splash", "Cannonball"),
|
|
("The Breeders", "Last Splash", "New Year")]
|
|
# Several songs: no dialog each, one review once both are in.
|
|
assert window.identified == []
|
|
_pump(qapp, lambda: window.reviews)
|
|
[analysis] = window.reviews
|
|
assert analysis.is_album and analysis.album == "Last Splash"
|
|
assert len(analysis.rows) == 2
|