diff --git a/.ytdl-archive.txt b/.ytdl-archive.txt new file mode 100644 index 0000000..8e9c4cf --- /dev/null +++ b/.ytdl-archive.txt @@ -0,0 +1,2 @@ +youtube DJVevQ3mMno +youtube rdi9RJrXgUQ diff --git a/rutube-download.py b/rutube-download.py new file mode 100644 index 0000000..064eed4 --- /dev/null +++ b/rutube-download.py @@ -0,0 +1,1113 @@ +#!/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())