132 lines
5.2 KiB
Python
132 lines
5.2 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
|
|
|
|
# Import the get_logger function from your custom logger module
|
|
from downloaderyoutube.utils.logger import get_logger
|
|
|
|
# Get a logger instance for this module
|
|
logger = get_logger(__name__)
|
|
|
|
@dataclass
|
|
class VideoFormat:
|
|
"""A simple class to hold structured information about a video format."""
|
|
format_id: str
|
|
resolution: str
|
|
ext: str
|
|
filesize: Optional[int]
|
|
note: str # e.g., "Video Only", "Audio Only", "Progressive"
|
|
|
|
def __str__(self) -> str:
|
|
"""User-friendly string representation for the combobox."""
|
|
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:
|
|
"""
|
|
Handles video operations using yt-dlp in separate threads to avoid
|
|
blocking the GUI.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.progress_callback: Optional[Callable[[int], None]] = None
|
|
self.completion_callback: Optional[Callable[[Optional[str]], None]] = None
|
|
self.formats_callback: Optional[Callable[[Optional[List[VideoFormat]]], None]] = None
|
|
|
|
def _progress_hook(self, d: Dict[str, Any]):
|
|
if d["status"] == "downloading":
|
|
if d.get("total_bytes") and d.get("downloaded_bytes"):
|
|
percentage = (d["downloaded_bytes"] / d["total_bytes"]) * 100
|
|
if self.progress_callback:
|
|
self.progress_callback(int(percentage))
|
|
elif d["status"] == "finished":
|
|
logger.info("yt-dlp finished downloading.")
|
|
|
|
def _get_formats_task(self, url: str):
|
|
"""Task to fetch video formats in a thread."""
|
|
try:
|
|
logger.info(f"Fetching formats for URL: {url}")
|
|
ydl_opts = {"noplaylist": True}
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
info = ydl.extract_info(url, download=False)
|
|
|
|
formats = []
|
|
for f in info.get("formats", []):
|
|
# We are interested in mp4 files that have video
|
|
if f.get("vcodec", "none") != "none" and f.get("ext") == "mp4":
|
|
note = "Progressive" if f.get("acodec", "none") != "none" else "Video Only"
|
|
formats.append(VideoFormat(
|
|
format_id=f["format_id"],
|
|
resolution=f.get("format_note", f.get("resolution", "N/A")),
|
|
ext=f["ext"],
|
|
filesize=f.get("filesize"),
|
|
note=note
|
|
))
|
|
|
|
# Sort formats: progressive first, then by resolution
|
|
formats.sort(key=lambda x: (x.note != "Progressive", -int(x.resolution.replace("p", "")) if x.resolution.replace("p", "").isdigit() else 0))
|
|
|
|
logger.info(f"Found {len(formats)} suitable formats.")
|
|
if self.formats_callback:
|
|
self.formats_callback(formats)
|
|
|
|
except Exception as e:
|
|
error_message = f"Failed to fetch formats: {e}"
|
|
logger.error(error_message)
|
|
logger.debug(traceback.format_exc())
|
|
if self.formats_callback:
|
|
self.formats_callback(None) # Indicate failure
|
|
|
|
def get_video_formats(self, url: str, formats_callback: Callable[[Optional[List[VideoFormat]]], None]):
|
|
"""Starts the format fetching process in a new thread."""
|
|
self.formats_callback = formats_callback
|
|
thread = Thread(target=self._get_formats_task, args=(url,), daemon=True)
|
|
thread.start()
|
|
|
|
def _download_task(self, url: str, download_path: str, format_id: str):
|
|
"""The actual download logic that runs in a separate thread."""
|
|
try:
|
|
logger.info(f"Starting download for URL: {url} with format: {format_id}")
|
|
ydl_opts = {
|
|
"format": format_id,
|
|
"outtmpl": f"{download_path}/%(title)s.%(ext)s",
|
|
"progress_hooks": [self._progress_hook],
|
|
"logger": logger,
|
|
"noplaylist": True,
|
|
}
|
|
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
ydl.download([url])
|
|
|
|
logger.info(f"Successfully downloaded video from URL: {url}")
|
|
if self.completion_callback:
|
|
self.completion_callback(None)
|
|
|
|
except Exception as e:
|
|
error_message = f"An unexpected error occurred: {e}"
|
|
logger.error(error_message)
|
|
logger.debug(traceback.format_exc())
|
|
if self.completion_callback:
|
|
self.completion_callback(error_message)
|
|
|
|
def download_video(
|
|
self,
|
|
url: str,
|
|
download_path: str,
|
|
format_id: str,
|
|
progress_callback: Callable[[int], None],
|
|
completion_callback: Callable[[Optional[str]], None],
|
|
):
|
|
"""Starts the video download in a new thread."""
|
|
self.progress_callback = progress_callback
|
|
self.completion_callback = completion_callback
|
|
thread = Thread(
|
|
target=self._download_task, args=(url, download_path, format_id), daemon=True
|
|
)
|
|
thread.start()
|