Refactor code style and improve documentation
- Standardized string quotes to single quotes across all files. - Added docstrings to several functions and classes for better clarity. - Updated mypy configuration in pyproject.toml for enhanced type checking. - Ignored specific linting rules for test files in ruff configuration. - Improved error messages in exception handling for better user feedback. - Cleaned up code formatting and structure for consistency.
This commit is contained in:
@@ -11,16 +11,16 @@ from .exceptions import (
|
||||
from .models import DownloadedVideo, SessionValidation
|
||||
|
||||
__all__ = [
|
||||
"DownloadedVideo",
|
||||
"InvalidSessionError",
|
||||
"InvalidUrlError",
|
||||
"JsRuntimeUnavailableError",
|
||||
"PathInput",
|
||||
"SessionValidation",
|
||||
"VideoDownloadError",
|
||||
"YtShortsDownloaderError",
|
||||
"download",
|
||||
"download_short",
|
||||
"download_to_path",
|
||||
"validate_session_file",
|
||||
'DownloadedVideo',
|
||||
'InvalidSessionError',
|
||||
'InvalidUrlError',
|
||||
'JsRuntimeUnavailableError',
|
||||
'PathInput',
|
||||
'SessionValidation',
|
||||
'VideoDownloadError',
|
||||
'YtShortsDownloaderError',
|
||||
'download',
|
||||
'download_short',
|
||||
'download_to_path',
|
||||
'validate_session_file',
|
||||
]
|
||||
|
||||
@@ -2,5 +2,5 @@ from __future__ import annotations
|
||||
|
||||
from .cli import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
if __name__ == '__main__':
|
||||
raise SystemExit(main())
|
||||
|
||||
@@ -13,11 +13,11 @@ from .services.session import validate_session_file
|
||||
PathInput = str | PathLike[str]
|
||||
|
||||
__all__ = [
|
||||
"PathInput",
|
||||
"download",
|
||||
"download_short",
|
||||
"download_to_path",
|
||||
"validate_session_file",
|
||||
'PathInput',
|
||||
'download',
|
||||
'download_short',
|
||||
'download_to_path',
|
||||
'validate_session_file',
|
||||
]
|
||||
|
||||
|
||||
@@ -25,10 +25,11 @@ def download(
|
||||
url: str,
|
||||
session_path: PathInput,
|
||||
) -> DownloadedVideo:
|
||||
"""Download a YouTube Short and return its MP4 bytes in memory."""
|
||||
validate_youtube_url(url)
|
||||
normalized_session_path = _normalize_session_path(session_path)
|
||||
|
||||
with TemporaryDirectory(prefix="yt-shorts-downloader-") as temporary_directory:
|
||||
with TemporaryDirectory(prefix='yt-shorts-downloader-') as temporary_directory:
|
||||
downloaded_file = download_video(
|
||||
url=url,
|
||||
output_dir=Path(temporary_directory),
|
||||
@@ -41,6 +42,7 @@ def download_short(
|
||||
url: str,
|
||||
session_path: PathInput,
|
||||
) -> DownloadedVideo:
|
||||
"""Alias for download kept for API readability."""
|
||||
return download(url=url, session_path=session_path)
|
||||
|
||||
|
||||
@@ -50,6 +52,7 @@ def download_to_path(
|
||||
*,
|
||||
output_dir: PathInput | None = None,
|
||||
) -> Path:
|
||||
"""Download a YouTube Short and persist the resulting MP4 to disk."""
|
||||
validate_youtube_url(url)
|
||||
normalized_session_path = _normalize_session_path(session_path)
|
||||
normalized_output_dir = _normalize_output_dir(output_dir)
|
||||
@@ -69,10 +72,7 @@ def _normalize_session_path(session_path: PathInput) -> Path:
|
||||
|
||||
|
||||
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 = Path.cwd() if output_dir is None else Path(output_dir).expanduser().resolve()
|
||||
|
||||
normalized_output_dir.mkdir(parents=True, exist_ok=True)
|
||||
return normalized_output_dir
|
||||
@@ -81,9 +81,9 @@ def _normalize_output_dir(output_dir: PathInput | None) -> Path:
|
||||
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")
|
||||
raise VideoDownloadError('Скачивание завершилось без итогового файла на диске')
|
||||
if normalized_downloaded_file.suffix.lower() != '.mp4':
|
||||
raise VideoDownloadError('Публичный API поддерживает только итоговый MP4')
|
||||
|
||||
return DownloadedVideo(
|
||||
filename=normalized_downloaded_file.name,
|
||||
|
||||
@@ -9,25 +9,27 @@ from .exceptions import YtShortsDownloaderError
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
"""Build the CLI argument parser."""
|
||||
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('url', help='Ссылка на YouTube Shorts')
|
||||
parser.add_argument(
|
||||
"session_path",
|
||||
'session_path',
|
||||
type=Path,
|
||||
help="Путь до session file в Netscape cookie format",
|
||||
help='Путь до session file в Netscape cookie format',
|
||||
)
|
||||
parser.add_argument(
|
||||
"--output-dir",
|
||||
'--output-dir',
|
||||
type=Path,
|
||||
default=Path("."),
|
||||
help="Папка, куда сохранить скачанный файл",
|
||||
default=Path('.'),
|
||||
help='Папка, куда сохранить скачанный файл',
|
||||
)
|
||||
return parser
|
||||
|
||||
|
||||
def main(argv: Sequence[str] | None = None) -> int:
|
||||
"""Run the CLI entrypoint and return the exit code."""
|
||||
parser = build_parser()
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
@@ -38,7 +40,7 @@ def main(argv: Sequence[str] | None = None) -> int:
|
||||
output_dir=args.output_dir,
|
||||
)
|
||||
except YtShortsDownloaderError as exc:
|
||||
parser.exit(status=1, message=f"[error] {exc}\n")
|
||||
parser.exit(status=1, message=f'[error] {exc}\n')
|
||||
|
||||
print(downloaded_file)
|
||||
return 0
|
||||
|
||||
@@ -5,8 +5,8 @@ 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",
|
||||
'JsRuntimeOptions',
|
||||
'download_video',
|
||||
'find_supported_js_runtimes',
|
||||
'validate_youtube_url',
|
||||
]
|
||||
|
||||
@@ -14,70 +14,67 @@ 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"
|
||||
_DEFAULT_OUTTMPL: Final[str] = '%(title).200B [%(id)s].%(ext)s'
|
||||
|
||||
|
||||
def download_video(url: str, output_dir: Path, session_path: Path) -> Path:
|
||||
"""Download a video with yt-dlp and return the final MP4 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
|
||||
raise InvalidUrlError(f'yt-dlp не поддерживает эту ссылку: {exc}') from exc
|
||||
except YtDlpDownloadError as exc:
|
||||
raise VideoDownloadError(f"Не удалось скачать видео: {exc}") from exc
|
||||
raise VideoDownloadError(f'Не удалось скачать видео: {exc}') from exc
|
||||
|
||||
if not isinstance(extracted_info, dict):
|
||||
raise VideoDownloadError("yt-dlp вернул неожиданный формат метаданных")
|
||||
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")
|
||||
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]"
|
||||
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,
|
||||
'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+."
|
||||
)
|
||||
raise JsRuntimeUnavailableError('Не найден поддерживаемый JavaScript runtime. Установите deno или Node.js 22+.')
|
||||
return js_runtimes
|
||||
|
||||
|
||||
def _unwrap_info(info: Metadata) -> Metadata:
|
||||
if info.get("_type") != "playlist":
|
||||
if info.get('_type') != 'playlist':
|
||||
return info
|
||||
|
||||
entries = info.get("entries")
|
||||
entries = info.get('entries')
|
||||
if not isinstance(entries, list):
|
||||
return info
|
||||
|
||||
@@ -90,30 +87,30 @@ def _unwrap_info(info: Metadata) -> Metadata:
|
||||
def _locate_downloaded_file(info: Metadata, output_dir: Path) -> Path | None:
|
||||
normalized_info = _unwrap_info(info)
|
||||
|
||||
requested_downloads = normalized_info.get("requested_downloads")
|
||||
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")
|
||||
filepath = item.get('filepath')
|
||||
if isinstance(filepath, str):
|
||||
candidate = Path(filepath)
|
||||
if candidate.exists():
|
||||
return candidate
|
||||
|
||||
for key in ("filepath", "_filename"):
|
||||
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")
|
||||
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),
|
||||
(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,
|
||||
)
|
||||
|
||||
@@ -7,41 +7,41 @@ from shutil import which
|
||||
type JsRuntimeOptions = dict[str, dict[str, str]]
|
||||
|
||||
_MIN_NODE_VERSION = (22, 0, 0)
|
||||
_SEMVER_PART_COUNT = 3
|
||||
|
||||
|
||||
def find_supported_js_runtimes() -> JsRuntimeOptions | None:
|
||||
deno_path = which("deno")
|
||||
"""Return the best supported JavaScript runtime for yt-dlp."""
|
||||
deno_path = which('deno')
|
||||
if deno_path is not None:
|
||||
return {"deno": {"path": deno_path}}
|
||||
return {'deno': {'path': deno_path}}
|
||||
|
||||
node_candidates: dict[str, tuple[int, int, int]] = {}
|
||||
node_on_path = which("node")
|
||||
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"
|
||||
nvm_versions_dir = Path.home() / '.nvm' / 'versions' / 'node'
|
||||
if nvm_versions_dir.exists():
|
||||
for candidate in nvm_versions_dir.glob("v*/bin/node"):
|
||||
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
|
||||
}
|
||||
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}}
|
||||
return {'node': {'path': best_node_path}}
|
||||
|
||||
|
||||
def _probe_executable_version(executable: str) -> tuple[int, int, int] | None:
|
||||
try:
|
||||
result = subprocess.run(
|
||||
[executable, "--version"],
|
||||
[executable, '--version'],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
text=True,
|
||||
@@ -58,12 +58,12 @@ def _probe_executable_version(executable: str) -> tuple[int, int, int] | None:
|
||||
|
||||
|
||||
def _parse_semver(value: str) -> tuple[int, int, int] | None:
|
||||
cleaned = value.strip().lstrip("vV")
|
||||
parts = cleaned.split(".")
|
||||
cleaned = value.strip().lstrip('vV')
|
||||
parts = cleaned.split('.')
|
||||
numbers: list[int] = []
|
||||
|
||||
for part in parts[:3]:
|
||||
digits = ""
|
||||
for part in parts[:_SEMVER_PART_COUNT]:
|
||||
digits = ''
|
||||
for character in part:
|
||||
if not character.isdigit():
|
||||
break
|
||||
@@ -72,7 +72,7 @@ def _parse_semver(value: str) -> tuple[int, int, int] | None:
|
||||
return None
|
||||
numbers.append(int(digits))
|
||||
|
||||
while len(numbers) < 3:
|
||||
while len(numbers) < _SEMVER_PART_COUNT:
|
||||
numbers.append(0)
|
||||
|
||||
return (numbers[0], numbers[1], numbers[2])
|
||||
|
||||
@@ -4,17 +4,18 @@ from urllib.parse import urlparse
|
||||
|
||||
from ..exceptions import InvalidUrlError
|
||||
|
||||
_YOUTUBE_HOST_SUFFIXES = ("youtube.com", "youtu.be")
|
||||
_YOUTUBE_HOST_SUFFIXES = ('youtube.com', 'youtu.be')
|
||||
|
||||
|
||||
def validate_youtube_url(url: str) -> None:
|
||||
"""Validate that the input URL points to a supported YouTube host."""
|
||||
parsed = urlparse(url)
|
||||
if parsed.scheme not in {"http", "https"}:
|
||||
raise InvalidUrlError("URL должна начинаться с http:// или https://")
|
||||
if parsed.scheme not in {'http', 'https'}:
|
||||
raise InvalidUrlError('URL должна начинаться с http:// или https://')
|
||||
|
||||
host = parsed.netloc.lower()
|
||||
if not host:
|
||||
raise InvalidUrlError("Не удалось определить домен URL")
|
||||
raise InvalidUrlError('Не удалось определить домен URL')
|
||||
|
||||
if not any(host == suffix or host.endswith(f".{suffix}") for suffix in _YOUTUBE_HOST_SUFFIXES):
|
||||
raise InvalidUrlError("Поддерживаются только ссылки YouTube")
|
||||
if not any(host == suffix or host.endswith(f'.{suffix}') for suffix in _YOUTUBE_HOST_SUFFIXES):
|
||||
raise InvalidUrlError('Поддерживаются только ссылки YouTube')
|
||||
|
||||
@@ -2,4 +2,4 @@ from __future__ import annotations
|
||||
|
||||
from .core.downloader import download_video
|
||||
|
||||
__all__ = ["download_video"]
|
||||
__all__ = ['download_video']
|
||||
|
||||
@@ -2,20 +2,30 @@ from __future__ import annotations
|
||||
|
||||
|
||||
class YtShortsDownloaderError(Exception):
|
||||
"""Base exception for package-specific failures."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class InvalidUrlError(YtShortsDownloaderError):
|
||||
"""Raised when the provided URL is not a supported YouTube URL."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class InvalidSessionError(YtShortsDownloaderError):
|
||||
"""Raised when the provided session file cannot be used."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class JsRuntimeUnavailableError(YtShortsDownloaderError):
|
||||
"""Raised when no supported JavaScript runtime is available."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class VideoDownloadError(YtShortsDownloaderError):
|
||||
"""Raised when yt-dlp cannot produce a valid MP4 result."""
|
||||
|
||||
pass
|
||||
|
||||
@@ -3,4 +3,4 @@ from __future__ import annotations
|
||||
from .download import DownloadedVideo
|
||||
from .session import SessionCookie, SessionValidation
|
||||
|
||||
__all__ = ["DownloadedVideo", "SessionCookie", "SessionValidation"]
|
||||
__all__ = ['DownloadedVideo', 'SessionCookie', 'SessionValidation']
|
||||
|
||||
@@ -5,6 +5,8 @@ from dataclasses import dataclass
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class DownloadedVideo:
|
||||
"""In-memory representation of a downloaded MP4 file."""
|
||||
|
||||
filename: str
|
||||
content: bytes
|
||||
media_type: str = "video/mp4"
|
||||
media_type: str = 'video/mp4'
|
||||
|
||||
@@ -5,6 +5,8 @@ from dataclasses import dataclass
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SessionCookie:
|
||||
"""Normalized cookie parsed from a Netscape session file."""
|
||||
|
||||
domain: str
|
||||
include_subdomains: bool
|
||||
path: str
|
||||
@@ -16,6 +18,8 @@ class SessionCookie:
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SessionValidation:
|
||||
"""Validation result for a candidate session cookie file."""
|
||||
|
||||
exists: bool
|
||||
structurally_valid: bool
|
||||
fresh: bool
|
||||
|
||||
@@ -2,4 +2,4 @@ from __future__ import annotations
|
||||
|
||||
from .core.runtime import JsRuntimeOptions, _parse_semver, find_supported_js_runtimes
|
||||
|
||||
__all__ = ["JsRuntimeOptions", "_parse_semver", "find_supported_js_runtimes"]
|
||||
__all__ = ['JsRuntimeOptions', '_parse_semver', 'find_supported_js_runtimes']
|
||||
|
||||
@@ -2,4 +2,4 @@ from __future__ import annotations
|
||||
|
||||
from .session import validate_session_file
|
||||
|
||||
__all__ = ["validate_session_file"]
|
||||
__all__ = ['validate_session_file']
|
||||
|
||||
@@ -5,27 +5,30 @@ from pathlib import Path
|
||||
|
||||
from ..models import SessionCookie, SessionValidation
|
||||
|
||||
_SESSION_DOMAIN_SUFFIXES = ("youtube.com", "google.com")
|
||||
_SESSION_DOMAIN_SUFFIXES = ('youtube.com', 'google.com')
|
||||
_AUTH_COOKIE_NAMES = {
|
||||
"SID",
|
||||
"HSID",
|
||||
"SSID",
|
||||
"APISID",
|
||||
"SAPISID",
|
||||
"__Secure-1PSID",
|
||||
"__Secure-3PSID",
|
||||
"LOGIN_INFO",
|
||||
'SID',
|
||||
'HSID',
|
||||
'SSID',
|
||||
'APISID',
|
||||
'SAPISID',
|
||||
'__Secure-1PSID',
|
||||
'__Secure-3PSID',
|
||||
'LOGIN_INFO',
|
||||
}
|
||||
_AUTH_COOKIE_ANCHORS = {
|
||||
"SID",
|
||||
"SAPISID",
|
||||
"__Secure-1PSID",
|
||||
"__Secure-3PSID",
|
||||
"LOGIN_INFO",
|
||||
'SID',
|
||||
'SAPISID',
|
||||
'__Secure-1PSID',
|
||||
'__Secure-3PSID',
|
||||
'LOGIN_INFO',
|
||||
}
|
||||
_MIN_FRESH_AUTH_COOKIES = 3
|
||||
_NETSCAPE_COOKIE_COLUMNS = 7
|
||||
|
||||
|
||||
def validate_session_file(path: Path) -> SessionValidation:
|
||||
"""Validate that a Netscape cookie file is usable for YouTube downloads."""
|
||||
normalized_path = path.expanduser().resolve()
|
||||
if not normalized_path.exists():
|
||||
return SessionValidation(
|
||||
@@ -33,7 +36,7 @@ def validate_session_file(path: Path) -> SessionValidation:
|
||||
structurally_valid=False,
|
||||
fresh=False,
|
||||
is_usable=False,
|
||||
message=f"Файл сессии не найден: {normalized_path}",
|
||||
message=f'Файл сессии не найден: {normalized_path}',
|
||||
)
|
||||
|
||||
try:
|
||||
@@ -44,7 +47,7 @@ def validate_session_file(path: Path) -> SessionValidation:
|
||||
structurally_valid=False,
|
||||
fresh=False,
|
||||
is_usable=False,
|
||||
message=f"Файл сессии не прошёл проверку формата: {exc}",
|
||||
message=f'Файл сессии не прошёл проверку формата: {exc}',
|
||||
)
|
||||
|
||||
if not cookies:
|
||||
@@ -53,19 +56,17 @@ def validate_session_file(path: Path) -> SessionValidation:
|
||||
structurally_valid=False,
|
||||
fresh=False,
|
||||
is_usable=False,
|
||||
message="Файл сессии пустой",
|
||||
message='Файл сессии пустой',
|
||||
)
|
||||
|
||||
relevant_cookies = [
|
||||
cookie for cookie in cookies if _matches_domain(cookie.domain, _SESSION_DOMAIN_SUFFIXES)
|
||||
]
|
||||
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",
|
||||
message='В файле сессии нет записей для YouTube или Google',
|
||||
)
|
||||
|
||||
now = int(time.time())
|
||||
@@ -83,26 +84,20 @@ def validate_session_file(path: Path) -> SessionValidation:
|
||||
soonest_expiry = cookie.expires
|
||||
|
||||
anchor_count = len(fresh_auth_names & _AUTH_COOKIE_ANCHORS)
|
||||
fresh = len(fresh_auth_names) >= 3 and anchor_count >= 1
|
||||
fresh = len(fresh_auth_names) >= _MIN_FRESH_AUTH_COOKIES 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 auth-cookie'),
|
||||
)
|
||||
|
||||
message = (
|
||||
"Файл сессии валиден и выглядит актуальным для YouTube: "
|
||||
f"найдено {len(fresh_auth_names)} свежих auth-cookie"
|
||||
)
|
||||
message = f'Файл сессии валиден и выглядит актуальным для YouTube: найдено {len(fresh_auth_names)} свежих auth-cookie'
|
||||
if soonest_expiry is not None:
|
||||
hours_left = max((soonest_expiry - now) // 3600, 0)
|
||||
message += f", ближайшее истечение примерно через {hours_left} ч"
|
||||
message += f', ближайшее истечение примерно через {hours_left} ч'
|
||||
|
||||
return SessionValidation(
|
||||
exists=True,
|
||||
@@ -116,26 +111,22 @@ def validate_session_file(path: Path) -> SessionValidation:
|
||||
def _parse_session_file(path: Path) -> list[SessionCookie]:
|
||||
cookies: list[SessionCookie] = []
|
||||
|
||||
with path.open("r", encoding="utf-8") as handle:
|
||||
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:
|
||||
is_httponly_cookie = stripped.startswith('#HttpOnly_')
|
||||
if stripped.startswith('#') and not is_httponly_cookie:
|
||||
continue
|
||||
|
||||
if is_httponly_cookie:
|
||||
stripped = stripped.removeprefix("#HttpOnly_")
|
||||
stripped = stripped.removeprefix('#HttpOnly_')
|
||||
|
||||
parts = stripped.split("\t")
|
||||
if len(parts) != 7:
|
||||
raise ValueError(
|
||||
"Строка "
|
||||
f"{line_number}: ожидалось 7 колонок Netscape cookie file, "
|
||||
f"получено {len(parts)}"
|
||||
)
|
||||
parts = stripped.split('\t')
|
||||
if len(parts) != _NETSCAPE_COOKIE_COLUMNS:
|
||||
raise ValueError(f'Строка {line_number}: ожидалось {_NETSCAPE_COOKIE_COLUMNS} колонок Netscape cookie file, получено {len(parts)}')
|
||||
|
||||
(
|
||||
domain,
|
||||
@@ -150,16 +141,14 @@ def _parse_session_file(path: Path) -> list[SessionCookie]:
|
||||
try:
|
||||
expires_at = int(expires)
|
||||
except ValueError as exc:
|
||||
raise ValueError(
|
||||
f"Строка {line_number}: expires должен быть UNIX timestamp"
|
||||
) from exc
|
||||
raise ValueError(f'Строка {line_number}: expires должен быть UNIX timestamp') from exc
|
||||
|
||||
cookies.append(
|
||||
SessionCookie(
|
||||
domain=domain,
|
||||
include_subdomains=include_subdomains.upper() == "TRUE",
|
||||
include_subdomains=include_subdomains.upper() == 'TRUE',
|
||||
path=cookie_path,
|
||||
secure=secure.upper() == "TRUE",
|
||||
secure=secure.upper() == 'TRUE',
|
||||
expires=expires_at,
|
||||
name=name,
|
||||
value=value,
|
||||
@@ -171,11 +160,8 @@ def _parse_session_file(path: Path) -> list[SessionCookie]:
|
||||
|
||||
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
|
||||
)
|
||||
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()
|
||||
return domain.removeprefix('#HttpOnly_').lstrip('.').lower()
|
||||
|
||||
@@ -2,4 +2,4 @@ from __future__ import annotations
|
||||
|
||||
from .services.session import validate_session_file
|
||||
|
||||
__all__ = ["validate_session_file"]
|
||||
__all__ = ['validate_session_file']
|
||||
|
||||
Reference in New Issue
Block a user