Andrey Lutsenko
1114 lines
39 KiB
Python
1114 lines
39 KiB
Python
#!/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 '<url>' [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<id>[^:\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())
|