Спойлер для обычных пользователей :
Если вы пришли сюда из Яндекса или Google с одной целью — просто и без заморочек скачать видео с Ютуба, сохранить музыку с ВК или вытащить рилс из Инстаграм, — вам не обязательно читать сложный технический код ниже.
Как разработчик, я устал от засилья спам-сайтов и рекламы, поэтому мы написали и запустили своего бесплатного бота. Просто переходите в Telegram, кидайте ему ссылку и забирайте готовый файл (никаких exe-шников, регистраций и водяных знаков).
👉 Ссылка на бесплатного бота: skachaesh_bot 👈
Что умеет наш бот (и для чего он создавался):
YouTube: чтобы вы могли за пару секунд скачать видео с ютуба по ссылке (вплоть до 1080p) или скачать музыку с ютуба в mp3.
ВКонтакте (VK): полная замена глючным сайтам вроде kissvk и savefrom. Бот позволяет скачать видео с вк онлайн бесплатно, вытащить клипы, фильмы, а также скачать песню с вк на телефон.
TikTok: главная киллер-фича — возможность скачать видео с тик тока без водяного знака (без плавающего логотипа). Также бот умеет вытаскивать звуки и музыку из трендовых роликов прямо вам в аудио.
Pinterest & Instagram: чтобы скачать видео с пинтерест в галерею без потери качества и логотипов. А из инсты — скачать рилс, сторис или фото.
Rutube, Boosty, Одноклассники (OK): поддержка отечественных платформ. Юзеры могут скачать видео с рутуба на андроид/пк, вытащить ролик с Одноклассников или даже скачать видео с бусти по ссылке.
Всё это работает быстро и в одном окне. А теперь для коллег-разработчиков: о том, как мы это спроектировали на aiogram 3, почему 90% таких ботов падают на втором юзере, и как мы боролись с блокировкой Event Loop при высоких нагрузках.
Главная проблема: блокирующий IO и смерть Event Loop
Новичок, написавший первого бота на aiogram, делает примерно так:
python
@router.message(F.text.startswith("http")) async def download_handler(message: Message) -> None: url = message.text with YoutubeDL({"outtmpl": "video.mp4"}) as ydl: ydl.download([url]) # 🔥 тут всё и умирает await message.answer_video(FSInputFile("video.mp4"))
На одном пользователе это даже работает. На втором — бот превращается в тыкву.
asyncio работает в одном потоке — это кооперативная многозадачность. А yt-dlp — полностью синхронная библиотека. Под капотом она ходит HTTP-запросами через urllib (блокирующий сокет), парсит страницы, скачивает чанки через блокирующие IO-операции и вызывает ffmpeg через subprocess.
Когда вы вызываете ydl.download([url]) в async def, вы буквально замораживаете Event Loop на всё время скачивания. Представьте: пользователь “А” прислал ссылку, чтобы скачать видео с рутуба весом в 2 ГБ. На ближайшие 10 минут ваш бот перестаёт отвечать абсолютно всем. Вывод: синхронный код в async-хендлере = DoS самого себя.
Архитектурное решение: выносим работу в ThreadPoolExecutor
Из трех классических путей (asyncio.to_thread, кастомный ThreadPoolExecutor, ProcessPoolExecutor) мы берем второй. Если пустить всё в дефолтный пул через to_thread, то при 50 параллельных запросах (например, когда толпа школьников решит скачать музыку с вк бесплатно), у вас улетят в небеса RAM и сеть. Нужен rate limiting.
Сервисный слой DownloaderService
from __future__ import annotations import asyncio import logging from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from pathlib import Path from typing import Any, Callable from yt_dlp import YoutubeDL from yt_dlp.utils import DownloadError logger = logging.getLogger(__name__) @dataclass(slots=True, frozen=True) class DownloadResult: file_path: Path title: str duration: int filesize: int class DownloaderService: """ Всю блокирующую работу выносим в отдельный ThreadPoolExecutor, чтобы не мешать Event Loop обрабатывать новые входящие ссылки. """ def __init__( self, download_dir: Path, max_workers: int = 4, ) -> None: self._download_dir = download_dir self._download_dir.mkdir(parents=True, exist_ok=True) # Ограничиваем количество потоков (rate limit) self._executor = ThreadPoolExecutor( max_workers=max_workers, thread_name_prefix="ytdlp-worker", ) async def download( self, url: str, progress_hook: Callable[[dict[str, Any]], None] | None = None, ) -> DownloadResult: loop = asyncio.get_running_loop() return await loop.run_in_executor( self._executor, self._blocking_download, url, progress_hook, ) def _blocking_download( self, url: str, progress_hook: Callable[[dict[str, Any]], None] | None, ) -> DownloadResult: # Настройки yt-dlp: лучшее mp4 до 1080p + лучшее аудио ydl_opts: dict[str, Any] = { "format": "bv*[height<=1080][ext=mp4]+ba[ext=m4a]/b[ext=mp4]/b", "merge_output_format": "mp4", "outtmpl": str(self._download_dir / "%(id)s.%(ext)s"), "quiet": True, "no_warnings": True, "noprogress": True, "concurrent_fragment_downloads": 4, "retries": 3, } if progress_hook is not None: ydl_opts["progress_hooks"] = [progress_hook] try: with YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(url, download=True) file_path = Path(ydl.prepare_filename(info)) except DownloadError as e: logger.warning("yt-dlp error for %s: %s", url, e) raise return DownloadResult( file_path=file_path, title=info.get("title", "video"), duration=int(info.get("duration") or 0), filesize=file_path.stat().st_size, ) async def shutdown(self) -> None: self._executor.shutdown(wait=True, cancel_futures=False)
Что здесь важно:
Мы не трогаем дефолтный executor (run_in_executor(self._executor, …)). max_workers=4 — пятый пользователь подождёт в очереди, а сервер не ляжет.
Прогресс-бар: как пробросить проценты из потока в корутину
Если юзер качает тяжелый ролик с бусти или пытается скачать фильм с вк, он хочет видеть прогресс-бар: 45% → 72% → 89%.
yt-dlp даёт синхронный коллбэк progress_hooks, который дёргается из рабочего потока. Вызывать оттуда await message.edit_text нельзя. Решение — asyncio.run_coroutine_threadsafe плюс жесткий троттлинг (не чаще раза в 2 секунды), иначе Telegram выдаст 429 Too Many Requests (флуд-бан).
Класс ProgressReporter
class ProgressReporter: __slots__ = ("_message", "_loop", "_min_interval", "_last_update", "_last_text") def __init__( self, message: Message, loop: asyncio.AbstractEventLoop, min_interval: float = 2.0, ) -> None: self._message = message self._loop = loop self._min_interval = min_interval self._last_update: float = 0.0 self._last_text: str = "" def __call__(self, d: dict[str, Any]) -> None: status = d.get("status") if status == "downloading": text = self._format_downloading(d) elif status == "finished": text = "🔧 Обработка (ffmpeg)..." else: return now = time.monotonic() if now - self._last_update < self._min_interval: return if text == self._last_text: return self._last_update = now self._last_text = text # Планируем корутину в чужой event loop — thread-safe asyncio.run_coroutine_threadsafe(self._safe_edit(text), self._loop) async def _safe_edit(self, text: str) -> None: try: await self._message.edit_text(text) except Exception: # Игнорируем TelegramBadRequest, если сообщение не изменилось pass @staticmethod def _format_downloading(d: dict[str, Any]) -> str: total = d.get("total_bytes") or d.get("total_bytes_estimate") or 0 downloaded = d.get("downloaded_bytes") or 0 speed = d.get("speed") or 0 if not total: return f"⬇️ Скачано {downloaded / 1024 / 1024:.1f} MB" percent = downloaded / total * 100 bar_len = 20 filled = int(bar_len * percent / 100) bar = "█" * filled + "░" * (bar_len - filled) speed_mb = speed / 1024 / 1024 if speed else 0 return ( f"⬇️ Загрузка\n" f"<code>[{bar}] {percent:.1f}%</code>\n" f"Скорость: {speed_mb:.2f} MB/s" )
Подводные камни и деплой
Пара моментов для тех, кто будет делать свои форки:
Лимит Telegram API: У ботов лимит на отправку файла — 50 МБ. Решается поднятием self-hosted Bot API (лимит расширяется до 2 ГБ).
FFmpeg: Обязан быть в PATH. Без него merge_output_format не сработает, и вместо того, чтобы, например, скачать видео с ютуба хорошего качества со звуком, вы получите отдельно немую картинку .mp4 и аудиофайл .m4a.
Очистка диска: Обязательно делайте .unlink() файлу в блоке finally вашего хендлера. Иначе сервер забьется кэшем за сутки.
Всё описанное крутится на скромной VPS, пережило апдейты платформ и мои DDoS-эксперименты. Архитектура надежна: Event Loop свободен, потоки работают автономно.
Можете потыкать вживую, посмотреть, как работает и проверить скорость скачивания — проект живёт тут: @skachaesh_bot.
Если статья зашла — пишите в комментариях, про что ещё покопать: отдельно про self-hosted Telegram Bot API и обход лимита 50 МБ, про FSM и очереди задач на Redis, или про то, как прикрутить сюда Celery и в какой момент это становится оверинжинирингом.
