107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
"""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) |