travandClaude Opus 5.5
ac682774e9
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 <noreply@anthropic.com>
542 lines
20 KiB
Python
542 lines
20 KiB
Python
"""File ▸ Import from URL… — trav's `song` shell helper, run from inside LinTunes.
|
||
|
||
`song` is ``yt-dlp --extract-audio --audio-format mp3 "$@"``. This module runs
|
||
exactly those flags; everything else on the command line is plumbing for
|
||
reading back what yt-dlp did. yt-dlp is a *runtime* CLI tool, detected with
|
||
``shutil.which`` like fpcalc and ffmpeg, never a pip dependency.
|
||
|
||
Shaped like ``fingerprint.py``: pure helpers first (the command line, the
|
||
marker parser, and where the songs should go), then the subprocess on a daemon
|
||
thread behind ``UrlImportWorker``.
|
||
|
||
Downloads land in a private temp dir and are handed to the GUI one at a time,
|
||
which imports each as it arrives. The normal importer *copies* it into the
|
||
organized tree, so the temp dir only ever holds our own downloads. The GUI
|
||
deletes it (``cleanup``) once it has seen ``finished`` or ``failed``: those
|
||
signals are queued after every ``downloaded``, so by then every file has been
|
||
copied. yt-dlp's default filename, ``Title [id].mp3``, is kept on purpose,
|
||
because it's exactly what ``filename_tags.parse_filename`` reads.
|
||
|
||
YouTube sometimes answers an anonymous request with "Sign in to confirm
|
||
you're not a bot" (it flags the *network*, so cell connections see it
|
||
most). The cure yt-dlp names is the browser's own YouTube cookies. So a run
|
||
that downloads nothing because of that check is retried once with
|
||
``--cookies-from-browser``, and the GUI remembers the browser in this
|
||
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
|
||
|
||
from lintunes.models.playlist import PlaylistType
|
||
|
||
# The `song` helper's own flags, verbatim.
|
||
SONG_ARGS = ["--extract-audio", "--audio-format", "mp3"]
|
||
|
||
# Line prefixes we ask yt-dlp to print, so its output can be read reliably
|
||
# instead of scraped.
|
||
START = "LTSTART"
|
||
FILE = "LTFILE"
|
||
PROGRESS = "LTPROG"
|
||
ENTRY = "LTENTRY"
|
||
|
||
|
||
# What YouTube says when it wants a signed-in browser. Matched on a fragment
|
||
# that sidesteps the apostrophe, which yt-dlp prints as a curly ’.
|
||
BOT_CHECK = "to confirm you"
|
||
|
||
# The config.json key naming the browser yt-dlp borrows cookies from.
|
||
COOKIES_KEY = "ytdlp_cookies_browser"
|
||
|
||
|
||
def is_bot_check(error: str) -> bool:
|
||
return BOT_CHECK in (error or "") and "bot" in error
|
||
|
||
|
||
def default_cookies_browser() -> str | None:
|
||
"""The browser whose cookies to borrow: the first one with a profile on
|
||
this machine, or None when there's nothing to borrow from."""
|
||
home = Path.home()
|
||
config = Path(os.environ.get("XDG_CONFIG_HOME", str(home / ".config")))
|
||
for browser, profile in (("firefox", home / ".mozilla" / "firefox"),
|
||
("chromium", config / "chromium"),
|
||
("chrome", config / "google-chrome")):
|
||
if profile.is_dir():
|
||
return browser
|
||
return None
|
||
|
||
|
||
def ytdlp_available() -> bool:
|
||
"""Whether the yt-dlp *binary* is on PATH (the fpcalc_available pattern)."""
|
||
return shutil.which("yt-dlp") is not None
|
||
|
||
|
||
def looks_like_url(text: str) -> bool:
|
||
text = (text or "").strip()
|
||
return (text.startswith(("http://", "https://"))
|
||
and len(text) > len("https://")
|
||
and not any(c.isspace() for c in text))
|
||
|
||
|
||
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).
|
||
|
||
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.
|
||
"""
|
||
if isinstance(urls, str):
|
||
urls = [urls]
|
||
return [
|
||
"yt-dlp", *SONG_ARGS,
|
||
*_cookie_args(cookies_browser),
|
||
"-P", str(dest_dir),
|
||
"--print",
|
||
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)
|
||
except (TypeError, ValueError):
|
||
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, 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", 3)
|
||
if len(parts) != 4:
|
||
return None
|
||
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
|
||
if line.startswith(PROGRESS + " "):
|
||
text = line[len(PROGRESS) + 1:].strip().rstrip("%")
|
||
try:
|
||
return ("progress", max(0, min(100, int(float(text)))))
|
||
except ValueError:
|
||
return None
|
||
return None
|
||
|
||
|
||
# "ERROR: [youtube] <id>: 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
|
||
class ImportTarget:
|
||
pid: str
|
||
position: int | None # playlist index to insert before; None appends
|
||
description: str # what the dialog tells the user
|
||
|
||
|
||
def accepts_adds(playlist) -> bool:
|
||
"""A playlist you can add songs to by hand: not a folder, not smart (its
|
||
membership comes from its rules), not a system list."""
|
||
return (playlist is not None
|
||
and playlist.playlist_type == PlaylistType.REGULAR
|
||
and not playlist.is_system)
|
||
|
||
|
||
def resolve_target(library, shown_pid: str, selected_row: int | None,
|
||
playing_context: str) -> ImportTarget | None:
|
||
"""Where "and add to current playlist?" puts the songs.
|
||
|
||
1. A playlist is shown and a song in it is selected: above that song.
|
||
2. Otherwise, the playlist that's playing: at its end.
|
||
3. Otherwise, the playlist that's shown: at its end.
|
||
4. Otherwise nowhere, and the checkbox is grayed out.
|
||
|
||
``selected_row`` is the selected song's row in the table's *source* model,
|
||
which is playlist order whatever the column sort is. That model skips
|
||
ids with no track behind them, so the row is mapped back to a real index
|
||
into ``track_ids``.
|
||
"""
|
||
shown = library.playlists.get(shown_pid) if shown_pid else None
|
||
if not accepts_adds(shown):
|
||
shown = None
|
||
|
||
if shown is not None and selected_row is not None:
|
||
present = [i for i, tid in enumerate(shown.track_ids)
|
||
if tid in library.tracks]
|
||
if 0 <= selected_row < len(present):
|
||
index = present[selected_row]
|
||
song = library.tracks[shown.track_ids[index]].name or "(untitled)"
|
||
return ImportTarget(shown.persistent_id, index,
|
||
f"Adds above “{song}” in “{shown.name}”")
|
||
|
||
if playing_context.startswith("playlist:"):
|
||
playing = library.playlists.get(playing_context.split(":", 1)[1])
|
||
if accepts_adds(playing):
|
||
return ImportTarget(playing.persistent_id, None,
|
||
f"Adds to the end of “{playing.name}” "
|
||
f"(playing)")
|
||
|
||
if shown is not None:
|
||
return ImportTarget(shown.persistent_id, None,
|
||
f"Adds to the end of “{shown.name}”")
|
||
return None
|
||
|
||
|
||
# ---- running yt-dlp ----
|
||
|
||
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."""
|
||
|
||
failed = pyqtSignal(str)
|
||
cookies_used = pyqtSignal(str)
|
||
|
||
def __init__(self, parent=None, *, cookies_browser: str | None = None):
|
||
super().__init__(parent)
|
||
self._cookies = cookies_browser
|
||
self._busy = False
|
||
self._cancel = threading.Event()
|
||
self._proc = None
|
||
|
||
def busy(self) -> bool:
|
||
return self._busy
|
||
|
||
def cancel(self):
|
||
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:
|
||
return
|
||
# yt-dlp runs ffmpeg as a child. Signal the whole session, so a cancel
|
||
# during conversion doesn't leave ffmpeg writing into the temp dir.
|
||
try:
|
||
os.killpg(proc.pid, signal.SIGTERM)
|
||
except (OSError, AttributeError):
|
||
proc.terminate()
|
||
|
||
def _start_thread(self):
|
||
self._busy = True
|
||
threading.Thread(target=self._run_guarded, daemon=True).start()
|
||
|
||
def _run_guarded(self):
|
||
try:
|
||
self._run()
|
||
except Exception as e: # never leave the GUI waiting forever
|
||
self._busy = False
|
||
self.failed.emit(str(e))
|
||
|
||
# -- 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
|
||
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:
|
||
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
|
||
|
||
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(
|
||
"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.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 → (results, ERROR lines, exit code), or None when
|
||
yt-dlp couldn't be started."""
|
||
errors: list[str] = []
|
||
count = 0
|
||
try:
|
||
proc = subprocess.Popen(
|
||
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.
|
||
stderr=subprocess.STDOUT,
|
||
text=True, encoding="utf-8", errors="replace", bufsize=1,
|
||
start_new_session=True)
|
||
except OSError as e:
|
||
self._busy = False
|
||
self.failed.emit(f"couldn't run yt-dlp: {e}")
|
||
return None
|
||
self._proc = proc
|
||
if self._cancel.is_set(): # cancelled before the process existed
|
||
self._terminate()
|
||
|
||
for line in proc.stdout:
|
||
parsed = parse_line(line)
|
||
if parsed is None:
|
||
if line.startswith("ERROR:"):
|
||
error = line[len("ERROR:"):].strip()
|
||
errors.append(error)
|
||
self._on_error(error)
|
||
continue
|
||
if self._on_parsed(parsed):
|
||
count += 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})
|