158 lines
8.2 KiB
Python
158 lines
8.2 KiB
Python
from __future__ import annotations
|
||
|
||
import math
|
||
import re
|
||
from pathlib import Path
|
||
from urllib.parse import parse_qs, urlsplit
|
||
|
||
HOST_NAME = "com.youtubedl.native"
|
||
VERSION = "1.3.2"
|
||
EXTENSION_ID = "ndnoeiaecgmjnjneekaeapeecmmjheli"
|
||
VIDEO_CONTAINERS = {"mp4", "mkv", "webm"}
|
||
AUDIO_CONTAINERS = {"m4a", "mp3", "opus", "flac", "wav"}
|
||
VIDEO_CODECS = {"auto", "copy", "h264", "hevc", "av1", "vp9"}
|
||
ENGINES = {"auto", "nvenc", "amf", "cpu"}
|
||
MODES = {"video", "video_only", "audio"}
|
||
FORMAT_ID = re.compile(r"^[A-Za-z0-9_.-]{1,80}$")
|
||
VIDEO_ID = re.compile(r"^[A-Za-z0-9_-]{11}$")
|
||
|
||
|
||
def youtube_url(value: object) -> str:
|
||
if not isinstance(value, str) or len(value) > 2048:
|
||
raise ValueError("Нужна ссылка на один ролик YouTube.")
|
||
try:
|
||
parsed = urlsplit(value.strip())
|
||
if parsed.scheme != "https" or parsed.username or parsed.password or parsed.port not in (None, 443):
|
||
raise ValueError()
|
||
host = (parsed.hostname or "").lower()
|
||
parts = parsed.path.strip("/").split("/")
|
||
if host == "youtu.be" and len(parts) == 1:
|
||
video_id = parts[0]
|
||
elif host in {"youtube.com", "www.youtube.com", "m.youtube.com", "music.youtube.com"}:
|
||
if parsed.path == "/watch":
|
||
video_id = parse_qs(parsed.query).get("v", [""])[0]
|
||
elif len(parts) == 2 and parts[0] in {"shorts", "live", "embed"}:
|
||
video_id = parts[1]
|
||
else:
|
||
raise ValueError()
|
||
else:
|
||
raise ValueError()
|
||
if not VIDEO_ID.fullmatch(video_id):
|
||
raise ValueError()
|
||
except (ValueError, TypeError):
|
||
raise ValueError("Поддерживаются HTTPS-ссылки на видео и Shorts с YouTube.") from None
|
||
return f"https://www.youtube.com/watch?v={video_id}"
|
||
|
||
|
||
def format_id(value: object) -> str:
|
||
if not isinstance(value, str) or not FORMAT_ID.fullmatch(value):
|
||
raise ValueError("Некорректный идентификатор дорожки.")
|
||
return value
|
||
|
||
|
||
def number(value, default=0):
|
||
return value if isinstance(value, (int, float)) and math.isfinite(value) else default
|
||
|
||
|
||
def video_formats(info: dict) -> list[dict]:
|
||
return [f for f in info.get("formats", []) if f.get("vcodec") not in (None, "none")
|
||
and number(f.get("height")) > 0 and not f.get("has_drm") and FORMAT_ID.fullmatch(str(f.get("format_id", "")))]
|
||
|
||
|
||
def audio_formats(info: dict) -> list[dict]:
|
||
return [f for f in info.get("formats", []) if f.get("vcodec") == "none"
|
||
and f.get("acodec") not in (None, "none") and not f.get("has_drm")
|
||
and FORMAT_ID.fullmatch(str(f.get("format_id", "")))]
|
||
|
||
|
||
def summarize(info: dict) -> dict:
|
||
if info.get("is_live"):
|
||
raise ValueError("Дождитесь окончания прямой трансляции. Сейчас поддерживаются готовые ролики.")
|
||
videos = []
|
||
for f in reversed(video_formats(info)):
|
||
videos.append({"id": str(f["format_id"]), "height": number(f.get("height")),
|
||
"width": number(f.get("width")), "fps": number(f.get("fps")),
|
||
"codec": f.get("vcodec", ""), "extension": f.get("ext", ""),
|
||
"hasAudio": f.get("acodec") not in (None, "none"),
|
||
"size": number(f.get("filesize") or f.get("filesize_approx")),
|
||
"hdr": f.get("dynamic_range", "SDR")})
|
||
audios = [{"id": str(f["format_id"]), "codec": f.get("acodec", ""),
|
||
"extension": f.get("ext", ""), "bitrate": number(f.get("abr") or f.get("tbr")),
|
||
"language": f.get("language") or "", "note": f.get("format_note") or "",
|
||
"size": number(f.get("filesize") or f.get("filesize_approx"))}
|
||
for f in reversed(audio_formats(info))]
|
||
if not videos and not audios:
|
||
raise ValueError("YouTube не вернул доступных дорожек для этого ролика.")
|
||
return {"id": str(info.get("id", "")), "url": youtube_url(info.get("webpage_url", "")),
|
||
"title": str(info.get("title", "Видео"))[:300],
|
||
"channel": str(info.get("uploader") or info.get("channel") or "")[:200],
|
||
"duration": number(info.get("duration")), "videos": videos[:180], "audios": audios[:100]}
|
||
|
||
|
||
def validate_download(payload: dict) -> dict:
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("Некорректная заявка.")
|
||
mode = payload.get("mode", "video")
|
||
container = payload.get("container", "mp4")
|
||
codec = payload.get("codec", "auto")
|
||
engine = payload.get("engine", "auto")
|
||
if mode not in MODES or container not in (AUDIO_CONTAINERS if mode == "audio" else VIDEO_CONTAINERS):
|
||
raise ValueError("Выберите корректный тип загрузки и формат файла.")
|
||
if codec not in VIDEO_CODECS or engine not in ENGINES or payload.get("preset", "balanced") not in {"fast", "balanced", "quality"}:
|
||
raise ValueError("Некорректные настройки кодирования.")
|
||
bitrate = payload.get("bitrate", 256)
|
||
if isinstance(bitrate, bool) or bitrate not in {128, 192, 256, 320}:
|
||
raise ValueError("Некорректный битрейт аудио.")
|
||
folder = payload.get("folder")
|
||
if not isinstance(folder, str) or not folder or len(folder) > 4096:
|
||
raise ValueError("Выберите папку сохранения.")
|
||
path = Path(folder).expanduser()
|
||
if not path.is_absolute() or not path.is_dir():
|
||
raise ValueError("Папка сохранения должна существовать и иметь абсолютный путь.")
|
||
work_folder = payload.get("workFolder", "")
|
||
if not isinstance(work_folder, str) or len(work_folder) > 4096:
|
||
raise ValueError("Выберите корректную рабочую папку.")
|
||
work_folder = work_folder or folder
|
||
work_path = Path(work_folder).expanduser()
|
||
if not work_path.is_absolute() or not work_path.is_dir():
|
||
raise ValueError("Рабочая папка должна существовать и иметь абсолютный путь.")
|
||
name = payload.get("filename", "")
|
||
if not isinstance(name, str) or len(name) > 180:
|
||
raise ValueError("Имя файла слишком длинное.")
|
||
return {"url": youtube_url(payload.get("url")), "mode": mode, "container": container,
|
||
"codec": codec, "engine": engine, "preset": payload.get("preset", "balanced"),
|
||
"bitrate": bitrate, "folder": str(path.resolve()), "workFolder": str(work_path.resolve()), "filename": name,
|
||
"videoId": format_id(payload.get("videoId", "best")),
|
||
"audioId": format_id(payload.get("audioId", "best"))}
|
||
|
||
|
||
def select_streams(info: dict, options: dict) -> str:
|
||
videos, audios = video_formats(info), audio_formats(info)
|
||
def choose(items, wanted, label):
|
||
if not items:
|
||
raise ValueError(f"Недоступна дорожка: {label}.")
|
||
if wanted == "best":
|
||
return items[-1]
|
||
for item in items:
|
||
if str(item["format_id"]) == wanted:
|
||
return item
|
||
raise ValueError("Выбранная дорожка исчезла. Обновите список качества.")
|
||
if options["mode"] == "audio":
|
||
# A progressive video is a valid fallback when separate audio is absent.
|
||
return str(choose(audios or [f for f in videos if f.get("acodec") not in (None, "none")],
|
||
options["audioId"], "аудио")["format_id"])
|
||
video = choose(videos, options["videoId"], "видео")
|
||
if options["mode"] == "video_only" or video.get("acodec") not in (None, "none"):
|
||
return str(video["format_id"])
|
||
audio = choose(audios, options["audioId"], "аудио")
|
||
return f"{video['format_id']}+{audio['format_id']}"
|
||
|
||
|
||
def safe_filename(value: str, fallback: str) -> str:
|
||
value = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", value).strip(" .")[:150].rstrip(" .")
|
||
if not value:
|
||
value = fallback
|
||
if value.split(".", 1)[0].upper() in {"CON", "PRN", "AUX", "NUL", *(f"COM{i}" for i in range(1, 10)), *(f"LPT{i}" for i in range(1, 10))}:
|
||
value = "_" + value
|
||
return value
|