trav
93399a1706
Import from URL now gets past YouTube's "Sign in to confirm you're not a bot": a run that downloads nothing for that reason retries once with yt-dlp's --cookies-from-browser (Firefox first), and a browser that worked is remembered in this machine's config.json so later imports send the cookies from the start.
343 lines
13 KiB
Python
343 lines
13 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.
|
||
"""
|
||
import os
|
||
import shutil
|
||
import signal
|
||
import subprocess
|
||
import tempfile
|
||
import threading
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
|
||
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"
|
||
|
||
|
||
# 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 build_command(url: str, dest_dir,
|
||
cookies_browser: str | None = None) -> list[str]:
|
||
"""yt-dlp with the `song` flags, downloading into ``dest_dir``.
|
||
|
||
``before_dl`` names each item as it starts (with its place in a playlist),
|
||
``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 [])
|
||
return [
|
||
"yt-dlp", *SONG_ARGS,
|
||
*cookies,
|
||
"-P", str(dest_dir),
|
||
"--print",
|
||
f"before_dl:{START} %(playlist_index|1)s\t%(n_entries|1)s\t%(title)s",
|
||
"--print", f"after_move:{FILE} %(filepath)s",
|
||
"--progress", "--newline",
|
||
"--progress-template", f"download:{PROGRESS} %(progress._percent_str)s",
|
||
"--", url,
|
||
]
|
||
|
||
|
||
def _int(text: str, default: int) -> int:
|
||
try:
|
||
return int(text)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def parse_line(line: str):
|
||
"""One line of yt-dlp output → ("start", i, n, title) | ("file", path) |
|
||
("progress", percent) | 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:
|
||
return None
|
||
return ("start", _int(parts[0], 1), _int(parts[1], 1), parts[2])
|
||
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
|
||
|
||
|
||
# ---- 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
|
||
|
||
|
||
# ---- the download ----
|
||
|
||
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.
|
||
|
||
``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):
|
||
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
|
||
|
||
def cancel(self):
|
||
self._cancel.set()
|
||
self._terminate()
|
||
|
||
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(self):
|
||
if self._busy:
|
||
return
|
||
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()
|
||
except Exception as e: # never leave the GUI waiting forever
|
||
self._busy = False
|
||
self.failed.emit(str(e))
|
||
|
||
def _run(self):
|
||
cookies = self._cookies
|
||
count, errors, returncode = self._attempt(cookies)
|
||
if returncode is None: # yt-dlp couldn't even start
|
||
return
|
||
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
|
||
if count:
|
||
self.cookies_used.emit(cookies)
|
||
|
||
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
|
||
self.failed.emit(
|
||
errors[-1] if errors else
|
||
f"yt-dlp found nothing to download (exit code "
|
||
f"{returncode})")
|
||
return
|
||
self.finished.emit({"downloaded": count, "errors": errors,
|
||
"cancelled": cancelled})
|
||
|
||
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."""
|
||
errors: list[str] = []
|
||
count = 0
|
||
try:
|
||
proc = subprocess.Popen(
|
||
build_command(self._url, self.temp_dir, 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 0, errors, 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:"):
|
||
errors.append(line[len("ERROR:"):].strip())
|
||
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():
|
||
count += 1
|
||
self.downloaded.emit(parsed[1])
|
||
proc.wait()
|
||
return count, errors, proc.returncode
|