travandClaude Opus 5
8188a5b389
File > Import from URL... runs trav's `song` helper (yt-dlp --extract-audio --audio-format mp3) from inside the app. Everything at the link downloads to a private temp dir and is imported as each song lands. "and add to current playlist?" puts the batch above the selected song, else at the end of the playing playlist, else at the end of the shown one, in the link's order. Each song goes straight into Identify Track. With no AcoustID key, the proposal comes from the filename alone. Album art now opens on a single click, in a window shaped like the cover and as big as the screen allows. Also fixed the dev fixture's play journal, which crashed startup (wrong shape) and would have been ignored anyway (wrong keys). The loader now skips a journal it can't read instead of aborting. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01R5m9mXFPHNro78BdD69mG2
475 lines
18 KiB
Python
475 lines
18 KiB
Python
"""Round 49 — Import from URL, and the album art on a single click.
|
|
|
|
trav's `song` shell helper is ``yt-dlp --extract-audio --audio-format mp3``,
|
|
and now File ▸ Import from URL… runs it for him. Everything at the link lands
|
|
in the library, optionally in the current playlist, and each song goes
|
|
through Identify Track as it arrives. The rule to protect is *where* the songs
|
|
go:
|
|
1. above the selected song in the shown playlist,
|
|
2. else at the end of the playlist that's playing,
|
|
3. else at the end of the one that's shown,
|
|
4. else nowhere, with the checkbox grayed out.
|
|
|
|
A batch keeps the link's order.
|
|
|
|
The side fix: the sidebar art opens its window on one click, sized to the
|
|
art's own shape and as big as the screen allows.
|
|
|
|
yt-dlp itself is never run here. A fake on PATH prints the same marker lines
|
|
the real command is told to print.
|
|
"""
|
|
import os
|
|
import shutil
|
|
import stat
|
|
import time
|
|
|
|
import pytest
|
|
from PyQt6.QtCore import QEvent, QObject, QPointF, QSize, Qt, pyqtSignal
|
|
from PyQt6.QtGui import QMouseEvent, QPixmap
|
|
|
|
from lintunes import fingerprint, url_import
|
|
from lintunes.library_manager import LibraryManager
|
|
from lintunes.models import Library, Track
|
|
from lintunes.models.playlist import Playlist, PlaylistType
|
|
from lintunes.preferences import Preferences
|
|
from lintunes.url_import import ImportTarget, UrlImportWorker
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# helpers
|
|
|
|
FAKE_YTDLP = '''#!/usr/bin/env python3
|
|
import os, sys, time
|
|
args = sys.argv[1:]
|
|
dest = args[args.index("-P") + 1]
|
|
url = args[-1]
|
|
mode = os.environ.get("FAKE_YTDLP_MODE", "ok")
|
|
if mode == "fail":
|
|
print("ERROR: [generic] Unsupported URL: " + url, file=sys.stderr, flush=True)
|
|
sys.exit(1)
|
|
src = os.environ.get("FAKE_YTDLP_AUDIO")
|
|
titles = ["First Song", "Second Song"]
|
|
for i, title in enumerate(titles, 1):
|
|
print(f"LTSTART {i}\\t{len(titles)}\\t{title}", flush=True)
|
|
print("[download] some noise yt-dlp prints", flush=True)
|
|
print("LTPROG 50.0%", flush=True)
|
|
path = os.path.join(dest, f"{title} [id{i}].mp3")
|
|
with open(path, "wb") as f:
|
|
f.write(open(src, "rb").read() if src else b"ID3fake")
|
|
print(f"LTFILE {path}", flush=True)
|
|
if mode == "slow":
|
|
time.sleep(30)
|
|
'''
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_ytdlp(tmp_path, monkeypatch):
|
|
bin_dir = tmp_path / "bin"
|
|
bin_dir.mkdir()
|
|
script = bin_dir / "yt-dlp"
|
|
script.write_text(FAKE_YTDLP)
|
|
script.chmod(script.stat().st_mode | stat.S_IEXEC)
|
|
monkeypatch.setenv("PATH", f"{bin_dir}{os.pathsep}{os.environ['PATH']}")
|
|
return script
|
|
|
|
|
|
@pytest.fixture
|
|
def isolated_config(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config"))
|
|
|
|
|
|
def _library(tmp_path):
|
|
"""Four tracks; regular playlists "a" (with a dangling id) and "b", plus
|
|
a smart one and a folder."""
|
|
media = tmp_path / "media"
|
|
library = Library(music_folder=str(media))
|
|
for tid in range(1, 5):
|
|
path = media / "Music" / f"Artist{tid}" / "Album" / f"T{tid}.mp3"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(b"audio")
|
|
library.tracks[tid] = Track(track_id=tid, name=f"T{tid}",
|
|
artist=f"Artist{tid}", location=str(path))
|
|
library.playlists["a"] = Playlist(name="Road Trip", persistent_id="a",
|
|
track_ids=[1, 99, 2, 3])
|
|
library.playlists["b"] = Playlist(name="Chill", persistent_id="b",
|
|
track_ids=[4])
|
|
library.playlists["s"] = Playlist(name="Smart", persistent_id="s",
|
|
playlist_type=PlaylistType.SMART,
|
|
track_ids=[1])
|
|
library.playlists["f"] = Playlist(name="Folder", persistent_id="f",
|
|
playlist_type=PlaylistType.FOLDER)
|
|
return library
|
|
|
|
|
|
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"
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# A. the command line and its output
|
|
|
|
|
|
class TestCommand:
|
|
def test_runs_the_song_flags_with_the_url_last(self, tmp_path):
|
|
cmd = url_import.build_command("https://example.com/v?id=1", tmp_path)
|
|
assert cmd[0] == "yt-dlp"
|
|
assert cmd[1:4] == ["--extract-audio", "--audio-format", "mp3"]
|
|
assert cmd[-2:] == ["--", "https://example.com/v?id=1"]
|
|
assert cmd[cmd.index("-P") + 1] == str(tmp_path)
|
|
|
|
def test_keeps_the_default_filename(self, tmp_path):
|
|
"""filename_tags reads yt-dlp's own "Title [id]", so no -o."""
|
|
cmd = url_import.build_command("https://x.test/v", tmp_path)
|
|
assert "-o" not in cmd and "--output" not in cmd
|
|
|
|
@pytest.mark.parametrize("line, expected", [
|
|
("LTSTART 2\t5\tSome Song\n", ("start", 2, 5, "Some Song")),
|
|
("LTSTART NA\tNA\tSolo\n", ("start", 1, 1, "Solo")),
|
|
("LTSTART 1\t1\tTab\tin title\n", ("start", 1, 1, "Tab\tin title")),
|
|
("LTFILE /tmp/x/Song [abc].mp3\n", ("file", "/tmp/x/Song [abc].mp3")),
|
|
("LTPROG 45.3%\n", ("progress", 45)),
|
|
("LTPROG 100.0%\r\n", ("progress", 100)),
|
|
("LTPROG N/A\n", None),
|
|
("LTFILE \n", None),
|
|
("[youtube] Extracting URL\n", None),
|
|
("ERROR: Video unavailable\n", None),
|
|
])
|
|
def test_parse_line(self, line, expected):
|
|
assert url_import.parse_line(line) == expected
|
|
|
|
@pytest.mark.parametrize("text, ok", [
|
|
("https://www.youtube.com/watch?v=abc", True),
|
|
(" http://soundcloud.com/x/y ", True),
|
|
("https://", False),
|
|
("youtube.com/watch?v=abc", False),
|
|
("https://a b", False),
|
|
("", False),
|
|
])
|
|
def test_looks_like_url(self, text, ok):
|
|
assert url_import.looks_like_url(text) is ok
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# B. where the songs go
|
|
|
|
|
|
class TestResolveTarget:
|
|
def test_above_the_selected_song(self, tmp_path):
|
|
library = _library(tmp_path)
|
|
# Source row 1 is T2: the dangling id 99 isn't a row, so the
|
|
# playlist index is 2, not 1.
|
|
target = url_import.resolve_target(library, "a", 1, "library")
|
|
assert (target.pid, target.position) == ("a", 2)
|
|
assert "T2" in target.description and "Road Trip" in target.description
|
|
|
|
def test_no_selection_goes_to_the_end_of_the_playing_playlist(self, tmp_path):
|
|
library = _library(tmp_path)
|
|
target = url_import.resolve_target(library, "a", None, "playlist:b")
|
|
assert (target.pid, target.position) == ("b", None)
|
|
assert "(playing)" in target.description
|
|
|
|
def test_selection_beats_playing(self, tmp_path):
|
|
library = _library(tmp_path)
|
|
target = url_import.resolve_target(library, "a", 0, "playlist:b")
|
|
assert (target.pid, target.position) == ("a", 0)
|
|
|
|
def test_shown_playlist_when_nothing_is_playing(self, tmp_path):
|
|
library = _library(tmp_path)
|
|
target = url_import.resolve_target(library, "a", None, "")
|
|
assert (target.pid, target.position) == ("a", None)
|
|
|
|
def test_smart_and_folder_are_never_targets(self, tmp_path):
|
|
library = _library(tmp_path)
|
|
assert url_import.resolve_target(library, "s", 0, "library") is None
|
|
assert url_import.resolve_target(library, "f", None,
|
|
"playlist:s") is None
|
|
# ...but a shown smart playlist falls through to a playing regular one.
|
|
target = url_import.resolve_target(library, "s", 0, "playlist:b")
|
|
assert target.pid == "b"
|
|
|
|
def test_nothing_shown_or_playing(self, tmp_path):
|
|
library = _library(tmp_path)
|
|
assert url_import.resolve_target(library, "", None, "") is None
|
|
assert url_import.resolve_target(library, "", None, "library") is None
|
|
|
|
def test_a_deleted_playing_playlist_falls_through(self, tmp_path):
|
|
library = _library(tmp_path)
|
|
target = url_import.resolve_target(library, "a", None, "playlist:gone")
|
|
assert target.pid == "a"
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# C. the worker, against a fake yt-dlp
|
|
|
|
|
|
def _run_worker(url, dest, cancel_on_first=False):
|
|
worker = UrlImportWorker(url)
|
|
worker.temp_dir = dest
|
|
events = []
|
|
worker.item_started.connect(lambda i, n, t: events.append(("start", i, n, t)))
|
|
worker.progress.connect(lambda p: events.append(("progress", p)))
|
|
|
|
def on_file(path):
|
|
events.append(("file", path))
|
|
if cancel_on_first:
|
|
worker.cancel()
|
|
worker.downloaded.connect(on_file)
|
|
worker.finished.connect(lambda d: events.append(("finished", d)))
|
|
worker.failed.connect(lambda m: events.append(("failed", m)))
|
|
worker._busy = True
|
|
worker._run()
|
|
return worker, events
|
|
|
|
|
|
class TestWorker:
|
|
def test_reports_each_song_in_order(self, tmp_path, fake_ytdlp):
|
|
dest = tmp_path / "dl"
|
|
dest.mkdir()
|
|
worker, events = _run_worker("https://x.test/list", dest)
|
|
assert [e[0] for e in events] == [
|
|
"start", "progress", "file", "start", "progress", "file",
|
|
"finished"]
|
|
assert events[0] == ("start", 1, 2, "First Song")
|
|
assert events[2] == ("file", str(dest / "First Song [id1].mp3"))
|
|
assert events[-1][1] == {"downloaded": 2, "errors": [],
|
|
"cancelled": False}
|
|
assert not worker.busy()
|
|
|
|
def test_nothing_downloaded_is_a_failure_with_yt_dlps_reason(
|
|
self, tmp_path, fake_ytdlp, monkeypatch):
|
|
monkeypatch.setenv("FAKE_YTDLP_MODE", "fail")
|
|
dest = tmp_path / "dl"
|
|
dest.mkdir()
|
|
_worker, events = _run_worker("https://x.test/nope", dest)
|
|
assert events == [("failed",
|
|
"[generic] Unsupported URL: https://x.test/nope")]
|
|
|
|
def test_cancel_stops_after_the_song_in_hand(self, tmp_path, fake_ytdlp,
|
|
monkeypatch):
|
|
monkeypatch.setenv("FAKE_YTDLP_MODE", "slow")
|
|
dest = tmp_path / "dl"
|
|
dest.mkdir()
|
|
started = time.monotonic()
|
|
_worker, events = _run_worker("https://x.test/list", dest,
|
|
cancel_on_first=True)
|
|
assert time.monotonic() - started < 10 # didn't sit out the sleep
|
|
assert events[-1] == ("finished", {"downloaded": 1, "errors": [],
|
|
"cancelled": True})
|
|
|
|
def test_threaded_run_and_temp_dir_lifecycle(self, qapp, fake_ytdlp):
|
|
worker = UrlImportWorker("https://x.test/list")
|
|
out = {}
|
|
worker.finished.connect(lambda d: out.setdefault("finished", d))
|
|
worker.start()
|
|
temp = worker.temp_dir
|
|
assert temp.name.startswith("lintunes-url-")
|
|
_pump(qapp, lambda: "finished" in out)
|
|
assert sorted(p.name for p in temp.iterdir()) == [
|
|
"First Song [id1].mp3", "Second Song [id2].mp3"]
|
|
worker.cleanup()
|
|
assert not temp.exists()
|
|
worker.cleanup() # idempotent
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# D. the window: import, order, identify
|
|
|
|
|
|
@pytest.fixture
|
|
def window(qapp, tmp_path, isolated_config, monkeypatch):
|
|
from lintunes.gui.main_window import MainWindow
|
|
manager = LibraryManager(_library(tmp_path), tmp_path / "data")
|
|
win = MainWindow(manager, Preferences(tmp_path / "data"))
|
|
queued = []
|
|
monkeypatch.setattr(win, "_enqueue_identify",
|
|
lambda tracks: queued.extend(tracks))
|
|
win.identify_queued = queued
|
|
yield win
|
|
win.close()
|
|
|
|
|
|
class TestWindow:
|
|
def test_import_files_returns_what_it_imported(self, window, mp3_file):
|
|
imported = window.import_files([mp3_file])
|
|
assert len(imported) == 1 and imported[0].location
|
|
|
|
def test_a_batch_lands_above_the_selected_song_in_order(
|
|
self, window, tmp_path, mp3_file):
|
|
manager = window._manager
|
|
window._url_target = ImportTarget("a", 2, "")
|
|
window._url_inserted = window._url_imported = 0
|
|
window._url_identify = True
|
|
for name in ("First Song [id1].mp3", "Second Song [id2].mp3"):
|
|
path = tmp_path / name
|
|
shutil.copy(mp3_file, path)
|
|
window._on_url_downloaded(str(path))
|
|
|
|
ids = manager.library.playlists["a"].track_ids
|
|
new = [t.track_id for t in window.identify_queued]
|
|
assert len(new) == 2
|
|
assert ids == [1, 99, *new, 2, 3]
|
|
assert [manager.library.tracks[t].name for t in new] == [
|
|
"First Song [id1]", "Second Song [id2]"]
|
|
|
|
def test_stop_identifying_still_imports(self, window, tmp_path, mp3_file):
|
|
window._url_target = None
|
|
window._url_identify = False
|
|
path = tmp_path / "Song [x].mp3"
|
|
shutil.copy(mp3_file, path)
|
|
window._on_url_downloaded(str(path))
|
|
assert window._url_imported == 1
|
|
assert window.identify_queued == []
|
|
|
|
def test_end_to_end_through_the_menu(self, window, qapp, tmp_path,
|
|
fake_ytdlp, mp3_file, monkeypatch):
|
|
from lintunes.gui import main_window as mw
|
|
monkeypatch.setenv("FAKE_YTDLP_AUDIO", str(mp3_file))
|
|
|
|
class _Dialog:
|
|
def __init__(self, description, parent=None):
|
|
assert "T2" in description
|
|
def exec(self):
|
|
return True
|
|
def url(self):
|
|
return "https://x.test/list"
|
|
def add_to_playlist(self):
|
|
return True
|
|
monkeypatch.setattr(mw, "UrlImportDialog", _Dialog)
|
|
|
|
window._show_playlist("a")
|
|
window._playlist_view.reveal_track(2)
|
|
window._import_from_url()
|
|
worker = window._url_worker
|
|
temp = worker.temp_dir
|
|
_pump(qapp, lambda: worker.temp_dir is None)
|
|
|
|
ids = window._manager.library.playlists["a"].track_ids
|
|
new = [t.track_id for t in window.identify_queued]
|
|
assert ids == [1, 99, *new, 2, 3] and len(new) == 2
|
|
assert not temp.exists()
|
|
assert window._sync_progress.isHidden()
|
|
|
|
def test_menu_action_sits_under_add_files(self, window):
|
|
texts = [a.text() for a in window.menuBar().actions()[0].menu().actions()]
|
|
at = texts.index("Add Files to Library…")
|
|
assert texts[at + 1] == "Import from URL…"
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# E. no AcoustID key: the filename, and nothing else
|
|
|
|
|
|
def test_no_key_proposes_from_the_filename_without_asking_anyone(
|
|
qapp, monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(fingerprint, "fingerprint_file",
|
|
lambda *a: calls.append("fpcalc"))
|
|
monkeypatch.setattr(fingerprint, "lookup_fingerprint",
|
|
lambda *a, **k: calls.append("network"))
|
|
track = Track(track_id=7, name="B. Clem - Zuuso [1025657891]",
|
|
location="/m/Unknown Artist/Unknown Album/"
|
|
"B. Clem - Zuuso [1025657891].mp3")
|
|
identifier = fingerprint.TrackIdentifier()
|
|
out = {}
|
|
identifier.finished.connect(lambda r: out.setdefault("r", r))
|
|
identifier.identify(track, "")
|
|
_pump(qapp, lambda: "r" in out)
|
|
assert calls == []
|
|
[candidate] = out["r"]["candidates"]
|
|
assert candidate.source == "filename"
|
|
assert (candidate.artist, candidate.name) == ("B. Clem", "Zuuso")
|
|
|
|
|
|
def test_a_journal_in_a_foreign_shape_never_aborts_startup(tmp_path):
|
|
"""Found verifying this round: Round 48's dev fixture wrapped the journal
|
|
as {"machine", "tracks"} and the loader died on int("tracks"), taking
|
|
startup with it. A journal it can't read is skipped; a good one beside it
|
|
still counts."""
|
|
import json
|
|
from lintunes.storage.play_journal import PlayJournal
|
|
plays = tmp_path / "plays"
|
|
plays.mkdir()
|
|
(plays / "wrapped.json").write_text(json.dumps(
|
|
{"machine": "wrapped", "tracks": {"1": {"plays": 30}}}))
|
|
(plays / "list.json").write_text("[1, 2, 3]")
|
|
(plays / "good.json").write_text(json.dumps({"1": {"plays": 2}}))
|
|
library = Library()
|
|
library.tracks[1] = Track(track_id=1, name="T1", play_count=1)
|
|
PlayJournal(machine="me").load(tmp_path, library)
|
|
assert library.tracks[1].play_count == 3 # base 1 + good 2
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# F. the album art window
|
|
|
|
|
|
class TestFitSize:
|
|
def test_square_cover_fills_the_height_and_stays_square(self):
|
|
size = fit_size_(QSize(600, 600), QSize(1920, 1080))
|
|
assert size.width() == size.height() == 1080 - 48
|
|
|
|
def test_a_small_cover_is_scaled_up(self):
|
|
assert fit_size_(QSize(300, 300), QSize(1920, 1080)) == QSize(1032, 1032)
|
|
|
|
def test_a_wide_cover_is_bounded_by_the_width(self):
|
|
assert fit_size_(QSize(2000, 500), QSize(1920, 1080)) == QSize(1920, 480)
|
|
|
|
def test_an_empty_cover_takes_the_room(self):
|
|
assert fit_size_(QSize(0, 0), QSize(800, 600)) == QSize(800, 552)
|
|
|
|
|
|
def fit_size_(art, available):
|
|
from lintunes.gui.art_window import fit_size
|
|
return fit_size(art, available)
|
|
|
|
|
|
class _FakePlayer(QObject):
|
|
track_changed = pyqtSignal(object)
|
|
|
|
|
|
def _release(button, x=5, y=5):
|
|
return QMouseEvent(QEvent.Type.MouseButtonRelease, QPointF(x, y),
|
|
QPointF(x, y), button, Qt.MouseButton.NoButton,
|
|
Qt.KeyboardModifier.NoModifier)
|
|
|
|
|
|
class TestSidebarArtClick:
|
|
def _art(self):
|
|
from lintunes.gui.sidebar import SidebarArt
|
|
art = SidebarArt(_FakePlayer())
|
|
art.resize(100, 100)
|
|
return art
|
|
|
|
def test_one_click_opens_the_art_sized_to_the_screen(self, qapp):
|
|
art = self._art()
|
|
pixmap = QPixmap(64, 64)
|
|
pixmap.fill(Qt.GlobalColor.red)
|
|
art._pixmap = pixmap
|
|
art.mouseReleaseEvent(_release(Qt.MouseButton.LeftButton))
|
|
window = art._art_window
|
|
try:
|
|
assert window.isVisible()
|
|
available = window.screen().availableGeometry().size()
|
|
assert window.size() == fit_size_(pixmap.size(), available)
|
|
assert window.width() == window.height()
|
|
finally:
|
|
window.hide()
|
|
|
|
def test_no_art_or_right_click_opens_nothing(self, qapp):
|
|
art = self._art()
|
|
art.mouseReleaseEvent(_release(Qt.MouseButton.LeftButton))
|
|
assert not art._art_window.isVisible()
|
|
art._pixmap = QPixmap(10, 10)
|
|
art.mouseReleaseEvent(_release(Qt.MouseButton.RightButton))
|
|
assert not art._art_window.isVisible()
|
|
|
|
def test_a_release_outside_the_art_is_not_a_click(self, qapp):
|
|
art = self._art()
|
|
art._pixmap = QPixmap(10, 10)
|
|
art.mouseReleaseEvent(_release(Qt.MouseButton.LeftButton, 500, 500))
|
|
assert not art._art_window.isVisible()
|