Добавить основную функциональность загрузки YouTube Shorts с использованием файла сессии и обработкой ошибок

This commit is contained in:
ВяткинАртём
2026-05-27 16:33:22 +03:00
parent 2d1f671d5a
commit d06bd989f3
23 changed files with 884 additions and 3 deletions
+4 -1
View File
@@ -58,6 +58,9 @@ cover/
# Django stuff: # Django stuff:
*.log *.log
app.log
logs/*
!logs/.gitkeep
local_settings.py local_settings.py
db.sqlite3 db.sqlite3
db.sqlite3-journal db.sqlite3-journal
@@ -173,4 +176,4 @@ cython_debug/
# PyPI configuration file # PyPI configuration file
.pypirc .pypirc
cookies.txt
+1
View File
@@ -0,0 +1 @@
3.12
+26
View File
@@ -0,0 +1,26 @@
# Makefile for uv + ruff + mypy + deptry
UV := uv
RUFF := ruff
MYPY := mypy
DEPTRY := deptry
.PHONY: full-lint clean-branches
full-lint:
@echo "Running ruff check and fix..."
$(UV) run $(RUFF) check . --fix
@echo "Running mypy..."
$(UV) run $(MYPY) . --config-file ./pyproject.toml --no-incremental
@echo "Running deptry..."
$(UV) run $(DEPTRY) . --config ./pyproject.toml
@echo "All checks completed!"
clean-branches:
ifeq ($(OS),Windows_NT)
@git fetch --prune
@cmd /C "for /f \"tokens=1\" %%i in ('git branch -vv ^| findstr ": gone]"') do git branch -D %%i" || ver > nul
else
git fetch --prune
git branch -vv | grep ': gone]' | awk '{print $$1}' | xargs -r git branch -D
endif
-2
View File
@@ -1,2 +0,0 @@
# yt-shorts-downloader
+4
View File
@@ -0,0 +1,4 @@
from yt_shorts_downloader.cli import main
if __name__ == "__main__":
raise SystemExit(main())
+7
View File
@@ -0,0 +1,7 @@
[project]
name = "yt-shorts-downloader"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = []
+23
View File
@@ -0,0 +1,23 @@
from __future__ import annotations
from .api import download, download_short, validate_session_file
from .exceptions import (
InvalidSessionError,
InvalidUrlError,
JsRuntimeUnavailableError,
VideoDownloadError,
YtShortsDownloaderError,
)
from .models import SessionValidation
__all__ = [
"InvalidSessionError",
"InvalidUrlError",
"JsRuntimeUnavailableError",
"SessionValidation",
"VideoDownloadError",
"YtShortsDownloaderError",
"download",
"download_short",
"validate_session_file",
]
+25
View File
@@ -0,0 +1,25 @@
from pathlib import Path
from .api import PathInput as PathInput
from .exceptions import InvalidSessionError as InvalidSessionError
from .exceptions import InvalidUrlError as InvalidUrlError
from .exceptions import JsRuntimeUnavailableError as JsRuntimeUnavailableError
from .exceptions import VideoDownloadError as VideoDownloadError
from .exceptions import YtShortsDownloaderError as YtShortsDownloaderError
from .models import SessionValidation as SessionValidation
def download(
url: str,
session_path: PathInput,
*,
output_dir: PathInput | None = None,
) -> Path: ...
def download_short(
url: str,
session_path: PathInput,
*,
output_dir: PathInput | None = None,
) -> Path: ...
def validate_session_file(path: Path) -> SessionValidation: ...
__all__: list[str]
+6
View File
@@ -0,0 +1,6 @@
from __future__ import annotations
from .cli import main
if __name__ == "__main__":
raise SystemExit(main())
+71
View File
@@ -0,0 +1,71 @@
from __future__ import annotations
from os import PathLike
from pathlib import Path
from urllib.parse import urlparse
from .downloader import download_video
from .exceptions import InvalidSessionError, InvalidUrlError
from .session import validate_session_file
PathInput = str | PathLike[str]
_YOUTUBE_HOST_SUFFIXES = ("youtube.com", "youtu.be")
__all__ = ["download", "download_short", "validate_session_file"]
def download(
url: str,
session_path: PathInput,
*,
output_dir: PathInput | None = None,
) -> Path:
_validate_youtube_url(url)
normalized_session_path = Path(session_path).expanduser().resolve()
session_validation = validate_session_file(normalized_session_path)
if not session_validation.is_usable:
raise InvalidSessionError(session_validation.message)
normalized_output_dir = _normalize_output_dir(output_dir)
return download_video(
url=url,
output_dir=normalized_output_dir,
session_path=normalized_session_path,
)
def download_short(
url: str,
session_path: PathInput,
*,
output_dir: PathInput | None = None,
) -> Path:
return download(url=url, session_path=session_path, output_dir=output_dir)
def _validate_youtube_url(url: str) -> None:
parsed = urlparse(url)
if parsed.scheme not in {"http", "https"}:
raise InvalidUrlError("URL должна начинаться с http:// или https://")
host = parsed.netloc.lower()
if not host:
raise InvalidUrlError("Не удалось определить домен URL")
if not any(
host == suffix or host.endswith(f".{suffix}")
for suffix in _YOUTUBE_HOST_SUFFIXES
):
raise InvalidUrlError("Поддерживаются только ссылки YouTube")
def _normalize_output_dir(output_dir: PathInput | None) -> Path:
if output_dir is None:
normalized_output_dir = Path.cwd()
else:
normalized_output_dir = Path(output_dir).expanduser().resolve()
normalized_output_dir.mkdir(parents=True, exist_ok=True)
return normalized_output_dir
+22
View File
@@ -0,0 +1,22 @@
from os import PathLike
from pathlib import Path
from .models import SessionValidation
type PathInput = str | PathLike[str]
def download(
url: str,
session_path: PathInput,
*,
output_dir: PathInput | None = None,
) -> Path: ...
def download_short(
url: str,
session_path: PathInput,
*,
output_dir: PathInput | None = None,
) -> Path: ...
def validate_session_file(path: Path) -> SessionValidation: ...
__all__: list[str]
+46
View File
@@ -0,0 +1,46 @@
from __future__ import annotations
import argparse
from collections.abc import Sequence
from pathlib import Path
from .api import download
from .exceptions import YtShortsDownloaderError
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description=(
"Скачать YouTube Shorts через session file в Netscape cookie format."
),
)
parser.add_argument("url", help="Ссылка на YouTube Shorts")
parser.add_argument(
"session_path",
type=Path,
help="Путь до session file в Netscape cookie format",
)
parser.add_argument(
"--output-dir",
type=Path,
default=Path("."),
help="Папка, куда сохранить скачанный файл",
)
return parser
def main(argv: Sequence[str] | None = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
try:
downloaded_file = download(
url=args.url,
session_path=args.session_path,
output_dir=args.output_dir,
)
except YtShortsDownloaderError as exc:
parser.exit(status=1, message=f"[error] {exc}\n")
print(downloaded_file)
return 0
+133
View File
@@ -0,0 +1,133 @@
from __future__ import annotations
from pathlib import Path
from shutil import which
from typing import cast
from yt_dlp import YoutubeDL
from yt_dlp.utils import DownloadError as YtDlpDownloadError
from yt_dlp.utils import UnsupportedError
from .exceptions import InvalidUrlError, JsRuntimeUnavailableError, VideoDownloadError
from .runtime import JsRuntimeOptions, find_supported_js_runtimes
type Metadata = dict[str, object]
type YtDlpOptions = dict[str, object]
_DEFAULT_OUTTMPL = "%(title).200B [%(id)s].%(ext)s"
def download_video(url: str, output_dir: Path, session_path: Path) -> Path:
options = _build_yt_dlp_options(output_dir=output_dir, session_path=session_path)
try:
with YoutubeDL(options) as youtube_downloader:
extracted_info = cast(
object,
youtube_downloader.extract_info(url, download=True),
)
except UnsupportedError as exc:
raise InvalidUrlError(f"yt-dlp не поддерживает эту ссылку: {exc}") from exc
except YtDlpDownloadError as exc:
raise VideoDownloadError(f"Не удалось скачать видео: {exc}") from exc
if not isinstance(extracted_info, dict):
raise VideoDownloadError("yt-dlp вернул неожиданный формат метаданных")
downloaded_file = _locate_downloaded_file(
info=cast(Metadata, extracted_info),
output_dir=output_dir,
)
if downloaded_file is None:
raise VideoDownloadError(
"Скачивание завершено, но итоговый путь к файлу определить не удалось"
)
return downloaded_file
def _build_yt_dlp_options(output_dir: Path, session_path: Path) -> YtDlpOptions:
js_runtimes = _get_supported_js_runtimes()
ffmpeg_available = which("ffmpeg") is not None
format_selector = (
"bv*[ext=mp4]+ba[ext=m4a]/bv*+ba/b[ext=mp4]/b"
if ffmpeg_available
else "b[ext=mp4]/best"
)
return {
"cookiefile": str(session_path),
"format": format_selector,
"js_runtimes": js_runtimes,
"merge_output_format": "mp4",
"no_warnings": True,
"noplaylist": True,
"noprogress": True,
"outtmpl": str(output_dir / _DEFAULT_OUTTMPL),
"overwrites": False,
"quiet": True,
}
def _get_supported_js_runtimes() -> JsRuntimeOptions:
js_runtimes = find_supported_js_runtimes()
if js_runtimes is None:
raise JsRuntimeUnavailableError(
"Не найден поддерживаемый JavaScript runtime. "
"Установите deno или Node.js 22+."
)
return js_runtimes
def _unwrap_info(info: Metadata) -> Metadata:
if info.get("_type") != "playlist":
return info
entries = info.get("entries")
if not isinstance(entries, list):
return info
for entry in entries:
if isinstance(entry, dict):
return cast(Metadata, entry)
return info
def _locate_downloaded_file(info: Metadata, output_dir: Path) -> Path | None:
normalized_info = _unwrap_info(info)
requested_downloads = normalized_info.get("requested_downloads")
if isinstance(requested_downloads, list):
for item in requested_downloads:
if not isinstance(item, dict):
continue
filepath = item.get("filepath")
if isinstance(filepath, str):
candidate = Path(filepath)
if candidate.exists():
return candidate
for key in ("filepath", "_filename"):
filepath = normalized_info.get(key)
if isinstance(filepath, str):
candidate = Path(filepath)
if candidate.exists():
return candidate
video_id = normalized_info.get("id")
if not isinstance(video_id, str):
return None
matches = sorted(
(
path
for path in output_dir.iterdir()
if path.is_file() and f"[{video_id}]" in path.name
),
key=lambda path: path.stat().st_mtime,
reverse=True,
)
if matches:
return matches[0]
return None
+21
View File
@@ -0,0 +1,21 @@
from __future__ import annotations
class YtShortsDownloaderError(Exception):
pass
class InvalidUrlError(YtShortsDownloaderError):
pass
class InvalidSessionError(YtShortsDownloaderError):
pass
class JsRuntimeUnavailableError(YtShortsDownloaderError):
pass
class VideoDownloadError(YtShortsDownloaderError):
pass
+5
View File
@@ -0,0 +1,5 @@
class YtShortsDownloaderError(Exception): ...
class InvalidUrlError(YtShortsDownloaderError): ...
class InvalidSessionError(YtShortsDownloaderError): ...
class JsRuntimeUnavailableError(YtShortsDownloaderError): ...
class VideoDownloadError(YtShortsDownloaderError): ...
+23
View File
@@ -0,0 +1,23 @@
from __future__ import annotations
from dataclasses import dataclass
@dataclass(frozen=True, slots=True)
class SessionCookie:
domain: str
include_subdomains: bool
path: str
secure: bool
expires: int
name: str
value: str
@dataclass(frozen=True, slots=True)
class SessionValidation:
exists: bool
structurally_valid: bool
fresh: bool
is_usable: bool
message: str
+19
View File
@@ -0,0 +1,19 @@
from dataclasses import dataclass
@dataclass(frozen=True, slots=True)
class SessionCookie:
domain: str
include_subdomains: bool
path: str
secure: bool
expires: int
name: str
value: str
@dataclass(frozen=True, slots=True)
class SessionValidation:
exists: bool
structurally_valid: bool
fresh: bool
is_usable: bool
message: str
+80
View File
@@ -0,0 +1,80 @@
from __future__ import annotations
import subprocess
from pathlib import Path
from shutil import which
type JsRuntimeOptions = dict[str, dict[str, str]]
_MIN_NODE_VERSION = (22, 0, 0)
def find_supported_js_runtimes() -> JsRuntimeOptions | None:
deno_path = which("deno")
if deno_path is not None:
return {"deno": {"path": deno_path}}
node_candidates: dict[str, tuple[int, int, int]] = {}
node_on_path = which("node")
if node_on_path is not None:
version = _probe_executable_version(node_on_path)
if version is not None:
node_candidates[str(Path(node_on_path).resolve())] = version
nvm_versions_dir = Path.home() / ".nvm" / "versions" / "node"
if nvm_versions_dir.exists():
for candidate in nvm_versions_dir.glob("v*/bin/node"):
version = _probe_executable_version(str(candidate))
if version is not None:
node_candidates[str(candidate.resolve())] = version
supported_nodes = {
path: version
for path, version in node_candidates.items()
if version >= _MIN_NODE_VERSION
}
if not supported_nodes:
return None
best_node_path = max(supported_nodes, key=lambda path: supported_nodes[path])
return {"node": {"path": best_node_path}}
def _probe_executable_version(executable: str) -> tuple[int, int, int] | None:
try:
result = subprocess.run(
[executable, "--version"],
capture_output=True,
check=False,
text=True,
timeout=5,
)
except (FileNotFoundError, OSError, subprocess.TimeoutExpired):
return None
output = (result.stdout or result.stderr).strip()
if not output:
return None
return _parse_semver(output.splitlines()[0])
def _parse_semver(value: str) -> tuple[int, int, int] | None:
cleaned = value.strip().lstrip("vV")
parts = cleaned.split(".")
numbers: list[int] = []
for part in parts[:3]:
digits = ""
for character in part:
if not character.isdigit():
break
digits += character
if not digits:
return None
numbers.append(int(digits))
while len(numbers) < 3:
numbers.append(0)
return (numbers[0], numbers[1], numbers[2])
+185
View File
@@ -0,0 +1,185 @@
from __future__ import annotations
import time
from pathlib import Path
from .models import SessionCookie, SessionValidation
_SESSION_DOMAIN_SUFFIXES = ("youtube.com", "google.com")
_AUTH_COOKIE_NAMES = {
"SID",
"HSID",
"SSID",
"APISID",
"SAPISID",
"__Secure-1PSID",
"__Secure-3PSID",
"LOGIN_INFO",
}
_AUTH_COOKIE_ANCHORS = {
"SID",
"SAPISID",
"__Secure-1PSID",
"__Secure-3PSID",
"LOGIN_INFO",
}
def validate_session_file(path: Path) -> SessionValidation:
normalized_path = path.expanduser().resolve()
if not normalized_path.exists():
return SessionValidation(
exists=False,
structurally_valid=False,
fresh=False,
is_usable=False,
message=f"Файл сессии не найден: {normalized_path}",
)
try:
cookies = _parse_session_file(normalized_path)
except ValueError as exc:
return SessionValidation(
exists=True,
structurally_valid=False,
fresh=False,
is_usable=False,
message=f"Файл сессии не прошёл проверку формата: {exc}",
)
if not cookies:
return SessionValidation(
exists=True,
structurally_valid=False,
fresh=False,
is_usable=False,
message="Файл сессии пустой",
)
relevant_cookies = [
cookie
for cookie in cookies
if _matches_domain(cookie.domain, _SESSION_DOMAIN_SUFFIXES)
]
if not relevant_cookies:
return SessionValidation(
exists=True,
structurally_valid=False,
fresh=False,
is_usable=False,
message="В файле сессии нет записей для YouTube или Google",
)
now = int(time.time())
fresh_auth_names: set[str] = set()
soonest_expiry: int | None = None
for cookie in relevant_cookies:
if cookie.name not in _AUTH_COOKIE_NAMES:
continue
if cookie.expires != 0 and cookie.expires <= now:
continue
fresh_auth_names.add(cookie.name)
if cookie.expires > 0 and (
soonest_expiry is None or cookie.expires < soonest_expiry
):
soonest_expiry = cookie.expires
anchor_count = len(fresh_auth_names & _AUTH_COOKIE_ANCHORS)
fresh = len(fresh_auth_names) >= 3 and anchor_count >= 1
if not fresh:
return SessionValidation(
exists=True,
structurally_valid=True,
fresh=False,
is_usable=False,
message=(
"Файл сессии в корректном формате, но в нём нет достаточного набора "
"актуальных YouTube auth-cookie"
),
)
message = (
"Файл сессии валиден и выглядит актуальным для YouTube: "
f"найдено {len(fresh_auth_names)} свежих auth-cookie"
)
if soonest_expiry is not None:
hours_left = max((soonest_expiry - now) // 3600, 0)
message += f", ближайшее истечение примерно через {hours_left} ч"
return SessionValidation(
exists=True,
structurally_valid=True,
fresh=True,
is_usable=True,
message=message,
)
def _parse_session_file(path: Path) -> list[SessionCookie]:
cookies: list[SessionCookie] = []
with path.open("r", encoding="utf-8") as handle:
for line_number, raw_line in enumerate(handle, start=1):
stripped = raw_line.strip()
if not stripped:
continue
is_httponly_cookie = stripped.startswith("#HttpOnly_")
if stripped.startswith("#") and not is_httponly_cookie:
continue
if is_httponly_cookie:
stripped = stripped.removeprefix("#HttpOnly_")
parts = stripped.split("\t")
if len(parts) != 7:
raise ValueError(
"Строка "
f"{line_number}: ожидалось 7 колонок Netscape cookie file, "
f"получено {len(parts)}"
)
(
domain,
include_subdomains,
cookie_path,
secure,
expires,
name,
value,
) = parts
try:
expires_at = int(expires)
except ValueError as exc:
raise ValueError(
f"Строка {line_number}: expires должен быть UNIX timestamp"
) from exc
cookies.append(
SessionCookie(
domain=domain,
include_subdomains=include_subdomains.upper() == "TRUE",
path=cookie_path,
secure=secure.upper() == "TRUE",
expires=expires_at,
name=name,
value=value,
)
)
return cookies
def _matches_domain(domain: str, suffixes: tuple[str, ...]) -> bool:
normalized_domain = _normalize_domain(domain)
return any(
normalized_domain == suffix or normalized_domain.endswith(f".{suffix}")
for suffix in suffixes
)
def _normalize_domain(domain: str) -> str:
return domain.removeprefix("#HttpOnly_").lstrip(".").lower()
+50
View File
@@ -0,0 +1,50 @@
from __future__ import annotations
from pathlib import Path
import pytest
from yt_shorts_downloader import api
from yt_shorts_downloader.exceptions import InvalidSessionError, InvalidUrlError
from yt_shorts_downloader.models import SessionValidation
def test_download_returns_path_from_downloader(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
expected_path = tmp_path / "video.mp4"
monkeypatch.setattr(
api,
"validate_session_file",
lambda path: SessionValidation(True, True, True, True, str(path)),
)
monkeypatch.setattr(
api,
"download_video",
lambda *, url, output_dir, session_path: expected_path,
)
downloaded_path = api.download(
"https://youtube.com/shorts/example",
session_path="cookies.txt",
output_dir=tmp_path,
)
assert downloaded_path == expected_path
def test_download_rejects_invalid_url() -> None:
with pytest.raises(InvalidUrlError):
api.download("https://example.com/watch?v=1", session_path="cookies.txt")
def test_download_rejects_invalid_session(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
api,
"validate_session_file",
lambda path: SessionValidation(False, False, False, False, "bad session"),
)
with pytest.raises(InvalidSessionError, match="bad session"):
api.download("https://youtube.com/shorts/example", session_path="cookies.txt")
+43
View File
@@ -0,0 +1,43 @@
from __future__ import annotations
from pathlib import Path
import pytest
from yt_shorts_downloader import cli
from yt_shorts_downloader.exceptions import InvalidSessionError
def test_cli_prints_downloaded_path(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
tmp_path: Path,
) -> None:
expected_path = tmp_path / "video.mp4"
monkeypatch.setattr(cli, "download", lambda **kwargs: expected_path)
exit_code = cli.main(
[
"https://youtube.com/shorts/example",
"cookies.txt",
"--output-dir",
str(tmp_path),
]
)
captured = capsys.readouterr()
assert exit_code == 0
assert captured.out.strip() == str(expected_path)
def test_cli_exits_with_error(monkeypatch: pytest.MonkeyPatch) -> None:
def raise_error(**kwargs: object) -> Path:
raise InvalidSessionError("session error")
monkeypatch.setattr(cli, "download", raise_error)
with pytest.raises(SystemExit) as exc_info:
cli.main(["https://youtube.com/shorts/example", "cookies.txt"])
assert exc_info.value.code == 1
+52
View File
@@ -0,0 +1,52 @@
from __future__ import annotations
from pathlib import Path
import pytest
from yt_shorts_downloader import runtime
def test_find_supported_js_runtimes_prefers_deno(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(
runtime,
"which",
lambda executable: "/usr/bin/deno" if executable == "deno" else None,
)
runtimes = runtime.find_supported_js_runtimes()
assert runtimes == {"deno": {"path": "/usr/bin/deno"}}
def test_find_supported_js_runtimes_uses_best_supported_node(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
first_node = tmp_path / ".nvm" / "versions" / "node" / "v22.1.0" / "bin" / "node"
second_node = tmp_path / ".nvm" / "versions" / "node" / "v22.9.0" / "bin" / "node"
second_node.parent.mkdir(parents=True)
first_node.parent.mkdir(parents=True)
first_node.write_text("", encoding="utf-8")
second_node.write_text("", encoding="utf-8")
monkeypatch.setattr(runtime, "which", lambda executable: None)
monkeypatch.setattr(Path, "home", lambda: tmp_path)
monkeypatch.setattr(
runtime,
"_probe_executable_version",
lambda executable: (22, 9, 0)
if executable.endswith("v22.9.0/bin/node")
else (22, 1, 0),
)
runtimes = runtime.find_supported_js_runtimes()
assert runtimes == {"node": {"path": str(second_node.resolve())}}
def test_parse_semver_handles_prefixed_versions() -> None:
assert runtime._parse_semver("v22.12.1") == (22, 12, 1)
assert runtime._parse_semver("node") is None
+38
View File
@@ -0,0 +1,38 @@
from __future__ import annotations
from pathlib import Path
from yt_shorts_downloader.session import validate_session_file
def test_validate_session_file_accepts_valid_youtube_session(tmp_path: Path) -> None:
session_file = tmp_path / "cookies.txt"
session_file.write_text(
"\n".join(
[
"# Netscape HTTP Cookie File",
".youtube.com\tTRUE\t/\tFALSE\t9999999999\tSID\tvalue1",
".youtube.com\tTRUE\t/\tTRUE\t9999999999\tSAPISID\tvalue2",
".youtube.com\tTRUE\t/\tTRUE\t9999999999\tLOGIN_INFO\tvalue3",
]
),
encoding="utf-8",
)
validation = validate_session_file(session_file)
assert validation.exists is True
assert validation.structurally_valid is True
assert validation.fresh is True
assert validation.is_usable is True
def test_validate_session_file_rejects_invalid_format(tmp_path: Path) -> None:
session_file = tmp_path / "cookies.txt"
session_file.write_text("broken\trow", encoding="utf-8")
validation = validate_session_file(session_file)
assert validation.exists is True
assert validation.structurally_valid is False
assert validation.is_usable is False