From e94b672644724400702a4d40d5a5d92c7fc9ed9b Mon Sep 17 00:00:00 2001 From: Conway Date: Thu, 11 Jun 2026 17:37:22 -0400 Subject: v1.7 - playlist distinction support + foolproofing Co-Authored-By: Claude Opus 4.8 --- app.py | 195 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 169 insertions(+), 26 deletions(-) (limited to 'app.py') diff --git a/app.py b/app.py index 1db2353..b6cff69 100644 --- a/app.py +++ b/app.py @@ -12,7 +12,9 @@ from __future__ import annotations import os import queue +import re import readline # noqa: F401 (importing enables line editing in input()) +import shutil import subprocess import sys import tempfile @@ -24,10 +26,18 @@ import yt_dlp from yt_dlp.postprocessor import EmbedThumbnailPP from yt_dlp.postprocessor.common import PostProcessor -VERSION = "1.5" +VERSION = "1.7" DOWNLOAD_DIR = Path.home() / "Downloads" PROMPT = "dlit> " +# A pasted link is first *resolved* (a quick, flat metadata fetch) to decide +# whether it is one video or a playlist. A playlist fans out into one queued +# job per entry, each labelled "(PlaylistName - i/N)". PLAYLIST_LIMIT bounds how +# many entries a single paste may expand into, so a pathological or effectively +# endless feed (e.g. youtube.com's recommendations) can't flood the queue. +PLAYLIST_LIMIT = 200 +SOCKET_TIMEOUT = 30 # seconds; bounds network stalls during resolve + download + PRESETS: dict[str, dict] = { "mp3": { "format": "bestaudio/best", @@ -49,9 +59,29 @@ PRESETS: dict[str, dict] = { "none": {}, } +class _QuietLogger: + """Swallow yt-dlp's own console output. + + yt-dlp writes errors/warnings straight to stderr even under quiet=True + (see YoutubeDL.to_stderr). Those bytes bypass our cursor save/restore line + management and land on top of the `dlit>` prompt, corrupting the display — + most visibly on an invalid URL, which emits an "ERROR: ..." line. Routing + everything through a logger keeps all output under our control; the failure + itself still surfaces via the DownloadError that worker() catches. + """ + + def debug(self, msg: str) -> None: ... + def info(self, msg: str) -> None: ... + def warning(self, msg: str) -> None: ... + def error(self, msg: str) -> None: ... + + +_QUIET_LOGGER = _QuietLogger() + _print_lock = threading.Lock() _line_distance: dict[int, int] = {} # job_id -> rows above the current cursor position _job_title: dict[int, str] = {} # job_id -> resolved title (once known) +_job_tag: dict[int, str] = {} # job_id -> "(PlaylistName - i/N) " prefix, "" for singles _RESET = "\x1b[0m" ICON_QUEUED = f"\x1b[35m~{_RESET}" # purple @@ -60,16 +90,64 @@ ICON_CONVERTING = f"\x1b[34m↻{_RESET}" # blue ICON_DONE = f"\x1b[32m✓{_RESET}" # green ICON_FAILED = f"\x1b[31m✗{_RESET}" # red +_SGR_RE = re.compile(r"\x1b\[[0-9;]*m") + + +def _clamp(text: str) -> str: + """Trim a status line to the terminal width, counting visible columns only. + + Job lines are rewritten in place by relying on each one occupying exactly + one terminal row (see update_job_line). A line wider than the terminal wraps + onto the row below — the prompt — and the cursor save/restore math no longer + lands where it should, corrupting the display. Clamping keeps every line to a + single row. SGR colour codes don't occupy columns, so they're copied through + without counting toward the budget. + """ + width = shutil.get_terminal_size((80, 24)).columns - 1 + if width <= 0: + return text + out: list[str] = [] + visible = i = 0 + while i < len(text): + m = _SGR_RE.match(text, i) + if m: + out.append(m.group()) + i = m.end() + continue + if visible >= width: + return "".join(out) + _RESET # reset in case we cut mid-colour + out.append(text[i]) + visible += 1 + i += 1 + return "".join(out) + + +def add_job_lines(items: list[tuple[int, str]]) -> None: + """Main thread: print one or more new tracked lines above the next prompt. + + Submitting a link can fan out into several lines at once (one per playlist + entry). Printing N lines pushes the next prompt down by N+1 rows — the N new + lines plus the just-submitted prompt line that stays on screen — so every + existing tracked line moves up by that much. The new lines, printed top to + bottom, sit at distances N, N-1, … 1 above the coming prompt. With N=1 this + reduces to the original "+= 2, new line at distance 1". + """ + with _print_lock: + n = len(items) + for k in _line_distance: + _line_distance[k] += n + 1 + for offset, (job_id, text) in enumerate(items): + _line_distance[job_id] = n - offset + sys.stdout.write(_clamp(text) + "\n") + sys.stdout.flush() + -def add_job_line(job_id: int, text: str) -> None: - """Main thread: print a new tracked line above the next prompt.""" +def notice(text: str) -> None: + """Main thread: print a one-off, untracked line above the next prompt.""" with _print_lock: - # Existing tracked lines move up by 2 rows: the just-submitted prompt line - # stays on screen, and the new tracked line we're about to print is one more. for k in _line_distance: _line_distance[k] += 2 - _line_distance[job_id] = 1 - sys.stdout.write(text + "\n") + sys.stdout.write(_clamp(text) + "\n") sys.stdout.flush() @@ -83,7 +161,7 @@ def update_job_line(job_id: int, text: str) -> None: "\x1b7" # save cursor (DECSC) f"\x1b[{d}A" # up d rows "\r\x1b[K" # clear that line - f"{text}" + f"{_clamp(text)}" "\x1b8" # restore cursor (DECRC) ) sys.stdout.flush() @@ -98,7 +176,8 @@ def make_opts(job_id: int, preset: str) -> dict: if title: _job_title[job_id] = title status = d.get("status") - shown_title = _job_title.get(job_id, "…")[:60] + tag = _job_tag.get(job_id, "") + shown_title = tag + _job_title.get(job_id, "…")[:60] if status == "downloading": total = d.get("total_bytes") or d.get("total_bytes_estimate") or 0 done = d.get("downloaded_bytes") or 0 @@ -122,11 +201,58 @@ def make_opts(job_id: int, preset: str) -> dict: "no_warnings": True, "no_color": True, "noprogress": True, + "logger": _QUIET_LOGGER, + "socket_timeout": SOCKET_TIMEOUT, "progress_hooks": [progress], **PRESETS[preset], } +def resolve_opts() -> dict: + """Options for the quick pre-download probe that classifies a pasted link. + + extract_flat keeps it cheap: playlist entries come back as bare references + (id/title/url) without fetching each video. noplaylist=True preserves the + download behaviour — a video that merely sits inside a playlist + (watch?v=…&list=…) resolves to the single video; only a pure playlist or feed + URL fans out. playlistend bounds how far an endless feed is walked. + """ + return { + "quiet": True, + "no_warnings": True, + "no_color": True, + "noplaylist": True, + "logger": _QUIET_LOGGER, + "socket_timeout": SOCKET_TIMEOUT, + "extract_flat": "in_playlist", + "playlistend": PLAYLIST_LIMIT, + } + + +def resolve(url: str) -> tuple[bool, str, list[tuple[str, str]]]: + """Classify a pasted link. Returns (is_playlist, name, [(entry_url, title)]). + + A single video comes back as (False, title, [(url, title)]). A playlist or + feed comes back as (True, playlist_name, [...]) — possibly empty, e.g. a + logged-out recommendations feed, in which case nothing is queued. + """ + with yt_dlp.YoutubeDL(resolve_opts()) as ydl: + info = ydl.extract_info(url, download=False) or {} + entries = info.get("entries") + if entries is None: + title = info.get("title") or url + return False, title, [(info.get("webpage_url") or url, title)] + name = info.get("title") or "playlist" + items = [] + for e in entries: + if not e: + continue + eurl = e.get("url") or e.get("webpage_url") or e.get("id") + if eurl: + items.append((eurl, e.get("title") or eurl)) + return True, name, items + + class SquareThumbnailCropper(PostProcessor): """Detect a square cover image padded into a 16:9 YouTube thumbnail and crop the bars off before EmbedThumbnail runs. @@ -271,6 +397,7 @@ def worker(jobs: "queue.Queue") -> None: if item is None: return job_id, url, preset = item + tag = _job_tag.get(job_id, "") try: with yt_dlp.YoutubeDL(make_opts(job_id, preset)) as ydl: if preset == "mp3": @@ -279,11 +406,11 @@ def worker(jobs: "queue.Queue") -> None: info = ydl.extract_info(url, download=True) title = (info or {}).get("title") or _job_title.get(job_id) or url _job_title[job_id] = title - update_job_line(job_id, f"[{ICON_DONE}] {title[:60]}") + update_job_line(job_id, f"[{ICON_DONE}] {tag}{title[:60]}") except Exception as exc: # noqa: BLE001 err = str(exc).splitlines()[0][:60] title = _job_title.get(job_id) or url - update_job_line(job_id, f"[{ICON_FAILED}] {title[:60]} {err}") + update_job_line(job_id, f"[{ICON_FAILED}] {tag}{title[:60]} {err}") def main() -> None: @@ -308,23 +435,39 @@ def main() -> None: break if cmd in PRESETS: preset = cmd - with _print_lock: - for k in _line_distance: - _line_distance[k] += 2 - sys.stdout.write(f" preset -> {preset}\n") - sys.stdout.flush() + notice(f" preset -> {preset}") else: - with _print_lock: - for k in _line_distance: - _line_distance[k] += 2 - sys.stdout.write( - f" unknown command: :{cmd} (try :mp3, :video, :none, :q)\n" - ) - sys.stdout.flush() + notice(f" unknown command: :{cmd} (try :mp3, :video, :none, :q)") continue - jobs.put((next_id, line, preset)) - add_job_line(next_id, f"[{ICON_QUEUED}] {line}") - next_id += 1 + + # A pasted link: resolve it (a quick network probe) to learn whether it's + # a single video or a playlist, then queue one job per resulting entry. + try: + is_playlist, name, entries = resolve(line) + except Exception as exc: # noqa: BLE001 + err = str(exc).splitlines()[0][:60] + add_job_lines([(next_id, f"[{ICON_FAILED}] {line} {err}")]) + next_id += 1 + continue + if not entries: + notice(f" nothing to download in '{name[:40]}'") + continue + + total = len(entries) + lines, pending = [], [] + for i, (entry_url, title) in enumerate(entries, 1): + _job_title[next_id] = title + if is_playlist: + _job_tag[next_id] = f"({name[:24]} - {i}/{total}) " + label = _job_tag.get(next_id, "") + title + lines.append((next_id, f"[{ICON_QUEUED}] {label}")) + pending.append((next_id, entry_url, preset)) + next_id += 1 + # Register the display lines before enqueuing, so a fast worker never + # tries to update a line whose distance isn't set yet. + add_job_lines(lines) + for job in pending: + jobs.put(job) if __name__ == "__main__": -- cgit v1.3.1