63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
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' and fmt.get('__needs_testing') == 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' and fmt.get('__needs_testing') == 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 |