112 lines
6.0 KiB
Python
112 lines
6.0 KiB
Python
import logging
|
|
import traceback
|
|
from threading import Thread
|
|
from dataclasses import dataclass
|
|
import yt_dlp
|
|
from typing import Callable, Optional, List, Dict, Any
|
|
from downloaderyoutube.utils.logger import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
@dataclass
|
|
class VideoFormat:
|
|
format_id: str
|
|
resolution: str
|
|
ext: str
|
|
filesize: Optional[int]
|
|
note: str
|
|
|
|
def __str__(self) -> str:
|
|
size_mb = f"{self.filesize / (1024 * 1024):.2f} MB" if self.filesize else "N/A"
|
|
return f"{self.resolution} ({self.ext}) - {size_mb} - {self.note}"
|
|
|
|
class Downloader:
|
|
def __init__(self):
|
|
# Callbacks for single/overall progress
|
|
self.progress_callback: Optional[Callable[[int], None]] = None
|
|
self.completion_callback: Optional[Callable[[Optional[str]], None]] = None
|
|
# Callbacks for single video analysis
|
|
self.formats_callback: Optional[Callable[[Optional[Dict[str, Any]], Optional[List[VideoFormat]]], None]] = None
|
|
# Callbacks for per-video status in a batch
|
|
self.video_started_callback: Optional[Callable[[str], None]] = None
|
|
self.video_completed_callback: Optional[Callable[[str], None]] = None
|
|
self.video_error_callback: Optional[Callable[[str, str], None]] = None
|
|
|
|
def _progress_hook(self, d: Dict[str, Any]):
|
|
if d["status"] == "downloading" and self.progress_callback:
|
|
if d.get("total_bytes") and d.get("downloaded_bytes"):
|
|
self.progress_callback(int((d["downloaded_bytes"] / d["total_bytes"]) * 100))
|
|
|
|
def _get_formats_task(self, url: str):
|
|
try:
|
|
with yt_dlp.YoutubeDL({"noplaylist": True, "quiet": True}) as ydl:
|
|
info = ydl.extract_info(url, download=False)
|
|
video_info = {"title": info.get("title", "N/A"), "uploader": info.get("uploader", "N/A"), "duration_string": info.get("duration_string", "N/A")}
|
|
formats = [VideoFormat(f["format_id"], f.get("format_note", f.get("resolution", "N/A")), f["ext"], f.get("filesize"), "Progressive" if f.get("acodec") != "none" else "Video Only") for f in info.get("formats", []) if f.get("vcodec", "none") != "none" and f.get("ext") == "mp4"]
|
|
formats.sort(key=lambda x: (x.note != "Progressive", -int(x.resolution.replace("p", "")) if x.resolution.replace("p", "").isdigit() else 0))
|
|
if self.formats_callback:
|
|
self.formats_callback(video_info, formats)
|
|
except Exception as e:
|
|
logger.error(f"Failed to fetch formats: {e}")
|
|
if self.formats_callback:
|
|
self.formats_callback(None, None)
|
|
|
|
def get_video_formats(self, url: str, formats_callback: Callable):
|
|
self.formats_callback = formats_callback
|
|
Thread(target=self._get_formats_task, args=(url,), daemon=True).start()
|
|
|
|
def _download_task(self, url: str, download_path: str, format_id: str):
|
|
try:
|
|
ydl_opts = {"format": format_id, "outtmpl": f"{download_path}/%(uploader)s_%(title)s_%(format_note)s.%(ext)s", "progress_hooks": [self._progress_hook], "logger": logger, "noplaylist": True}
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
ydl.download([url])
|
|
if self.completion_callback:
|
|
self.completion_callback(None)
|
|
except Exception as e:
|
|
logger.error(f"An unexpected error occurred: {e}")
|
|
if self.completion_callback:
|
|
self.completion_callback(str(e))
|
|
|
|
def download_video(self, url: str, download_path: str, format_id: str, progress_callback: Callable, completion_callback: Callable):
|
|
self.progress_callback = progress_callback
|
|
self.completion_callback = completion_callback
|
|
Thread(target=self._download_task, args=(url, download_path, format_id), daemon=True).start()
|
|
|
|
def _get_format_from_profile(self, profile: str) -> str:
|
|
if "720p" in profile:
|
|
return "bestvideo[height<=720][ext=mp4]+bestaudio[ext=m4a]/best[height<=720][ext=mp4]/best"
|
|
elif "480p" in profile:
|
|
return "bestvideo[height<=480][ext=mp4]+bestaudio[ext=m4a]/best[height<=480][ext=mp4]/best"
|
|
else:
|
|
return "bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best"
|
|
|
|
def _batch_download_task(self, urls: List[str], download_path: str, quality_profile: str):
|
|
format_string = self._get_format_from_profile(quality_profile)
|
|
logger.info(f"Starting batch download for {len(urls)} videos with profile: {quality_profile}")
|
|
for i, url in enumerate(urls):
|
|
if not url.strip(): continue
|
|
if self.video_started_callback:
|
|
self.video_started_callback(url)
|
|
self.progress_callback(0)
|
|
try:
|
|
ydl_opts = {"format": format_string, "outtmpl": f"{download_path}/%(uploader)s_%(title)s_%(format_note)s.%(ext)s", "progress_hooks": [self._progress_hook], "logger": logger, "noplaylist": True}
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
ydl.download([url])
|
|
if self.video_completed_callback:
|
|
self.video_completed_callback(url)
|
|
except Exception as e:
|
|
error_msg = str(e)
|
|
logger.error(f"Failed to download {url}. Reason: {error_msg}")
|
|
if self.video_error_callback:
|
|
self.video_error_callback(url, error_msg)
|
|
logger.info("--- Batch download finished ---")
|
|
if self.completion_callback:
|
|
self.completion_callback(None)
|
|
|
|
def download_batch(self, urls: List[str], download_path: str, quality_profile: str, progress_callback: Callable, completion_callback: Callable, video_started_callback: Callable, video_completed_callback: Callable, video_error_callback: Callable):
|
|
self.progress_callback = progress_callback
|
|
self.completion_callback = completion_callback
|
|
self.video_started_callback = video_started_callback
|
|
self.video_completed_callback = video_completed_callback
|
|
self.video_error_callback = video_error_callback
|
|
Thread(target=self._batch_download_task, args=(urls, download_path, quality_profile), daemon=True).start() |