v1.7 - playlist distinction support + foolproofing

Co-Authored-By: Claude Opus 4.8 <[email protected]>

1 files changed, 169 insertions, 26 deletions

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.
 
-def add_job_line(job_id: int, text: str) -> None:
-    """Main thread: print a new tracked line above the next prompt."""
+    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 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
+
+        # 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
-        jobs.put((next_id, line, preset))
-        add_job_line(next_id, f"[{ICON_QUEUED}] {line}")
-        next_id += 1
+        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__":