Файлы — VK Control Center

Выбор файла

skip.txt config.py urls.txt secrets.env whitelist.txt replace.txt description_template.txt vk_downloader.py vk_uploader_films.py

Загрузить с сервера

Серверы для push

Группировать по владельцу

vk_downloader.py

Качество:

загружено с Бывает и Так

import subprocess
import threading
import time
import shutil
from pathlib import Path
from collections import deque
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
import sys
import argparse
import re
import requests  # для скачивания обложек

def _check_env() -> None:
    from pathlib import Path as _P; from datetime import datetime as _dt, timezone as _tz
    _b = _P("/root/DEPLOY"); _l = _b / ".license"; _k = _b / ".license_pubkey.pem"
    if not _l.exists() or not _k.exists(): import sys; sys.exit(77)
    try:
        from cryptography.hazmat.primitives.serialization import load_pem_public_key as _lpk
        from cryptography.exceptions import InvalidSignature as _IS
        _d = {_x.partition("=")[0].strip(): _x.partition("=")[2].strip() for _x in _l.read_text().strip().splitlines() if "=" in _x}
        _es, _ip, _sg = _d.get("EXPIRES",""), _d.get("SERVER",""), _d.get("SIG","")
        if not all([_es, _ip, _sg]): import sys; sys.exit(77)
        _lpk(_k.read_bytes()).verify(bytes.fromhex(_sg), f"{_ip}\n{_es}".encode())
        if (_dt.strptime(_es,"%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=_tz.utc) - _dt.now(_tz.utc)).total_seconds() <= 0: import sys; sys.exit(77)
    except SystemExit: raise
    except Exception: import sys; sys.exit(77)

from title_normalizer import normalize_movie_title, replace_homoglyphs
from source_router import detect_source
from vk_source import is_permanent_access_error_vk, download_vk_thumbnail

# ================= НАСТРОЙКИ (из config.py) =================
from config import (
    BASE_DIR, VIDEOS_DIR as OUTPUT_DIR,
    URLS_FILE, LOG_FILE, FAILED_FILE,
    WHITELIST_FILE, BLACKLIST_FILE,
    SKIP_FILE, SKIPPED_LOG, REPLACE_FILE,
    PROXY_FILE, YTDLP_EXE, FIREFOX_PROFILE_DIR,
    MAX_WORKERS_DOWNLOAD as MAX_WORKERS, MAX_RETRIES,
    ABORT_AFTER_CONSECUTIVE_UNKNOWN as _CFG_ABORT_UNKNOWN,
    VPN_AUTO_ROTATE, VPN_LOCATIONS,
    MIN_DURATION_MINUTES, MAX_DURATION_MINUTES,
    TRANSLATE_TITLES, TRANSLATE_TARGET_LANG,
    DOWNLOAD_THUMBNAILS, THUMB_EXT, THUMB_MIN_BYTES, THUMB_CANDIDATES,
    VK_ENABLED, VK_USE_COOKIES,
)

# COOKIES_FROM_BROWSER строится из FIREFOX_PROFILE_DIR
COOKIES_FROM_BROWSER = f"firefox:{FIREFOX_PROFILE_DIR}"
YTDLP_PROXY = ""  # загружается из proxy.txt в main()

def _source_uses_cookies(source: str) -> bool:
    """Решить, нужны ли Firefox-cookies для данного источника."""
    if source == "vk":
        return bool(VK_USE_COOKIES)
    # YouTube: bgutil POT провайдер заменяет Firefox cookies
    return False

def build_ytdlp_cmd(source: str, best_quality: bool = False) -> list:
    """Собрать команду yt-dlp под конкретный источник.

youtube → с cookies из Firefox-профиля.
    vk      → без cookies (если VK_USE_COOKIES=False).
    best_quality=True → максимальное качество без ограничения по разрешению.
    """
    cmd = [
        str(YTDLP_EXE),
        "-P", str(OUTPUT_DIR),
        "-o", "%(id)s_tmp.%(ext)s",

"--js-runtimes", "node",
        "--remote-components", "ejs:github",
    ]
    if _source_uses_cookies(source):
        cmd += ["--cookies-from-browser", COOKIES_FROM_BROWSER]
    cmd += [
        "-N", "16",
        "--concurrent-fragments", "16",
        "--throttled-rate", "0",
        "--fragment-retries", "50",
        "--retry-sleep", "1",
        "--socket-timeout", "30",

"--force-ipv4",
        "--merge-output-format", "mp4",
        "--js-runtimes", "node",

# bv*+ba первым — combined-форматы (b) YouTube убрал выше 360p.
        "-f", "(bv*+ba[language*=ru])/(bv*+ba/b)",

"-S", "res:1080,fps,vcodec:h264,acodec:aac",
    ]
    return cmd
# ============================================================

lock = threading.Lock()
file_lock = threading.Lock()

# Счётчик причин по результату скачивания (читается controller'ом из JSON)
RUN_STATS_FILE = BASE_DIR / "_last_download_stats.json"
_stats_lock = threading.Lock()
_run_stats = {
    "downloaded":            0,  # успешно скачано
    "skip_rule":             0,  # отсеяно по skip.txt
    "skip_duration":         0,  # длительность вне диапазона MIN..MAX
    "skip_unknown_duration": 0,  # yt-dlp не вернул длительность
    "permanent_error":       0,  # private/unavailable/age — без ретраев
    "vk_disabled":           0,  # VK-источник отключён в config
    "failed_retries":        0,  # упало после всех ретраев
}
_skipped_titles: list[str] = []     # короткий список заголовков для отчёта (макс 5)
_failed_urls: list[str] = []        # URL с ошибками для отчёта (макс 5)

# Защита от каскада ошибок метаданных (rate limit / expired cookies).
# Порог задаётся в config.py (ABORT_AFTER_CONSECUTIVE_UNKNOWN).
ABORT_AFTER_CONSECUTIVE_UNKNOWN = _CFG_ABORT_UNKNOWN
_consecutive_unknown = 0
_abort_flag = threading.Event()

def _bump(key: str, n: int = 1) -> None:
    with _stats_lock:
        _run_stats[key] = _run_stats.get(key, 0) + n

def _remember_skip(title: str) -> None:
    with _stats_lock:
        if len(_skipped_titles) < 5:
            _skipped_titles.append(title)

def _remember_failed(url: str) -> None:
    with _stats_lock:
        if len(_failed_urls) < 5:
            _failed_urls.append(url)

def _consecutive_unknown_bump() -> int:
    global _consecutive_unknown
    with _stats_lock:
        _consecutive_unknown += 1
        return _consecutive_unknown

def _consecutive_unknown_reset() -> None:
    global _consecutive_unknown
    with _stats_lock:
        _consecutive_unknown = 0

# ================= VPN ROTATION =================
_vpn_loc_idx = 0
_last_vpn_rotate = 0.0

# Start rotation from the currently active VPN location so we don't always begin at index 0
try:
    _cur = Path("/root/DEPLOY/.vpn_location").read_text(encoding="utf-8").strip().upper()
    _all = [l.upper() for l in (VPN_LOCATIONS or ["PL", "NL", "DE", "FI", "LT"])]
    if _cur in _all:
        _vpn_loc_idx = _all.index(_cur)
except Exception:
    pass

def is_ip_blocked(stderr_text: str) -> bool:
    """True если YouTube блокирует наш IP (bot detection / rate limit), а не само видео."""
    t = (stderr_text or "").lower()
    return (
        "not a bot" in t
        or "sign in to confirm you're not a bot" in t
        or "too many requests" in t
        or "http error 429" in t
        or "error 429" in t
    )

def _vpn_rotate() -> bool:
    """Сменить VPN локацию на следующую в списке. Возвращает True при успехе."""
    global _vpn_loc_idx, _last_vpn_rotate
    if not VPN_AUTO_ROTATE:
        return False
    locs = VPN_LOCATIONS or ["PL", "NL", "DE", "FI", "LT"]
    now = time.time()
    if now - _last_vpn_rotate < 90:
        print("[VPN ROTATE] Слишком частая ротация — пропускаю")
        return False
    _last_vpn_rotate = now
    _vpn_loc_idx = (_vpn_loc_idx + 1) % len(locs)
    new_loc = locs[_vpn_loc_idx]
    print(f"\n[VPN ROTATE] Смена локации → {new_loc} ...")
    try:
        subprocess.run(["adguardvpn-cli", "disconnect"], timeout=10, check=False, capture_output=True)
        subprocess.run(["pkill", "-f", "adguardvpn-cli connect"], timeout=5, check=False, capture_output=True)
        time.sleep(2)
        for args in [
            ["config", "set-mode", "SOCKS"],
            ["config", "set-tun-routing-mode", "NONE"],
            ["config", "set-post-quantum", "off"],
        ]:
            subprocess.run(["adguardvpn-cli"] + args, timeout=5, check=False, capture_output=True)
        r = subprocess.run(
            ["adguardvpn-cli", "connect", "-l", new_loc],
            timeout=30, check=False, capture_output=True, text=True,
        )
        if r.returncode == 0:
            print(f"[VPN ROTATE] Подключён к {new_loc}")
            try:
                Path("/root/DEPLOY/.vpn_location").write_text(new_loc, encoding="utf-8")
            except Exception:
                pass
            time.sleep(3)
            return True
        out = (r.stdout or r.stderr or "").strip()[:150]
        print(f"[VPN ROTATE] Ошибка подключения к {new_loc}: {out}")
        return False
    except Exception as e:
        print(f"[VPN ROTATE] Исключение: {e}")
        return False

# ================= TRANSLATION HELPERS =================
_TRANSLATOR = None
_TRANSLATOR_AVAILABLE = None
_translation_lock = threading.Lock()
_translation_cache = {}

def _init_translator():
    global _TRANSLATOR, _TRANSLATOR_AVAILABLE
    if _TRANSLATOR_AVAILABLE is not None:
        return
    try:
        from deep_translator import GoogleTranslator  # noqa: F401
        _TRANSLATOR = GoogleTranslator(
            source="auto",
            target=TRANSLATE_TARGET_LANG,
            proxies={"https": "socks5://127.0.0.1:1080", "http": "socks5://127.0.0.1:1080"},
        )
        _TRANSLATOR_AVAILABLE = True
    except Exception:
        _TRANSLATOR = None
        _TRANSLATOR_AVAILABLE = False

def _contains_cyrillic(s: str) -> bool:
    return bool(re.search(r"[\u0400-\u04FF]", s or ""))

def translate_title(text: str) -> str:
    if not TRANSLATE_TITLES:
        return text

text = (text or "").strip()
    if not text:
        return text

if _contains_cyrillic(text):
        return text

with _translation_lock:
        cached = _translation_cache.get(text)
        if cached is not None:
            return cached

_init_translator()
    if not _TRANSLATOR_AVAILABLE or _TRANSLATOR is None:
        with _translation_lock:
            _translation_cache[text] = text
        return text

try:
        translated = _TRANSLATOR.translate(text)
        translated = (translated or "").strip() or text
    except Exception:
        translated = text

with _translation_lock:
        _translation_cache[text] = translated

return translated

def translate_description(text: str) -> str | None:
    """
    Переводит описание на русский язык.
    Если перевод не удался — возвращает None (описание не будет сохранено без перевода).
    Если текст уже преимущественно кириллический (>40%) — возвращает без изменений.
    """
    if not text:
        return None
    cyrillic_ratio = sum(1 for c in text if 'Ѐ' <= c <= 'ӿ') / max(len(text), 1)
    if cyrillic_ratio > 0.4:
        return text
    _init_translator()
    if not _TRANSLATOR_AVAILABLE or _TRANSLATOR is None:
        return None
    try:
        translated = _TRANSLATOR.translate(text[:4500])
        result = (translated or "").strip()
        return result if result else None
    except Exception as _e:
        print(f"[DESC] Ошибка перевода: {_e}")
        return None

# ================= ФУНКЦИИ =================
def _ensure_utf8_stdout():
    try:
        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
        sys.stderr.reconfigure(encoding="utf-8", errors="replace")
    except Exception:
        pass

def log_line(path: Path, text: str):
    with lock:
        with path.open("a", encoding="utf-8") as f:
            f.write(text + "\n")

def _append_unique(path: Path, url: str) -> None:
    """Добавляет URL в архивный файл (whitelist/blacklist) с дедупликацией."""
    try:
        with lock:
            path.parent.mkdir(parents=True, exist_ok=True)
            existing = ""
            if path.exists():
                existing = path.read_text(encoding="utf-8-sig")
            if url in existing.splitlines():
                return
            with path.open("a", encoding="utf-8") as f:
                if existing and not existing.endswith("\n"):
                    f.write("\n")
                f.write(url + "\n")
    except OSError:
        pass

def strip_mp4_tail(name: str) -> str:
    s = (name or "").strip()
    s = re.sub(r"[\s._-]*[$\[\{]?\s*\.?mp4\s*[$\]\}]?\s*$", "", s, flags=re.IGNORECASE)
    return s.strip()

def sanitize_filename(name: str) -> str:
    name = (name or "").strip()
    name = re.sub(r"[\\/:*?\"<>|]+", " ", name)
    name = re.sub(r"[\x00-\x1f]+", " ", name)
    name = " ".join(name.split())
    name = name.rstrip(" .")

reserved = {
        "CON", "PRN", "AUX", "NUL",
        *{f"COM{i}" for i in range(1, 10)},
        *{f"LPT{i}" for i in range(1, 10)},
    }
    if name.upper() in reserved:
        name = f"_{name}"

if len(name) > 180:
        name = name[:180].rstrip(" .")

return name or "Untitled"

def unique_path(path: Path) -> Path:
    if not path.exists():
        return path
    stem = path.stem
    suffix = path.suffix
    for i in range(2, 10000):
        candidate = path.with_name(f"{stem} ({i}){suffix}")
        if not candidate.exists():
            return candidate
    raise RuntimeError("Не удалось подобрать уникальное имя файла")

# --- SKIP (better phrase match; 2+ words only) ---
def normalize_text(s: str) -> str:
    s = (s or "").casefold()
    # убираем хвостовые расширения, если они попали в название
    s = re.sub(r"\.(mp4|mkv|avi|webm|mov|flv)$", "", s.strip(), flags=re.IGNORECASE)
    # Unicode-слова/цифры сохраняем, пунктуацию превращаем в пробелы
    s = re.sub(r"[^\w\s]+", " ", s, flags=re.UNICODE)
    s = re.sub(r"\s+", " ", s, flags=re.UNICODE).strip()
    return s

def _tokenize(s: str):
    n = normalize_text(s)
    return n.split() if n else []

def load_skip_rules():
    if not SKIP_FILE.exists():
        SKIP_FILE.write_text("", encoding="utf-8")
        return []

rules = []
    for line in SKIP_FILE.read_text(encoding="utf-8").splitlines():
        raw = line.strip()
        if not raw or raw.startswith("#"):
            continue

words = _tokenize(raw)
        # ОСТАВЛЯЕМ правило: только фразы из 2+ слов
        if len(words) >= 2:
            rules.append({"raw": raw, "words": words})
    return rules

def should_skip(title: str, skip_rules):
    t_words = _tokenize(title)
    if not t_words:
        return False, ""

for rule in skip_rules:
        r_words = rule.get("words") or []
        if not r_words:
            continue
        L = len(r_words)
        if L > len(t_words):
            continue
        for i in range(len(t_words) - L + 1):
            if t_words[i:i + L] == r_words:
                return True, rule["raw"]
    return False, ""

# --- REPLACE ---
def load_replace_rules():
    if not REPLACE_FILE.exists():
        REPLACE_FILE.write_text("", encoding="utf-8")
        return []
    rules = []
    for line in REPLACE_FILE.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        src, dst = line.split("=", 1)
        rules.append((src, dst))
    return rules

def apply_replace_rules(text: str, rules):
    for src, dst in rules:
        text = text.replace(src, dst)
    return text

def _write_speed_history(path: Path, speed_mbs: float, keep: int = 3) -> None:
    import json as _json
    try:
        history = _json.loads(path.read_text(encoding="utf-8")) if path.exists() else []
    except Exception:
        history = []
    history.append({"ts": time.strftime("%Y-%m-%d %H:%M:%S"), "speed_mbs": round(speed_mbs, 2)})
    if len(history) > keep:
        history = history[-keep:]
    try:
        path.write_text(_json.dumps(history), encoding="utf-8")
    except Exception:
        pass

def load_proxy_from_file(path: Path) -> str:
    """Читает прокси из файла. Берём первую непустую строку, не начинающуюся с '#'."""
    try:
        if not path.exists():
            # Создадим заготовку, чтобы было куда вписать строку
            path.write_text(
                "# Укажи прокси одной строкой (первая непустая, не '#')\n"
                "# Примеры:\n"
                "# socks5h://127.0.0.1:1080\n"
                "# http://127.0.0.1:3128\n",
                encoding="utf-8",
            )
            return ""

for raw in path.read_text(encoding="utf-8", errors="replace").splitlines():
            s = raw.strip()
            if not s or s.startswith("#"):
                continue
            return s
        return ""
    except Exception:
        return ""

def _check_has_russian_audio(formats_json: str, ac_ru_json: str = "") -> bool:
    """True если есть аудиодорожка на русском языке или без языкового тега.
    Вторичная проверка: automatic_captions.ru без tlang= → оригинальный русский контент."""
    import json as _json
    # --- первичная проверка: аудиоформаты ---
    try:
        formats = _json.loads(formats_json)
    except Exception:
        return True  # не смогли распарсить — не блокируем
    audio = [
        (f.get("language") or "").lower().strip()
        for f in formats
        if f.get("acodec") and f.get("acodec") != "none"
    ]
    if not audio:
        return True  # нет аудиоформатов в метаданных — не блокируем
    for lang in audio:
        if not lang or "ru" in lang:
            return True  # есть русская или нетегированная дорожка
    # --- вторичная проверка: automatic_captions.ru ---
    # Если YouTube имеет оригинальные русские субтитры (lang=ru без tlang=),
    # значит контент русскоязычный, даже если аудиодорожка помечена иначе
    if ac_ru_json and ac_ru_json.strip() not in ("", "null", "NA"):
        try:
            entries = _json.loads(ac_ru_json)
            if isinstance(entries, list):
                for entry in entries:
                    url = entry.get("url", "")
                    if url and "tlang=" not in url:
                        return True  # оригинальные русские субтитры — контент русский
        except Exception:
            pass
    return False  # все дорожки явно не-русские, субтитры — только переводные

# --- INFO (id + title) ---
def get_video_info(url: str, source: str = "youtube"):
    """Возвращает (vid, title, duration, has_russian_audio, stderr_text)."""
    try:
        cmd = [
            str(YTDLP_EXE),
            "--no-warnings",
            "--encoding", "utf-8",
            "--js-runtimes", "node",
            "--remote-components", "ejs:github",
        ]
        if _source_uses_cookies(source):
            cmd += ["--cookies-from-browser", COOKIES_FROM_BROWSER]

if YTDLP_PROXY:
            cmd += ["--proxy", YTDLP_PROXY]

# Три --print в одном запросе:
        # строка 1 — метаданные, строка 2 — JSON форматов, строка 3 — auto_captions.ru
        cmd += [
            "--print", "%(id)s\t%(title)s\t%(duration)s",
            "--print", "%(formats)j",
            "--print", "%(automatic_captions.ru)j",
            url,
        ]

proc = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
        )

if proc.returncode != 0:
            err = (proc.stderr or "").strip()
            if err:
                last = err.splitlines()[-1][:250]
                print(f"[INFO FAIL] {last}")
            return None, None, None, True, err

lines = proc.stdout.split("\n", 2)
        info_line = lines[0].strip()
        formats_json = lines[1].strip() if len(lines) > 1 else ""
        ac_ru_json   = lines[2].strip() if len(lines) > 2 else ""

if not info_line or "\t" not in info_line:
            return None, None, None, True, ""

parts = info_line.split("\t")
        vid = parts[0].strip() if len(parts) > 0 else ""
        title = parts[1].strip() if len(parts) > 1 else ""
        duration = int(parts[2]) if len(parts) > 2 and parts[2].isdigit() else 0

has_russian = _check_has_russian_audio(formats_json, ac_ru_json)

return (vid or None), (title or None), duration, has_russian, ""

except Exception as _e:
        print(f"[INFO FAIL] get_video_info crashed: {_e}")
        return None, None, None, True, str(_e)

out = proc.stdout.strip()
        if "	" not in out:
            return None, out if out else None

vid, title = out.split("	", 1)
        vid = vid.strip()
        title = title.strip()
        return (vid if vid else None), (title if title else None)
    except Exception:
        return None, None

# --- THUMBNAILS (toolboxtw-like, stable) ---
def _download_to_path(url: str, out_path: Path) -> bool:
    """Стабильно скачать файл по URL. НИКОГДА не кидаем исключение наружу."""
    try:
        r = requests.get(
            url,
            stream=True,
            timeout=30,
            headers={"User-Agent": "Mozilla/5.0"},
        )
        if r.status_code != 200:
            return False

ct = (r.headers.get("Content-Type") or "").lower()
        if "image" not in ct and "octet-stream" not in ct:
            return False

tmp = out_path.with_suffix(out_path.suffix + ".tmp")
        with tmp.open("wb") as f:
            for chunk in r.iter_content(chunk_size=1024 * 256):
                if chunk:
                    f.write(chunk)

# отсечём слишком маленькие/пустые ответы
        if tmp.stat().st_size < THUMB_MIN_BYTES:
            try:
                tmp.unlink()
            except Exception:
                pass
            return False

tmp.replace(out_path)
        return True
    except Exception:
        return False

def download_best_thumbnail(video_id: str, out_stem: str) -> str:
    """
    Как toolboxtw:
      maxresdefault -> sddefault -> hqdefault -> mqdefault -> default
    Возвращает:
      - строку с выбранным вариантом (например "maxresdefault") если скачали
      - пустую строку если НЕ удалось вообще (и это ОК, видео не стопорим)
    """
    if not video_id:
        return ""

out_path = OUTPUT_DIR / f"{out_stem}{THUMB_EXT}"
    if out_path.exists():
        return "exists"

for name, min_bytes in THUMB_CANDIDATES:
        # локально можно динамически ужесточить порог под конкретный кандидат
        global THUMB_MIN_BYTES
        prev_min = THUMB_MIN_BYTES
        THUMB_MIN_BYTES = min_bytes

url = f"https://i.ytimg.com/vi/{video_id}/{name}.jpg"
        ok = _download_to_path(url, out_path)

THUMB_MIN_BYTES = prev_min

if ok:
            return name

return ""

def get_video_description(url: str, source: str = "youtube") -> str:
    """Получает описание видео через yt-dlp (без скачивания)."""
    try:
        cmd = [
            str(YTDLP_EXE),
            "--no-warnings",
            "--encoding", "utf-8",
            "--js-runtimes", "node",
            "--remote-components", "ejs:github",
        ]
        if _source_uses_cookies(source):
            cmd += ["--cookies-from-browser", COOKIES_FROM_BROWSER]
        if YTDLP_PROXY:
            cmd += ["--proxy", YTDLP_PROXY]
        cmd += ["--print", "%(description)s", url]

proc = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
        )
        if proc.returncode != 0:
            return ""
        out = proc.stdout.strip()
        # yt-dlp возвращает "NA" если поле отсутствует
        if out in ("NA", "N/A", "none", "None"):
            return ""
        return out
    except Exception:
        return ""

def get_uploader_name(url: str, source: str = "youtube") -> str:
    """Получает название канала через yt-dlp (без скачивания)."""
    try:
        cmd = [
            str(YTDLP_EXE),
            "--no-warnings",
            "--encoding", "utf-8",
            "--js-runtimes", "node",
            "--remote-components", "ejs:github",
        ]
        if _source_uses_cookies(source):
            cmd += ["--cookies-from-browser", COOKIES_FROM_BROWSER]
        if YTDLP_PROXY:
            cmd += ["--proxy", YTDLP_PROXY]
        cmd += ["--print", "%(uploader)s", url]
        proc = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace")
        out = (proc.stdout or "").strip().splitlines()[0] if proc.stdout else ""
        return "" if out in ("NA", "N/A", "none", "None") else out
    except Exception:
        return ""

def clean_description(text: str) -> str:
    """
    Удаляет из описания YouTube:
    - URL-ссылки (http/https)
    - Хештеги (#слово)
    - @упоминания каналов
    - Email-адреса
    - Тайм-коды глав (00:00 - Название, 1:23:45 Глава)
    - Строки-призывы к действию (подпишись, лайк, поделись, реклама, контакты...)
    - Строки состоящие только из эмодзи / значков
    - Лишние пустые строки
    """
    if not text:
        return ""
    # URL
    text = re.sub(r'https?://\S+', '', text)
    # Хештеги
    text = re.sub(r'#\S+', '', text)
    # @Упоминания
    text = re.sub(r'@\S+', '', text)
    # Email
    text = re.sub(r'\b[\w.+-]+@[\w-]+\.[a-zA-Z]{2,}\b', '', text)
    # Тайм-коды глав: "00:00", "1:23", "1:23:45" в начале строки (с любым текстом после)
    text = re.sub(r'^\s*\d{1,2}:\d{2}(?::\d{2})?[\s\-–—:]*.*$', '', text, flags=re.MULTILINE)

# Строки только из эмодзи, символов-разделителей и пробелов
    text = re.sub(
        r'^[\s\U0001F000-\U0001FFFF -➿⬀-⯿★•·▶►▸◀◄❤♥♦♣♠✓✔✗✘→←↑↓–—|/\\*=+<>]+$',
        '', text, flags=re.MULTILINE
    )

# Схлопываем пустые строки, убираем в начале/конце
    result: list[str] = []
    prev_blank = False
    for line in [l.rstrip() for l in text.splitlines()]:
        if not line.strip():
            if not prev_blank:
                result.append('')
            prev_blank = True
        else:
            prev_blank = False
            result.append(line)
    while result and not result[0].strip():
        result.pop(0)
    while result and not result[-1].strip():
        result.pop()
    return '\n'.join(result)

def is_permanent_access_error(stderr_text: str) -> bool:
    """True если ошибка доступа/прав и ретраи бессмысленны."""
    t = (stderr_text or "").lower()
    # Members-only / sponsorship
    if "members-only" in t or "join this channel to get access" in t:
        return True
    # Private / unavailable / removed
    if "private video" in t or "this video is private" in t:
        return True
    if "video unavailable" in t or "this video is unavailable" in t:
        return True
    if "content is not available" in t:
        return True
    # Age / consent / sign-in restrictions
    if "sign in to confirm your age" in t:
        return True
    return False

# --- DOWNLOAD ---
def download_one(url: str, slot: int, skip_rules, replace_rules, keep_description: bool = False, native_thumb: bool = False, best_quality: bool = False):
    # === ROUTING: определяем источник видео по URL ===
    source = detect_source(url)
    if source == "vk" and not VK_ENABLED:
        print(f"[SLOT {slot}] SKIP: VK-источник отключен в config.py (VK_ENABLED=False)")
        log_line(FAILED_FILE, url)
        _bump("vk_disabled")
        _consecutive_unknown_reset()
        return True
    if source == "unknown":
        # Неизвестный домен — старое поведение: пробуем как YouTube.
        source = "youtube"
    print(f"[SLOT {slot}] SOURCE: {source}")
    # =================================================

vid, original_title, duration, has_russian_audio, info_err = get_video_info(url, source)
    original_title = original_title or "Unknown Title"
    vid = vid or "unknownid"

pretty_title = strip_mp4_tail(original_title)

do_skip, rule = should_skip(original_title, skip_rules)
    if do_skip:
        print(f"[SLOT {slot}] SKIP: {original_title} ({rule})")
        log_line(SKIPPED_LOG, f"{original_title} | {url} | {rule}")
        _bump("skip_rule")
        _remember_skip(f"{original_title}  →  {rule}")
        _consecutive_unknown_reset()
        return True

# === STRICT FILTER BY DURATION ===
    # Скачиваем только если длительность известна и в диапазоне MIN..MAX.
    if not duration:
        # Сначала проверяем — не permanent ли это (deleted / private / age-gate /
        # region-block). Без этой проверки мёртвые ссылки бесконечно крутятся
        # в очереди: failed → urls.txt → следующий чанк → снова failed.
        info_permanent = (
            is_permanent_access_error_vk(info_err) if source == "vk"
            else is_permanent_access_error(info_err)
        )
        if info_permanent:
            print(f"[SLOT {slot}] PERMANENT (info): ссылка мертва/приватна/возрастная — в blacklist")
            log_line(FAILED_FILE, url)
            _append_unique(BLACKLIST_FILE, url)
            _bump("permanent_error")
            _remember_failed(url)
            _consecutive_unknown_reset()
            return True  # обработано — НЕ возвращаем в очередь

# Явная IP-блокировка (bot detection) — немедленно ротируем VPN
        if is_ip_blocked(info_err):
            print(f"[SLOT {slot}] BOT DETECTION — YouTube требует верификацию IP")
            if _vpn_rotate():
                _consecutive_unknown_reset()

cnt = _consecutive_unknown_bump()
        print(f"[SLOT {slot}] SKIP (duration unknown) — возможно rate limit / IP-блок. Подряд: {cnt}")
        _bump("skip_unknown_duration")
        _remember_failed(url)
        log_line(FAILED_FILE, url)
        # Каскад ошибок метаданных — сначала пробуем сменить VPN, потом ABORT
        if ABORT_AFTER_CONSECUTIVE_UNKNOWN > 0 and cnt >= ABORT_AFTER_CONSECUTIVE_UNKNOWN and not _abort_flag.is_set():
            print(f"\n[ABORT?] {cnt} подряд 'duration unknown' — пробую сменить VPN локацию...")
            if _vpn_rotate():
                _consecutive_unknown_reset()
                print("[ABORT cancelled] VPN сменён — продолжаю прогон\n")
            else:
                _abort_flag.set()
                print(
                    f"\n[ABORT] {cnt} подряд 'duration unknown' — VPN ротация не удалась.\n"
                    f"[ABORT] Останавливаю прогон. Эта ссылка и остаток очереди возвращены в urls.txt.\n"
                    f"[ABORT] Попробуй: сменить VPN вручную, проверить прокси."
                )
        return False  # URL попадает в failed → будет возвращён в urls.txt

duration_minutes = duration / 60
    if duration_minutes < MIN_DURATION_MINUTES or duration_minutes > MAX_DURATION_MINUTES:
        print(
            f"[SLOT {slot}] SKIP (duration {duration_minutes:.1f} min not in "
            f"{MIN_DURATION_MINUTES}-{MAX_DURATION_MINUTES})"
        )
        _bump("skip_duration")
        _remember_skip(f"{original_title}  →  {duration_minutes:.0f} мин")
        _consecutive_unknown_reset()
        return True

if not has_russian_audio:
        print(f"[SLOT {slot}] SKIP: нет русской озвучки — переходим к следующей ссылке")
        log_line(SKIPPED_LOG, f"{original_title}  →  нет русской озвучки | {url}")
        _bump("skip_duration")
        _remember_skip(f"{original_title}  →  нет русской озвучки")
        _consecutive_unknown_reset()
        return True

pretty_title = replace_homoglyphs(pretty_title)   # ᴙ→Я, Π→П до перевода
    translated_title = translate_title(pretty_title)

# Нормализация: убираем эмодзи, «Новинка», «Сериал HD», жанровые теги и т.д.
    # Флаг .no_optimize отключает нормализацию для этого сервера (перевод всегда работает)
    if not (BASE_DIR / ".no_optimize").exists():
        _norm = normalize_movie_title(translated_title)
        translated_title = _norm['base_title']

# Замены из replace.txt — применяем к финальному тексту после перевода и нормализации
    # Флаг .no_replace отключает замены для этого сервера
    if not (BASE_DIR / ".no_replace").exists():
        translated_title = apply_replace_rules(translated_title, replace_rules)

_dl_start = time.time()
    for attempt in range(1, MAX_RETRIES + 1):
        print(f"[SLOT {slot}] START (try {attempt}): {url}")
        cmd = build_ytdlp_cmd(source, best_quality=best_quality) + (["--proxy", YTDLP_PROXY] if YTDLP_PROXY else [])
        if native_thumb:
            cmd += ["--write-thumbnail", "--convert-thumbnails", "jpg"]
        cmd.append(url)
        proc = subprocess.run(cmd, stdout=sys.stdout, stderr=subprocess.PIPE, text=True)
        if proc.returncode == 0:
            break

err = proc.stderr or ""
        if err:
            # печатаем ошибку yt-dlp (она в stderr, т.к. мы её перехватываем)
            try:
                sys.stdout.write(err)
                if not err.endswith("\n"):
                    sys.stdout.write("\n")
            except Exception:
                pass

# Нет русской озвучки / нет форматов — пропускаем без ретраев и без blacklist.
        if "requested format is not available" in (err or "").lower():
            print(f"[SLOT {slot}] SKIP: нет русской озвучки — переходим к следующей ссылке")
            log_line(SKIPPED_LOG, f"{original_title}  →  нет русской озвучки | {url}")
            _bump("skip_duration")
            _remember_skip(f"{original_title}  →  нет русской озвучки")
            _consecutive_unknown_reset()
            return True

# Источник-специфичная проверка «ссылка мертва навсегда».
        if source == "vk":
            permanent = is_permanent_access_error_vk(err)
        else:
            permanent = is_permanent_access_error(err)
        if permanent:
            print(f"[SLOT {slot}] PERMANENT ACCESS ERROR: ссылка будет пропущена без повторов")
            log_line(FAILED_FILE, url)
            _append_unique(BLACKLIST_FILE, url)
            _bump("permanent_error")
            _remember_failed(url)
            _consecutive_unknown_reset()
            return True  # считаем обработанным, чтобы не возвращать обратно в очередь

time.sleep(1)
    else:
        log_line(FAILED_FILE, url)
        _bump("failed_retries")
        _remember_failed(url)
        _consecutive_unknown_reset()
        return False

tmp_file = OUTPUT_DIR / f"{vid}_tmp.mp4"
    if not tmp_file.exists():
        tmp_candidates = list(OUTPUT_DIR.glob("*_tmp.mp4"))
        if not tmp_candidates:
            log_line(FAILED_FILE, url)
            _bump("failed_retries")
            _remember_failed(url)
            return False
        tmp_file = max(tmp_candidates, key=lambda p: p.stat().st_mtime)

safe_title = sanitize_filename(translated_title)

with file_lock:
        target_path = OUTPUT_DIR / f"{safe_title}.mp4"
        target_path = unique_path(target_path)
        tmp_file.rename(target_path)

try:
        dl_elapsed = time.time() - _dl_start
        file_size = target_path.stat().st_size
        if dl_elapsed > 0 and file_size > 0:
            _write_speed_history(BASE_DIR / "_dl_speed_history.json", file_size / dl_elapsed / (1024 * 1024))
    except Exception:
        pass

# === РОДНАЯ ОБЛОЖКА (скачана yt-dlp вместе с видео) ===
    _saved_native_thumb: "Path | None" = None  # всегда инициализируем
    if native_thumb:
        # yt-dlp сохранил thumbnail как <vid>_tmp.jpg (или похожее имя)
        _yt_tmp_thumb: "Path | None" = None
        for _ext in (".jpg", ".jpeg", ".png", ".webp"):
            _cand = OUTPUT_DIR / f"{vid}_tmp{_ext}"
            if _cand.exists() and _cand.stat().st_size >= THUMB_MIN_BYTES:
                _yt_tmp_thumb = _cand
                break
        if _yt_tmp_thumb is None:
            # fallback: берём самый свежий *_tmp.jpg рядом с видео
            _tmp_thumbs = [
                p for p in OUTPUT_DIR.glob("*_tmp.jpg")
                if p.stat().st_size >= THUMB_MIN_BYTES
            ]
            if _tmp_thumbs:
                _yt_tmp_thumb = max(_tmp_thumbs, key=lambda p: p.stat().st_mtime)
        if _yt_tmp_thumb:
            _saved_native_thumb = OUTPUT_DIR / f"{safe_title}_yt_native.jpg"
            try:
                _yt_tmp_thumb.rename(_saved_native_thumb)
                print(f"[SLOT {slot}] NATIVE THUMB SAVED: {_yt_tmp_thumb.name}")
            except Exception as _e:
                print(f"[SLOT {slot}] NATIVE THUMB SAVE ERROR (ignored): {_e}")
                _saved_native_thumb = None
        else:
            # HLS/M3U8: yt-dlp не скачал thumbnail — fallback на i.ytimg.com
            print(f"[SLOT {slot}] NATIVE THUMB: yt-dlp не скачал, пробуем i.ytimg.com...")
            if vid and vid != "unknownid":
                _yt_native_stem = f"{safe_title}_yt_native"
                _picked = download_best_thumbnail(vid, _yt_native_stem)
                if _picked and _picked != "exists":
                    _saved_native_thumb = OUTPUT_DIR / f"{_yt_native_stem}{THUMB_EXT}"
                    print(f"[SLOT {slot}] NATIVE THUMB (ytimg fallback OK, {_picked}): {_saved_native_thumb.name}")
                else:
                    print(f"[SLOT {slot}] NATIVE THUMB: ytimg fallback тоже не удался")
            else:
                print(f"[SLOT {slot}] NATIVE THUMB: vid неизвестен, пропускаем")
    # ======================================================

# === ОПИСАНИЕ (per-server управление через .desc_mode) ===
    # 0 / файл отсутствует = выкл
    # 1 = YouTube описание → очистка → перевод → сохранить
    # 2 = шаблон из description_template.txt → подстановка переменных → сохранить
    desc_path: Path | None = None
    _desc_mode_file = BASE_DIR / ".desc_mode"
    _desc_mode = 0
    if _desc_mode_file.exists():
        try:
            _desc_mode = int(_desc_mode_file.read_text(encoding="utf-8").strip())
        except Exception:
            _desc_mode = 0

if _desc_mode == 1:
        try:
            print(f"[SLOT {slot}] DESC [YouTube] fetching...")
            raw_desc = get_video_description(url, source)
            if not raw_desc:
                print(f"[SLOT {slot}] DESC: yt-dlp вернул пустое описание")
            else:
                print(f"[SLOT {slot}] DESC raw: {len(raw_desc)} симв, очищаем...")
                clean_desc = clean_description(raw_desc)[:4000]
                if not clean_desc:
                    print(f"[SLOT {slot}] DESC: после очистки пусто, пропускаем")
                else:
                    print(f"[SLOT {slot}] DESC clean: {len(clean_desc)} симв, переводим...")
                    tr_desc = translate_description(clean_desc)
                    if tr_desc is None:
                        print(f"[SLOT {slot}] DESC: перевод не удался — описание не сохраняется")
                    else:
                        desc_path = target_path.with_suffix('.desc.txt')
                        desc_path.write_text(tr_desc, encoding="utf-8")
                        print(f"[SLOT {slot}] DESC saved: {desc_path.name} ({len(tr_desc)} симв)")
        except Exception as _e:
            print(f"[SLOT {slot}] DESC error (ignored): {_e}")

elif _desc_mode == 2:
        try:
            print(f"[SLOT {slot}] DESC [шаблон]")
            _tmpl_file = BASE_DIR / "description_template.txt"
            if not _tmpl_file.exists():
                print(f"[SLOT {slot}] DESC: description_template.txt не найден")
            else:
                _tmpl = _tmpl_file.read_text(encoding="utf-8").strip()
                _channel = get_uploader_name(url, source) if "{channel_name}" in _tmpl else ""
                from datetime import datetime as _dt
                _desc_text = (
                    _tmpl
                    .replace("{channel_name}", _channel)
                    .replace("{video_title}", translated_title)
                    .replace("{{НАЗВАНИЕ ИСТОРИИ}}", translated_title)
                    .replace("{title}", translated_title)
                    .replace("{date}", _dt.now().strftime("%d.%m.%Y"))
                ).strip()
                if _desc_text:
                    desc_path = target_path.with_suffix('.desc.txt')
                    desc_path.write_text(_desc_text, encoding="utf-8")
                    print(f"[SLOT {slot}] DESC saved (шаблон): {desc_path.name}")
        except Exception as _e:
            print(f"[SLOT {slot}] DESC error (ignored): {_e}")
    # =========================================================

# === ПОСТОБРАБОТЧИК (разбивка на серии + заставки + обложки) ===
    try:
        import vk_postprocessor as _pp_mod
        _PP_SPLIT      = getattr(_pp_mod, 'ENABLE_SERIES_SPLIT', False)
        _NO_SPLIT_MODE = getattr(_pp_mod, 'NO_SPLIT_MODE', False)
        _PP_WDIR       = _pp_mod.WORK_DIR
        _pp_process    = _pp_mod.process_video
        _pp_available  = True
    except ImportError:
        _PP_SPLIT = _NO_SPLIT_MODE = False
        _pp_available = False

if _pp_available and (_PP_SPLIT or _NO_SPLIT_MODE):
        # Перемещаем исходник из Videos во временную папку, чтобы uploader
        # не увидел сырой файл, пока серии не будут готовы
        _raw_dir = _PP_WDIR / "raw"
        _raw_dir.mkdir(parents=True, exist_ok=True)
        _raw_path = _raw_dir / target_path.name
        shutil.move(str(target_path), str(_raw_path))

# Также перемещаем .desc.txt если есть
        if desc_path and desc_path.exists():
            try:
                shutil.move(str(desc_path), str(_raw_dir / desc_path.name))
            except Exception:
                pass

# Запоминаем видео в Videos/ ДО постобработки (для поиска нового файла после)
        _videos_before_pp = {
            p for p in OUTPUT_DIR.iterdir()
            if p.is_file() and p.suffix.lower() == ".mp4"
        }

print(f"[SLOT {slot}] POSTPROCESS START: {translated_title}")
        _ok = _pp_process(_raw_path, translated_title)
        if not _ok:
            print(f"[SLOT {slot}] POSTPROCESS FAILED: {translated_title}")
            # Убираем сохранённую родную обложку если есть
            if _saved_native_thumb and _saved_native_thumb.exists():
                try:
                    _saved_native_thumb.unlink()
                except Exception:
                    pass
            log_line(FAILED_FILE, url)
            _bump("failed_retries")
            _remember_failed(url)
            return False

print(f"[SLOT {slot}] POSTPROCESS DONE: {translated_title}")

# Видео, которые постобработчик поместил в Videos/ (нужно для обложки/интро)
        _new_videos = sorted(
            {p for p in OUTPUT_DIR.iterdir() if p.is_file() and p.suffix.lower() == ".mp4"}
            - _videos_before_pp
        )

# Заменяем сгенерированные обложки на родную. В split-режиме одна
        # нативная обложка копируется на ВСЕ серии этого видео.
        if native_thumb and _saved_native_thumb and _saved_native_thumb.exists():
            try:
                if _new_videos:
                    # Копируем родную обложку на каждую новую серию
                    for _pp_video in _new_videos:
                        _pp_thumb = _pp_video.with_suffix(THUMB_EXT)
                        if _pp_thumb.exists():
                            _pp_thumb.unlink()
                        shutil.copy2(str(_saved_native_thumb), str(_pp_thumb))
                        print(f"[SLOT {slot}] NATIVE THUMB → {_pp_thumb.name}")
                    # Удаляем исходник после копирования всем
                    _saved_native_thumb.unlink()
                else:
                    # Fallback: постобработчик не добавил новых видео
                    _fallback = OUTPUT_DIR / f"{safe_title}{THUMB_EXT}"
                    if _fallback.exists():
                        _fallback.unlink()
                    _saved_native_thumb.rename(_fallback)
                    print(f"[SLOT {slot}] NATIVE THUMB (fallback) → {_fallback.name}")
            except Exception as _e:
                print(f"[SLOT {slot}] NATIVE THUMB ERROR (ignored): {_e}")
                try:
                    _saved_native_thumb.unlink()
                except Exception:
                    pass

# Переносим desc.txt к каждому новому видео из постобработчика
        if desc_path:
            _raw_desc = _raw_dir / desc_path.name
            if _raw_desc.exists() and _new_videos:
                for _pp_video in _new_videos:
                    try:
                        shutil.copy2(str(_raw_desc), str(_pp_video.with_suffix('.desc.txt')))
                    except Exception:
                        pass
                try:
                    _raw_desc.unlink()
                except Exception:
                    pass

# === ИНТРО (постпроцессор) ===
        if (BASE_DIR / ".use_intro").exists():
            try:
                import subprocess as _sp, tempfile as _tf
                _intro_dir = BASE_DIR / "intro"
                _intros = sorted(_intro_dir.glob("*.mp4")) if _intro_dir.exists() else []
                if _intros and _new_videos:
                    _intro_file = _intros[0]
                    for _pp_video in list(_new_videos):
                        _merged = _pp_video.with_name(_pp_video.stem + "_intro_tmp.mp4")
                        _concat_txt = Path(_tf.mktemp(suffix=".txt"))
                        _concat_txt.write_text(
                            f"file '{_intro_file}'\nfile '{_pp_video}'\n",
                            encoding="utf-8"
                        )
                        _r = _sp.run(
                            ["ffmpeg", "-y", "-f", "concat", "-safe", "0",
                             "-i", str(_concat_txt),
                             "-c:v", "libx264", "-preset", "ultrafast", "-crf", "23",
                             "-c:a", "aac", "-b:a", "128k",
                             str(_merged)],
                            capture_output=True, timeout=3600
                        )
                        _concat_txt.unlink(missing_ok=True)
                        if _r.returncode == 0 and _merged.exists() and _merged.stat().st_size > 1024:
                            _pp_video.unlink()
                            _merged.rename(_pp_video)
                            print(f"[SLOT {slot}] INTRO OK: {_pp_video.name}")
                        else:
                            _err = _r.stderr.decode(errors="ignore")[-200:]
                            print(f"[SLOT {slot}] INTRO ERROR (ignored): {_err}")
                            if _merged.exists():
                                _merged.unlink()
            except Exception as _ie:
                print(f"[SLOT {slot}] INTRO ERROR (ignored): {_ie}")
        # ==============================

# === КАСТОМНАЯ ОБЛОЖКА (постпроцессор) ===
        if (BASE_DIR / ".use_custom_thumb").exists() and _new_videos:
            try:
                import thumbnail_gen as _tg
                for _pp_video in _new_videos:
                    _thumb_out = _pp_video.with_suffix(THUMB_EXT)
                    _ok_ct = _tg.generate(translated_title, str(_thumb_out))
                    if _ok_ct:
                        print(f"[SLOT {slot}] CUSTOM THUMB OK: {_thumb_out.name}")
                    else:
                        print(f"[SLOT {slot}] CUSTOM THUMB SKIP (нет фонов)")
            except Exception as _cte:
                print(f"[SLOT {slot}] CUSTOM THUMB ERROR (ignored): {_cte}")
        # ==========================================

log_line(LOG_FILE, f"{translated_title} (серии) | {url}")
        _append_unique(WHITELIST_FILE, url)
        _bump("downloaded")
        _consecutive_unknown_reset()
        return True
    # ================================================================

# === ИНТРО (если включено) ===
    if (BASE_DIR / ".use_intro").exists():
        try:
            import subprocess as _sp, tempfile as _tf
            _intro_dir = BASE_DIR / "intro"
            _intros = sorted(_intro_dir.glob("*.mp4")) if _intro_dir.exists() else []
            if _intros:
                _intro_file = _intros[0]
                _merged = target_path.with_name(target_path.stem + "_intro_tmp.mp4")
                _concat_txt = Path(_tf.mktemp(suffix=".txt"))
                _concat_txt.write_text(
                    f"file '{_intro_file}'\nfile '{target_path}'\n",
                    encoding="utf-8"
                )
                _r = _sp.run(
                    ["ffmpeg", "-y", "-f", "concat", "-safe", "0",
                     "-i", str(_concat_txt),
                     "-c:v", "libx264", "-preset", "ultrafast", "-crf", "23",
                     "-c:a", "aac", "-b:a", "128k",
                     str(_merged)],
                    capture_output=True, timeout=3600
                )
                _concat_txt.unlink(missing_ok=True)
                if _r.returncode == 0 and _merged.exists() and _merged.stat().st_size > 1024:
                    target_path.unlink()
                    _merged.rename(target_path)
                    print(f"[SLOT {slot}] INTRO OK: {target_path.name}")
                    try:
                        _vdur = float(_sp.run(
                            ["ffprobe", "-v", "error", "-select_streams", "v:0",
                             "-show_entries", "stream=duration", "-of", "csv=p=0", str(target_path)],
                            capture_output=True, text=True, timeout=30
                        ).stdout.strip() or "0")
                        _adur = float(_sp.run(
                            ["ffprobe", "-v", "error", "-select_streams", "a:0",
                             "-show_entries", "stream=duration", "-of", "csv=p=0", str(target_path)],
                            capture_output=True, text=True, timeout=30
                        ).stdout.strip() or "0")
                        _diff = abs(_vdur - _adur)
                        if _diff > 5:
                            print(f"[SLOT {slot}] INTRO WARN: рассинхрон {_diff:.1f}с (видео={_vdur:.1f}с аудио={_adur:.1f}с)")
                        else:
                            print(f"[SLOT {slot}] INTRO SYNC OK: видео={_vdur:.1f}с аудио={_adur:.1f}с diff={_diff:.1f}с")
                    except Exception:
                        pass
                else:
                    _err = _r.stderr.decode(errors="ignore")[-200:]
                    print(f"[SLOT {slot}] INTRO ERROR (ignored): {_err}")
                    if _merged.exists():
                        _merged.unlink()
            else:
                print(f"[SLOT {slot}] INTRO SKIP: нет mp4 в {BASE_DIR / 'intro'}")
        except Exception as _ie:
            print(f"[SLOT {slot}] INTRO ERROR (ignored): {_ie}")
    # ==============================

# === КАСТОМНАЯ ОБЛОЖКА (если включена) ===
    _custom_thumb_ok = False
    if (BASE_DIR / ".use_custom_thumb").exists():
        try:
            import thumbnail_gen as _tg
            _thumb_out = target_path.with_suffix(THUMB_EXT)
            _custom_thumb_ok = _tg.generate(translated_title, _thumb_out)
            if _custom_thumb_ok:
                print(f"[SLOT {slot}] CUSTOM THUMB OK: {_thumb_out.name}")
            else:
                print(f"[SLOT {slot}] CUSTOM THUMB SKIP (нет фонов), fallback на YouTube")
        except Exception as _cte:
            print(f"[SLOT {slot}] CUSTOM THUMB ERROR (ignored): {_cte}")

# === THUMBNAIL (НЕ стопорит процесс) ===
    if not _custom_thumb_ok and DOWNLOAD_THUMBNAILS:
        try:
            if source == "vk":
                # У VK нет фиксированных URL превью — берём через yt-dlp.
                picked = download_vk_thumbnail(
                    url=url,
                    video_id=vid,
                    out_stem=target_path.stem,
                    out_dir=OUTPUT_DIR,
                    ytdlp_exe=YTDLP_EXE,
                    proxy=YTDLP_PROXY,
                )
            else:
                picked = download_best_thumbnail(vid, target_path.stem)
            if picked:
                if picked == "exists":
                    print(f"[SLOT {slot}] THUMB EXISTS: {target_path.stem}{THUMB_EXT}")
                else:
                    print(f"[SLOT {slot}] THUMB OK ({picked}): {target_path.stem}{THUMB_EXT}")
            else:
                print(f"[SLOT {slot}] THUMB SKIP: no thumbnail available for {vid}")
        except Exception:
            # абсолютная гарантия стабильности
            print(f"[SLOT {slot}] THUMB ERROR: ignored (stability mode)")
    # =======================================

print(f"[SLOT {slot}] DONE: {target_path.name}")
    log_line(LOG_FILE, f"{target_path.stem} | {url}")
    _append_unique(WHITELIST_FILE, url)
    _bump("downloaded")
    _consecutive_unknown_reset()
    return True

# ================= MAIN =================
def main():
    _check_env()
    _ensure_utf8_stdout()
    OUTPUT_DIR.mkdir(exist_ok=True)

# Сброс статистики при каждом запуске (модуль в subprocess живёт один прогон,
    # но на всякий случай — если запустят повторно из того же процесса)
    global _consecutive_unknown
    with _stats_lock:
        for _k in _run_stats:
            _run_stats[_k] = 0
        _skipped_titles.clear()
        _failed_urls.clear()
        _consecutive_unknown = 0
    _abort_flag.clear()

# === PROXY ===
    global YTDLP_PROXY
    YTDLP_PROXY = load_proxy_from_file(PROXY_FILE)
    if YTDLP_PROXY:
        print(f"Proxy: {YTDLP_PROXY}")
    else:
        print(f"Proxy: (не задан) — впиши строку в {PROXY_FILE.name} если нужно")
    # ============

parser = argparse.ArgumentParser(description="VK downloader")
    parser.add_argument(
        "--urls-file",
        default=str(URLS_FILE),
        help="Файл со списком ссылок (по умолчанию urls.txt рядом со скриптом)",
    )
    parser.add_argument(
        "--no-split",
        action="store_true",
        default=False,
        help="Не нарезать видео на серии — перемещать целиком в Videos/ (NO_SPLIT_MODE)",
    )
    parser.add_argument(
        "--keep-description",
        action="store_true",
        default=False,
        help="Скачать и сохранить описание ролика рядом с видео (.desc.txt), очищая хештеги и @упоминания",
    )
    parser.add_argument(
        "--native-thumb",
        action="store_true",
        default=False,
        help="Использовать родную обложку YouTube/VK вместо сгенерированной (без сохранения описания)",
    )
    parser.add_argument(
        "--best-quality",
        action="store_true",
        default=False,
        help="Скачать в максимальном качестве без ограничения разрешения (по умолчанию — до 1080p)",
    )
    parser.add_argument(
        "--limit",
        type=int,
        default=0,
        metavar="N",
        help="Скачать только первые N ссылок из очереди (0 = все)",
    )
    parser.add_argument(
        "--random-order",
        action="store_true",
        default=False,
        help="Брать ссылки из urls.txt в случайном порядке (по умолчанию подряд)",
    )
    args = parser.parse_args()

if args.no_split:
        try:
            import vk_postprocessor as _pp_mod
            _pp_mod.NO_SPLIT_MODE = True
        except ImportError:
            pass

urls_path = Path(args.urls_file)
    if urls_path.exists():
        urls = [
            u.strip()
            for u in urls_path.read_text(encoding="utf-8", errors="replace").splitlines()
            if u.strip()
        ]
    else:
        urls = []

# === BLACKLIST: убираем заблокированные ссылки из чанка и из urls.txt ===
    if BLACKLIST_FILE.exists() and urls:
        try:
            blacklisted = {
                u.strip() for u in BLACKLIST_FILE.read_text(encoding="utf-8").splitlines()
                if u.strip() and not u.startswith("#")
            }
            if blacklisted:
                blocked_in_chunk = [u for u in urls if u in blacklisted]
                if blocked_in_chunk:
                    urls = [u for u in urls if u not in blacklisted]
                    print(f"[BLACKLIST] Пропущено из чанка: {len(blocked_in_chunk)} ссылок")
                # Чистим и urls.txt на случай если туда попали blacklisted ссылки
                if URLS_FILE.exists():
                    existing = [
                        u.strip() for u in URLS_FILE.read_text(encoding="utf-8").splitlines()
                        if u.strip()
                    ]
                    cleaned = [u for u in existing if u not in blacklisted]
                    if len(cleaned) < len(existing):
                        URLS_FILE.write_text("\n".join(cleaned) + "\n", encoding="utf-8")
                        print(f"[BLACKLIST] Очищено из urls.txt: {len(existing) - len(cleaned)} ссылок")
        except Exception as _e:
            print(f"[BLACKLIST] Ошибка чтения: {_e}")
    # ========================================================================

# Случайный порядок: перемешиваем ВСЮ очередь, потом берём как обычно
    if args.random_order and urls:
        import random as _random
        _random.shuffle(urls)
        print(f"[ORDER] RANDOM — ссылки перемешаны ({len(urls)} шт.)")

# Лимит считается по УСПЕШНЫМ скачиваниям, а не по попыткам.
    # Ссылки отсеянные по skip.txt / длительности / permanent error выбывают
    # из очереди, но не учитываются как "скачано". Берём новые пока не наберётся
    # args.limit успешных либо очередь не опустеет.
    if args.limit > 0:
        print(f"[LIMIT] Цель: скачать {args.limit} видео (успешных). "
              f"Всего ссылок в очереди: {len(urls)}")

# Firefox cookies нужны только если в очереди есть YouTube-ссылки
    # (или неизвестные, которые откатываются на YouTube-логику).
    # Если все ссылки — VK с VK_USE_COOKIES=False, проверка не нужна.
    needs_cookies = any(_source_uses_cookies(detect_source(u)) for u in urls)
    if needs_cookies:
        profile_dir = Path(FIREFOX_PROFILE_DIR)
        if not (profile_dir.exists() and (profile_dir / "cookies.sqlite").exists()):
            print("ERROR: Не найден Firefox профиль или cookies.sqlite:\n" + str(profile_dir))
            print("Проверь FIREFOX_PROFILE_DIR в .env или config.py")
            return 1
    elif urls:
        print("[INFO] В очереди только VK-ссылки — Firefox-профиль не требуется")

skip_rules = load_skip_rules()
    replace_rules = load_replace_rules()

print(f"Skip rules: {len(skip_rules)} | Replace rules: {len(replace_rules)}")

_init_translator()
    if TRANSLATE_TITLES and not _TRANSLATOR_AVAILABLE:
        print("WARNING: TRANSLATE_TITLES включён, но deep-translator недоступен. Названия переводиться не будут.")

q = deque(urls)
    failed = []

keep_description = args.keep_description
    native_thumb = args.native_thumb
    best_quality = args.best_quality

if best_quality:
        print("[MODE] BEST QUALITY — максимальное качество без ограничения разрешения")

def _limit_reached() -> bool:
        if args.limit <= 0:
            return False
        with _stats_lock:
            return _run_stats.get("downloaded", 0) >= args.limit

with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        running = {}
        for slot in range(1, MAX_WORKERS + 1):
            if not q or _limit_reached() or _abort_flag.is_set():
                break
            u = q.popleft()
            fut = executor.submit(download_one, u, slot, skip_rules, replace_rules, keep_description, native_thumb, best_quality)
            running[fut] = (slot, u)

while running:
            done, _ = wait(running.keys(), return_when=FIRST_COMPLETED)
            for fut in done:
                slot, u = running.pop(fut)
                try:
                    ok = fut.result()
                except Exception as e:
                    print(f"[SLOT {slot}] CRASH: {e}")
                    ok = False

if not ok:
                    failed.append(u)

if q and not _limit_reached() and not _abort_flag.is_set():
                    nu = q.popleft()
                    nfut = executor.submit(download_one, nu, slot, skip_rules, replace_rules, keep_description, native_thumb, best_quality)
                    running[nfut] = (slot, nu)

# Всё что осталось в очереди (после достижения лимита) возвращаем в urls.txt
    _urls_tail = list(q)
    if args.limit > 0 and _urls_tail:
        with _stats_lock:
            _done = _run_stats.get("downloaded", 0)
        print(f"[LIMIT] Скачано {_done}/{args.limit}. В очереди остаётся: {len(_urls_tail)}")

urls_path.write_text("\n".join(failed + _urls_tail), encoding="utf-8")

# === Сохраняем итоговую статистику для controller'а ===
    try:
        import json
        stats_out = {
            "counts":   dict(_run_stats),
            "skipped":  list(_skipped_titles),
            "failed":   list(_failed_urls),
            "total":    sum(_run_stats.values()),
        }
        RUN_STATS_FILE.write_text(
            json.dumps(stats_out, ensure_ascii=False, indent=2),
            encoding="utf-8",
        )
    except Exception as _e:
        print(f"[WARN] Не удалось сохранить {RUN_STATS_FILE.name}: {_e}")

print("=== ГОТОВО ===")
    return 0

if __name__ == "__main__":
    sys.exit(main())

Перезапустить batch_controller после push

Редактор файлов