vpn: require an active WireGuard handshake before downloading (refuse a dead tunnel)

yt_downloader/progress.py

@@ -0,0 +1,107 @@
+"""Minimal terminal progress bar for yt-dlp downloads (stdlib only)."""
+
+from __future__ import annotations
+
+import sys
+import time
+from pathlib import Path
+
+
+def _fmt_bytes(n: float) -> str:
+    n = float(n or 0)
+    for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
+        if n < 1024 or unit == "TiB":
+            return f"{int(n)} B" if unit == "B" else f"{n:.1f} {unit}"
+        n /= 1024
+    return "0 B"
+
+
+def _fmt_eta(sec: float) -> str:
+    sec = int(sec or 0)
+    if sec < 60:
+        return f"{sec:02d}s"
+    m, s = divmod(sec, 60)
+    if m < 60:
+        return f"{m:02d}:{s:02d}"
+    h, m = divmod(m, 60)
+    return f"{h:02d}:{m:02d}:{s:02d}"
+
+
+class DownloadProgress:
+    """Draws a one-line progress bar from yt-dlp `progress_hooks` data.
+
+    On a TTY the bar redraws in place; when piped to a file it falls back to
+    a sparse log line every few seconds (like yt-dlp's own ``[download]``).
+    A bestvideo+bestaudio download reports each stream as its own file, so the
+    bar resets per stream and shows the stream's filename.
+    """
+
+    _BAR = 24
+
+    def __init__(self, stream=None):
+        self._stream = stream or sys.stderr
+        self._tty = self._stream.isatty()
+        self._file: str | None = None
+        self._last = 0.0
+        self._last_log = 0.0
+        self._line_open = False
+        self._min_interval = 0.05
+
+    def hook(self, data: dict) -> None:
+        status = data.get("status")
+        if status == "downloading":
+            self._on_downloading(data)
+        elif status == "finished":
+            self._on_finished()
+
+    def _on_downloading(self, data: dict) -> None:
+        fname = Path(data.get("filename", "")).name
+        now = time.monotonic()
+        if fname != self._file:
+            if self._tty and self._line_open:
+                self._stream.write("\n")
+            self._file = fname
+            self._line_open = False
+        if not self._tty:
+            if now - self._last_log < 5.0:
+                return
+            self._last_log = now
+            self._stream.write(self._text(data) + "\n")
+            self._stream.flush()
+            return
+        if now - self._last < self._min_interval:
+            return
+        self._last = now
+        self._stream.write("\r\x1b[K" + self._text(data))
+        self._stream.flush()
+        self._line_open = True
+
+    def _on_finished(self) -> None:
+        if self._tty and self._line_open:
+            self._stream.write("\n")
+            self._stream.flush()
+        self._file = None
+        self._line_open = False
+
+    def _text(self, data: dict) -> str:
+        done = data.get("downloaded_bytes") or 0
+        total = data.get("total_bytes") or data.get("total_bytes_estimate") or 0
+        speed = data.get("speed")
+        eta = data.get("eta")
+        pct = (done / total * 100) if total else None
+        parts = []
+        if pct is not None:
+            filled = int(self._BAR * pct / 100)
+            bar = "\u2588" * filled + "\u2591" * (self._BAR - filled)
+            parts.append(f"[{bar}] {pct:5.1f}%")
+        else:
+            parts.append(f"[{'\u2591' * self._BAR}] {_fmt_bytes(done)}")
+        if total:
+            parts.append(f"{_fmt_bytes(done)}/{_fmt_bytes(total)}")
+        if speed:
+            parts.append(f"{_fmt_bytes(speed)}/s")
+        if eta is not None:
+            parts.append(f"ETA {_fmt_eta(eta)}")
+        if self._file:
+            parts.append(self._file)
+        return " ".join(parts)