import asyncio import os import uuid import yt_dlp from typing import Optional, Tuple from .base import Downloader from bot.utils import run_blocking class YouTubeDownloader(Downloader): """Загрузчик для YouTube""" def is_supported_url(self, url: str) -> bool: return "youtube.com" in url or "youtu.be" in url async def download_audio(self, url: str, config: dict) -> Tuple[Optional[str], Optional[str], Optional[int]]: """Скачивает аудио с YouTube""" # Настройки yt-dlp ydl_opts = { 'format': 'bestaudio/best', 'outtmpl': os.path.join(config['tmp_dir'], f'%(id)s.%(ext)s'), 'postprocessors': [{ 'key': 'FFmpegExtractAudio', 'preferredcodec': 'mp3', 'preferredquality': str(config.get('audio_quality', 192)), }], 'quiet': True, 'no_warnings': True, 'proxy': config.get('proxy'), 'max_filesize': 50 * 1024 * 1024, # 100MB 'noplaylist': True, } # Ограничение длительности if max_duration := config.get('max_duration'): ydl_opts['match_filter'] = yt_dlp.match_filter_func( f"duration < {max_duration}" ) try: # Запускаем в отдельном процессе result = await run_blocking(self._download, url, ydl_opts) return result except Exception as e: return None, str(e) def _download(self, url: str, opts: dict) -> Tuple[str, str, int]: """Синхронная загрузка (выполняется в отдельном процессе)""" with yt_dlp.YoutubeDL(opts) as ydl: info = ydl.extract_info(url, download=True) filename = ydl.prepare_filename(info) base, _ = os.path.splitext(filename) mp3_path = base + '.mp3' # Получаем название title = info.get('title', 'audio') or 'audio' clean_title = self.sanitize_filename(title) # Получаем длительность в миллисекундах duration_ms = int(info.get('duration', 0) * 1000) return mp3_path, clean_title, duration_ms