Refactor code structure for improved readability and maintainability
This commit is contained in:
ВяткинАртём
@@ -1,6 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from .api import download, download_short, validate_session_file
|
||||
from .api import PathInput, download, download_short, download_to_path, validate_session_file
|
||||
from .exceptions import (
|
||||
InvalidSessionError,
|
||||
InvalidUrlError,
|
||||
@@ -8,16 +8,19 @@ from .exceptions import (
|
||||
VideoDownloadError,
|
||||
YtShortsDownloaderError,
|
||||
)
|
||||
from .models import SessionValidation
|
||||
from .models import DownloadedVideo, SessionValidation
|
||||
|
||||
__all__ = [
|
||||
"DownloadedVideo",
|
||||
"InvalidSessionError",
|
||||
"InvalidUrlError",
|
||||
"JsRuntimeUnavailableError",
|
||||
"PathInput",
|
||||
"SessionValidation",
|
||||
"VideoDownloadError",
|
||||
"YtShortsDownloaderError",
|
||||
"download",
|
||||
"download_short",
|
||||
"download_to_path",
|
||||
"validate_session_file",
|
||||
]
|
||||
|
||||
@@ -1,25 +0,0 @@
|
||||
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]
|
||||
@@ -2,32 +2,56 @@ from __future__ import annotations
|
||||
|
||||
from os import PathLike
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlparse
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
from .downloader import download_video
|
||||
from .exceptions import InvalidSessionError, InvalidUrlError
|
||||
from .session import validate_session_file
|
||||
from .core.downloader import download_video
|
||||
from .core.urls import validate_youtube_url
|
||||
from .exceptions import InvalidSessionError, VideoDownloadError
|
||||
from .models import DownloadedVideo
|
||||
from .services.session import validate_session_file
|
||||
|
||||
PathInput = str | PathLike[str]
|
||||
|
||||
_YOUTUBE_HOST_SUFFIXES = ("youtube.com", "youtu.be")
|
||||
|
||||
__all__ = ["download", "download_short", "validate_session_file"]
|
||||
__all__ = [
|
||||
"PathInput",
|
||||
"download",
|
||||
"download_short",
|
||||
"download_to_path",
|
||||
"validate_session_file",
|
||||
]
|
||||
|
||||
|
||||
def download(
|
||||
url: str,
|
||||
session_path: PathInput,
|
||||
) -> DownloadedVideo:
|
||||
validate_youtube_url(url)
|
||||
normalized_session_path = _normalize_session_path(session_path)
|
||||
|
||||
with TemporaryDirectory(prefix="yt-shorts-downloader-") as temporary_directory:
|
||||
downloaded_file = download_video(
|
||||
url=url,
|
||||
output_dir=Path(temporary_directory),
|
||||
session_path=normalized_session_path,
|
||||
)
|
||||
return _read_downloaded_video(downloaded_file)
|
||||
|
||||
|
||||
def download_short(
|
||||
url: str,
|
||||
session_path: PathInput,
|
||||
) -> DownloadedVideo:
|
||||
return download(url=url, session_path=session_path)
|
||||
|
||||
|
||||
def download_to_path(
|
||||
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)
|
||||
|
||||
validate_youtube_url(url)
|
||||
normalized_session_path = _normalize_session_path(session_path)
|
||||
normalized_output_dir = _normalize_output_dir(output_dir)
|
||||
return download_video(
|
||||
url=url,
|
||||
@@ -36,29 +60,12 @@ def download(
|
||||
)
|
||||
|
||||
|
||||
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_session_path(session_path: PathInput) -> Path:
|
||||
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)
|
||||
return normalized_session_path
|
||||
|
||||
|
||||
def _normalize_output_dir(output_dir: PathInput | None) -> Path:
|
||||
@@ -69,3 +76,16 @@ def _normalize_output_dir(output_dir: PathInput | None) -> Path:
|
||||
|
||||
normalized_output_dir.mkdir(parents=True, exist_ok=True)
|
||||
return normalized_output_dir
|
||||
|
||||
|
||||
def _read_downloaded_video(downloaded_file: Path) -> DownloadedVideo:
|
||||
normalized_downloaded_file = downloaded_file.expanduser().resolve()
|
||||
if not normalized_downloaded_file.exists():
|
||||
raise VideoDownloadError("Скачивание завершилось без итогового файла на диске")
|
||||
if normalized_downloaded_file.suffix.lower() != ".mp4":
|
||||
raise VideoDownloadError("Публичный API поддерживает только итоговый MP4")
|
||||
|
||||
return DownloadedVideo(
|
||||
filename=normalized_downloaded_file.name,
|
||||
content=normalized_downloaded_file.read_bytes(),
|
||||
)
|
||||
|
||||
@@ -1,22 +0,0 @@
|
||||
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]
|
||||
@@ -4,15 +4,13 @@ import argparse
|
||||
from collections.abc import Sequence
|
||||
from pathlib import Path
|
||||
|
||||
from .api import download
|
||||
from .api import download_to_path
|
||||
from .exceptions import YtShortsDownloaderError
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(
|
||||
description=(
|
||||
"Скачать YouTube Shorts через session file в Netscape cookie format."
|
||||
),
|
||||
description=("Скачать YouTube Shorts через session file в Netscape cookie format."),
|
||||
)
|
||||
parser.add_argument("url", help="Ссылка на YouTube Shorts")
|
||||
parser.add_argument(
|
||||
@@ -34,7 +32,7 @@ def main(argv: Sequence[str] | None = None) -> int:
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
try:
|
||||
downloaded_file = download(
|
||||
downloaded_file = download_to_path(
|
||||
url=args.url,
|
||||
session_path=args.session_path,
|
||||
output_dir=args.output_dir,
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from .downloader import download_video
|
||||
from .runtime import JsRuntimeOptions, find_supported_js_runtimes
|
||||
from .urls import validate_youtube_url
|
||||
|
||||
__all__ = [
|
||||
"JsRuntimeOptions",
|
||||
"download_video",
|
||||
"find_supported_js_runtimes",
|
||||
"validate_youtube_url",
|
||||
]
|
||||
@@ -0,0 +1,123 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from shutil import which
|
||||
from typing import Final, 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: Final[str] = "%(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 = 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(
|
||||
"Скачивание завершено, но итоговый путь к файлу определить не удалось"
|
||||
)
|
||||
if downloaded_file.suffix.lower() != ".mp4":
|
||||
raise VideoDownloadError("Итоговый файл не является MP4")
|
||||
|
||||
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]/b[ext=mp4]" if ffmpeg_available else "b[ext=mp4]"
|
||||
|
||||
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
|
||||
@@ -0,0 +1,78 @@
|
||||
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])
|
||||
@@ -0,0 +1,20 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from ..exceptions import InvalidUrlError
|
||||
|
||||
_YOUTUBE_HOST_SUFFIXES = ("youtube.com", "youtu.be")
|
||||
|
||||
|
||||
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")
|
||||
@@ -1,133 +1,5 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from shutil import which
|
||||
from typing import cast
|
||||
from .core.downloader import download_video
|
||||
|
||||
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
|
||||
__all__ = ["download_video"]
|
||||
|
||||
@@ -1,5 +0,0 @@
|
||||
class YtShortsDownloaderError(Exception): ...
|
||||
class InvalidUrlError(YtShortsDownloaderError): ...
|
||||
class InvalidSessionError(YtShortsDownloaderError): ...
|
||||
class JsRuntimeUnavailableError(YtShortsDownloaderError): ...
|
||||
class VideoDownloadError(YtShortsDownloaderError): ...
|
||||
@@ -1,19 +0,0 @@
|
||||
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
|
||||
@@ -0,0 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from .download import DownloadedVideo
|
||||
from .session import SessionCookie, SessionValidation
|
||||
|
||||
__all__ = ["DownloadedVideo", "SessionCookie", "SessionValidation"]
|
||||
@@ -0,0 +1,10 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class DownloadedVideo:
|
||||
filename: str
|
||||
content: bytes
|
||||
media_type: str = "video/mp4"
|
||||
@@ -1,80 +1,5 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from shutil import which
|
||||
from .core.runtime import JsRuntimeOptions, _parse_semver, find_supported_js_runtimes
|
||||
|
||||
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])
|
||||
__all__ = ["JsRuntimeOptions", "_parse_semver", "find_supported_js_runtimes"]
|
||||
|
||||
@@ -0,0 +1,5 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from .session import validate_session_file
|
||||
|
||||
__all__ = ["validate_session_file"]
|
||||
@@ -0,0 +1,181 @@
|
||||
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()
|
||||
@@ -1,185 +1,5 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from pathlib import Path
|
||||
from .services.session import validate_session_file
|
||||
|
||||
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()
|
||||
__all__ = ["validate_session_file"]
|
||||
|
||||
Reference in New Issue
Block a user