From ac682774e9a3bd3834a1a8b3ba3852ecd3ba1432 Mon Sep 17 00:00:00 2001 From: trav Date: Fri, 2 Oct 2026 17:41:55 -0700 Subject: [PATCH] v0.36.0: a playlist link says so before it downloads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pasting a song from a YouTube Mix (watch?v=…&list=RD…) imported song after song: yt-dlp takes the whole list by default, and a Mix runs to thousands of rows and loops back on itself, so the linked song came in again and again with nothing on screen saying why. Now anything but a plain song link is listed first (--flat-playlist, streamed into the dialog). One song goes straight through. Several become a checklist: a song-in-a-list link ticks only its song and offers Just This Song without waiting, and a playlist link ticks everything. Rows are de-duplicated by id. The chosen songs download by their own pages, never the list link, and more than one gets a per-song progress window (Hide keeps downloading; the status-bar line brings it back). Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 14 + lintunes/__init__.py | 2 +- lintunes/gui/main_window.py | 63 ++++- lintunes/gui/url_import_dialog.py | 277 +++++++++++++++++- lintunes/gui/url_import_progress.py | 135 +++++++++ lintunes/url_import.py | 359 ++++++++++++++++++------ tests/test_round49.py | 19 +- tests/test_round59.py | 2 +- tests/test_round75.py | 416 ++++++++++++++++++++++++++++ 9 files changed, 1185 insertions(+), 102 deletions(-) create mode 100644 lintunes/gui/url_import_progress.py create mode 100644 tests/test_round75.py diff --git a/CLAUDE.md b/CLAUDE.md index 07b98cf..f277df5 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -351,6 +351,20 @@ persistence) → GUI (Qt widgets that read the manager and connect to its signal Firefox first), and a browser that worked is saved as `ytdlp_cookies_browser` in **`config.json`**, not preferences — which browser holds a YouTube login is this machine's business (Round 59). + **A link is listed before it downloads** (Round 75): trav pasted a song + from a YouTube Mix (`watch?v=…&list=RD…`), yt-dlp took the whole list, + and songs kept arriving with no word why. `link_kind` (URL alone) lets + only a plain song link skip the listing; everything else goes through + `PlaylistProbe` (`--flat-playlist`, one `LTENTRY` line per song, streamed + into the dialog's checklist). A song-in-a-list ticks only its song (plus a + **Just This Song** button that doesn't wait), a playlist link ticks all. + Measured on the real Mix: ~36 s to the first row, 2,967 rows, the linked + song **8 times** (a Mix loops), so rows are de-duplicated by id. Chosen + songs download by **their own page URLs**, never the list link or + `--playlist-items`, since a Mix reshuffles on every listing. Several songs + get `gui/url_import_progress.py`, a non-modal per-song list (Hide keeps + downloading; clicking the status-bar line reopens it). `LTSTART` carries + the id, and `ERROR: [site] : …` lines become `item_failed`. - **`lintunes/mpris.py`** — registers `org.mpris.MediaPlayer2.lintunes` over D-Bus so the desktop's media keys / now-playing popup control playback. Spacebar and diff --git a/lintunes/__init__.py b/lintunes/__init__.py index bd73aac..c55d6d9 100644 --- a/lintunes/__init__.py +++ b/lintunes/__init__.py @@ -1,3 +1,3 @@ """LinTunes — iTunes-style music library manager and player for Linux.""" -__version__ = "0.35.0" +__version__ = "0.36.0" diff --git a/lintunes/gui/main_window.py b/lintunes/gui/main_window.py index d93c901..17fbfc4 100644 --- a/lintunes/gui/main_window.py +++ b/lintunes/gui/main_window.py @@ -39,6 +39,7 @@ from lintunes.gui.library_view import LibraryView from lintunes.gui.playlist_view import PlaylistView from lintunes.gui.transport import TransportBar from lintunes.gui.url_import_dialog import UrlImportDialog +from lintunes.gui.url_import_progress import UrlImportProgress from lintunes.gui.info_dialog import InfoDialog from lintunes.gui.music_folder_dialog import MusicFolderDialog from lintunes.gui.preferences_dialog import PreferencesDialog @@ -57,6 +58,17 @@ def album_tracks(library, artist: str, album: str) -> list: and (t.album_artist or t.artist).casefold() == artist_cf] +class _StatusLabel(QLabel): + """The status bar's transfer line, which also answers a click.""" + + clicked = pyqtSignal() + + def mouseReleaseEvent(self, event): + if event.button() == Qt.MouseButton.LeftButton: + self.clicked.emit() + super().mouseReleaseEvent(event) + + class _CancelSyncButton(QPushButton): """Compact "✕" in the status bar's right corner that expands to say what it does ("cancel transfer") while hovered, then shrinks back. Expanding @@ -197,8 +209,10 @@ class MainWindow(QMainWindow): # is added *last* owns the corner and never moves. The cancel button # goes there deliberately: it is the one widget the user aims at, and # the label beside it is as wide as the song title of the moment. - self._sync_label = QLabel() + self._sync_label = _StatusLabel() self._sync_label.hide() + # A click brings back a hidden link-import progress window. + self._sync_label.clicked.connect(self._show_url_progress) self.statusBar().addPermanentWidget(self._sync_label) self._sync_progress = QProgressBar() self._sync_progress.setFixedWidth(160) @@ -235,6 +249,7 @@ class MainWindow(QMainWindow): self._url_imported = 0 self._url_identify = True # False after "Stop Identifying" self._url_note = "" + self._url_progress = None # UrlImportProgress, for several songs # Separate instance from the playback inhibitor: pausing music # mid-sync must not drop the sync's hold. Logout flag included so # GNOME's shutdown/restart dialog names the transfer as the blocker. @@ -1305,11 +1320,18 @@ class MainWindow(QMainWindow): "package)\nffmpeg: your distro's “ffmpeg” package") return target = self._url_import_target() - dialog = UrlImportDialog(target.description if target else None, self) - if not dialog.exec(): + cookies = load_config().get(url_import.COOKIES_KEY) + dialog = UrlImportDialog(target.description if target else None, self, + cookies_browser=cookies) + accepted = dialog.exec() + if dialog.cookies_used: # the listing needed them + self._remember_cookies_browser(dialog.cookies_used) + cookies = dialog.cookies_used + if not accepted: return if self._ensure_music_dir() is None: return + chosen = dialog.chosen_entries() self._url_target = target if dialog.add_to_playlist() else None self._url_inserted = 0 self._url_imported = 0 @@ -1318,9 +1340,22 @@ class MainWindow(QMainWindow): self._url_note = (f"tags proposed from filenames only ({problem})" if problem else "") + # The chosen songs by their own pages, never the list link: that + # would download the whole list again, and a Mix reshuffles. worker = url_import.UrlImportWorker( - dialog.url(), self, - cookies_browser=load_config().get(url_import.COOKIES_KEY)) + dialog.download_urls(), self, + cookies_browser=cookies) + if self._url_progress is not None: + self._url_progress.close() + self._url_progress.deleteLater() + self._url_progress = None + if len(chosen) > 1: + self._url_progress = UrlImportProgress(chosen, self) + self._url_progress.stop_requested.connect( + self._confirm_cancel_url_import) + worker.item_failed.connect(self._url_progress.failed) + worker.progress.connect(self._url_progress.progress) + self._url_progress.show() worker.cookies_used.connect(self._remember_cookies_browser) worker.item_started.connect(self._on_url_item_started) worker.progress.connect(self._sync_progress.setValue) @@ -1336,6 +1371,13 @@ class MainWindow(QMainWindow): self._sync_progress.show() worker.start() + def _show_url_progress(self): + if (self._url_progress is not None + and self._url_worker is not None and self._url_worker.busy()): + self._url_progress.show() + self._url_progress.raise_() + self._url_progress.activateWindow() + def _remember_cookies_browser(self, browser: str): """A bot-checked link got through on this browser's cookies: send them from the start next time. config.json, not preferences — the @@ -1345,7 +1387,10 @@ class MainWindow(QMainWindow): config[url_import.COOKIES_KEY] = browser save_config(config) - def _on_url_item_started(self, index: int, count: int, title: str): + def _on_url_item_started(self, index: int, count: int, item_id: str, + title: str): + if self._url_progress is not None: + self._url_progress.started(item_id) short = title if len(title) <= 40 else title[:39] + "…" where = f"{index} of {count} · " if count > 1 else "" self._sync_label.setText(f"Downloading {where}{short}") @@ -1366,6 +1411,8 @@ class MainWindow(QMainWindow): if pid: self._url_inserted += len(imported) self._url_imported += len(imported) + if self._url_progress is not None: + self._url_progress.landed(bool(imported)) if imported and self._url_identify: self._enqueue_identify(imported) @@ -1376,6 +1423,8 @@ class MainWindow(QMainWindow): def _on_url_finished(self, summary: dict): self._finish_url_import() + if self._url_progress is not None: + self._url_progress.finish(summary["cancelled"]) count = self._url_imported if summary["cancelled"]: msg = f"Import cancelled — {count} song(s) imported" @@ -1395,6 +1444,8 @@ class MainWindow(QMainWindow): def _on_url_failed(self, message: str): self._finish_url_import() + if self._url_progress is not None: + self._url_progress.finish(False) QMessageBox.warning(self, "Import from URL", f"Couldn't import from that link:\n\n{message}") diff --git a/lintunes/gui/url_import_dialog.py b/lintunes/gui/url_import_dialog.py index d7ad27b..eb98482 100644 --- a/lintunes/gui/url_import_dialog.py +++ b/lintunes/gui/url_import_dialog.py @@ -1,22 +1,74 @@ """File ▸ Import from URL…: paste a link, optionally add to the current playlist. Passive, like IdentifyDialog: the caller works out where the songs would go -(``url_import.resolve_target``) and passes its description in. The dialog only -reports the link and whether the box was ticked. +(``url_import.resolve_target``) and passes its description in. The dialog +reports the link, whether the box was ticked, and which songs to take. + +Since Round 75 it also says when a link is a playlist, *before* anything +downloads. A song link opened from a YouTube Mix carries ``list=RD…``, and +yt-dlp's default is to take the whole list, which runs to hundreds of songs. +Pasting one imported song after song with no hint why. So anything but a +plain song link is listed first (``url_import.PlaylistProbe``). One song +goes straight through. Several are shown as a checklist that fills while +yt-dlp is still listing. A link to a song *in* a list ticks only that song, +and a playlist link ticks them all. + +Two things measured against the real Mix: listing it took ~36 s before the +first row and returned 2,967 rows in which the linked song appeared 8 times +(the Mix loops back on itself). So rows are de-duplicated by id, and a +song-in-a-list link gets **Just This Song**, which works from the first +moment and downloads the song's own page with the list stripped off. """ +from urllib.parse import parse_qs, urlsplit + +from PyQt6.QtCore import Qt from PyQt6.QtWidgets import ( - QApplication, QCheckBox, QDialog, QDialogButtonBox, QLabel, QLineEdit, + QApplication, QCheckBox, QDialog, QDialogButtonBox, QHBoxLayout, QLabel, + QLineEdit, QListWidget, QListWidgetItem, QProgressBar, QPushButton, QVBoxLayout, ) -from lintunes.url_import import looks_like_url +from lintunes.url_import import PlaylistProbe, link_kind, looks_like_url + +ENTRY_ROLE = Qt.ItemDataRole.UserRole + + +def format_duration(seconds) -> str: + if not seconds: + return "" + seconds = int(seconds) + if seconds >= 3600: + return f"{seconds // 3600}:{seconds % 3600 // 60:02d}:{seconds % 60:02d}" + return f"{seconds // 60}:{seconds % 60:02d}" + + +def linked_song_id(url: str) -> str: + """The ``v=`` a song-in-a-list link points at ("" when it has none).""" + try: + parts = urlsplit(url) + except ValueError: + return "" + if (parts.hostname or "").lower() == "youtu.be": + return parts.path.strip("/") + return (parse_qs(parts.query).get("v") or [""])[0] class UrlImportDialog(QDialog): - def __init__(self, target_description: str | None, parent=None): + def __init__(self, target_description: str | None, parent=None, *, + cookies_browser: str | None = None): super().__init__(parent) self.setWindowTitle("Import from URL") self.setMinimumWidth(460) + self._cookies = cookies_browser + self._probe = None + self._kind = "song" + self._linked = "" + self._entries: list[dict] = [] # everything the probe found + self._picking = False # the checklist is showing + self._single = None # the one entry of a one-song link + self._seen: set[str] = set() # ids already listed (a Mix loops) + self._just_song = False # "Just This Song" was pressed + self.cookies_used = None # a browser the probe needed layout = QVBoxLayout(self) layout.addWidget(QLabel("Paste a link to a song, or to a playlist of " @@ -29,6 +81,37 @@ class UrlImportDialog(QDialog): self._url.selectAll() layout.addWidget(self._url) + # Listing what's at the link: a line of text and a busy bar. + self._status = QLabel() + self._status.setWordWrap(True) + self._status.hide() + layout.addWidget(self._status) + self._busy = QProgressBar() + self._busy.setRange(0, 0) + self._busy.setTextVisible(False) + self._busy.setMaximumHeight(8) + self._busy.hide() + layout.addWidget(self._busy) + + # The checklist, once the link turns out to hold several songs. + self._list = QListWidget() + self._list.setMinimumHeight(280) + self._list.setUniformItemSizes(True) + self._list.hide() + self._list.itemChanged.connect(self._update_ok) + layout.addWidget(self._list, 1) + self._pick_row = QHBoxLayout() + self._all = QPushButton("Select All") + self._none = QPushButton("Select None") + self._all.clicked.connect(lambda: self._check_all(True)) + self._none.clicked.connect(lambda: self._check_all(False)) + self._pick_row.addWidget(self._all) + self._pick_row.addWidget(self._none) + self._pick_row.addStretch(1) + for button in (self._all, self._none): + button.hide() + layout.addLayout(self._pick_row) + self._add = QCheckBox("and add to current playlist?") layout.addWidget(self._add) self._hint = QLabel(target_description @@ -46,18 +129,196 @@ class UrlImportDialog(QDialog): | QDialogButtonBox.StandardButton.Cancel) self._ok = buttons.button(QDialogButtonBox.StandardButton.Ok) self._ok.setText("Import") - buttons.accepted.connect(self.accept) + self._just = buttons.addButton("Just This Song", + QDialogButtonBox.ButtonRole.ActionRole) + self._just.setToolTip("Import only the song the link points at, " + "without waiting for the list") + self._just.hide() + self._just.clicked.connect(self._on_just_song) + buttons.accepted.connect(self._on_import) buttons.rejected.connect(self.reject) layout.addWidget(buttons) self._url.textChanged.connect(self._update_ok) self._update_ok() - def _update_ok(self): - self._ok.setEnabled(looks_like_url(self._url.text())) + # ---- what the caller reads ---- def url(self) -> str: return self._url.text().strip() def add_to_playlist(self) -> bool: return self._add.isEnabled() and self._add.isChecked() + + def download_urls(self) -> list[str]: + """What to hand yt-dlp: the ticked songs' own pages, the linked song + alone, or the link itself — never a list link the user didn't see.""" + if self._just_song: + return [f"https://www.youtube.com/watch?v={self._linked}"] + chosen = self.chosen_entries() + return [e["url"] for e in chosen] if chosen else [self.url()] + + def chosen_entries(self) -> list[dict]: + """The ticked songs, in the list's order. Empty means "download the + link itself" (one song, or a link that was never listed).""" + if not self._picking or self._just_song: + return [] + return [self._list.item(i).data(ENTRY_ROLE) + for i in range(self._list.count()) + if self._list.item(i).checkState() == Qt.CheckState.Checked] + + # ---- the states ---- + + def _checked_count(self) -> int: + return sum(1 for i in range(self._list.count()) + if self._list.item(i).checkState() == Qt.CheckState.Checked) + + def _update_ok(self, *_): + if self._picking: + n = self._checked_count() + self._ok.setText(f"Import {n} Song{'' if n == 1 else 's'}") + self._ok.setEnabled(n > 0) + elif self._probe is not None: + self._ok.setEnabled(False) + else: + self._ok.setText("Import") + self._ok.setEnabled(looks_like_url(self._url.text())) + + def _on_import(self): + if self._picking: + self._stop_probe() + self.accept() + return + url = self.url() + self._kind = link_kind(url) + if self._kind == "song": + self.accept() + return + self._linked = linked_song_id(url) if self._kind == "song_in_list" \ + else "" + self._entries = [] + self._seen = set() + self._single = None + self._url.setEnabled(False) + self._just.setVisible(bool(self._linked)) + self._status.setText("Checking what's at the link…") + self._status.show() + self._busy.show() + # No parent: a cancelled probe's thread can outlive this dialog, + # and must not be emitting from a deleted QObject when it does. + probe = PlaylistProbe(url, cookies_browser=self._cookies) + probe.entry.connect(self._on_entry) + probe.finished.connect(self._on_probe_finished) + probe.failed.connect(self._on_probe_failed) + probe.cookies_used.connect(self._on_cookies_used) + self._probe = probe + self._update_ok() + probe.start() + + def _on_just_song(self): + self._just_song = True + self._stop_probe() + self.accept() + + def _on_cookies_used(self, browser: str): + self.cookies_used = browser + self._cookies = browser + + def _on_entry(self, entry: dict): + if self.sender() is not self._probe: + return + if entry.get("id") in self._seen: + return + self._seen.add(entry.get("id")) + self._entries.append(entry) + if len(self._entries) == 1: + return # one song is not a playlist (yet) + if not self._picking: + self._show_picker() + for earlier in self._entries[:-1]: + self._add_row(earlier) + self._add_row(entry) + self._status.setText(self._count_text(listing=True)) + + def _show_picker(self): + self._picking = True + self._list.show() + self._all.show() + self._none.show() + self.resize(max(self.width(), 560), max(self.height(), 520)) + + def _add_row(self, entry: dict): + title = entry.get("title") or entry.get("id") or "(untitled)" + length = format_duration(entry.get("duration")) + item = QListWidgetItem(f"{title} {length}" if length else title) + item.setData(ENTRY_ROLE, entry) + item.setFlags(item.flags() | Qt.ItemFlag.ItemIsUserCheckable) + if self._kind == "song_in_list": + checked = entry.get("id") == self._linked + else: + checked = True + self._list.blockSignals(True) + item.setCheckState(Qt.CheckState.Checked if checked + else Qt.CheckState.Unchecked) + self._list.addItem(item) + self._list.blockSignals(False) + self._update_ok() + + def _count_text(self, listing: bool) -> str: + n = len(self._entries) + text = f"This link is a playlist: {n} songs" + if listing: + text += " so far, still listing…" + if self._kind == "song_in_list": + text += ("\nOnly the song you linked is ticked. Tick more to " + "import them too.") + return text + + def _on_probe_finished(self, count: int): + if self.sender() is not self._probe: + return + self._probe = None + self._busy.hide() + if self._picking: + self._status.setText(self._count_text(listing=False)) + if self._checked_count() == 0 and self._list.count(): + # The linked song wasn't in the list it came with. + self._list.item(0).setCheckState(Qt.CheckState.Checked) + self._update_ok() + return + # One song (or none): nothing to choose, so import it as linked. + self._single = self._entries[0] if self._entries else None + self.accept() + + def _on_probe_failed(self, message: str): + if self.sender() is not self._probe: + return + self._probe = None + self._busy.hide() + if self._picking: # it listed some songs, then gave up + self._status.setText(self._count_text(listing=False) + + f"\n(yt-dlp stopped listing: {message})") + self._update_ok() + return + self._url.setEnabled(True) + self._just.hide() + self._status.setText(f"Couldn't read that link:\n{message}") + self._update_ok() + + def _check_all(self, on: bool): + state = Qt.CheckState.Checked if on else Qt.CheckState.Unchecked + self._list.blockSignals(True) + for i in range(self._list.count()): + self._list.item(i).setCheckState(state) + self._list.blockSignals(False) + self._update_ok() + + def _stop_probe(self): + if self._probe is not None: + probe, self._probe = self._probe, None + probe.cancel() + self._busy.hide() + + def done(self, result): + self._stop_probe() + super().done(result) diff --git a/lintunes/gui/url_import_progress.py b/lintunes/gui/url_import_progress.py new file mode 100644 index 0000000..af569fe --- /dev/null +++ b/lintunes/gui/url_import_progress.py @@ -0,0 +1,135 @@ +"""The per-song list for a link import of more than one song (Round 75). + +The status bar has room for one title and one bar. That was fine for a song +and useless for a playlist: there was no way to see what was coming, what had +landed, or what yt-dlp had given up on. This window lists every chosen song +with where it's up to. It is non-modal, and Hide only closes it: the download +carries on, and clicking the status-bar line brings the window back. Stop +asks the same question the status bar's cancel button does. + +Rows are keyed by the song's id, which yt-dlp names when each song starts and +in each error it pins on one. A file landing belongs to the song that started +last, since yt-dlp downloads one at a time. +""" +from PyQt6.QtCore import Qt, pyqtSignal +from PyQt6.QtWidgets import ( + QDialog, QHBoxLayout, QHeaderView, QLabel, QPushButton, QTreeWidget, + QTreeWidgetItem, QVBoxLayout, +) + +WAITING = "Waiting" +IMPORTED = "Imported" +SKIPPED = "Skipped" +NOT_IMPORTED = "Downloaded, not imported" +NOT_DOWNLOADED = "Not downloaded" + + +class UrlImportProgress(QDialog): + stop_requested = pyqtSignal() + + def __init__(self, entries: list[dict], parent=None): + super().__init__(parent) + self.setWindowTitle("Importing from URL") + self.setModal(False) + self.resize(620, 460) + self._rows: dict[str, QTreeWidgetItem] = {} + self._current = "" + self._done = False + + layout = QVBoxLayout(self) + self._summary = QLabel() + layout.addWidget(self._summary) + self._tree = QTreeWidget() + self._tree.setColumnCount(2) + self._tree.setHeaderLabels(["Song", "Status"]) + self._tree.setRootIsDecorated(False) + self._tree.setUniformRowHeights(True) + header = self._tree.header() + header.setSectionResizeMode(0, QHeaderView.ResizeMode.Stretch) + header.setSectionResizeMode(1, QHeaderView.ResizeMode.ResizeToContents) + header.setStretchLastSection(False) + for entry in entries: + item = QTreeWidgetItem( + [entry.get("title") or entry.get("id") or "(untitled)", + WAITING]) + item.setToolTip(0, item.text(0)) + self._tree.addTopLevelItem(item) + if entry.get("id"): + self._rows[entry["id"]] = item + layout.addWidget(self._tree, 1) + + buttons = QHBoxLayout() + buttons.addStretch(1) + self._stop = QPushButton("Stop") + self._stop.clicked.connect(self.stop_requested) + self._hide = QPushButton("Hide") + self._hide.setToolTip("The download carries on. Click the progress " + "in the status bar to see this again.") + self._hide.clicked.connect(self.close) + buttons.addWidget(self._stop) + buttons.addWidget(self._hide) + layout.addLayout(buttons) + self._refresh_summary() + + # ---- fed by the worker, through MainWindow ---- + + def status(self, item_id: str) -> str: + item = self._rows.get(item_id) + return item.text(1) if item is not None else "" + + def _set(self, item_id: str, text: str): + item = self._rows.get(item_id) + if item is None: + return + item.setText(1, text) + item.setToolTip(1, text) + self._refresh_summary() + + def started(self, item_id: str): + self._current = item_id + self._set(item_id, "Downloading…") + item = self._rows.get(item_id) + if item is not None: + self._tree.scrollToItem(item) + + def progress(self, percent: int): + if self._current and self.status(self._current).startswith("Download"): + self._set(self._current, f"Downloading {percent}%") + + def landed(self, imported: bool): + """The file for the song that started last is here.""" + if self._current: + self._set(self._current, IMPORTED if imported else NOT_IMPORTED) + self._current = "" + + def failed(self, item_id: str, reason: str): + self._set(item_id, f"Couldn't download: {reason}") + if item_id == self._current: + self._current = "" + + def finish(self, cancelled: bool): + """Every song not accounted for was skipped (a stop) or quietly + dropped by yt-dlp.""" + self._done = True + leftover = SKIPPED if cancelled else NOT_DOWNLOADED + for item_id, item in self._rows.items(): + if item.text(1) == WAITING or item.text(1).startswith( + "Downloading"): + self._set(item_id, leftover) + self._stop.setEnabled(False) + self._hide.setText("Close") + self._refresh_summary() + + def _refresh_summary(self): + total = self._tree.topLevelItemCount() + statuses = [self._tree.topLevelItem(i).text(1) for i in range(total)] + imported = statuses.count(IMPORTED) + problems = sum(1 for s in statuses + if s.startswith("Couldn't") or s in (NOT_IMPORTED, + NOT_DOWNLOADED)) + text = f"{imported} of {total} songs imported" + if problems: + text += f" · {problems} couldn't be" + if self._done: + text = "Done: " + text + self._summary.setText(text) diff --git a/lintunes/url_import.py b/lintunes/url_import.py index 19e6074..90e7135 100644 --- a/lintunes/url_import.py +++ b/lintunes/url_import.py @@ -25,15 +25,23 @@ that downloads nothing because of that check is retried once with machine's ``config.json`` (``ytdlp_cookies_browser``) so later imports send the cookies from the start. Which browser holds a YouTube login is a fact about this machine, which is why it isn't in the synced preferences. + +Anything but a plain song link is *listed* before it downloads +(``link_kind``, ``PlaylistProbe``): a song opened from a YouTube Mix carries +``list=RD…``, and yt-dlp's default is to take the whole list. The songs the +user picks are then downloaded by their own pages, never by playlist +position, because a Mix comes back reshuffled every time it's listed. """ import os import shutil import signal import subprocess +import re import tempfile import threading from dataclasses import dataclass from pathlib import Path +from urllib.parse import parse_qs, urlsplit from PyQt6.QtCore import QObject, pyqtSignal @@ -47,6 +55,7 @@ SONG_ARGS = ["--extract-audio", "--audio-format", "mp3"] START = "LTSTART" FILE = "LTFILE" PROGRESS = "LTPROG" +ENTRY = "LTENTRY" # What YouTube says when it wants a signed-in browser. Matched on a fragment @@ -86,31 +95,85 @@ def looks_like_url(text: str) -> bool: and not any(c.isspace() for c in text)) -def build_command(url: str, dest_dir, - cookies_browser: str | None = None) -> list[str]: - """yt-dlp with the `song` flags, downloading into ``dest_dir``. +def link_kind(url: str) -> str: + """What a link names, from the URL alone: "song", "song_in_list" (a song + opened from a playlist or a YouTube Mix — yt-dlp would take the whole + list), "list", or "unknown" (another site; only a probe can tell). - ``before_dl`` names each item as it starts (with its place in a playlist), + Only "song" skips the probe, so a plain song link imports as fast as + ever. A Mix is the one that bites: it's a song link with ``list=RD…`` + on the end, and the list behind it runs to hundreds of songs. + """ + try: + parts = urlsplit((url or "").strip()) + except ValueError: + return "unknown" + host = (parts.hostname or "").lower() + if host.startswith("www."): + host = host[4:] + if host.startswith("m."): + host = host[2:] + query = parse_qs(parts.query) + has_list = bool(query.get("list")) + if host == "youtu.be": + return "song_in_list" if has_list else "song" + if host in ("youtube.com", "music.youtube.com"): + if parts.path == "/watch" and query.get("v"): + return "song_in_list" if has_list else "song" + if parts.path.startswith("/shorts/"): + return "song" + if parts.path == "/playlist" and has_list: + return "list" + return "unknown" + + +def build_command(urls, dest_dir, + cookies_browser: str | None = None) -> list[str]: + """yt-dlp with the `song` flags, downloading ``urls`` (one link, or a + list of them) into ``dest_dir``. + + ``before_dl`` names each item as it starts (its place in a playlist, its + id, its title), ``after_move`` gives the final mp3's path once ffmpeg has finished with it, and the progress template reduces the progress bar to one number per line. yt-dlp's default already carries on past an unavailable video in a playlist, so "import everything at the link" needs no extra flag. """ - cookies = (["--cookies-from-browser", cookies_browser] - if cookies_browser else []) + if isinstance(urls, str): + urls = [urls] return [ "yt-dlp", *SONG_ARGS, - *cookies, + *_cookie_args(cookies_browser), "-P", str(dest_dir), "--print", - f"before_dl:{START} %(playlist_index|1)s\t%(n_entries|1)s\t%(title)s", + f"before_dl:{START} %(playlist_index|1)s\t%(n_entries|1)s\t" + f"%(id)s\t%(title)s", "--print", f"after_move:{FILE} %(filepath)s", "--progress", "--newline", "--progress-template", f"download:{PROGRESS} %(progress._percent_str)s", + "--", *urls, + ] + + +def build_probe_command(url: str, + cookies_browser: str | None = None) -> list[str]: + """List what's at a link without downloading any of it: one ENTRY line + per song, printed as yt-dlp finds it. ``webpage_url`` is each song's own + page, for a playlist entry and a lone song alike, and the download is + then handed those pages, never playlist positions: a Mix comes back in + a different order every time it's listed.""" + return [ + "yt-dlp", "--flat-playlist", *_cookie_args(cookies_browser), + "--print", + f"{ENTRY} %(id)s\t%(duration|)s\t%(webpage_url|)s\t%(title)s", "--", url, ] +def _cookie_args(browser: str | None) -> list[str]: + return ["--cookies-from-browser", browser] if browser else [] + + def _int(text: str, default: int) -> int: try: return int(text) @@ -118,15 +181,30 @@ def _int(text: str, default: int) -> int: return default +def _duration(text: str) -> int | None: + try: + return int(float(text)) + except (TypeError, ValueError): + return None + + def parse_line(line: str): - """One line of yt-dlp output → ("start", i, n, title) | ("file", path) | - ("progress", percent) | None for everything else.""" + """One line of yt-dlp output → ("start", i, n, id, title) | + ("file", path) | ("progress", percent) | ("entry", {id, duration, url, + title}) | None for everything else.""" line = line.rstrip("\r\n") if line.startswith(START + " "): - parts = line[len(START) + 1:].split("\t", 2) - if len(parts) != 3: + parts = line[len(START) + 1:].split("\t", 3) + if len(parts) != 4: return None - return ("start", _int(parts[0], 1), _int(parts[1], 1), parts[2]) + return ("start", _int(parts[0], 1), _int(parts[1], 1), parts[2], + parts[3]) + if line.startswith(ENTRY + " "): + parts = line[len(ENTRY) + 1:].split("\t", 3) + if len(parts) != 4 or not parts[2].strip(): + return None + return ("entry", {"id": parts[0], "duration": _duration(parts[1]), + "url": parts[2], "title": parts[3]}) if line.startswith(FILE + " "): path = line[len(FILE) + 1:] return ("file", path) if path.strip() else None @@ -139,6 +217,16 @@ def parse_line(line: str): return None +# "ERROR: [youtube] : Video unavailable" — the id says which song. +_ERROR_ID = re.compile(r"^\[[^\]]+\] ([\w-]+): (.+)$") + + +def error_item(error: str) -> tuple[str, str] | None: + """(id, reason) for an error yt-dlp pinned on one song, else None.""" + m = _ERROR_ID.match(error or "") + return (m.group(1), m.group(2)) if m else None + + # ---- where the songs go ---- @dataclass @@ -196,35 +284,23 @@ def resolve_target(library, shown_pid: str, selected_row: int | None, return None -# ---- the download ---- +# ---- running yt-dlp ---- -class UrlImportWorker(QObject): - """Runs yt-dlp on a daemon thread, reporting each finished file as it - lands. Threading matches ExportWorker: a plain daemon thread, not a - QThread. +class _YtdlpRun(QObject): + """What the probe and the download share: one yt-dlp at a time on a + daemon thread (ExportWorker's threading, not a QThread), a cancel that + takes ffmpeg down with it, and the bot-check retry with a browser's + cookies. Subclasses supply the command and read the marker lines.""" - ``finished`` carries {"downloaded": int, "errors": [str], "cancelled": - bool}. ``failed`` means nothing downloaded at all, and carries yt-dlp's - last error line. ``cookies_used`` names the browser whose cookies got a - bot-checked link through, so the caller can remember it. - """ - - item_started = pyqtSignal(int, int, str) # index, count, title - progress = pyqtSignal(int) # percent of the current item - downloaded = pyqtSignal(str) # final path of one mp3 - finished = pyqtSignal(dict) failed = pyqtSignal(str) cookies_used = pyqtSignal(str) - def __init__(self, url: str, parent=None, *, - cookies_browser: str | None = None): + def __init__(self, parent=None, *, cookies_browser: str | None = None): super().__init__(parent) - self._url = url self._cookies = cookies_browser self._busy = False self._cancel = threading.Event() self._proc = None - self.temp_dir: Path | None = None def busy(self) -> bool: return self._busy @@ -233,6 +309,9 @@ class UrlImportWorker(QObject): self._cancel.set() self._terminate() + def cancelled(self) -> bool: + return self._cancel.is_set() + def _terminate(self): proc = self._proc if proc is None or proc.poll() is not None: @@ -244,20 +323,10 @@ class UrlImportWorker(QObject): except (OSError, AttributeError): proc.terminate() - def start(self): - if self._busy: - return + def _start_thread(self): self._busy = True - self.temp_dir = Path(tempfile.mkdtemp(prefix="lintunes-url-")) threading.Thread(target=self._run_guarded, daemon=True).start() - def cleanup(self): - """Remove our temp dir. Called by the GUI once it has imported every - file, meaning after ``finished`` or ``failed``.""" - if self.temp_dir is not None: - shutil.rmtree(self.temp_dir, ignore_errors=True) - self.temp_dir = None - def _run_guarded(self): try: self._run() @@ -265,51 +334,62 @@ class UrlImportWorker(QObject): self._busy = False self.failed.emit(str(e)) - def _run(self): + # -- subclass hooks -- + def _command(self, cookies_browser) -> list[str]: + raise NotImplementedError + + def _on_parsed(self, parsed) -> bool: + """One marker line. True when it counts as a result (a song found, + a song downloaded), which is what decides a bot-check retry.""" + raise NotImplementedError + + def _on_error(self, error: str): + pass + + def _run_with_retry(self): + """→ (results, ERROR lines, cookies used, exit code), or None when + yt-dlp couldn't be started (``failed`` has been emitted).""" cookies = self._cookies - count, errors, returncode = self._attempt(cookies) - if returncode is None: # yt-dlp couldn't even start - return + attempt = self._attempt(cookies) + if attempt is None: + return None + count, errors, returncode = attempt if (count == 0 and not self._cancel.is_set() and not cookies and any(is_bot_check(e) for e in errors)): # Nothing landed, so a second pass can't duplicate a song. cookies = default_cookies_browser() if cookies: - count, errors, returncode = self._attempt(cookies) - if returncode is None: - return + attempt = self._attempt(cookies) + if attempt is None: + return None + count, errors, returncode = attempt if count: self.cookies_used.emit(cookies) + return count, errors, cookies, returncode - cancelled = self._cancel.is_set() - self._busy = False - if count == 0 and not cancelled: - if errors and is_bot_check(errors[-1]): - self.failed.emit( - "YouTube wants to see a signed-in browser before it " - "will hand this over" - + (f" (tried {cookies}'s cookies)" if cookies else "") - + ".\n\nSign in to YouTube in " - + (cookies.title() if cookies else "your browser") - + " and try again.\n\nyt-dlp said: " + errors[-1]) - return + def _fail_nothing(self, errors, cookies, returncode, what: str): + """Nothing came back: say why, in words a bot check deserves.""" + if errors and is_bot_check(errors[-1]): self.failed.emit( - errors[-1] if errors else - f"yt-dlp found nothing to download (exit code " - f"{returncode})") + "YouTube wants to see a signed-in browser before it " + "will hand this over" + + (f" (tried {cookies}'s cookies)" if cookies else "") + + ".\n\nSign in to YouTube in " + + (cookies.title() if cookies else "your browser") + + " and try again.\n\nyt-dlp said: " + errors[-1]) return - self.finished.emit({"downloaded": count, "errors": errors, - "cancelled": cancelled}) + self.failed.emit( + errors[-1] if errors else + f"yt-dlp found nothing to {what} (exit code {returncode})") def _attempt(self, cookies_browser): - """One yt-dlp run → (songs downloaded, ERROR lines, exit code). The - exit code is None when yt-dlp couldn't be started, in which case - ``failed`` has already been emitted.""" + """One yt-dlp run → (results, ERROR lines, exit code), or None when + yt-dlp couldn't be started.""" errors: list[str] = [] count = 0 try: proc = subprocess.Popen( - build_command(self._url, self.temp_dir, cookies_browser), + self._command(cookies_browser), stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, # One stream: the markers are on stdout, the ERROR: lines # on stderr, and one reader can't deadlock on the other. @@ -319,7 +399,7 @@ class UrlImportWorker(QObject): except OSError as e: self._busy = False self.failed.emit(f"couldn't run yt-dlp: {e}") - return 0, errors, None + return None self._proc = proc if self._cancel.is_set(): # cancelled before the process existed self._terminate() @@ -328,15 +408,134 @@ class UrlImportWorker(QObject): parsed = parse_line(line) if parsed is None: if line.startswith("ERROR:"): - errors.append(line[len("ERROR:"):].strip()) + error = line[len("ERROR:"):].strip() + errors.append(error) + self._on_error(error) continue - kind = parsed[0] - if kind == "start": - self.item_started.emit(parsed[1], parsed[2], parsed[3]) - elif kind == "progress": - self.progress.emit(parsed[1]) - elif kind == "file" and not self._cancel.is_set(): + if self._on_parsed(parsed): count += 1 - self.downloaded.emit(parsed[1]) proc.wait() return count, errors, proc.returncode + + +class PlaylistProbe(_YtdlpRun): + """Lists the songs at a link before anything downloads, so a playlist + — or a song link that is secretly a 900-song Mix — is shown as one + and picked from, instead of imported whole. + + ``entry`` streams each song as yt-dlp finds it ({id, duration, url, + title}); a long Mix takes a while to list, and the dialog fills as it + goes. ``finished`` carries how many were found (cancelled or not); + ``failed`` means none were. + """ + + entry = pyqtSignal(dict) + finished = pyqtSignal(int) + + def __init__(self, url: str, parent=None, *, + cookies_browser: str | None = None): + super().__init__(parent, cookies_browser=cookies_browser) + self._url = url + + def start(self): + if not self._busy: + self._start_thread() + + def _command(self, cookies_browser): + return build_probe_command(self._url, cookies_browser) + + def _on_parsed(self, parsed): + if parsed[0] != "entry" or self._cancel.is_set(): + return False + self.entry.emit(parsed[1]) + return True + + def _run(self): + result = self._run_with_retry() + if result is None: + return + count, errors, cookies, returncode = result + self._busy = False + if count == 0 and not self._cancel.is_set(): + self._fail_nothing(errors, cookies, returncode, "import") + return + self.finished.emit(count) + + +class UrlImportWorker(_YtdlpRun): + """Runs yt-dlp on a daemon thread, reporting each finished file as it + lands. + + ``urls`` is one link, or the chosen songs' own pages after a probe. + ``item_started`` is (index, count, id, title): over several links the + worker counts them itself, since each link is a playlist of one to + yt-dlp. ``item_failed`` is (id, reason) for a song yt-dlp named in an + error. ``finished`` carries {"downloaded": int, "errors": [str], + "cancelled": bool}. ``failed`` means nothing downloaded at all, and + carries yt-dlp's last error line. ``cookies_used`` names the browser + whose cookies got a bot-checked link through, so the caller can + remember it. + """ + + item_started = pyqtSignal(int, int, str, str) # index, count, id, title + item_failed = pyqtSignal(str, str) # id, reason + progress = pyqtSignal(int) # percent of current item + downloaded = pyqtSignal(str) # final path of one mp3 + finished = pyqtSignal(dict) + + def __init__(self, urls, parent=None, *, + cookies_browser: str | None = None): + super().__init__(parent, cookies_browser=cookies_browser) + self._urls = [urls] if isinstance(urls, str) else list(urls) + self._started = 0 + self.temp_dir: Path | None = None + + def start(self): + if self._busy: + return + self.temp_dir = Path(tempfile.mkdtemp(prefix="lintunes-url-")) + self._start_thread() + + def cleanup(self): + """Remove our temp dir. Called by the GUI once it has imported every + file, meaning after ``finished`` or ``failed``.""" + if self.temp_dir is not None: + shutil.rmtree(self.temp_dir, ignore_errors=True) + self.temp_dir = None + + def _command(self, cookies_browser): + self._started = 0 + return build_command(self._urls, self.temp_dir, cookies_browser) + + def _on_parsed(self, parsed): + kind = parsed[0] + if kind == "start": + _, index, count, item_id, title = parsed + self._started += 1 + if len(self._urls) > 1: + index, count = self._started, len(self._urls) + self.item_started.emit(index, count, item_id, title) + elif kind == "progress": + self.progress.emit(parsed[1]) + elif kind == "file" and not self._cancel.is_set(): + self.downloaded.emit(parsed[1]) + return True + return False + + def _on_error(self, error): + item = error_item(error) + if item is not None: + self.item_failed.emit(*item) + + def _run(self): + result = self._run_with_retry() + if result is None: + return + count, errors, cookies, returncode = result + cancelled = self._cancel.is_set() + self._busy = False + if count == 0 and not cancelled: + self._fail_nothing(errors, cookies, returncode, "download") + return + self.finished.emit({"downloaded": count, "errors": errors, + "cancelled": cancelled}) diff --git a/tests/test_round49.py b/tests/test_round49.py index 112ba25..0dd4de2 100644 --- a/tests/test_round49.py +++ b/tests/test_round49.py @@ -50,7 +50,7 @@ if mode == "fail": 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(f"LTSTART {i}\\t{len(titles)}\\tid{i}\\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") @@ -127,9 +127,10 @@ class TestCommand: 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")), + ("LTSTART 2\t5\tab1\tSome Song\n", ("start", 2, 5, "ab1", "Some Song")), + ("LTSTART NA\tNA\tx\tSolo\n", ("start", 1, 1, "x", "Solo")), + ("LTSTART 1\t1\tx\tTab\tin title\n", + ("start", 1, 1, "x", "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)), @@ -210,7 +211,8 @@ 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.item_started.connect( + lambda i, n, vid, t: events.append(("start", i, n, t))) worker.progress.connect(lambda p: events.append(("progress", p))) def on_file(path): @@ -330,8 +332,13 @@ class TestWindow: monkeypatch.setenv("FAKE_YTDLP_AUDIO", str(mp3_file)) class _Dialog: - def __init__(self, description, parent=None): + cookies_used = None + def __init__(self, description, parent=None, **_kw): assert "T2" in description + def chosen_entries(self): + return [] + def download_urls(self): + return [self.url()] def exec(self): return True def url(self): diff --git a/tests/test_round59.py b/tests/test_round59.py index ce5e7bc..c4655c1 100644 --- a/tests/test_round59.py +++ b/tests/test_round59.py @@ -30,7 +30,7 @@ if "--cookies-from-browser" not in args or os.environ.get("FAKE_ALWAYS_BOT"): dest = args[args.index("-P") + 1] path = os.path.join(dest, "Song [m-7NEY4p5s0].mp3") open(path, "wb").write(b"ID3fake") -print("LTSTART 1\\t1\\tSong", flush=True) +print("LTSTART 1\\t1\\tm-7NEY4p5s0\\tSong", flush=True) print(f"LTFILE {path}", flush=True) ''' diff --git a/tests/test_round75.py b/tests/test_round75.py new file mode 100644 index 0000000..0eec876 --- /dev/null +++ b/tests/test_round75.py @@ -0,0 +1,416 @@ +"""Round 75 — a link that is a playlist says so before it downloads. + +trav pasted a song from a YouTube Mix (``watch?v=…&list=RD…``). yt-dlp's +default is to take the whole list, so LinTunes started importing song after +song from a list of over 900, with nothing on screen saying why. Now: + +- only a plain song link skips the listing (``link_kind``); +- anything else is listed first (``PlaylistProbe``). One song goes straight + through, and several become a checklist: a song-in-a-list link ticks only + its song, a playlist link ticks them all; +- the chosen songs are downloaded by their own pages, never the list link; +- an import of several songs gets a per-song progress window. + +yt-dlp itself is never run. A fake on PATH speaks the same marker lines. +""" +import os +import stat +import sys +import time + +import pytest +from PyQt6.QtCore import Qt + +from lintunes import url_import +from lintunes.gui.url_import_dialog import ( + UrlImportDialog, format_duration, linked_song_id, +) +from lintunes.gui.url_import_progress import ( + IMPORTED, NOT_DOWNLOADED, SKIPPED, WAITING, UrlImportProgress, +) +from lintunes.library_manager import LibraryManager +from lintunes.models import Library +from lintunes.preferences import Preferences +from lintunes.url_import import ( + PlaylistProbe, UrlImportWorker, error_item, link_kind, +) + +MIX = "https://www.youtube.com/watch?app=desktop&v=bbb&list=RDbbb&start_radio=1" + +# Lists $FAKE_ENTRIES songs (aaa, bbb, ccc, …) for --flat-playlist; downloads +# each URL it's given otherwise, failing any whose id starts with "bad". +# Every run appends its argv to $FAKE_YTDLP_LOG. +FAKE_YTDLP = '''#!/usr/bin/env python3 +import 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: + n = int(os.environ.get("FAKE_ENTRIES", "3")) + loops = int(os.environ.get("FAKE_LOOPS", "1")) + for i in list(range(n)) * loops: + vid = chr(ord("a") + i) * 3 + print(f"LTENTRY {vid}\\t{60 * (i + 1)}.0\\t" + f"https://www.youtube.com/watch?v={vid}\\tSong {vid}", flush=True) + if n == 0: + print("ERROR: [generic] Unsupported URL: " + urls[0], file=sys.stderr, + flush=True) + sys.exit(1) + 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}\\tSong {vid}", flush=True) + print("LTPROG 50.0%", flush=True) + path = os.path.join(dest, f"Song {vid} [{vid}].mp3") + src = os.environ.get("FAKE_YTDLP_AUDIO") + with open(path, "wb") as f: + f.write(open(src, "rb").read() if src else b"ID3fake") + print(f"LTFILE {path}", flush=True) +''' + + +@pytest.fixture +def fake(tmp_path, monkeypatch): + 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("XDG_CONFIG_HOME", str(tmp_path / "config")) + return lambda: ([line.split("\x1f") for line in + log.read_text().splitlines()] if log.exists() else []) + + +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. what a link names, from the URL alone + + +@pytest.mark.parametrize("url, kind", [ + ("https://www.youtube.com/watch?v=abc", "song"), + ("https://m.youtube.com/watch?v=abc&t=30", "song"), + ("https://music.youtube.com/watch?v=abc", "song"), + ("https://youtu.be/abc", "song"), + ("https://youtu.be/abc?list=PL1", "song_in_list"), + ("https://www.youtube.com/shorts/abc", "song"), + (MIX, "song_in_list"), + ("https://www.youtube.com/watch?v=abc&list=PLxyz&index=4", "song_in_list"), + ("https://www.youtube.com/playlist?list=PLxyz", "list"), + ("https://soundcloud.com/someone/sets/an-album", "unknown"), + ("https://soundcloud.com/someone/a-song", "unknown"), + ("https://www.youtube.com/@channel", "unknown"), + ("not a url", "unknown"), +]) +def test_link_kind(url, kind): + assert link_kind(url) == kind + + +def test_linked_song_id(): + assert linked_song_id(MIX) == "bbb" + assert linked_song_id("https://youtu.be/xyz?list=PL") == "xyz" + assert linked_song_id("https://soundcloud.com/a/b") == "" + + +def test_format_duration(): + assert format_duration(None) == "" + assert format_duration(65) == "1:05" + assert format_duration(3725) == "1:02:05" + + +def test_entry_and_error_lines(): + assert url_import.parse_line( + "LTENTRY abc\t195.0\thttps://y.test/watch?v=abc\tA: Song\tX\n") == ( + "entry", {"id": "abc", "duration": 195, + "url": "https://y.test/watch?v=abc", "title": "A: Song\tX"}) + assert url_import.parse_line("LTENTRY abc\tNA\thttps://y/v\tT\n")[1][ + "duration"] is None + # no page to download it from: not an entry we can use + assert url_import.parse_line("LTENTRY abc\t1\t\tT\n") is None + assert error_item("[youtube] -5o7lcpMDqs: Video unavailable") == ( + "-5o7lcpMDqs", "Video unavailable") + assert error_item("Unable to download webpage") is None + + +def test_several_urls_all_follow_the_double_dash(tmp_path): + cmd = url_import.build_command(["https://a/1", "https://a/2"], tmp_path) + assert cmd[cmd.index("--"):] == ["--", "https://a/1", "https://a/2"] + probe = url_import.build_probe_command(MIX, "firefox") + assert "--flat-playlist" in probe and probe[-2:] == ["--", MIX] + assert probe[probe.index("--cookies-from-browser") + 1] == "firefox" + assert "--extract-audio" not in probe # listing downloads nothing + + +# -------------------------------------------------------------------------- +# B. the probe and the worker + + +def test_probe_streams_entries_in_order(fake): + probe = PlaylistProbe(MIX) + got = [] + probe.entry.connect(lambda e: got.append(e["id"])) + probe.finished.connect(lambda n: got.append(("finished", n))) + probe._run() + assert got == ["aaa", "bbb", "ccc", ("finished", 3)] + assert not probe.busy() + + +def test_a_probe_that_finds_nothing_fails(fake, monkeypatch): + monkeypatch.setenv("FAKE_ENTRIES", "0") + probe = PlaylistProbe("https://x.test/nothing") + got = [] + probe.failed.connect(got.append) + probe._run() + assert got == ["[generic] Unsupported URL: https://x.test/nothing"] + + +def test_worker_counts_across_urls_and_names_failures(fake, tmp_path): + worker = UrlImportWorker(["https://y/watch?v=aaa", + "https://y/watch?v=badone", + "https://y/watch?v=ccc"]) + worker.temp_dir = tmp_path / "dl" + worker.temp_dir.mkdir() + events = [] + worker.item_started.connect( + lambda i, n, vid, t: events.append(("start", i, n, vid))) + worker.item_failed.connect(lambda vid, why: events.append(("bad", vid))) + worker.downloaded.connect(lambda p: events.append(("file",))) + worker.finished.connect(lambda d: events.append(("done", d["downloaded"]))) + worker._run() + assert events == [("start", 1, 3, "aaa"), ("file",), ("bad", "badone"), + ("start", 2, 3, "ccc"), ("file",), ("done", 2)] + + +# -------------------------------------------------------------------------- +# C. the dialog + + +def _dialog(qapp, url): + qapp.clipboard().setText(url) + dialog = UrlImportDialog("Adds to the end of “Chill”") + assert dialog.url() == url + return dialog + + +def test_a_plain_song_link_is_never_listed(qapp, fake): + dialog = _dialog(qapp, "https://www.youtube.com/watch?v=abc") + dialog._on_import() + assert dialog.result() == dialog.DialogCode.Accepted + assert dialog.chosen_entries() == [] + assert fake() == [] # yt-dlp was never run + + +def test_a_mix_ticks_only_the_linked_song(qapp, fake): + dialog = _dialog(qapp, MIX) + dialog._on_import() + assert not dialog._ok.isEnabled() # nothing to import while checking + _pump(qapp, lambda: dialog._probe is None) + assert dialog.result() != dialog.DialogCode.Accepted + assert dialog._list.count() == 3 + assert dialog._ok.text() == "Import 1 Song" + assert [e["id"] for e in dialog.chosen_entries()] == ["bbb"] + assert "3 songs" in dialog._status.text() + + dialog._list.item(2).setCheckState(Qt.CheckState.Checked) + assert dialog._ok.text() == "Import 2 Songs" + dialog._check_all(False) + assert not dialog._ok.isEnabled() + dialog._check_all(True) + assert dialog._ok.text() == "Import 3 Songs" + dialog._on_import() + assert dialog.result() == dialog.DialogCode.Accepted + assert [e["url"] for e in dialog.chosen_entries()] == [ + "https://www.youtube.com/watch?v=aaa", + "https://www.youtube.com/watch?v=bbb", + "https://www.youtube.com/watch?v=ccc"] + + +def test_a_looping_mix_lists_each_song_once(qapp, fake, monkeypatch): + monkeypatch.setenv("FAKE_LOOPS", "3") + dialog = _dialog(qapp, MIX) + dialog._on_import() + _pump(qapp, lambda: dialog._probe is None) + assert dialog._list.count() == 3 + assert [e["id"] for e in dialog.chosen_entries()] == ["bbb"] + + +def test_just_this_song_never_waits_for_the_list(qapp, fake): + dialog = _dialog(qapp, MIX) + dialog._on_import() + assert dialog._just.isVisibleTo(dialog) + dialog._just.click() # before a single row has arrived + assert dialog.result() == dialog.DialogCode.Accepted + assert dialog._probe is None + assert dialog.chosen_entries() == [] + assert dialog.download_urls() == ["https://www.youtube.com/watch?v=bbb"] + + +def test_download_urls(qapp, fake): + song = _dialog(qapp, "https://www.youtube.com/watch?v=abc") + song._on_import() + assert song.download_urls() == ["https://www.youtube.com/watch?v=abc"] + playlist = _dialog(qapp, "https://www.youtube.com/playlist?list=PLx") + playlist._on_import() + assert not playlist._just.isVisibleTo(playlist) # no song to pick + _pump(qapp, lambda: playlist._probe is None) + assert len(playlist.download_urls()) == 3 + + +def test_a_playlist_link_ticks_everything(qapp, fake): + dialog = _dialog(qapp, "https://www.youtube.com/playlist?list=PLx") + dialog._on_import() + _pump(qapp, lambda: dialog._probe is None) + assert len(dialog.chosen_entries()) == 3 + + +def test_a_one_song_listing_goes_straight_through(qapp, fake, monkeypatch): + monkeypatch.setenv("FAKE_ENTRIES", "1") + dialog = _dialog(qapp, "https://soundcloud.com/someone/a-song") + dialog._on_import() + _pump(qapp, lambda: dialog.result() == dialog.DialogCode.Accepted) + assert dialog.chosen_entries() == [] # the link itself downloads + + +def test_a_failed_listing_stays_open_and_says_why(qapp, fake, monkeypatch): + monkeypatch.setenv("FAKE_ENTRIES", "0") + dialog = _dialog(qapp, "https://x.test/nothing") + dialog._on_import() + _pump(qapp, lambda: dialog._probe is None) + assert dialog.result() != dialog.DialogCode.Accepted + assert "Unsupported URL" in dialog._status.text() + assert dialog._ok.isEnabled() and dialog._url.isEnabled() + + +# -------------------------------------------------------------------------- +# D. the progress window + + +def test_progress_rows_follow_the_download(qapp): + window = UrlImportProgress([{"id": "a", "title": "A"}, + {"id": "b", "title": "B"}, + {"id": "c", "title": "C"}, + {"id": "d", "title": "D"}]) + assert window.status("a") == WAITING + window.started("a") + window.progress(43) + assert window.status("a") == "Downloading 43%" + window.landed(True) + assert window.status("a") == IMPORTED + window.failed("b", "Video unavailable") + assert window.status("b") == "Couldn't download: Video unavailable" + window.started("c") + window.finish(cancelled=True) + assert window.status("c") == SKIPPED and window.status("d") == SKIPPED + assert window._summary.text().startswith("Done: 1 of 4 songs imported") + assert not window._stop.isEnabled() + + quiet = UrlImportProgress([{"id": "x", "title": "X"}]) + quiet.finish(cancelled=False) + assert quiet.status("x") == NOT_DOWNLOADED + + +# -------------------------------------------------------------------------- +# E. through the window + + +@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")) + monkeypatch.setattr(win, "_enqueue_identify", lambda tracks: None) + yield win + win.close() + + +def test_only_the_chosen_songs_download(window, qapp, fake, mp3_file, + monkeypatch): + from lintunes.gui import main_window as mw + monkeypatch.setenv("FAKE_YTDLP_AUDIO", str(mp3_file)) + chosen = [{"id": "aaa", "title": "Song aaa", + "url": "https://www.youtube.com/watch?v=aaa"}, + {"id": "ccc", "title": "Song ccc", + "url": "https://www.youtube.com/watch?v=ccc"}] + + class _Dialog: + cookies_used = None + def __init__(self, description, parent=None, **_kw): + pass + def exec(self): + return True + def url(self): + return MIX + def add_to_playlist(self): + return False + def chosen_entries(self): + return chosen + def download_urls(self): + return [c["url"] for c in chosen] + monkeypatch.setattr(mw, "UrlImportDialog", _Dialog) + + window._import_from_url() + worker = window._url_worker + _pump(qapp, lambda: worker.temp_dir is None) + + [argv] = fake() + assert argv[argv.index("--") + 1:] == [c["url"] for c in chosen] + assert MIX not in argv # never the whole list + progress = window._url_progress + assert progress is not None and progress.isVisible() + assert progress.status("aaa") == IMPORTED + assert progress.status("ccc") == IMPORTED + assert len(window._manager.library.tracks) == 2 + + # Hidden, it comes back from the status bar only while downloading. + progress.close() + window._sync_label.clicked.emit() + assert not progress.isVisible() # done downloading: stays hidden + worker._busy = True # as if still going + window._sync_label.clicked.emit() + assert progress.isVisible() + worker._busy = False + + +def test_one_song_gets_no_progress_window(window, qapp, fake, mp3_file, + monkeypatch): + from lintunes.gui import main_window as mw + monkeypatch.setenv("FAKE_YTDLP_AUDIO", str(mp3_file)) + + class _Dialog: + cookies_used = None + def __init__(self, *a, **kw): + pass + def exec(self): + return True + def url(self): + return "https://www.youtube.com/watch?v=aaa" + def add_to_playlist(self): + return False + def chosen_entries(self): + return [] + def download_urls(self): + return [self.url()] + monkeypatch.setattr(mw, "UrlImportDialog", _Dialog) + window._import_from_url() + worker = window._url_worker + _pump(qapp, lambda: worker.temp_dir is None) + assert window._url_progress is None + assert len(window._manager.library.tracks) == 1