initial commit

app/ytdlp.py

@@ -0,0 +1,63 @@
+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