import yt_dlp import asyncio def get_formats(url: str): ydl = yt_dlp.YoutubeDL() info = ydl.extract_info(url, download=False) video_options = [] audio_options = [] for fmt in info['formats']: # Video-only if fmt.get('vcodec') != 'none' and fmt.get('acodec') == 'none': video_options.append({ 'height': fmt.get('height'), 'resolution': fmt.get('resolution'), 'format_id': fmt.get('format_id'), 'ext': fmt.get('ext'), 'tbr': fmt.get('tbr'), }) # Audio-only elif fmt.get('acodec') != 'none' and fmt.get('vcodec') == 'none': audio_options.append({ 'format': fmt.get('format'), 'format_id': fmt.get('format_id'), 'ext': fmt.get('ext'), 'tbr': fmt.get('tbr'), 'language': fmt.get('language') or fmt.get('language_preference'), }) return { 'video_options': video_options, 'audio_options': audio_options, } def download_video(url: str, format: str, out_path: str, temp_path: str, progress_callback, loop): """ Downloads a video and sends a progress callback as it downloads Returns the format used """ progress = {'percent': 0} def hook(d): if d['status'] == 'downloading': total = d.get('total_bytes') or d.get('total_bytes_estimate') downloaded = d.get('downloaded_bytes') if total and downloaded: percent = int(downloaded / total * 100) if percent != progress['percent']: progress['percent'] = percent asyncio.run_coroutine_threadsafe(progress_callback(percent), loop) elif d['status'] == 'finished': asyncio.run_coroutine_threadsafe(progress_callback(100), loop) opts = { 'format': format, 'paths': {'home': out_path, 'temp': temp_path}, 'progress_hooks': [hook] } with yt_dlp.YoutubeDL(opts) as ydl: ydl.download([url]) return format