From fb273db3b11c4c513ea8d8291780f9a8b464ee8a Mon Sep 17 00:00:00 2001 From: just n Date: Wed, 17 Dec 2025 17:15:59 +0000 Subject: [PATCH] Update ytdlp_handler.py --- ytdlp_handler.py | 19 ------------------- 1 file changed, 19 deletions(-) diff --git a/ytdlp_handler.py b/ytdlp_handler.py index ee00506..aa8cbba 100644 --- a/ytdlp_handler.py +++ b/ytdlp_handler.py @@ -1,10 +1,4 @@ #!/usr/bin/env python3 -""" -YouTube-DLP Integration Module. - -Provides async wrappers for searching and downloading content via yt-dlp. -Includes parsing logic for yt-dlp's JSON output and progress updates. -""" import os import asyncio @@ -18,10 +12,8 @@ import logging logger = logging.getLogger(__name__) - @dataclass class SearchResult: - """Represents a single search result from yt-dlp.""" id: str title: str duration: float @@ -32,7 +24,6 @@ class SearchResult: @property def duration_str(self) -> str: - """Formats duration seconds into HH:MM:SS string.""" try: total_seconds = int(self.duration) if self.duration else 0 @@ -61,10 +52,8 @@ class SearchResult: 'view_count': self.view_count } - @dataclass class DownloadProgress: - """Real-time status of an active download.""" filename: str status: str percent: float = 0.0 @@ -72,9 +61,7 @@ class DownloadProgress: eta: str = "" error: str = "" - class YtDlpHandler: - """Manages yt-dlp subprocesses.""" def __init__( self, @@ -94,7 +81,6 @@ class YtDlpHandler: self._active_downloads: Dict[str, asyncio.subprocess.Process] = {} def _find_ytdlp(self) -> str: - """Locates the yt-dlp binary in the system path.""" for name in ['yt-dlp', 'yt-dlp.exe', 'youtube-dl']: try: result = subprocess.run( @@ -111,7 +97,6 @@ class YtDlpHandler: raise RuntimeError("yt-dlp not found. Install with: pip install yt-dlp") async def search(self, query: str, source: str = "youtube") -> List[SearchResult]: - """Performs a non-download search using yt-dlp's internal search operators.""" search_prefix = { "youtube": "ytsearch", "soundcloud": "scsearch", @@ -187,7 +172,6 @@ class YtDlpHandler: url: str, progress_callback: Optional[Callable[[DownloadProgress], None]] = None ) -> Optional[str]: - """Downloads audio and converts it to Opus.""" output_template = str(self.download_directory / "%(title)s.%(ext)s") cmd = [ @@ -229,7 +213,6 @@ class YtDlpHandler: if progress: progress_callback(progress) - # Parse stdout to find where the file is being saved if '[download] Destination:' in line: output_file = line.split('Destination:', 1)[1].strip() elif 'has already been downloaded' in line: @@ -246,7 +229,6 @@ class YtDlpHandler: if process.returncode == 0: if not output_file or not os.path.exists(output_file): - # Fallback: check most recent file in download dir recent_files = sorted( self.download_directory.glob('*.opus'), key=lambda x: x.stat().st_mtime, @@ -273,7 +255,6 @@ class YtDlpHandler: return None def _parse_progress(self, line: str) -> Optional[DownloadProgress]: - """Parses standard yt-dlp stdout progress lines.""" percent_match = re.search(r'(\d+\.?\d*)%', line) speed_match = re.search(r'at\s+(\S+/s)', line) eta_match = re.search(r'ETA\s+(\S+)', line)