2819 lines
74 KiB
Python
2819 lines
74 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Рекурсивное скачивание защищённой публичной папки Яндекс Диска.
|
||
|
||
Возможности:
|
||
- все пользовательские настройки читаются из
|
||
yandex_downloader.env рядом со скриптом;
|
||
- passToken автоматически получается по паролю через Chromium;
|
||
- постоянный профиль Chromium повторно использует действующие cookies;
|
||
- папка скачивается пофайлово, без формирования ZIP;
|
||
- сохраняется структура вложенных каталогов;
|
||
- wget продолжает недокачанные файлы;
|
||
- полностью скачанные файлы пропускаются;
|
||
- при CAPTCHA весь процесс немедленно останавливается;
|
||
- ссылка CAPTCHA сохраняется в файл;
|
||
- повторный запуск продолжает загрузку;
|
||
- ITEMS_PER_PAGE не используется: размер страницы определяет Яндекс.
|
||
|
||
Зависимости:
|
||
|
||
sudo apt install -y python3 python3-requests wget
|
||
python3 -m pip install --user playwright
|
||
python3 -m playwright install chromium
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import getpass
|
||
import html
|
||
import json
|
||
import random
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Any, Iterable
|
||
from urllib.parse import quote, urlparse
|
||
|
||
import requests
|
||
from requests.adapters import HTTPAdapter
|
||
from urllib3.util.retry import Retry
|
||
|
||
# ========================== НАСТРОЙКИ ==========================
|
||
|
||
# Файл со всеми пользовательскими настройками.
|
||
# Путь вычисляется относительно самого скрипта, поэтому запускать его
|
||
# можно из любого текущего каталога.
|
||
ENV_FILE = Path(__file__).resolve().with_name(
|
||
".env"
|
||
)
|
||
|
||
# ===============================================================
|
||
|
||
|
||
USER_AGENT = (
|
||
"Mozilla/5.0 (X11; Linux x86_64) "
|
||
"AppleWebKit/537.36 "
|
||
"(KHTML, like Gecko) "
|
||
"Chrome/142.0.0.0 Safari/537.36"
|
||
)
|
||
|
||
|
||
class YandexDownloadError(RuntimeError):
|
||
"""Обычная ошибка взаимодействия с Яндекс Диском."""
|
||
|
||
|
||
class CaptchaRequired(YandexDownloadError):
|
||
"""Яндекс потребовал ручное прохождение CAPTCHA."""
|
||
|
||
def __init__(
|
||
self,
|
||
captcha_url: str,
|
||
response_data: dict[str, Any] | None = None,
|
||
) -> None:
|
||
self.captcha_url = captcha_url
|
||
self.response_data = response_data or {}
|
||
|
||
super().__init__(
|
||
"Яндекс потребовал CAPTCHA:\n"
|
||
f"{captcha_url}"
|
||
)
|
||
|
||
|
||
def parse_env_value(
|
||
raw_value: str,
|
||
variable_name: str,
|
||
line_number: int,
|
||
) -> str:
|
||
"""
|
||
Разобрать значение из env-файла.
|
||
|
||
Поддерживаются:
|
||
|
||
NAME=value
|
||
NAME="value with spaces"
|
||
NAME='value with spaces'
|
||
|
||
Символ # внутри значения сохраняется. Комментариями считаются
|
||
только строки, которые начинаются с #.
|
||
"""
|
||
|
||
value = raw_value.strip()
|
||
|
||
if not value:
|
||
return ""
|
||
|
||
if value[0] not in {
|
||
'"',
|
||
"'",
|
||
}:
|
||
return value
|
||
|
||
quote_character = value[0]
|
||
|
||
if (
|
||
len(value) < 2
|
||
or value[-1] != quote_character
|
||
):
|
||
raise YandexDownloadError(
|
||
f"{ENV_FILE}:{line_number}: "
|
||
f"у {variable_name} не закрыта кавычка."
|
||
)
|
||
|
||
value = value[1:-1]
|
||
|
||
if quote_character == "'":
|
||
return value
|
||
|
||
# В двойных кавычках поддерживаем наиболее полезные
|
||
# escape-последовательности env-файлов.
|
||
result: list[str] = []
|
||
index = 0
|
||
|
||
escape_sequences = {
|
||
"n": "\n",
|
||
"r": "\r",
|
||
"t": "\t",
|
||
"\\": "\\",
|
||
'"': '"',
|
||
}
|
||
|
||
while index < len(value):
|
||
character = value[index]
|
||
|
||
if character != "\\":
|
||
result.append(character)
|
||
index += 1
|
||
continue
|
||
|
||
if index + 1 >= len(value):
|
||
result.append("\\")
|
||
index += 1
|
||
continue
|
||
|
||
next_character = value[index + 1]
|
||
result.append(
|
||
escape_sequences.get(
|
||
next_character,
|
||
next_character,
|
||
)
|
||
)
|
||
index += 2
|
||
|
||
return "".join(result)
|
||
|
||
|
||
def load_env_file(
|
||
env_file: Path,
|
||
) -> dict[str, str]:
|
||
"""
|
||
Прочитать настройки загрузчика из отдельного env-файла.
|
||
|
||
Системные переменные окружения намеренно не подмешиваются:
|
||
конфигурация запуска полностью определяется одним файлом.
|
||
"""
|
||
|
||
try:
|
||
lines = env_file.read_text(
|
||
encoding="utf-8-sig"
|
||
).splitlines()
|
||
|
||
except FileNotFoundError as exc:
|
||
raise YandexDownloadError(
|
||
"Файл настроек не найден:\n"
|
||
f"{env_file}\n\n"
|
||
"Положи yandex_downloader.env "
|
||
"рядом со скриптом."
|
||
) from exc
|
||
|
||
except OSError as exc:
|
||
raise YandexDownloadError(
|
||
"Не удалось прочитать файл настроек "
|
||
f"{env_file}: {exc}"
|
||
) from exc
|
||
|
||
result: dict[str, str] = {}
|
||
|
||
for line_number, original_line in enumerate(
|
||
lines,
|
||
start=1,
|
||
):
|
||
line = original_line.strip()
|
||
|
||
if (
|
||
not line
|
||
or line.startswith("#")
|
||
):
|
||
continue
|
||
|
||
if line.startswith("export "):
|
||
line = line[7:].lstrip()
|
||
|
||
if "=" not in line:
|
||
raise YandexDownloadError(
|
||
f"{env_file}:{line_number}: "
|
||
"ожидалась строка ИМЯ=ЗНАЧЕНИЕ."
|
||
)
|
||
|
||
variable_name, raw_value = line.split(
|
||
"=",
|
||
1,
|
||
)
|
||
|
||
variable_name = variable_name.strip()
|
||
|
||
if not re.fullmatch(
|
||
r"[A-Za-z_][A-Za-z0-9_]*",
|
||
variable_name,
|
||
):
|
||
raise YandexDownloadError(
|
||
f"{env_file}:{line_number}: "
|
||
f"некорректное имя переменной "
|
||
f"{variable_name!r}."
|
||
)
|
||
|
||
result[variable_name] = parse_env_value(
|
||
raw_value,
|
||
variable_name,
|
||
line_number,
|
||
)
|
||
|
||
required_variables = (
|
||
"TARGET_URL",
|
||
"PUBLIC_LINK_PASSWORD",
|
||
"PASS_TOKEN",
|
||
"OUTPUT_DIR",
|
||
"PLAYWRIGHT_PROFILE_DIR",
|
||
"PLAYWRIGHT_HEADLESS",
|
||
"PLAYWRIGHT_TIMEOUT",
|
||
"FILE_DELAY_MIN",
|
||
"FILE_DELAY_MAX",
|
||
"PAGE_DELAY_MIN",
|
||
"PAGE_DELAY_MAX",
|
||
"COOLDOWN_EVERY_FILES",
|
||
"COOLDOWN_MIN",
|
||
"COOLDOWN_MAX",
|
||
"CONNECT_TIMEOUT",
|
||
"READ_TIMEOUT",
|
||
"HTTP_RETRIES",
|
||
)
|
||
|
||
missing_variables = [
|
||
variable_name
|
||
for variable_name in required_variables
|
||
if variable_name not in result
|
||
]
|
||
|
||
if missing_variables:
|
||
raise YandexDownloadError(
|
||
"В файле настроек отсутствуют переменные: "
|
||
+ ", ".join(missing_variables)
|
||
)
|
||
|
||
return result
|
||
|
||
|
||
def env_bool(
|
||
env_values: dict[str, str],
|
||
variable_name: str,
|
||
) -> bool:
|
||
"""Прочитать логическое значение из env-файла."""
|
||
|
||
value = env_values[variable_name].strip().lower()
|
||
|
||
if value in {
|
||
"1",
|
||
"true",
|
||
"yes",
|
||
"on",
|
||
}:
|
||
return True
|
||
|
||
if value in {
|
||
"0",
|
||
"false",
|
||
"no",
|
||
"off",
|
||
}:
|
||
return False
|
||
|
||
raise YandexDownloadError(
|
||
f"{variable_name} должен иметь значение "
|
||
"true или false."
|
||
)
|
||
|
||
|
||
def env_int(
|
||
env_values: dict[str, str],
|
||
variable_name: str,
|
||
) -> int:
|
||
"""Прочитать целое число из env-файла."""
|
||
|
||
value = env_values[variable_name].strip()
|
||
|
||
try:
|
||
return int(value)
|
||
|
||
except ValueError as exc:
|
||
raise YandexDownloadError(
|
||
f"{variable_name} должен быть целым числом, "
|
||
f"получено: {value!r}."
|
||
) from exc
|
||
|
||
|
||
def env_float(
|
||
env_values: dict[str, str],
|
||
variable_name: str,
|
||
) -> float:
|
||
"""Прочитать число из env-файла."""
|
||
|
||
value = env_values[variable_name].strip()
|
||
|
||
try:
|
||
return float(value)
|
||
|
||
except ValueError as exc:
|
||
raise YandexDownloadError(
|
||
f"{variable_name} должен быть числом, "
|
||
f"получено: {value!r}."
|
||
) from exc
|
||
|
||
|
||
def validate_settings(
|
||
*,
|
||
playwright_timeout: int,
|
||
file_delay_min: float,
|
||
file_delay_max: float,
|
||
page_delay_min: float,
|
||
page_delay_max: float,
|
||
cooldown_every_files: int,
|
||
cooldown_min: float,
|
||
cooldown_max: float,
|
||
connect_timeout: int,
|
||
read_timeout: int,
|
||
http_retries: int,
|
||
) -> None:
|
||
"""Проверить числовые настройки до начала загрузки."""
|
||
|
||
positive_values = {
|
||
"PLAYWRIGHT_TIMEOUT": playwright_timeout,
|
||
"CONNECT_TIMEOUT": connect_timeout,
|
||
"READ_TIMEOUT": read_timeout,
|
||
}
|
||
|
||
for variable_name, value in positive_values.items():
|
||
if value <= 0:
|
||
raise YandexDownloadError(
|
||
f"{variable_name} должен быть больше нуля."
|
||
)
|
||
|
||
non_negative_values = {
|
||
"FILE_DELAY_MIN": file_delay_min,
|
||
"FILE_DELAY_MAX": file_delay_max,
|
||
"PAGE_DELAY_MIN": page_delay_min,
|
||
"PAGE_DELAY_MAX": page_delay_max,
|
||
"COOLDOWN_MIN": cooldown_min,
|
||
"COOLDOWN_MAX": cooldown_max,
|
||
"COOLDOWN_EVERY_FILES": cooldown_every_files,
|
||
"HTTP_RETRIES": http_retries,
|
||
}
|
||
|
||
for variable_name, value in non_negative_values.items():
|
||
if value < 0:
|
||
raise YandexDownloadError(
|
||
f"{variable_name} не может быть отрицательным."
|
||
)
|
||
|
||
ranges = (
|
||
(
|
||
"FILE_DELAY_MIN",
|
||
file_delay_min,
|
||
"FILE_DELAY_MAX",
|
||
file_delay_max,
|
||
),
|
||
(
|
||
"PAGE_DELAY_MIN",
|
||
page_delay_min,
|
||
"PAGE_DELAY_MAX",
|
||
page_delay_max,
|
||
),
|
||
(
|
||
"COOLDOWN_MIN",
|
||
cooldown_min,
|
||
"COOLDOWN_MAX",
|
||
cooldown_max,
|
||
),
|
||
)
|
||
|
||
for (
|
||
minimum_name,
|
||
minimum_value,
|
||
maximum_name,
|
||
maximum_value,
|
||
) in ranges:
|
||
if minimum_value > maximum_value:
|
||
raise YandexDownloadError(
|
||
f"{minimum_name} не может быть больше "
|
||
f"{maximum_name}."
|
||
)
|
||
|
||
|
||
def get_pass_token_by_password(
|
||
target_url: str,
|
||
password: str,
|
||
profile_directory: Path,
|
||
headless: bool,
|
||
timeout_seconds: int,
|
||
) -> str:
|
||
"""
|
||
Открыть защищённую публичную ссылку в Chromium,
|
||
ввести пароль при необходимости и вернуть cookie passToken.
|
||
|
||
Постоянный профиль сохраняет cookies и состояние CAPTCHA
|
||
между запусками.
|
||
"""
|
||
|
||
try:
|
||
from playwright.sync_api import (
|
||
TimeoutError as PlaywrightTimeoutError,
|
||
)
|
||
from playwright.sync_api import sync_playwright
|
||
|
||
except ImportError as exc:
|
||
raise YandexDownloadError(
|
||
"Для автоматического получения passToken "
|
||
"нужен Playwright.\n"
|
||
"Установка:\n\n"
|
||
"python3 -m pip install --user playwright\n"
|
||
"python3 -m playwright install chromium"
|
||
) from exc
|
||
|
||
target_url = target_url.strip()
|
||
|
||
if not target_url:
|
||
raise YandexDownloadError(
|
||
"TARGET_URL не задан."
|
||
)
|
||
|
||
if timeout_seconds <= 0:
|
||
raise YandexDownloadError(
|
||
"PLAYWRIGHT_TIMEOUT должен быть больше нуля."
|
||
)
|
||
|
||
profile_directory = (
|
||
profile_directory
|
||
.expanduser()
|
||
.resolve()
|
||
)
|
||
|
||
profile_directory.mkdir(
|
||
parents=True,
|
||
exist_ok=True,
|
||
)
|
||
|
||
timeout_ms = timeout_seconds * 1000
|
||
|
||
with sync_playwright() as playwright:
|
||
try:
|
||
context = (
|
||
playwright.chromium
|
||
.launch_persistent_context(
|
||
user_data_dir=str(
|
||
profile_directory
|
||
),
|
||
headless=headless,
|
||
viewport={
|
||
"width": 1280,
|
||
"height": 900,
|
||
},
|
||
locale="ru-RU",
|
||
user_agent=USER_AGENT,
|
||
args=[
|
||
(
|
||
"--disable-blink-features="
|
||
"AutomationControlled"
|
||
),
|
||
],
|
||
)
|
||
)
|
||
|
||
except Exception as exc:
|
||
raise YandexDownloadError(
|
||
"Не удалось запустить Chromium через "
|
||
"Playwright. Проверь установку командой:\n\n"
|
||
"python3 -m playwright install chromium\n\n"
|
||
f"Подробности: {exc}"
|
||
) from exc
|
||
|
||
try:
|
||
context.set_default_timeout(
|
||
timeout_ms
|
||
)
|
||
|
||
if context.pages:
|
||
page = context.pages[0]
|
||
else:
|
||
page = context.new_page()
|
||
|
||
def find_pass_token() -> str | None:
|
||
cookies = context.cookies(
|
||
[
|
||
"https://disk.yandex.ru",
|
||
"https://yandex.ru",
|
||
]
|
||
)
|
||
|
||
for cookie in cookies:
|
||
if cookie.get("name") != "passToken":
|
||
continue
|
||
|
||
value = cookie.get("value")
|
||
|
||
if isinstance(value, str) and value:
|
||
return value
|
||
|
||
return None
|
||
|
||
print(
|
||
"[AUTH] Открываю защищённую ссылку "
|
||
"в Chromium..."
|
||
)
|
||
|
||
page.goto(
|
||
target_url,
|
||
wait_until="domcontentloaded",
|
||
timeout=timeout_ms,
|
||
)
|
||
|
||
try:
|
||
page.wait_for_load_state(
|
||
"networkidle",
|
||
timeout=min(
|
||
timeout_ms,
|
||
10_000,
|
||
),
|
||
)
|
||
|
||
except PlaywrightTimeoutError:
|
||
# Фоновые запросы Яндекс Диска могут не дать
|
||
# состоянию networkidle наступить. Форма при этом
|
||
# уже доступна, поэтому продолжаем.
|
||
pass
|
||
|
||
def first_visible(
|
||
locator: Any,
|
||
) -> Any | None:
|
||
"""Вернуть первый видимый элемент локатора."""
|
||
|
||
try:
|
||
count = locator.count()
|
||
|
||
except Exception:
|
||
return None
|
||
|
||
for index in range(count):
|
||
candidate = locator.nth(index)
|
||
|
||
try:
|
||
if candidate.is_visible(
|
||
timeout=250
|
||
):
|
||
return candidate
|
||
|
||
except Exception:
|
||
continue
|
||
|
||
return None
|
||
|
||
def find_password_input() -> tuple[
|
||
Any | None,
|
||
str | None,
|
||
]:
|
||
"""
|
||
Найти поле пароля при разных вариантах разметки.
|
||
|
||
Яндекс может использовать обычный type=text,
|
||
поэтому одного input[type=password] недостаточно.
|
||
"""
|
||
|
||
selectors = (
|
||
'input[type="password"]',
|
||
'input[autocomplete="current-password"]',
|
||
'input[autocomplete="new-password"]',
|
||
'input[name*="password" i]',
|
||
'input[id*="password" i]',
|
||
'input[placeholder*="парол" i]',
|
||
'input[aria-label*="парол" i]',
|
||
'form input:visible',
|
||
'main input:visible',
|
||
(
|
||
'input:visible:not([type="checkbox"])'
|
||
':not([type="radio"])'
|
||
':not([type="hidden"])'
|
||
':not([type="search"])'
|
||
),
|
||
(
|
||
'[contenteditable="true"]'
|
||
'[role="textbox"]:visible'
|
||
),
|
||
)
|
||
|
||
for selector in selectors:
|
||
candidate = first_visible(
|
||
page.locator(selector)
|
||
)
|
||
|
||
if candidate is not None:
|
||
return candidate, selector
|
||
|
||
return None, None
|
||
|
||
def password_wall_is_visible() -> bool:
|
||
"""Проверить, что страница всё ещё просит пароль."""
|
||
|
||
for text_value in (
|
||
"Введите пароль",
|
||
"Enter password",
|
||
):
|
||
heading = first_visible(
|
||
page.get_by_text(
|
||
text_value,
|
||
exact=True,
|
||
)
|
||
)
|
||
|
||
if heading is not None:
|
||
return True
|
||
|
||
return False
|
||
|
||
def find_open_button() -> Any | None:
|
||
"""Найти кнопку отправки формы пароля."""
|
||
|
||
for button_name in (
|
||
re.compile(
|
||
r"^\s*Открыть\s*$",
|
||
re.IGNORECASE,
|
||
),
|
||
re.compile(
|
||
r"^\s*Open\s*$",
|
||
re.IGNORECASE,
|
||
),
|
||
):
|
||
button = first_visible(
|
||
page.get_by_role(
|
||
"button",
|
||
name=button_name,
|
||
)
|
||
)
|
||
|
||
if button is not None:
|
||
return button
|
||
|
||
return first_visible(
|
||
page.locator(
|
||
'button[type="submit"]:visible'
|
||
)
|
||
)
|
||
|
||
password_input: Any | None = None
|
||
matched_selector: str | None = None
|
||
|
||
form_deadline = (
|
||
time.monotonic()
|
||
+ min(
|
||
timeout_seconds,
|
||
20,
|
||
)
|
||
)
|
||
|
||
while time.monotonic() < form_deadline:
|
||
(
|
||
password_input,
|
||
matched_selector,
|
||
) = find_password_input()
|
||
|
||
if password_input is not None:
|
||
break
|
||
|
||
existing_token = find_pass_token()
|
||
|
||
if (
|
||
existing_token
|
||
and not password_wall_is_visible()
|
||
):
|
||
print(
|
||
"[AUTH] Используется passToken "
|
||
"из сохранённого профиля."
|
||
)
|
||
|
||
return existing_token
|
||
|
||
time.sleep(0.25)
|
||
|
||
if password_input is not None:
|
||
if not password:
|
||
if not sys.stdin.isatty():
|
||
raise YandexDownloadError(
|
||
"Сохранённый passToken недействителен, "
|
||
"а PUBLIC_LINK_PASSWORD не задан "
|
||
"и интерактивный ввод недоступен."
|
||
)
|
||
|
||
password = getpass.getpass(
|
||
"Пароль публичной ссылки: "
|
||
)
|
||
|
||
if not password:
|
||
raise YandexDownloadError(
|
||
"Пароль публичной ссылки не задан."
|
||
)
|
||
|
||
print(
|
||
"[AUTH] Поле пароля найдено "
|
||
f"({matched_selector})."
|
||
)
|
||
|
||
print(
|
||
"[AUTH] Ввожу пароль публичной ссылки..."
|
||
)
|
||
|
||
try:
|
||
password_input.click()
|
||
password_input.fill(password)
|
||
|
||
try:
|
||
entered_password = (
|
||
password_input.input_value()
|
||
)
|
||
|
||
except Exception:
|
||
entered_password = (
|
||
password_input.inner_text()
|
||
)
|
||
|
||
if entered_password != password:
|
||
password_input.press(
|
||
"Control+A"
|
||
)
|
||
password_input.type(
|
||
password,
|
||
delay=25,
|
||
)
|
||
|
||
try:
|
||
entered_password = (
|
||
password_input.input_value()
|
||
)
|
||
|
||
except Exception:
|
||
entered_password = (
|
||
password_input.inner_text()
|
||
)
|
||
|
||
if entered_password != password:
|
||
raise YandexDownloadError(
|
||
"Поле пароля найдено, но Chromium "
|
||
"не смог записать в него пароль."
|
||
)
|
||
|
||
except YandexDownloadError:
|
||
raise
|
||
|
||
except Exception as exc:
|
||
raise YandexDownloadError(
|
||
"Не удалось заполнить поле пароля "
|
||
f"на странице Яндекса: {exc}"
|
||
) from exc
|
||
|
||
print(
|
||
"[AUTH] Пароль введён; отправляю форму..."
|
||
)
|
||
|
||
open_button = find_open_button()
|
||
|
||
if open_button is not None:
|
||
button_deadline = (
|
||
time.monotonic()
|
||
+ 5
|
||
)
|
||
|
||
while (
|
||
time.monotonic()
|
||
< button_deadline
|
||
):
|
||
try:
|
||
if open_button.is_enabled():
|
||
break
|
||
|
||
except Exception:
|
||
break
|
||
|
||
time.sleep(0.1)
|
||
|
||
try:
|
||
open_button.click(
|
||
timeout=5000
|
||
)
|
||
|
||
except Exception:
|
||
password_input.press(
|
||
"Enter"
|
||
)
|
||
|
||
else:
|
||
password_input.press(
|
||
"Enter"
|
||
)
|
||
|
||
elif password_wall_is_visible():
|
||
raise YandexDownloadError(
|
||
"Страница просит пароль, но поле ввода "
|
||
"не найдено. Яндекс снова изменил "
|
||
"разметку формы."
|
||
)
|
||
|
||
deadline = (
|
||
time.monotonic()
|
||
+ timeout_seconds
|
||
)
|
||
|
||
while time.monotonic() < deadline:
|
||
current_url = page.url
|
||
lowered_url = current_url.lower()
|
||
|
||
if (
|
||
"/showcaptcha" in lowered_url
|
||
or "captcha" in lowered_url
|
||
):
|
||
if headless:
|
||
raise CaptchaRequired(
|
||
current_url
|
||
)
|
||
|
||
print(
|
||
"[AUTH] Пройди CAPTCHA "
|
||
"в открытом окне Chromium."
|
||
)
|
||
|
||
try:
|
||
page_text = page.locator(
|
||
"body"
|
||
).inner_text(
|
||
timeout=2000
|
||
)
|
||
|
||
except Exception:
|
||
page_text = ""
|
||
|
||
lowered_text = page_text.lower()
|
||
|
||
if (
|
||
"неверный пароль" in lowered_text
|
||
or "пароль указан неверно" in lowered_text
|
||
or "wrong password" in lowered_text
|
||
):
|
||
raise YandexDownloadError(
|
||
"Яндекс отклонил пароль "
|
||
"публичной ссылки."
|
||
)
|
||
|
||
token = find_pass_token()
|
||
|
||
form_still_visible = (
|
||
password_wall_is_visible()
|
||
)
|
||
|
||
if token and not form_still_visible:
|
||
print(
|
||
"[AUTH] passToken получен."
|
||
)
|
||
|
||
return token
|
||
|
||
time.sleep(1)
|
||
|
||
screenshot_path = (
|
||
profile_directory
|
||
/ "pass-token-error.png"
|
||
)
|
||
|
||
try:
|
||
page.screenshot(
|
||
path=str(screenshot_path),
|
||
full_page=True,
|
||
)
|
||
|
||
except Exception:
|
||
pass
|
||
|
||
raise YandexDownloadError(
|
||
"Не удалось получить passToken "
|
||
f"за {timeout_seconds} секунд.\n"
|
||
"Возможные причины: неверный пароль, "
|
||
"CAPTCHA или изменение страницы Яндекса.\n"
|
||
"Снимок страницы: "
|
||
f"{screenshot_path}"
|
||
)
|
||
|
||
finally:
|
||
context.close()
|
||
|
||
|
||
def deep_find_first(
|
||
obj: Any,
|
||
key: str,
|
||
) -> Any | None:
|
||
"""Рекурсивно найти первое значение указанного ключа."""
|
||
|
||
if isinstance(obj, dict):
|
||
if key in obj:
|
||
return obj[key]
|
||
|
||
for value in obj.values():
|
||
found = deep_find_first(
|
||
value,
|
||
key,
|
||
)
|
||
|
||
if found is not None:
|
||
return found
|
||
|
||
elif isinstance(obj, list):
|
||
for value in obj:
|
||
found = deep_find_first(
|
||
value,
|
||
key,
|
||
)
|
||
|
||
if found is not None:
|
||
return found
|
||
|
||
return None
|
||
|
||
|
||
def safe_component(name: str) -> str:
|
||
"""Преобразовать имя в безопасный компонент пути."""
|
||
|
||
name = name.replace("/", "_")
|
||
name = name.replace("\x00", "_")
|
||
|
||
if name in {
|
||
"",
|
||
".",
|
||
"..",
|
||
}:
|
||
return "_"
|
||
|
||
return name
|
||
|
||
|
||
def human_size(
|
||
value: int | None,
|
||
) -> str:
|
||
"""Преобразовать размер файла в читаемый формат."""
|
||
|
||
if value is None or value < 0:
|
||
return "размер неизвестен"
|
||
|
||
units = (
|
||
"B",
|
||
"KiB",
|
||
"MiB",
|
||
"GiB",
|
||
"TiB",
|
||
)
|
||
|
||
number = float(value)
|
||
|
||
for unit in units:
|
||
if number < 1024 or unit == units[-1]:
|
||
return f"{number:.2f} {unit}"
|
||
|
||
number /= 1024
|
||
|
||
return f"{value} B"
|
||
|
||
|
||
def resource_identifier(
|
||
item: dict[str, Any],
|
||
) -> str | None:
|
||
"""
|
||
Получить идентификатор файла или каталога.
|
||
|
||
Яндекс может использовать:
|
||
- path;
|
||
- hash;
|
||
- id.
|
||
"""
|
||
|
||
for key in (
|
||
"path",
|
||
"hash",
|
||
"id",
|
||
):
|
||
value = item.get(key)
|
||
|
||
if isinstance(value, str) and value:
|
||
return value
|
||
|
||
return None
|
||
|
||
|
||
def make_page_signature(
|
||
items: Iterable[dict[str, Any]],
|
||
) -> tuple[str, ...]:
|
||
"""
|
||
Создать подпись страницы.
|
||
|
||
Нужна для обнаружения ситуации, когда API повторно
|
||
возвращает одну и ту же страницу.
|
||
"""
|
||
|
||
result: list[str] = []
|
||
|
||
for item in items:
|
||
identifier = resource_identifier(item)
|
||
|
||
if identifier is None:
|
||
identifier = str(
|
||
item.get("name") or ""
|
||
)
|
||
|
||
result.append(identifier)
|
||
|
||
return tuple(result)
|
||
|
||
|
||
class YandexProtectedFolderDownloader:
|
||
def __init__(
|
||
self,
|
||
target_url: str,
|
||
pass_token: str,
|
||
output_dir: Path,
|
||
*,
|
||
file_delay_min: float,
|
||
file_delay_max: float,
|
||
page_delay_min: float,
|
||
page_delay_max: float,
|
||
cooldown_every_files: int,
|
||
cooldown_min: float,
|
||
cooldown_max: float,
|
||
connect_timeout: int,
|
||
read_timeout: int,
|
||
http_retries: int,
|
||
) -> None:
|
||
parsed_url = urlparse(target_url)
|
||
|
||
if parsed_url.scheme != "https":
|
||
raise ValueError(
|
||
"TARGET_URL должен начинаться с https://"
|
||
)
|
||
|
||
if parsed_url.hostname != "disk.yandex.ru":
|
||
raise ValueError(
|
||
"Ожидалась ссылка вида "
|
||
"https://disk.yandex.ru/..."
|
||
)
|
||
|
||
self.target_url = target_url
|
||
|
||
self.base_url = (
|
||
f"{parsed_url.scheme}://"
|
||
f"{parsed_url.netloc}"
|
||
)
|
||
|
||
self.pass_token = (
|
||
self.normalize_pass_token(
|
||
pass_token
|
||
)
|
||
)
|
||
|
||
self.output_dir = output_dir
|
||
self.file_delay_min = file_delay_min
|
||
self.file_delay_max = file_delay_max
|
||
self.page_delay_min = page_delay_min
|
||
self.page_delay_max = page_delay_max
|
||
self.cooldown_every_files = (
|
||
cooldown_every_files
|
||
)
|
||
self.cooldown_min = cooldown_min
|
||
self.cooldown_max = cooldown_max
|
||
self.connect_timeout = connect_timeout
|
||
self.read_timeout = read_timeout
|
||
self.http_retries = http_retries
|
||
|
||
self.session = requests.Session()
|
||
|
||
self.session.headers.update(
|
||
{
|
||
"User-Agent": USER_AGENT,
|
||
"Accept-Language": (
|
||
"ru-RU,ru;q=0.9,"
|
||
"en-US;q=0.7,en;q=0.6"
|
||
),
|
||
}
|
||
)
|
||
|
||
self.session.cookies.set(
|
||
"passToken",
|
||
self.pass_token,
|
||
domain=".yandex.ru",
|
||
path="/",
|
||
)
|
||
|
||
retries = Retry(
|
||
total=self.http_retries,
|
||
connect=self.http_retries,
|
||
read=self.http_retries,
|
||
status=self.http_retries,
|
||
backoff_factor=1.0,
|
||
status_forcelist=(
|
||
429,
|
||
500,
|
||
502,
|
||
503,
|
||
504,
|
||
),
|
||
allowed_methods=frozenset(
|
||
{
|
||
"GET",
|
||
"POST",
|
||
}
|
||
),
|
||
respect_retry_after_header=True,
|
||
)
|
||
|
||
adapter = HTTPAdapter(
|
||
max_retries=retries
|
||
)
|
||
|
||
self.session.mount(
|
||
"https://",
|
||
adapter,
|
||
)
|
||
|
||
self.sk = ""
|
||
self.root_hash = ""
|
||
|
||
self.visited_directories: set[str] = set()
|
||
self.seen_files: set[str] = set()
|
||
|
||
self.file_count = 0
|
||
self.downloaded_count = 0
|
||
self.skipped_count = 0
|
||
self.failed_count = 0
|
||
|
||
self.total_known_bytes = 0
|
||
self.download_url_request_count = 0
|
||
|
||
@staticmethod
|
||
def normalize_pass_token(
|
||
pass_token: str,
|
||
) -> str:
|
||
"""
|
||
Нормализовать passToken.
|
||
|
||
Допустимые варианты:
|
||
|
||
abcdef123
|
||
|
||
passToken=abcdef123
|
||
|
||
passToken=abcdef123; Path=/;
|
||
"""
|
||
|
||
value = pass_token.strip()
|
||
|
||
if value.startswith("passToken="):
|
||
value = value.split(
|
||
"=",
|
||
1,
|
||
)[1]
|
||
|
||
if ";" in value:
|
||
value = value.split(
|
||
";",
|
||
1,
|
||
)[0]
|
||
|
||
return value.strip()
|
||
|
||
def api_headers(
|
||
self,
|
||
referer: str,
|
||
) -> dict[str, str]:
|
||
"""Получить заголовки для внутренних API-запросов."""
|
||
|
||
return {
|
||
"Accept": (
|
||
"application/json, "
|
||
"text/plain, */*"
|
||
),
|
||
"Origin": self.base_url,
|
||
"Referer": referer,
|
||
"X-Requested-With": "XMLHttpRequest",
|
||
"X-Retpath-Y": referer,
|
||
"Content-Type": "text/plain",
|
||
}
|
||
|
||
@staticmethod
|
||
def raise_if_captcha(
|
||
result: dict[str, Any],
|
||
) -> None:
|
||
"""
|
||
Немедленно остановить процесс,
|
||
если API вернул CAPTCHA.
|
||
"""
|
||
|
||
if result.get("type") != "captcha":
|
||
return
|
||
|
||
captcha_data = result.get("captcha")
|
||
|
||
if not isinstance(
|
||
captcha_data,
|
||
dict,
|
||
):
|
||
captcha_data = {}
|
||
|
||
captcha_url = captcha_data.get(
|
||
"captcha-page"
|
||
)
|
||
|
||
if (
|
||
not isinstance(captcha_url, str)
|
||
or not captcha_url
|
||
):
|
||
captcha_url = (
|
||
"https://disk.yandex.ru/showcaptcha"
|
||
)
|
||
|
||
raise CaptchaRequired(
|
||
captcha_url=captcha_url,
|
||
response_data=result,
|
||
)
|
||
|
||
def load_initial_state(self) -> None:
|
||
"""
|
||
Открыть исходную ссылку и получить:
|
||
- sk;
|
||
- hash корневой папки.
|
||
"""
|
||
|
||
response = self.session.get(
|
||
self.target_url,
|
||
timeout=(
|
||
self.connect_timeout,
|
||
self.read_timeout,
|
||
),
|
||
)
|
||
|
||
response.raise_for_status()
|
||
|
||
body = response.text
|
||
|
||
if "/showcaptcha" in response.url:
|
||
raise CaptchaRequired(
|
||
response.url
|
||
)
|
||
|
||
if (
|
||
"SmartCaptcha" in body
|
||
or "smart-captcha" in body
|
||
):
|
||
raise CaptchaRequired(
|
||
"https://disk.yandex.ru/showcaptcha"
|
||
)
|
||
|
||
patterns = (
|
||
(
|
||
r'<script\s+type="application/json"'
|
||
r'[^>]*id="store-prefetch"[^>]*>'
|
||
r"(.*?)"
|
||
r"</script>"
|
||
),
|
||
(
|
||
r'<script[^>]*id="store-prefetch"'
|
||
r'[^>]*type="application/json"[^>]*>'
|
||
r"(.*?)"
|
||
r"</script>"
|
||
),
|
||
)
|
||
|
||
raw_state: str | None = None
|
||
|
||
for pattern in patterns:
|
||
match = re.search(
|
||
pattern,
|
||
body,
|
||
flags=(
|
||
re.DOTALL
|
||
| re.IGNORECASE
|
||
),
|
||
)
|
||
|
||
if match:
|
||
raw_state = match.group(1)
|
||
break
|
||
|
||
if raw_state is None:
|
||
if (
|
||
"password-protected" in body
|
||
or "Введите пароль" in body
|
||
or "enter-password" in body
|
||
):
|
||
raise YandexDownloadError(
|
||
"passToken не принят или истёк: "
|
||
"страница снова требует пароль."
|
||
)
|
||
|
||
raise YandexDownloadError(
|
||
"На странице не найден "
|
||
"JSON store-prefetch."
|
||
)
|
||
|
||
try:
|
||
state = json.loads(
|
||
html.unescape(raw_state)
|
||
)
|
||
|
||
except json.JSONDecodeError as exc:
|
||
raise YandexDownloadError(
|
||
"Не удалось разобрать "
|
||
f"store-prefetch: {exc}"
|
||
) from exc
|
||
|
||
sk = state.get("sk")
|
||
|
||
if not isinstance(sk, str) or not sk:
|
||
sk = deep_find_first(
|
||
state,
|
||
"sk",
|
||
)
|
||
|
||
if not isinstance(sk, str) or not sk:
|
||
match_sk = re.search(
|
||
r'"sk"\s*:\s*"([^"]+)"',
|
||
body,
|
||
)
|
||
|
||
if match_sk:
|
||
sk = match_sk.group(1)
|
||
|
||
if not isinstance(sk, str) or not sk:
|
||
raise YandexDownloadError(
|
||
"Не удалось получить параметр sk."
|
||
)
|
||
|
||
resources = state.get("resources")
|
||
|
||
if not isinstance(resources, dict):
|
||
resources = deep_find_first(
|
||
state,
|
||
"resources",
|
||
)
|
||
|
||
if not isinstance(resources, dict):
|
||
raise YandexDownloadError(
|
||
"В store-prefetch отсутствует "
|
||
"словарь resources."
|
||
)
|
||
|
||
if "password-protected" in resources:
|
||
raise YandexDownloadError(
|
||
"passToken не открыл папку. "
|
||
"Токен истёк либо относится "
|
||
"к другой ссылке."
|
||
)
|
||
|
||
root_hash = (
|
||
self.find_current_directory_hash(
|
||
state,
|
||
resources,
|
||
)
|
||
)
|
||
|
||
self.sk = sk
|
||
|
||
self.root_hash = (
|
||
self.normalize_directory_hash(
|
||
root_hash
|
||
)
|
||
)
|
||
|
||
@staticmethod
|
||
def find_current_directory_hash(
|
||
state: dict[str, Any],
|
||
resources: dict[str, Any],
|
||
) -> str:
|
||
"""Определить hash текущей открытой папки."""
|
||
|
||
current_resource_id = state.get(
|
||
"currentResourceId"
|
||
)
|
||
|
||
if current_resource_id is None:
|
||
current_resource_id = deep_find_first(
|
||
state,
|
||
"currentResourceId",
|
||
)
|
||
|
||
if current_resource_id is not None:
|
||
current_id = str(
|
||
current_resource_id
|
||
)
|
||
|
||
direct_resource = resources.get(
|
||
current_id
|
||
)
|
||
|
||
if isinstance(
|
||
direct_resource,
|
||
dict,
|
||
):
|
||
identifier = resource_identifier(
|
||
direct_resource
|
||
)
|
||
|
||
if identifier:
|
||
return identifier
|
||
|
||
for resource in resources.values():
|
||
if not isinstance(
|
||
resource,
|
||
dict,
|
||
):
|
||
continue
|
||
|
||
if str(
|
||
resource.get("id")
|
||
) != current_id:
|
||
continue
|
||
|
||
if resource.get("type") != "dir":
|
||
continue
|
||
|
||
identifier = resource_identifier(
|
||
resource
|
||
)
|
||
|
||
if identifier:
|
||
return identifier
|
||
|
||
current_resource = state.get(
|
||
"currentResource"
|
||
)
|
||
|
||
if not isinstance(
|
||
current_resource,
|
||
dict,
|
||
):
|
||
current_resource = deep_find_first(
|
||
state,
|
||
"currentResource",
|
||
)
|
||
|
||
if isinstance(
|
||
current_resource,
|
||
dict,
|
||
):
|
||
identifier = resource_identifier(
|
||
current_resource
|
||
)
|
||
|
||
if identifier:
|
||
return identifier
|
||
|
||
directory_candidates: list[
|
||
dict[str, Any]
|
||
] = []
|
||
|
||
for resource in resources.values():
|
||
if not isinstance(
|
||
resource,
|
||
dict,
|
||
):
|
||
continue
|
||
|
||
if resource.get("type") == "dir":
|
||
directory_candidates.append(
|
||
resource
|
||
)
|
||
|
||
for resource in directory_candidates:
|
||
if resource.get("parent") not in {
|
||
None,
|
||
"",
|
||
}:
|
||
continue
|
||
|
||
identifier = resource_identifier(
|
||
resource
|
||
)
|
||
|
||
if identifier:
|
||
return identifier
|
||
|
||
for resource in directory_candidates:
|
||
identifier = resource_identifier(
|
||
resource
|
||
)
|
||
|
||
if identifier:
|
||
return identifier
|
||
|
||
raise YandexDownloadError(
|
||
"Не удалось определить hash "
|
||
"текущей папки."
|
||
)
|
||
|
||
@staticmethod
|
||
def normalize_directory_hash(
|
||
directory_hash: str,
|
||
) -> str:
|
||
"""
|
||
Привести идентификатор папки к формату API.
|
||
|
||
Корневая папка:
|
||
|
||
PUBLIC_HASH:
|
||
|
||
Вложенная папка:
|
||
|
||
PUBLIC_HASH:/folder/subfolder
|
||
"""
|
||
|
||
value = directory_hash.strip()
|
||
|
||
if ":" not in value:
|
||
value += ":"
|
||
|
||
return value
|
||
|
||
def fetch_list(
|
||
self,
|
||
directory_hash: str,
|
||
offset: int,
|
||
) -> tuple[list[dict[str, Any]], bool]:
|
||
"""Получить очередную страницу содержимого папки."""
|
||
|
||
payload = {
|
||
"hash": (
|
||
self.normalize_directory_hash(
|
||
directory_hash
|
||
)
|
||
),
|
||
"offset": offset,
|
||
"withSizes": True,
|
||
"sk": self.sk,
|
||
"options": {
|
||
"hasExperimentVideoWithoutPreview": True,
|
||
},
|
||
}
|
||
|
||
raw_json = json.dumps(
|
||
payload,
|
||
ensure_ascii=False,
|
||
separators=(
|
||
",",
|
||
":",
|
||
),
|
||
)
|
||
|
||
encoded_payload = quote(
|
||
raw_json,
|
||
safe="",
|
||
).encode("utf-8")
|
||
|
||
endpoint = (
|
||
f"{self.base_url}"
|
||
"/public/api/fetch-list"
|
||
)
|
||
|
||
response = self.session.post(
|
||
endpoint,
|
||
data=encoded_payload,
|
||
headers=self.api_headers(
|
||
self.target_url
|
||
),
|
||
timeout=(
|
||
self.connect_timeout,
|
||
self.read_timeout,
|
||
),
|
||
)
|
||
|
||
response.raise_for_status()
|
||
|
||
try:
|
||
result = response.json()
|
||
|
||
except ValueError as exc:
|
||
preview = response.text[:500]
|
||
|
||
raise YandexDownloadError(
|
||
"fetch-list вернул не JSON. "
|
||
f"Ответ: {preview!r}"
|
||
) from exc
|
||
|
||
if not isinstance(result, dict):
|
||
raise YandexDownloadError(
|
||
"fetch-list вернул JSON, "
|
||
"но корневой объект "
|
||
"не является словарём."
|
||
)
|
||
|
||
self.raise_if_captcha(result)
|
||
|
||
data = result.get("data")
|
||
|
||
if not isinstance(data, dict):
|
||
data = {}
|
||
|
||
if result.get("error") is True:
|
||
code = (
|
||
result.get("code")
|
||
or result.get("statusCode")
|
||
or data.get("code")
|
||
)
|
||
|
||
raise YandexDownloadError(
|
||
"Ошибка fetch-list: "
|
||
f"code={code}, ответ={result}"
|
||
)
|
||
|
||
items = result.get("resources")
|
||
|
||
if not isinstance(items, list):
|
||
items = data.get("resources")
|
||
|
||
if not isinstance(items, list):
|
||
items = result.get("items")
|
||
|
||
if not isinstance(items, list):
|
||
items = data.get("items")
|
||
|
||
if not isinstance(items, list):
|
||
raise YandexDownloadError(
|
||
"fetch-list не вернул список "
|
||
f"resources/items: {result}"
|
||
)
|
||
|
||
filtered_items = [
|
||
item
|
||
for item in items
|
||
if isinstance(item, dict)
|
||
]
|
||
|
||
completed_value = result.get(
|
||
"completed"
|
||
)
|
||
|
||
if completed_value is None:
|
||
completed_value = data.get(
|
||
"completed"
|
||
)
|
||
|
||
return (
|
||
filtered_items,
|
||
bool(completed_value),
|
||
)
|
||
|
||
def get_download_url(
|
||
self,
|
||
resource_hash: str,
|
||
) -> str:
|
||
"""Получить временную прямую ссылку на файл."""
|
||
|
||
payload = {
|
||
"hash": resource_hash,
|
||
"sk": self.sk,
|
||
"passToken": self.pass_token,
|
||
}
|
||
|
||
endpoint = (
|
||
f"{self.base_url}"
|
||
"/public/api/download-url"
|
||
)
|
||
|
||
response = self.session.post(
|
||
endpoint,
|
||
data=json.dumps(
|
||
payload,
|
||
ensure_ascii=False,
|
||
separators=(
|
||
",",
|
||
":",
|
||
),
|
||
).encode("utf-8"),
|
||
headers=self.api_headers(
|
||
self.target_url
|
||
),
|
||
timeout=(
|
||
self.connect_timeout,
|
||
self.read_timeout,
|
||
),
|
||
)
|
||
|
||
response.raise_for_status()
|
||
|
||
try:
|
||
result = response.json()
|
||
|
||
except ValueError as exc:
|
||
preview = response.text[:500]
|
||
|
||
raise YandexDownloadError(
|
||
"download-url вернул не JSON. "
|
||
f"Ответ: {preview!r}"
|
||
) from exc
|
||
|
||
if not isinstance(result, dict):
|
||
raise YandexDownloadError(
|
||
"download-url вернул JSON, "
|
||
"но корневой объект "
|
||
"не является словарём."
|
||
)
|
||
|
||
self.raise_if_captcha(result)
|
||
|
||
data = result.get("data")
|
||
|
||
if not isinstance(data, dict):
|
||
data = {}
|
||
|
||
error = result.get("error")
|
||
|
||
if error not in {
|
||
False,
|
||
None,
|
||
}:
|
||
code = (
|
||
data.get("code")
|
||
or result.get("code")
|
||
or result.get("statusCode")
|
||
)
|
||
|
||
if code == 309:
|
||
raise YandexDownloadError(
|
||
"passToken истёк или "
|
||
"не подходит к этой папке."
|
||
)
|
||
|
||
raise YandexDownloadError(
|
||
"Ошибка download-url: "
|
||
f"code={code}, ответ={result}"
|
||
)
|
||
|
||
if data.get("read_only") is True:
|
||
raise YandexDownloadError(
|
||
"Владелец публичной ссылки "
|
||
"запретил скачивание."
|
||
)
|
||
|
||
direct_url = data.get("url")
|
||
|
||
if (
|
||
not isinstance(direct_url, str)
|
||
or not direct_url
|
||
):
|
||
direct_url = result.get("url")
|
||
|
||
if (
|
||
not isinstance(direct_url, str)
|
||
or not direct_url
|
||
):
|
||
raise YandexDownloadError(
|
||
"В ответе download-url "
|
||
f"отсутствует URL: {result}"
|
||
)
|
||
|
||
return direct_url
|
||
|
||
@staticmethod
|
||
def get_file_size(
|
||
item: dict[str, Any],
|
||
) -> int | None:
|
||
"""Получить ожидаемый размер файла."""
|
||
|
||
meta = item.get("meta")
|
||
|
||
if isinstance(meta, dict):
|
||
size = meta.get("size")
|
||
|
||
if isinstance(size, int):
|
||
return size
|
||
|
||
size = item.get("size")
|
||
|
||
if isinstance(size, int):
|
||
return size
|
||
|
||
return None
|
||
|
||
def wait_before_download_url(self) -> None:
|
||
"""Ограничить частоту запросов download-url."""
|
||
|
||
delay = random.uniform(
|
||
self.file_delay_min,
|
||
self.file_delay_max,
|
||
)
|
||
|
||
print(
|
||
f"[WAIT] {delay:.1f} с "
|
||
"перед запросом ссылки"
|
||
)
|
||
|
||
time.sleep(delay)
|
||
|
||
self.download_url_request_count += 1
|
||
|
||
if (
|
||
self.cooldown_every_files > 0
|
||
and self.download_url_request_count
|
||
% self.cooldown_every_files == 0
|
||
):
|
||
cooldown = random.uniform(
|
||
self.cooldown_min,
|
||
self.cooldown_max,
|
||
)
|
||
|
||
print(
|
||
"[WAIT] Длительный перерыв после "
|
||
f"{self.download_url_request_count} "
|
||
"запросов: "
|
||
f"{cooldown:.1f} с"
|
||
)
|
||
|
||
time.sleep(cooldown)
|
||
|
||
def download_file(
|
||
self,
|
||
item: dict[str, Any],
|
||
relative_directory: Path,
|
||
) -> None:
|
||
"""Скачать один файл через wget."""
|
||
|
||
name = safe_component(
|
||
str(
|
||
item.get("name")
|
||
or "unnamed-file"
|
||
)
|
||
)
|
||
|
||
resource_hash = (
|
||
item.get("path")
|
||
or item.get("hash")
|
||
)
|
||
|
||
if (
|
||
not isinstance(resource_hash, str)
|
||
or not resource_hash
|
||
):
|
||
raise YandexDownloadError(
|
||
f"У файла {name!r} "
|
||
"отсутствует path/hash."
|
||
)
|
||
|
||
if resource_hash in self.seen_files:
|
||
return
|
||
|
||
self.seen_files.add(resource_hash)
|
||
|
||
self.file_count += 1
|
||
|
||
expected_size = self.get_file_size(
|
||
item
|
||
)
|
||
|
||
if expected_size is not None:
|
||
self.total_known_bytes += (
|
||
expected_size
|
||
)
|
||
|
||
destination_directory = (
|
||
self.output_dir
|
||
/ relative_directory
|
||
)
|
||
|
||
destination_directory.mkdir(
|
||
parents=True,
|
||
exist_ok=True,
|
||
)
|
||
|
||
destination = (
|
||
destination_directory
|
||
/ name
|
||
)
|
||
|
||
if destination.exists():
|
||
if not destination.is_file():
|
||
raise YandexDownloadError(
|
||
"Локальный путь существует, "
|
||
"но не является файлом: "
|
||
f"{destination}"
|
||
)
|
||
|
||
current_size = (
|
||
destination.stat().st_size
|
||
)
|
||
|
||
if (
|
||
expected_size is not None
|
||
and current_size
|
||
== expected_size
|
||
):
|
||
self.skipped_count += 1
|
||
|
||
print(
|
||
f"[SKIP] {destination} "
|
||
f"({human_size(expected_size)})"
|
||
)
|
||
|
||
return
|
||
|
||
if (
|
||
expected_size is not None
|
||
and current_size
|
||
> expected_size
|
||
):
|
||
raise YandexDownloadError(
|
||
"Локальный файл больше "
|
||
"ожидаемого:\n"
|
||
f" файл: {destination}\n"
|
||
f" локально: {current_size}\n"
|
||
f" ожидается: {expected_size}"
|
||
)
|
||
|
||
print(
|
||
f"[GET ] {destination} "
|
||
f"({human_size(expected_size)})"
|
||
)
|
||
|
||
self.wait_before_download_url()
|
||
|
||
direct_url = self.get_download_url(
|
||
resource_hash
|
||
)
|
||
|
||
command = [
|
||
"wget",
|
||
"--inet4-only",
|
||
"--continue",
|
||
"--tries=20",
|
||
"--timeout=60",
|
||
"--read-timeout=120",
|
||
"--waitretry=5",
|
||
"--retry-connrefused",
|
||
"--no-verbose",
|
||
"--output-document",
|
||
str(destination),
|
||
direct_url,
|
||
]
|
||
|
||
completed_process = subprocess.run(
|
||
command,
|
||
check=False,
|
||
)
|
||
|
||
if completed_process.returncode != 0:
|
||
raise YandexDownloadError(
|
||
"wget завершился с кодом "
|
||
f"{completed_process.returncode} "
|
||
f"для файла {destination}"
|
||
)
|
||
|
||
if not destination.is_file():
|
||
raise YandexDownloadError(
|
||
"После wget файл не найден: "
|
||
f"{destination}"
|
||
)
|
||
|
||
actual_size = (
|
||
destination.stat().st_size
|
||
)
|
||
|
||
if (
|
||
expected_size is not None
|
||
and actual_size != expected_size
|
||
):
|
||
raise YandexDownloadError(
|
||
"Размер файла после скачивания "
|
||
"не совпадает:\n"
|
||
f" файл: {destination}\n"
|
||
f" получено: {actual_size}\n"
|
||
f" ожидалось: {expected_size}"
|
||
)
|
||
|
||
self.downloaded_count += 1
|
||
|
||
def process_page(
|
||
self,
|
||
items: Iterable[dict[str, Any]],
|
||
relative_directory: Path,
|
||
) -> list[tuple[str, str]]:
|
||
"""
|
||
Обработать страницу.
|
||
|
||
Возвращает список вложенных каталогов:
|
||
|
||
[
|
||
(hash_каталога, локальное_имя),
|
||
...
|
||
]
|
||
"""
|
||
|
||
subdirectories: list[
|
||
tuple[str, str]
|
||
] = []
|
||
|
||
for item in items:
|
||
item_type = item.get("type")
|
||
|
||
name = safe_component(
|
||
str(
|
||
item.get("name")
|
||
or "unnamed"
|
||
)
|
||
)
|
||
|
||
if item_type == "dir":
|
||
identifier = resource_identifier(
|
||
item
|
||
)
|
||
|
||
if identifier is None:
|
||
print(
|
||
"[WARN] У каталога "
|
||
f"{name!r} отсутствует "
|
||
"path/hash/id.",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
continue
|
||
|
||
subdirectories.append(
|
||
(
|
||
identifier,
|
||
name,
|
||
)
|
||
)
|
||
|
||
continue
|
||
|
||
if item_type == "file":
|
||
try:
|
||
self.download_file(
|
||
item,
|
||
relative_directory,
|
||
)
|
||
|
||
except CaptchaRequired:
|
||
# CAPTCHA должна остановить
|
||
# весь процесс, а не один файл.
|
||
raise
|
||
|
||
except Exception as exc:
|
||
local_path = (
|
||
self.output_dir
|
||
/ relative_directory
|
||
/ name
|
||
)
|
||
|
||
print(
|
||
f"[ERR ] {local_path}: "
|
||
f"{exc}",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
self.failed_count += 1
|
||
|
||
return subdirectories
|
||
|
||
def split_fetch_list_items(
|
||
self,
|
||
items: list[dict[str, Any]],
|
||
directory_hash: str,
|
||
) -> tuple[
|
||
list[dict[str, Any]],
|
||
list[dict[str, Any]],
|
||
]:
|
||
"""
|
||
Отделить реальные дочерние файлы и каталоги
|
||
от служебного элемента текущей папки.
|
||
|
||
Внутренний fetch-list Яндекс Диска обычно возвращает:
|
||
|
||
40 дочерних элементов
|
||
+
|
||
1 описание текущего каталога
|
||
|
||
Служебный элемент нельзя учитывать при увеличении offset.
|
||
"""
|
||
|
||
normalized_directory_hash = (
|
||
self.normalize_directory_hash(
|
||
directory_hash
|
||
)
|
||
)
|
||
|
||
child_items: list[
|
||
dict[str, Any]
|
||
] = []
|
||
|
||
service_items: list[
|
||
dict[str, Any]
|
||
] = []
|
||
|
||
for item in items:
|
||
identifier = resource_identifier(
|
||
item
|
||
)
|
||
|
||
is_current_directory = False
|
||
|
||
if (
|
||
item.get("type") == "dir"
|
||
and isinstance(identifier, str)
|
||
and identifier
|
||
):
|
||
normalized_identifier = (
|
||
self.normalize_directory_hash(
|
||
identifier
|
||
)
|
||
)
|
||
|
||
if (
|
||
normalized_identifier
|
||
== normalized_directory_hash
|
||
):
|
||
is_current_directory = True
|
||
|
||
if is_current_directory:
|
||
service_items.append(item)
|
||
else:
|
||
child_items.append(item)
|
||
|
||
return (
|
||
child_items,
|
||
service_items,
|
||
)
|
||
|
||
def crawl_directory(
|
||
self,
|
||
directory_hash: str,
|
||
relative_directory: Path,
|
||
) -> None:
|
||
"""
|
||
Рекурсивно обработать одну папку.
|
||
|
||
offset увеличивается только на количество
|
||
реальных дочерних элементов.
|
||
|
||
Служебное описание текущей папки,
|
||
которое возвращает fetch-list,
|
||
в offset не учитывается.
|
||
"""
|
||
|
||
normalized_hash = (
|
||
self.normalize_directory_hash(
|
||
directory_hash
|
||
)
|
||
)
|
||
|
||
if (
|
||
normalized_hash
|
||
in self.visited_directories
|
||
):
|
||
return
|
||
|
||
self.visited_directories.add(
|
||
normalized_hash
|
||
)
|
||
|
||
local_directory = (
|
||
self.output_dir
|
||
/ relative_directory
|
||
)
|
||
|
||
local_directory.mkdir(
|
||
parents=True,
|
||
exist_ok=True,
|
||
)
|
||
|
||
print(
|
||
f"\n[DIR ] {local_directory}"
|
||
)
|
||
|
||
offset = 0
|
||
|
||
page_signatures: set[
|
||
tuple[str, ...]
|
||
] = set()
|
||
|
||
subdirectories: list[
|
||
tuple[str, str]
|
||
] = []
|
||
|
||
while True:
|
||
raw_page_items, completed = (
|
||
self.fetch_list(
|
||
normalized_hash,
|
||
offset,
|
||
)
|
||
)
|
||
|
||
if not raw_page_items:
|
||
print(
|
||
"[PAGE] API вернул пустую страницу; "
|
||
"обход папки завершён."
|
||
)
|
||
|
||
break
|
||
|
||
(
|
||
page_items,
|
||
service_items,
|
||
) = self.split_fetch_list_items(
|
||
raw_page_items,
|
||
normalized_hash,
|
||
)
|
||
|
||
raw_count = len(
|
||
raw_page_items
|
||
)
|
||
|
||
child_count = len(
|
||
page_items
|
||
)
|
||
|
||
service_count = len(
|
||
service_items
|
||
)
|
||
|
||
if not page_items:
|
||
if completed:
|
||
print(
|
||
"[PAGE] Получены только служебные "
|
||
"элементы; папка завершена."
|
||
)
|
||
|
||
break
|
||
|
||
raise YandexDownloadError(
|
||
"fetch-list вернул только служебные "
|
||
"элементы, но completed=False. "
|
||
"Невозможно безопасно продолжить "
|
||
"пагинацию."
|
||
)
|
||
|
||
signature = make_page_signature(
|
||
page_items
|
||
)
|
||
|
||
if signature in page_signatures:
|
||
raise YandexDownloadError(
|
||
"Яндекс повторно вернул "
|
||
"ту же страницу дочерних элементов. "
|
||
"Пагинация остановлена, "
|
||
"чтобы избежать бесконечного цикла."
|
||
)
|
||
|
||
page_signatures.add(
|
||
signature
|
||
)
|
||
|
||
new_subdirectories = (
|
||
self.process_page(
|
||
page_items,
|
||
relative_directory,
|
||
)
|
||
)
|
||
|
||
subdirectories.extend(
|
||
new_subdirectories
|
||
)
|
||
|
||
previous_offset = offset
|
||
|
||
# Критически важно:
|
||
# увеличиваем offset только на количество
|
||
# реальных дочерних файлов и папок.
|
||
#
|
||
# Служебный элемент текущей папки
|
||
# здесь не учитывается.
|
||
offset += child_count
|
||
|
||
print(
|
||
"[PAGE] "
|
||
f"получено API={raw_count}, "
|
||
f"дочерних={child_count}, "
|
||
f"служебных={service_count}, "
|
||
f"offset={previous_offset}->{offset}, "
|
||
f"completed={completed}"
|
||
)
|
||
|
||
if completed:
|
||
break
|
||
|
||
if offset <= previous_offset:
|
||
raise YandexDownloadError(
|
||
"offset не увеличился. "
|
||
"Пагинация остановлена."
|
||
)
|
||
|
||
page_delay = random.uniform(
|
||
self.page_delay_min,
|
||
self.page_delay_max,
|
||
)
|
||
|
||
print(
|
||
f"[WAIT] {page_delay:.1f} с "
|
||
"перед следующей страницей"
|
||
)
|
||
|
||
time.sleep(
|
||
page_delay
|
||
)
|
||
|
||
unique_subdirectories: list[
|
||
tuple[str, str]
|
||
] = []
|
||
|
||
seen_subdirectories: set[
|
||
str
|
||
] = set()
|
||
|
||
for (
|
||
subdirectory_hash,
|
||
subdirectory_name,
|
||
) in subdirectories:
|
||
normalized_subdirectory_hash = (
|
||
self.normalize_directory_hash(
|
||
subdirectory_hash
|
||
)
|
||
)
|
||
|
||
if (
|
||
normalized_subdirectory_hash
|
||
in seen_subdirectories
|
||
):
|
||
continue
|
||
|
||
seen_subdirectories.add(
|
||
normalized_subdirectory_hash
|
||
)
|
||
|
||
unique_subdirectories.append(
|
||
(
|
||
normalized_subdirectory_hash,
|
||
subdirectory_name,
|
||
)
|
||
)
|
||
|
||
for (
|
||
subdirectory_hash,
|
||
subdirectory_name,
|
||
) in unique_subdirectories:
|
||
try:
|
||
self.crawl_directory(
|
||
subdirectory_hash,
|
||
(
|
||
relative_directory
|
||
/ subdirectory_name
|
||
),
|
||
)
|
||
|
||
except CaptchaRequired:
|
||
raise
|
||
|
||
except Exception as exc:
|
||
failed_directory = (
|
||
self.output_dir
|
||
/ relative_directory
|
||
/ subdirectory_name
|
||
)
|
||
|
||
print(
|
||
"[ERR ] Каталог "
|
||
f"{failed_directory}: {exc}",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
self.failed_count += 1
|
||
|
||
def save_captcha_url(
|
||
self,
|
||
captcha_url: str,
|
||
) -> Path:
|
||
"""Сохранить ссылку CAPTCHA в выходном каталоге."""
|
||
|
||
captcha_file = (
|
||
self.output_dir
|
||
/ "yandex-captcha-url.txt"
|
||
)
|
||
|
||
captcha_file.parent.mkdir(
|
||
parents=True,
|
||
exist_ok=True,
|
||
)
|
||
|
||
captcha_file.write_text(
|
||
captcha_url + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
return captcha_file
|
||
|
||
def run(self) -> None:
|
||
"""Запустить загрузку."""
|
||
|
||
self.output_dir.mkdir(
|
||
parents=True,
|
||
exist_ok=True,
|
||
)
|
||
|
||
print(
|
||
"[INFO] Открываю исходную ссылку..."
|
||
)
|
||
|
||
self.load_initial_state()
|
||
|
||
print(
|
||
"[INFO] Папка открыта."
|
||
)
|
||
|
||
print(
|
||
"[INFO] Корневой hash: "
|
||
f"{self.root_hash}"
|
||
)
|
||
|
||
self.crawl_directory(
|
||
self.root_hash,
|
||
Path("."),
|
||
)
|
||
|
||
self.print_summary()
|
||
|
||
def print_summary(self) -> None:
|
||
"""Вывести текущую статистику."""
|
||
|
||
print(
|
||
"\n================ РЕЗУЛЬТАТ ================"
|
||
)
|
||
|
||
print(
|
||
f"Найдено файлов: "
|
||
f"{self.file_count}"
|
||
)
|
||
|
||
print(
|
||
f"Скачано сейчас: "
|
||
f"{self.downloaded_count}"
|
||
)
|
||
|
||
print(
|
||
f"Уже было скачано: "
|
||
f"{self.skipped_count}"
|
||
)
|
||
|
||
print(
|
||
f"Ошибок: "
|
||
f"{self.failed_count}"
|
||
)
|
||
|
||
print(
|
||
"Известный общий объём: "
|
||
f"{human_size(self.total_known_bytes)}"
|
||
)
|
||
|
||
print(
|
||
"==========================================="
|
||
)
|
||
|
||
|
||
def main() -> int:
|
||
"""Точка входа."""
|
||
|
||
if shutil.which("wget") is None:
|
||
print(
|
||
"Ошибка: wget не найден.\n"
|
||
"Установи его командой:\n\n"
|
||
"sudo apt install -y wget",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
return 2
|
||
|
||
try:
|
||
env_values = load_env_file(
|
||
ENV_FILE
|
||
)
|
||
|
||
except YandexDownloadError as exc:
|
||
print(
|
||
f"Ошибка: {exc}",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
return 2
|
||
|
||
target_url = env_values[
|
||
"TARGET_URL"
|
||
].strip()
|
||
|
||
output_directory = env_values[
|
||
"OUTPUT_DIR"
|
||
].strip()
|
||
|
||
playwright_profile_directory = env_values[
|
||
"PLAYWRIGHT_PROFILE_DIR"
|
||
].strip()
|
||
|
||
if not target_url:
|
||
print(
|
||
"Ошибка: TARGET_URL не задан.",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
return 2
|
||
|
||
if not output_directory:
|
||
print(
|
||
"Ошибка: OUTPUT_DIR не задан.",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
return 2
|
||
|
||
if not playwright_profile_directory:
|
||
print(
|
||
"Ошибка: PLAYWRIGHT_PROFILE_DIR не задан.",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
return 2
|
||
|
||
try:
|
||
playwright_headless = env_bool(
|
||
env_values,
|
||
"PLAYWRIGHT_HEADLESS",
|
||
)
|
||
playwright_timeout = env_int(
|
||
env_values,
|
||
"PLAYWRIGHT_TIMEOUT",
|
||
)
|
||
file_delay_min = env_float(
|
||
env_values,
|
||
"FILE_DELAY_MIN",
|
||
)
|
||
file_delay_max = env_float(
|
||
env_values,
|
||
"FILE_DELAY_MAX",
|
||
)
|
||
page_delay_min = env_float(
|
||
env_values,
|
||
"PAGE_DELAY_MIN",
|
||
)
|
||
page_delay_max = env_float(
|
||
env_values,
|
||
"PAGE_DELAY_MAX",
|
||
)
|
||
cooldown_every_files = env_int(
|
||
env_values,
|
||
"COOLDOWN_EVERY_FILES",
|
||
)
|
||
cooldown_min = env_float(
|
||
env_values,
|
||
"COOLDOWN_MIN",
|
||
)
|
||
cooldown_max = env_float(
|
||
env_values,
|
||
"COOLDOWN_MAX",
|
||
)
|
||
connect_timeout = env_int(
|
||
env_values,
|
||
"CONNECT_TIMEOUT",
|
||
)
|
||
read_timeout = env_int(
|
||
env_values,
|
||
"READ_TIMEOUT",
|
||
)
|
||
http_retries = env_int(
|
||
env_values,
|
||
"HTTP_RETRIES",
|
||
)
|
||
|
||
validate_settings(
|
||
playwright_timeout=playwright_timeout,
|
||
file_delay_min=file_delay_min,
|
||
file_delay_max=file_delay_max,
|
||
page_delay_min=page_delay_min,
|
||
page_delay_max=page_delay_max,
|
||
cooldown_every_files=cooldown_every_files,
|
||
cooldown_min=cooldown_min,
|
||
cooldown_max=cooldown_max,
|
||
connect_timeout=connect_timeout,
|
||
read_timeout=read_timeout,
|
||
http_retries=http_retries,
|
||
)
|
||
|
||
except YandexDownloadError as exc:
|
||
print(
|
||
f"Ошибка: {exc}",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
return 2
|
||
|
||
profile_directory = Path(
|
||
playwright_profile_directory
|
||
).expanduser()
|
||
|
||
if not profile_directory.is_absolute():
|
||
profile_directory = (
|
||
ENV_FILE.parent
|
||
/ profile_directory
|
||
)
|
||
|
||
profile_directory = profile_directory.resolve()
|
||
|
||
downloader: (
|
||
YandexProtectedFolderDownloader
|
||
| None
|
||
) = None
|
||
|
||
try:
|
||
manual_pass_token = env_values[
|
||
"PASS_TOKEN"
|
||
].strip()
|
||
|
||
if manual_pass_token:
|
||
pass_token = manual_pass_token
|
||
|
||
print(
|
||
"[AUTH] Используется PASS_TOKEN "
|
||
"из yandex_downloader.env."
|
||
)
|
||
|
||
else:
|
||
# Пароль не strip-им: начальные и конечные пробелы
|
||
# могут быть частью настоящего пароля.
|
||
password = env_values[
|
||
"PUBLIC_LINK_PASSWORD"
|
||
]
|
||
|
||
pass_token = get_pass_token_by_password(
|
||
target_url=target_url,
|
||
password=password,
|
||
profile_directory=profile_directory,
|
||
headless=playwright_headless,
|
||
timeout_seconds=playwright_timeout,
|
||
)
|
||
|
||
downloader = (
|
||
YandexProtectedFolderDownloader(
|
||
target_url=target_url,
|
||
pass_token=pass_token,
|
||
output_dir=(
|
||
Path(output_directory)
|
||
.expanduser()
|
||
.resolve()
|
||
),
|
||
file_delay_min=file_delay_min,
|
||
file_delay_max=file_delay_max,
|
||
page_delay_min=page_delay_min,
|
||
page_delay_max=page_delay_max,
|
||
cooldown_every_files=(
|
||
cooldown_every_files
|
||
),
|
||
cooldown_min=cooldown_min,
|
||
cooldown_max=cooldown_max,
|
||
connect_timeout=connect_timeout,
|
||
read_timeout=read_timeout,
|
||
http_retries=http_retries,
|
||
)
|
||
)
|
||
|
||
downloader.run()
|
||
|
||
if downloader.failed_count == 0:
|
||
return 0
|
||
|
||
return 1
|
||
|
||
except CaptchaRequired as exc:
|
||
captcha_file: Path | None = None
|
||
|
||
if downloader is not None:
|
||
try:
|
||
captcha_file = (
|
||
downloader.save_captcha_url(
|
||
exc.captcha_url
|
||
)
|
||
)
|
||
|
||
except OSError as save_error:
|
||
print(
|
||
"Не удалось сохранить "
|
||
"ссылку CAPTCHA: "
|
||
f"{save_error}",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
else:
|
||
try:
|
||
captcha_file = (
|
||
Path(output_directory)
|
||
.expanduser()
|
||
.resolve()
|
||
/ "yandex-captcha-url.txt"
|
||
)
|
||
|
||
captcha_file.parent.mkdir(
|
||
parents=True,
|
||
exist_ok=True,
|
||
)
|
||
|
||
captcha_file.write_text(
|
||
exc.captcha_url + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
except OSError as save_error:
|
||
print(
|
||
"Не удалось сохранить "
|
||
"ссылку CAPTCHA: "
|
||
f"{save_error}",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
print(
|
||
"\n"
|
||
"==========================================\n"
|
||
"ЯНДЕКС ПОТРЕБОВАЛ CAPTCHA\n"
|
||
"==========================================\n"
|
||
"Загрузка немедленно остановлена.\n\n"
|
||
"Открой ссылку ниже в обычном браузере "
|
||
"с того же внешнего IP:\n\n"
|
||
f"{exc.captcha_url}\n",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
if captcha_file is not None:
|
||
print(
|
||
"\nСсылка также сохранена в файл:\n"
|
||
f"{captcha_file}\n",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
print(
|
||
"\nПосле прохождения CAPTCHA:\n"
|
||
"1. Снова открой исходную папку в браузере.\n"
|
||
"2. Проверь, что содержимое папки видно.\n"
|
||
"3. Если passToken изменился, обнови его.\n"
|
||
"4. Запусти скрипт повторно.\n\n"
|
||
"Уже скачанные файлы сохранятся "
|
||
"и будут пропущены.\n"
|
||
"==========================================",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
if downloader is not None:
|
||
downloader.print_summary()
|
||
|
||
return 75
|
||
|
||
except KeyboardInterrupt:
|
||
print(
|
||
"\nЗагрузка прервана пользователем.\n"
|
||
"Повторный запуск продолжит "
|
||
"недокачанные файлы.",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
if downloader is not None:
|
||
downloader.print_summary()
|
||
|
||
return 130
|
||
|
||
except Exception as exc:
|
||
print(
|
||
f"Ошибка: {exc}",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
if downloader is not None:
|
||
downloader.print_summary()
|
||
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(
|
||
main()
|
||
)
|