diff --git a/.gitignore b/.gitignore index e405477..b1e5f48 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,8 @@ -.DS_Store +.DS_Store CalculatorApp/ # Артефакты локальной сборки (релизные архивы грузятся в GitHub Releases, в git не коммитим) ssh-monitor-*.tar.gz + +# Локальный загрузчик Rutube / yt-dlp — живёт в другом репозитории +rutube-download.py +.ytdl-archive.txt diff --git a/.ytdl-archive.txt b/.ytdl-archive.txt deleted file mode 100644 index 8e9c4cf..0000000 --- a/.ytdl-archive.txt +++ /dev/null @@ -1,2 +0,0 @@ -youtube DJVevQ3mMno -youtube rdi9RJrXgUQ diff --git a/rutube-download.py b/rutube-download.py deleted file mode 100644 index 064eed4..0000000 --- a/rutube-download.py +++ /dev/null @@ -1,1113 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -""" -rutube-download.py — порт rutube-download.ps1 (вер. 0.8). - -Те же возможности: URL, папка или полный путь к .mp4, качество best|1080|720, ---non-interactive, --force-redownload, --ru-lang, проверка/установка зависимостей, -архив загрузок, временная папка yt-dlp, YouTube + node/ejs, один итоговый MP4. - -Запуск: - python3 rutube-download.py '' [output_dir_or_file.mp4] [--quality best|1080|720] ... - В zsh/fish вместо '...' можно взять URL в "..." (если в ссылке есть ?). - -В zsh и fish URL с символом «?» (YouTube и др.) нужно брать в кавычки, иначе оболочка -воспримет «?» как шаблон (глоб). Подойдут одинарные или двойные: 'https://...?v=...' или "https://...?v=..." - -Соответствие параметров PowerShell: - -Url → первый позиционный аргумент - -OutputDir → второй позиционный аргумент - -Quality → --quality - -NonInteractive → --non-interactive - -ForceRedownload → --force-redownload - -RuLang → --ru-lang -""" - -from __future__ import annotations - -import argparse -import os -import re -import shutil -import subprocess -import sys -from datetime import datetime, timedelta -from pathlib import Path -from typing import Callable, Iterable, Sequence - -SCRIPT_VERSION = "0.8" - -# --- цвета (ANSI) --- -_USE_COLOR = sys.stdout.isatty() - - -def _c(code: str) -> Callable[[str], str]: - def wrap(s: str) -> str: - if not _USE_COLOR: - return s - return f"\033[{code}m{s}\033[0m" - - return wrap - - -yellow = _c("93") -# Баннер приложения: жирный оранжевый (256-color 208). Чёрный bold «1;30» на тёмном фоне почти не виден. -banner = _c("1;38;5;208") -cyan = _c("96") -green = _c("92") -red = _c("91") -dim = _c("90") -magenta = _c("35") - - -def info(msg: str) -> None: - print(cyan(msg)) - - -def warn(msg: str) -> None: - print(yellow(msg)) - - -def ok(msg: str) -> None: - print(green(msg)) - - -def fail(msg: str) -> None: - print(red(msg)) - - -def dim_print(msg: str) -> None: - print(dim(msg)) - - -def key_value_pink(key: str, value: str) -> None: - if _USE_COLOR: - print(f"{key} \033[35m{value}\033[0m") - else: - print(f"{key} {value}") - - -def is_windows() -> bool: - return sys.platform == "win32" - - -def is_darwin() -> bool: - return sys.platform == "darwin" - - -def t(ru: bool, en: str, ru_s: str) -> str: - return ru_s if ru else en - - -def format_file_size(n: int) -> str: - gb = 1024**3 - mb = 1024**2 - kb = 1024 - if n >= gb: - return f"{n / gb:.2f} GB" - if n >= mb: - return f"{n / mb:.2f} MB" - if n >= kb: - return f"{n / kb:.2f} KB" - return f"{n} B" - - -def format_dt(value: datetime) -> str: - return value.strftime("%Y-%m-%d %H:%M:%S") - - -def ask_yes_no(question: str, auto_yes: bool, ru: bool) -> bool: - if auto_yes: - print(f"{question} ({t(ru, 'auto: yes', 'авто: да')})") - return True - yn = t(ru, "(yes/no)", "(да/нет)") - while True: - try: - answer = input(f"{question} {yn} ").strip().lower() - except EOFError: - return False - if answer in ("yes", "y", "да", "д"): - return True - if answer in ("no", "n", "нет", "н"): - return False - print(t(ru, "Type yes/no.", "Введите да или нет.")) - - -def refresh_path_windows() -> None: - import winreg - - parts: list[str] = [] - for root, subkey in ( - (winreg.HKEY_LOCAL_MACHINE, r"SYSTEM\CurrentControlSet\Control\Session Manager\Environment"), - (winreg.HKEY_CURRENT_USER, "Environment"), - ): - try: - with winreg.OpenKey(root, subkey) as k: - parts.append(str(winreg.QueryValueEx(k, "Path")[0])) - except OSError: - pass - if parts: - os.environ["Path"] = ";".join(parts) - la = os.environ.get("LOCALAPPDATA", "") - if la: - winget_links = str(Path(la) / "Microsoft" / "WinGet" / "Links") - if Path(winget_links).is_dir(): - cur = os.environ.get("Path", "") - if winget_links.lower() not in cur.lower(): - os.environ["Path"] = cur + ";" + winget_links - - -def prepend_path(dir_path: Path) -> None: - if not dir_path.is_dir(): - return - d = str(dir_path.resolve()) - sep = ";" if is_windows() else ":" - cur = os.environ.get("PATH", "") - if d not in cur.split(sep): - os.environ["PATH"] = d + sep + cur - - -def candidate_extra_paths(name: str) -> list[Path]: - """Доп. пути как в Test-Command ps1 + типичные для macOS.""" - out: list[Path] = [] - n = name.lower() - if is_windows(): - la = os.environ.get("LOCALAPPDATA", "") - pf = os.environ.get("ProgramFiles", "C:\\Program Files") - if la: - out.append(Path(la) / "Microsoft" / "WinGet" / "Links" / f"{name}.exe") - if n == "node": - out.extend( - [ - Path(pf) / "nodejs" / "node.exe", - Path(la) / "Programs" / "nodejs" / "node.exe" if la else Path(), - ] - ) - if n == "ffmpeg": - out.extend( - [ - Path(pf) / "ffmpeg" / "bin" / "ffmpeg.exe", - Path(pf) / "FFmpeg" / "bin" / "ffmpeg.exe", - ] - ) - else: - for base in ("/opt/homebrew/bin", "/usr/local/bin"): - p = Path(base) / name - if p not in out: - out.append(p) - return [p for p in out if p and p.name] - - -def command_resolved(name: str) -> Path | None: - """Имя без .exe на Windows тоже ищем через shutil.which.""" - which = shutil.which(name) - if which: - return Path(which) - if is_windows() and not name.lower().endswith(".exe"): - which = shutil.which(name + ".exe") - if which: - return Path(which) - for p in candidate_extra_paths(name if not is_windows() else name + (".exe" if is_windows() else "")): - if is_windows() and not str(p).lower().endswith(".exe"): - p = p.with_suffix(".exe") if p.suffix == "" else p - try: - if p.is_file(): - prepend_path(p.parent) - return p - except OSError: - continue - return None - - -def find_python_executable() -> Path | None: - for cand in ("python3", "python"): - p = command_resolved(cand) - if p: - return p - return None - - -def run_cmd( - cmd: Sequence[str | Path], - *, - capture: bool = False, - check: bool = False, -) -> subprocess.CompletedProcess[str]: - kwargs: dict = { - "text": True, - "encoding": "utf-8", - "errors": "replace", - } - if capture: - kwargs["stdout"] = subprocess.PIPE - kwargs["stderr"] = subprocess.STDOUT - else: - kwargs["stdout"] = None - kwargs["stderr"] = None - r = subprocess.run([str(x) for x in cmd], **kwargs) - if check and r.returncode != 0: - raise RuntimeError(f"Command failed: {cmd} exit={r.returncode}") - return r # type: ignore[return-value] - - -def run_cmd_live(cmd: Sequence[str | Path]) -> int: - """Запуск с наследованием stdout/stderr.""" - r = subprocess.run([str(x) for x in cmd], text=True, encoding="utf-8", errors="replace") - return r.returncode - - -def which_brew() -> Path | None: - p = shutil.which("brew") - return Path(p) if p else None - - -def ensure_python(non_interactive: bool, ru: bool) -> Path | None: - py = find_python_executable() - if py: - dim_print(t(ru, "Python found.", "Python найден.")) - return py - - warn(t(ru, "Python not found in PATH.", "Python не найден в PATH.")) - if not ask_yes_no(t(ru, "Install Python now?", "Установить Python сейчас?"), non_interactive, ru): - fail(t(ru, "Cannot continue without Python.", "Без Python продолжить нельзя.")) - return None - - if is_windows(): - if not shutil.which("winget"): - fail( - t( - ru, - "winget not found. Install Python manually: https://www.python.org/downloads/windows/", - "winget не найден. Установите Python вручную: https://www.python.org/downloads/windows/", - ) - ) - return None - info(t(ru, "Installing Python via winget...", "Установка Python через winget...")) - rc = run_cmd_live( - [ - "winget", - "install", - "-e", - "--id", - "Python.Python.3.12", - "--accept-source-agreements", - "--accept-package-agreements", - ] - ) - if rc != 0: - fail(t(ru, "Failed to install Python via winget.", "Не удалось установить Python через winget.")) - return None - refresh_path_windows() - py = find_python_executable() - if py: - dim_print(t(ru, "Python installed.", "Python установлен.")) - return py - warn( - t( - ru, - "Python installed but not available in this shell. Restart terminal and run script again.", - "Python установлен, но недоступен в этой сессии. Перезапустите терминал и снова запустите скрипт.", - ) - ) - return None - - if is_darwin(): - brew = which_brew() - if not brew: - fail( - t( - ru, - "Homebrew not found. Install from https://brew.sh then rerun, or install Python from python.org.", - "Homebrew не найден. Установите с https://brew.sh и повторите, либо поставьте Python с python.org.", - ) - ) - return None - info(t(ru, "Installing Python via brew...", "Установка Python через brew...")) - rc = run_cmd_live([str(brew), "install", "python@3.12"]) - if rc != 0: - fail(t(ru, "brew install python failed.", "Ошибка brew install python.")) - return None - for p in (Path("/opt/homebrew/opt/python@3.12/bin"), Path("/usr/local/opt/python@3.12/bin")): - prepend_path(p) - py = find_python_executable() - if py: - dim_print(t(ru, "Python installed.", "Python установлен.")) - return py - warn( - t( - ru, - "Python installed but not in PATH. Open a new terminal or add brew python to PATH.", - "Python установлен, но не в PATH. Откройте новый терминал или добавьте python из brew в PATH.", - ) - ) - return None - - fail( - t( - ru, - "Unsupported OS for auto-install. Install python3 and pip, then rerun.", - "Автоустановка для этой ОС не настроена. Установите python3 и pip вручную.", - ) - ) - return None - - -def ensure_pip(python_exe: Path, non_interactive: bool, ru: bool) -> bool: - if shutil.which("pip") or shutil.which("pip3"): - dim_print(t(ru, "pip found.", "pip найден.")) - return True - r = run_cmd([str(python_exe), "-m", "pip", "--version"], capture=True) - if r.returncode == 0: - ok(t(ru, "pip is available via python -m pip.", "pip доступен как python -m pip.")) - return True - - warn(t(ru, "pip not found.", "pip не найден.")) - if not ask_yes_no( - t(ru, "Install pip now via ensurepip?", "Установить pip через ensurepip?"), - non_interactive, - ru, - ): - fail(t(ru, "Cannot install yt-dlp automatically without pip.", "Без pip нельзя поставить yt-dlp автоматически.")) - return False - - if run_cmd_live([str(python_exe), "-m", "ensurepip", "--upgrade"]) != 0: - fail(t(ru, "Failed to install pip.", "Не удалось установить pip.")) - return False - - r2 = run_cmd([str(python_exe), "-m", "pip", "--version"], capture=True) - if r2.returncode == 0: - ok(t(ru, "pip installed.", "pip установлен.")) - return True - fail(t(ru, "pip is still unavailable.", "pip всё ещё недоступен.")) - return False - - -def ensure_tool_winget( - command_name: str, - winget_id: str, - display_name: str, - non_interactive: bool, - ru: bool, -) -> bool: - if command_resolved(command_name): - dim_print(t(ru, f"{display_name} found.", f"{display_name} найден.")) - return True - - warn(t(ru, f"{display_name} not found.", f"{display_name} не найден.")) - if not ask_yes_no(t(ru, f"Install {display_name} now?", f"Установить {display_name} сейчас?"), non_interactive, ru): - fail(t(ru, f"{display_name} is required for full functionality.", f"Для полной работы нужен {display_name}.")) - return False - - if not shutil.which("winget"): - fail( - t( - ru, - f"winget not found. Install {display_name} manually and rerun script.", - f"winget не найден. Установите {display_name} вручную и перезапустите скрипт.", - ) - ) - return False - - info(t(ru, f"Installing {display_name} via winget...", f"Установка {display_name} через winget...")) - rc = run_cmd_live( - ["winget", "install", "-e", "--id", winget_id, "--accept-source-agreements", "--accept-package-agreements"] - ) - if rc != 0: - fail(t(ru, f"Failed to install {display_name}.", f"Не удалось установить {display_name}.")) - return False - refresh_path_windows() - if command_resolved(command_name): - dim_print(t(ru, f"{display_name} installed.", f"{display_name} установлен.")) - return True - warn( - t( - ru, - f"{display_name} installed but not available in this shell. Restart terminal and run script again.", - f"{display_name} установлен, но недоступен в этой сессии. Перезапустите терминал.", - ) - ) - return False - - -def ensure_tool_brew( - command_name: str, - brew_formula: str, - display_name: str, - non_interactive: bool, - ru: bool, -) -> bool: - if command_resolved(command_name): - dim_print(t(ru, f"{display_name} found.", f"{display_name} найден.")) - return True - - warn(t(ru, f"{display_name} not found.", f"{display_name} не найден.")) - if not ask_yes_no(t(ru, f"Install {display_name} now?", f"Установить {display_name} сейчас?"), non_interactive, ru): - fail(t(ru, f"{display_name} is required for full functionality.", f"Нужен {display_name}.")) - return False - - brew = which_brew() - if not brew: - fail( - t( - ru, - f"Homebrew not found. Install {display_name} manually (e.g. brew install {brew_formula}).", - f"Homebrew не найден. Установите {display_name} вручную (например brew install {brew_formula}).", - ) - ) - return False - - info(t(ru, f"Installing {display_name} via brew...", f"Установка {display_name} через brew...")) - if run_cmd_live([str(brew), "install", brew_formula]) != 0: - fail(t(ru, f"Failed to install {display_name}.", f"Не удалось установить {display_name}.")) - return False - prepend_path(Path("/opt/homebrew/bin")) - prepend_path(Path("/usr/local/bin")) - if command_resolved(command_name): - dim_print(t(ru, f"{display_name} installed.", f"{display_name} установлен.")) - return True - warn( - t( - ru, - f"{display_name} installed but not in PATH. Open a new terminal.", - f"{display_name} установлен, но не в PATH. Откройте новый терминал.", - ) - ) - return False - - -def ensure_tool_cross( - command_name: str, - *, - winget_id: str, - brew_formula: str, - display_name: str, - non_interactive: bool, - ru: bool, -) -> bool: - if is_windows(): - return ensure_tool_winget(command_name, winget_id, display_name, non_interactive, ru) - if is_darwin(): - return ensure_tool_brew(command_name, brew_formula, display_name, non_interactive, ru) - # Linux: только подсказка - if command_resolved(command_name): - dim_print(t(ru, f"{display_name} found.", f"{display_name} найден.")) - return True - warn(t(ru, f"{display_name} not found.", f"{display_name} не найден.")) - fail( - t( - ru, - f"Install {display_name} with your package manager (e.g. apt install {brew_formula}) and rerun.", - f"Установите {display_name} через пакетный менеджер и перезапустите скрипт.", - ) - ) - return False - - -# Regex для полных путей к .mp4 в логах yt-dlp (Windows + Unix) -_RE_MP4_WIN = re.compile(r"([A-Za-z]:\\[^\r\n]+?\.mp4)", re.IGNORECASE) -_RE_MP4_UNIX = re.compile(r"(/[^\r\n]+\.mp4)") - - -def extract_mp4_paths_from_line(line: str) -> list[str]: - found: list[str] = [] - found.extend(m.group(1).strip() for m in _RE_MP4_WIN.finditer(line)) - found.extend(m.group(1).strip() for m in _RE_MP4_UNIX.finditer(line)) - # Убираем дубликаты с сохранением порядка - seen: set[str] = set() - out: list[str] = [] - for p in found: - if p not in seen: - seen.add(p) - out.append(p) - return out - - -def expected_path_from_probe_lines(lines: Iterable[str], out_dir: Path) -> str | None: - """Как Resolve-ExpectedMp4Path: полный путь в логе; иначе односегментное имя *.mp4 → папка вывода.""" - for line in lines: - for p in extract_mp4_paths_from_line(line): - if p.lower().endswith(".mp4"): - return p - stripped = line.strip() - if stripped.lower().endswith(".mp4") and re.fullmatch( - r"[^\\/:;\r\n]+\.mp4", stripped, flags=re.IGNORECASE - ): - return str((out_dir / stripped).resolve()) - return None - - -def build_ytdlp_prefix(python_exe: Path, use_module: bool) -> list[str]: - if use_module: - return [str(python_exe), "-m", "yt_dlp"] - return ["yt-dlp"] - - -def probe_video_id(prefix: list[str], video_url: str, is_youtube: bool) -> str | None: - args = ["--no-download", "--no-playlist", "--print", "id", video_url] - if is_youtube: - args = ["--js-runtimes", "node", "--remote-components", "ejs:github"] + args - _code, lines = run_ytdlp_capture(prefix, args) - for line in lines: - s = line.strip() - if s and not s.startswith("["): - return s - return None - - -def prune_archive_for_video_id(archive_path: Path, video_id: str) -> int: - """Удаляет из download-archive строки с этим id (формат yt-dlp: «extractor id»).""" - if not video_id or not archive_path.is_file(): - return 0 - id_l = video_id.strip().lower() - raw_lines = archive_path.read_text(encoding="utf-8", errors="replace").splitlines() - kept: list[str] = [] - removed = 0 - for line in raw_lines: - s = line.strip() - if not s: - continue - parts = s.split(None, 1) - if len(parts) >= 2 and parts[1].lower() == id_l: - removed += 1 - continue - if len(parts) == 1 and parts[0].lower() == id_l: - removed += 1 - continue - if len(parts) == 1 and ":" in parts[0]: - _left, _sep, right = parts[0].partition(":") - if right.lower() == id_l: - removed += 1 - continue - kept.append(line.rstrip("\r")) - if removed: - archive_path.write_text("\n".join(kept) + ("\n" if kept else ""), encoding="utf-8") - return removed - - -_RE_ARCHIVE_SKIP = re.compile( - r"\[download\]\s*(?P[^:\s]+)\s*:\s*has already been recorded in the archive", - re.IGNORECASE, -) - - -def extract_id_from_archive_skip_line(line: str) -> str | None: - m = _RE_ARCHIVE_SKIP.search(line) - return m.group("id").strip() if m else None - - -def resolve_final_from_ytdlp_lines( - yt_lines: list[str], - *, - known_target: str | None, - custom_target: Path | None, - expected_mp4: str | None, - run_start: datetime, - output_root: Path, -) -> tuple[Path | None, str | None]: - """ - Итоговый .mp4 после прогона yt-dlp. - Второе значение: None — ок или «нет файла»; иначе код фатальной ошибки как в main. - """ - mp4_paths: list[str] = [] - for line in yt_lines: - mp4_paths.extend(extract_mp4_paths_from_line(line)) - mp4_paths = list(dict.fromkeys(mp4_paths)) - - if known_target and Path(known_target).is_file(): - return Path(known_target), None - if custom_target and custom_target.is_file(): - return custom_target.resolve(), None - if not custom_target and len(mp4_paths) == 1 and Path(mp4_paths[0]).is_file(): - return Path(mp4_paths[0]), None - if not custom_target and expected_mp4 and Path(expected_mp4).is_file(): - return Path(expected_mp4), None - if not custom_target and len(mp4_paths) > 1: - return None, "multi_mp4" - if custom_target: - return None, "no_custom_target" - - recent: list[Path] = [] - cutoff = run_start - timedelta(minutes=2) - for p in output_root.glob("*.mp4"): - if not p.is_file(): - continue - m = datetime.fromtimestamp(p.stat().st_mtime) - if m >= cutoff: - recent.append(p) - recent.sort(key=lambda x: x.stat().st_mtime, reverse=True) - if len(recent) == 1: - return recent[0], None - if len(recent) > 1: - return None, "multi_recent" - return None, None - - -def run_ytdlp_capture( - prefix: list[str], - args: list[str], -) -> tuple[int, list[str]]: - cmd = prefix + args - r = run_cmd(cmd, capture=True) - text = r.stdout or "" - lines = text.splitlines() - return r.returncode, lines - - -def run_ytdlp_stream_filtered( - prefix: list[str], - args: list[str], -) -> tuple[int, list[str]]: - """Полный лог в collected для парсинга MP4. [youtube]/[rutube] не печатаем. [download] — одна строка с \\r, без спама.""" - cmd = prefix + args - collected: list[str] = [] - noise = re.compile(r"^\[(youtube|rutube)\]", re.IGNORECASE) - download_tag = re.compile(r"^\[download\]", re.IGNORECASE) - env = os.environ.copy() - if len(prefix) >= 2 and Path(prefix[0]).name.lower().startswith("python"): - env.setdefault("PYTHONUNBUFFERED", "1") - - proc = subprocess.Popen( - [str(x) for x in cmd], - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - encoding="utf-8", - errors="replace", - bufsize=1, - env=env, - ) - - def term_w() -> int: - try: - return max(40, shutil.get_terminal_size().columns) - except OSError: - return 100 - - progress_open = False - tty_out = sys.stdout.isatty() - - def close_progress_line() -> None: - nonlocal progress_open - if progress_open: - sys.stdout.write("\n") - sys.stdout.flush() - progress_open = False - - assert proc.stdout is not None - for line in proc.stdout: - line_no_nl = line.rstrip("\n\r") - collected.append(line_no_nl) - - if download_tag.match(line_no_nl): - rest = download_tag.sub("", line_no_nl).strip() - rest = re.sub(r"\s+", " ", rest) - if not tty_out: - print(line_no_nl) - continue - max_rest = max(20, term_w() - 18) - if len(rest) > max_rest: - rest = rest[: max_rest - 1] + "…" - if _USE_COLOR: - sys.stdout.write(f"\r\033[96m[download]\033[0m {rest}\033[K") - else: - sys.stdout.write(f"\r[download] {rest}\033[K") - sys.stdout.flush() - progress_open = True - continue - - if noise.match(line_no_nl): - continue - - close_progress_line() - print(line_no_nl) - - proc.wait() - close_progress_line() - return proc.returncode, collected - - -def resolve_expected_mp4_path( - prefix: list[str], - video_url: str, - out_dir: Path, - output_template: str, - is_youtube: bool, -) -> str | None: - probe_args = [ - "--no-download", - "--no-playlist", - "--paths", - str(out_dir), - "-o", - output_template, - "--merge-output-format", - "mp4", - "--restrict-filenames", - "--print", - "filename", - video_url, - ] - if is_youtube: - probe_args = ["--js-runtimes", "node", "--remote-components", "ejs:github"] + probe_args - code, lines = run_ytdlp_capture(prefix, probe_args) - _ = code # как в PS1: не обязательно 0 для парсинга - return expected_path_from_probe_lines(lines, out_dir) - - -def main() -> int: - script_root = Path(__file__).resolve().parent - default_out = script_root / "downloads" - default_temp = script_root / ".ytdl-temp" - archive_file = script_root / ".ytdl-archive.txt" - - parser = argparse.ArgumentParser( - description=( - "Youtube / RuTube downloader (yt-dlp), Python port of rutube-download.ps1.\n" - "Shell tip: in zsh/fish, quote the URL in single or double quotes when it contains ? " - "(e.g. youtube.com/watch?v=...), or the shell may report \"no matches found\"." - ), - formatter_class=argparse.RawDescriptionHelpFormatter, - epilog=""" -Shell (zsh / fish): wrap URLs in single or double quotes — YouTube links contain ? which triggers globbing. - Пример: python3 rutube-download.py "https://www.youtube.com/watch?v=ID" ~/Downloads/YT-Video - Example: python3 rutube-download.py 'https://www.youtube.com/watch?v=ID' ~/Downloads/YT-Video - -Examples: - python3 rutube-download.py 'https://rutube.ru/video/ID' - python3 rutube-download.py "https://www.youtube.com/watch?v=ID" --quality 1080 - python3 rutube-download.py 'https://rutube.ru/video/ID' --non-interactive - python3 rutube-download.py "https://rutube.ru/video/ID" --force-redownload -""", - ) - parser.add_argument( - "url", - nargs="?", - help="Video URL; in zsh/fish use quotes if the URL contains ? (e.g. 'https://...' or \"https://...\")", - ) - parser.add_argument( - "output_dir", - nargs="?", - help="Output folder OR full path to target file.mp4", - ) - parser.add_argument("--quality", choices=("best", "1080", "720"), default="best") - parser.add_argument("--non-interactive", action="store_true", help="Auto-yes for install prompts") - parser.add_argument("--force-redownload", action="store_true", help="Ignore archive, overwrite") - parser.add_argument("--ru-lang", action="store_true", help="Russian messages") - - args = parser.parse_args() - ru = args.ru_lang - ni = args.non_interactive - - print(banner(f"Youtube / RuTube downloader by Papa Tramp ver. {SCRIPT_VERSION}")) - - if not args.url: - shell_hint = t( - args.ru_lang, - "Tip (zsh/fish): put the URL in single or double quotes — YouTube links contain ? and the shell " - "would otherwise try to glob-match.\n" - " Example: python3 rutube-download.py \"https://www.youtube.com/watch?v=...\" ~/Downloads/out\n", - "Подсказка (zsh/fish): заключите URL в одинарные или двойные кавычки — в ссылках YouTube есть «?», " - "иначе оболочка воспримет его как шаблон (ошибка «no matches found»).\n" - " Пример: python3 rutube-download.py 'https://www.youtube.com/watch?v=...' ~/Downloads/out\n", - ) - dim_print(shell_hint) - parser.print_help() - return 0 - - url = args.url.strip() - if not url: - fail(t(ru, "Error: url is required.", "Ошибка: нужен URL.")) - parser.print_help() - return 1 - - output_dir_arg = args.output_dir - output_root = Path(output_dir_arg.strip() if output_dir_arg else default_out).expanduser() - output_template = "%(title)s.%(ext)s" - custom_target: Path | None = None - - if re.search(r"\.mp4$", str(output_root), re.IGNORECASE): - custom_target = output_root.resolve() - output_template = custom_target.name - parent = custom_target.parent - # Как Split-Path -Parent в PS1: только имя файла → текущая папка - if parent == Path(".") or str(parent) in (".", ""): - output_root = Path.cwd().resolve() - else: - output_root = parent.resolve() - - output_root = output_root.resolve() - default_temp.mkdir(parents=True, exist_ok=True) - output_root.mkdir(parents=True, exist_ok=True) - - python_exe = ensure_python(ni, ru) - if not python_exe: - return 1 - if not ensure_pip(python_exe, ni, ru): - return 1 - - # yt-dlp: исполняемый или модуль - use_ytdlp_exe = command_resolved("yt-dlp") is not None - if use_ytdlp_exe: - ytdlp_prefix = ["yt-dlp"] - dim_print(t(ru, "yt-dlp found in PATH.", "yt-dlp найден в PATH.")) - else: - r = run_cmd([str(python_exe), "-m", "yt_dlp", "--version"], capture=True) - if r.returncode == 0: - dim_print(t(ru, "yt-dlp available via python -m yt_dlp.", "yt-dlp доступен как python -m yt_dlp.")) - ytdlp_prefix = [str(python_exe), "-m", "yt_dlp"] - else: - warn(t(ru, "yt-dlp not found.", "yt-dlp не найден.")) - if not ask_yes_no(t(ru, "Install yt-dlp now?", "Установить yt-dlp сейчас?"), ni, ru): - fail(t(ru, "Download canceled: yt-dlp not installed.", "Скачивание отменено: yt-dlp не установлен.")) - return 1 - info( - t( - ru, - "Installing/updating yt-dlp and dependencies...", - "Установка/обновление yt-dlp и зависимостей...", - ) - ) - if ( - run_cmd_live( - [ - str(python_exe), - "-m", - "pip", - "install", - "-U", - "yt-dlp", - "requests", - "urllib3", - "charset-normalizer", - "chardet", - ] - ) - != 0 - ): - fail(t(ru, "Failed to install yt-dlp/dependencies via pip.", "Не удалось установить yt-dlp через pip.")) - return 1 - ytdlp_prefix = [str(python_exe), "-m", "yt_dlp"] - - is_youtube = bool(re.search(r"(youtube\.com|youtu\.be)", url, re.IGNORECASE)) - - if not ensure_tool_cross("ffmpeg", winget_id="Gyan.FFmpeg", brew_formula="ffmpeg", display_name="FFmpeg", non_interactive=ni, ru=ru): - return 1 - - if is_youtube: - if not ensure_tool_cross( - "node", - winget_id="OpenJS.NodeJS.LTS", - brew_formula="node", - display_name="Node.js", - non_interactive=ni, - ru=ru, - ): - return 1 - - format_selector = { - "1080": "bv*[height<=1080]+ba/b[height<=1080]/b", - "720": "bv*[height<=720]+ba/b[height<=720]/b", - "best": "bv*+ba/b", - }[args.quality] - - download_args: list[str] = [ - "-f", - format_selector, - "--merge-output-format", - "mp4", - "--restrict-filenames", - "--no-playlist", - "--paths", - str(output_root), - "--paths", - f"temp:{default_temp}", - "-o", - output_template, - url, - ] - - if args.force_redownload: - download_args = ["--force-overwrites", "--no-continue"] + download_args - else: - if custom_target: - download_args = ["--continue", "--no-overwrites"] + download_args - else: - download_args = ["--continue", "--no-overwrites", "--download-archive", str(archive_file)] + download_args - - if is_youtube: - download_args = ["--js-runtimes", "node", "--remote-components", "ejs:github"] + download_args - - expected_mp4 = resolve_expected_mp4_path( - ytdlp_prefix, - url, - output_root, - output_template, - is_youtube, - ) - known_target: str | None = None - if custom_target: - known_target = str(custom_target.resolve()) - elif expected_mp4: - known_target = expected_mp4 - - print(f"Downloading: {url}") - print(f"Output folder: {output_root}") - if custom_target: - print(f"Target file: {custom_target}") - print(f"Temp folder: {default_temp.resolve()}") - print(f"Quality: {args.quality}") - dim_print( - t( - args.ru_lang, - "yt-dlp: [download] progress is shown on one updating line; other messages print normally. " - "Merging after 100% may take a while.", - "yt-dlp: прогресс [download] — одна обновляемая строка; остальные сообщения как обычно. " - "После 100% может идти склейка форматов.", - ) - ) - - if not args.force_redownload and known_target and Path(known_target).is_file(): - existing = Path(known_target) - warn(t(ru, "Already downloaded. Skipping.", "Уже скачано. Пропуск.")) - print(f"Path: {existing.resolve()}") - key_value_pink(t(ru, "Size:", "Размер:"), format_file_size(existing.stat().st_size)) - key_value_pink(t(ru, "Saved at:", "Сохранено:"), format_dt(datetime.fromtimestamp(existing.stat().st_mtime))) - return 0 - - using_archive = not args.force_redownload and custom_target is None - - run_start = datetime.now() - custom_before: os.stat_result | None = None - if custom_target and custom_target.is_file(): - custom_before = custom_target.stat() - - # Файл удалили, а id остался в .ytdl-archive.txt — yt-dlp «успешно» ничего не качает. - if using_archive and known_target and not Path(known_target).is_file(): - vid0 = probe_video_id(ytdlp_prefix, url, is_youtube) - if vid0: - n0 = prune_archive_for_video_id(archive_file, vid0) - if n0: - dim_print( - t( - ru, - f"Removed {n0} stale archive line(s) for id {vid0} (expected .mp4 is missing).", - f"Удалено {n0} устаревших строк архива для id {vid0} — ожидаемого .mp4 нет на диске.", - ) - ) - - code, yt_lines = run_ytdlp_stream_filtered(ytdlp_prefix, download_args) - final, ferr = resolve_final_from_ytdlp_lines( - yt_lines, - known_target=known_target, - custom_target=custom_target, - expected_mp4=expected_mp4, - run_start=run_start, - output_root=output_root, - ) - - if ( - code == 0 - and final is None - and ferr is None - and using_archive - and any("has already been recorded in the archive" in x for x in yt_lines) - ): - vid_retry: str | None = None - for ln in yt_lines: - vid_retry = extract_id_from_archive_skip_line(ln) - if vid_retry: - break - if not vid_retry: - vid_retry = probe_video_id(ytdlp_prefix, url, is_youtube) - if vid_retry: - n1 = prune_archive_for_video_id(archive_file, vid_retry) - if n1 > 0: - dim_print( - t( - ru, - "Stale archive entry removed; retrying download…", - "Удалена устаревшая запись архива; повторная загрузка…", - ) - ) - code, yt_lines = run_ytdlp_stream_filtered(ytdlp_prefix, download_args) - if code == 0: - final, ferr = resolve_final_from_ytdlp_lines( - yt_lines, - known_target=known_target, - custom_target=custom_target, - expected_mp4=expected_mp4, - run_start=run_start, - output_root=output_root, - ) - - if code != 0: - fail(t(ru, f"Download failed. Exit code: {code}", f"Ошибка скачивания. Код выхода: {code}")) - return 1 - - if ferr == "multi_mp4": - fail( - t( - ru, - "Expected exactly one MP4 file, but detected multiple paths in yt-dlp output.", - "Ожидался один MP4, в выводе yt-dlp несколько путей.", - ) - ) - return 1 - if ferr == "no_custom_target": - fail( - t( - ru, - f"Target file was requested but not created: {custom_target}", - f"Целевой файл не создан: {custom_target}", - ) - ) - return 1 - if ferr == "multi_recent": - fail( - t( - ru, - "Expected exactly one MP4 file for this run, but detected multiple recent MP4 files.", - "Ожидался один недавний MP4, найдено несколько файлов.", - ) - ) - return 1 - - if final is not None: - st = final.stat() - size_text = format_file_size(st.st_size) - if ( - custom_before - and custom_target is not None - and final.resolve() == custom_target.resolve() - and st.st_size == custom_before.st_size - and st.st_mtime_ns == custom_before.st_mtime_ns - and not args.force_redownload - ): - print( - green( - t( - ru, - "Done. Target file already existed; download skipped.", - "Готово. Целевой файл уже существовал; скачивание пропущено.", - ) - ) - ) - else: - print(green(t(ru, "Done. MP4 file ready.", "Готово. MP4 готов."))) - print(green(f"Path: {final.resolve()}")) - key_value_pink(t(ru, "Size:", "Размер:"), size_text) - key_value_pink(t(ru, "Saved at:", "Сохранено:"), format_dt(datetime.fromtimestamp(st.st_mtime))) - return 0 - - fail( - t( - ru, - "Download finished, but no MP4 output was detected. Check yt-dlp output above.", - "Скачивание завершилось, но MP4 не обнаружен. Смотрите вывод yt-dlp выше.", - ) - ) - return 1 - - -if __name__ == "__main__": - sys.exit(main())