import gc import os import re import shlex # For safely parsing command arguments import signal import subprocess # For direct CLI command execution import sys import time from pathlib import Path from typing import Optional, List, Set from PySide6.QtCore import QObject, QThread, Signal from .ytsage_yt_dlp import get_yt_dlp_path from ..utils.ytsage_constants import ( SUBPROCESS_CREATIONFLAGS, VIDEO_EXTENSIONS, AUDIO_EXTENSIONS, SUBTITLE_EXTENSIONS, MEDIA_EXTENSIONS, ) from ..utils.ytsage_localization import LocalizationManager from ..utils.ytsage_logger import logger # Shorthand for localization _ = LocalizationManager.get_text class SignalManager(QObject): update_formats = Signal(list) update_status = Signal(str) update_progress = Signal(float) playlist_info_label_visible = Signal(bool) playlist_info_label_text = Signal(str) selected_subs_label_text = Signal(str) playlist_select_btn_visible = Signal(bool) playlist_select_btn_text = Signal(str) class DownloadThread(QThread): progress_signal = Signal(float) status_signal = Signal(str) finished_signal = Signal() error_signal = Signal(str) file_exists_signal = Signal(str) # New signal for file existence update_details = Signal(str) # New signal for filename, speed, ETA def __init__( self, url, path, format_id, is_audio_only=False, format_has_audio=False, subtitle_langs=None, is_playlist=False, merge_subs=False, enable_sponsorblock=False, sponsorblock_categories=None, resolution="", playlist_items=None, save_description=False, embed_chapters=False, cookie_file=None, browser_cookies=None, rate_limit=None, download_section=None, force_keyframes=False, proxy_url=None, geo_proxy_url=None, force_output_format=False, preferred_output_format="mp4", force_audio_format=False, preferred_audio_format="best", audio_normalization=False, filename_format=None, concurrent_fragments=1, ) -> None: super().__init__() self.url = url self.path = Path(path) self.format_id = format_id self.is_audio_only = is_audio_only self.format_has_audio = format_has_audio self.subtitle_langs = subtitle_langs if subtitle_langs else [] self.is_playlist = is_playlist self.merge_subs = merge_subs self.enable_sponsorblock = enable_sponsorblock self.sponsorblock_categories = sponsorblock_categories if sponsorblock_categories else ["sponsor"] self.resolution = resolution self.playlist_items = playlist_items self.save_description = save_description self.embed_chapters = embed_chapters self.cookie_file = cookie_file self.browser_cookies = browser_cookies self.rate_limit = rate_limit self.download_section = download_section self.force_keyframes = force_keyframes self.proxy_url = proxy_url self.geo_proxy_url = geo_proxy_url self.force_output_format = force_output_format self.preferred_output_format = preferred_output_format self.force_audio_format = force_audio_format self.preferred_audio_format = preferred_audio_format self.audio_normalization = audio_normalization self.filename_format = filename_format self.concurrent_fragments = concurrent_fragments self.paused: bool = False self.cancelled: bool = False self.process: Optional[subprocess.Popen] = None self.current_filename: Optional[str] = None # Initialize filename storage self.last_file_path: Optional[str] = None # Initialize full file path storage self.subtitle_files: List[str] = [] # Track subtitle files that are created self.initial_subtitle_files: Set[Path] = set() # Track initial subtitle files before download self.download_files: Set[Path] = set() # Every destination path this download wrote to self.expected_phases: int = 1 # 2 when video and audio download separately before merge self._media_phase: int = 0 # Index of the media stream currently downloading def cleanup_partial_files(self) -> None: """Delete partial files (.part/.ytdl and unmerged .fNNN. streams), but only those belonging to destinations this download actually wrote — the download directory may contain unrelated files from other applications.""" try: pattern = re.compile(r"\.f\d+\.") # Pattern to match format codes like .f243. for dest in self.download_files: candidates = [dest.with_name(dest.name + ".part"), dest.with_name(dest.name + ".ytdl")] if dest.suffix == ".part" or pattern.search(dest.name): candidates.append(dest) for file_path in candidates: if file_path.exists(): self._safe_delete_with_retry(file_path) except Exception as e: logger.exception(f"Error cleaning partial files: {e}") def _safe_delete_with_retry(self, file_path: Path, max_retries: int = 5, delay: float = 2.0) -> None: """Safely delete a file with retry mechanism for file locking issues across platforms""" for attempt in range(max_retries): try: # Force garbage collection to release any Python-held file handles gc.collect() if file_path.exists(): file_path.unlink(missing_ok=True) logger.info(f"Successfully deleted {file_path.name}") return except PermissionError as e: if attempt < max_retries - 1: logger.warning(f"File {file_path.name} is locked, retrying in {delay} seconds... (attempt {attempt + 1}/{max_retries})") time.sleep(delay) delay = min(delay * 1.5, 5.0) # Exponential backoff, capped at 5 seconds else: logger.error(f"Failed to delete {file_path.name} after {max_retries} attempts: {e}") return except Exception as e: logger.error(f"Error deleting {file_path.name}: {e}") return def _terminate_process_tree(self, process: subprocess.Popen) -> None: """Terminate a process and all its children across platforms""" pid = process.pid try: if sys.platform == "win32": # Windows: Use taskkill to kill the entire process tree # /T = kill child processes, /F = force kill # Use subprocess.run with no encoding to avoid codec issues subprocess.run( ["taskkill", "/F", "/T", "/PID", str(pid)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, creationflags=SUBPROCESS_CREATIONFLAGS, ) logger.debug(f"Killed process tree on Windows (PID: {pid})") else: # Unix-like systems: Kill the process group try: # Try to kill the process group os.killpg(os.getpgid(pid), signal.SIGTERM) time.sleep(0.5) # Force kill if still running os.killpg(os.getpgid(pid), signal.SIGKILL) except (ProcessLookupError, PermissionError): # Process already terminated or no permission pass logger.debug(f"Killed process group on Unix (PID: {pid})") except Exception as e: logger.warning(f"Error killing process tree: {e}") # Fallback to standard termination try: process.terminate() process.wait(timeout=2) except Exception: try: process.kill() process.wait() except Exception: pass # Ensure process is waited on to avoid zombies try: process.wait(timeout=3) except Exception: pass def cleanup_subtitle_files(self) -> None: """Delete subtitle files after they have been merged into the video file""" deleted_count: List[int] = [0, 0] def safe_delete(path: Path) -> bool: try: # Check if file exists before trying to delete if path.exists(): path.unlink(missing_ok=True) logger.debug(f"Deleted subtitle file: {path.name}") return True return False except Exception as e: logger.exception(f"Error deleting subtitle file {path}: {e}") return False try: # --- Method 1: Delete tracked subtitle files --- for f in self.subtitle_files or []: deleted_count[0] += safe_delete(path=Path(f)) else: logger.debug(f"Deleted {deleted_count[0]} of {len(self.subtitle_files)} tracked subtitle files") # --- Method 2: Delete new subtitle files not in initial set --- # Only touch subtitles belonging to files this download wrote; the # directory may contain subtitle files from other processes. download_stems = {p.stem for p in self.download_files} | {Path(f).stem for f in self.subtitle_files or []} new_subtitle_files: Set[Path] = { f for f in Path(self.path).rglob("*") if f.suffix in [".vtt", ".srt"] and f not in self.initial_subtitle_files and any(f.name.startswith(stem) for stem in download_stems if stem) } for subtitle_file in new_subtitle_files: deleted_count[1] += safe_delete(path=subtitle_file) else: logger.debug(f"Deleted {deleted_count[1]} of {len(new_subtitle_files)} new subtitle files") except Exception as e: logger.exception(f"Error cleaning subtitle files: {e}") def _build_yt_dlp_command(self) -> List[str]: """Build the yt-dlp command line with all options for direct execution.""" yt_dlp_path: str = get_yt_dlp_path() if str(yt_dlp_path) == "yt-dlp": # Sentinel: no managed binary and no opted-in system binary. # Never exec a bare command name from PATH. raise FileNotFoundError("yt-dlp is not installed - run the yt-dlp setup first") # Build the command line array cmd: List[str] = [str(yt_dlp_path)] logger.debug(f"Using yt-dlp from: {yt_dlp_path}") # Add concurrent fragments setting if self.concurrent_fragments: cmd.extend(["-N", str(self.concurrent_fragments)]) logger.debug(f"Using {self.concurrent_fragments} concurrent connections") # Format selection strategy - use format ID if provided or fallback to resolution if self.is_playlist: # For playlists, specific format_id from the first video often fails for subsequent videos. # Instead, we rely on dynamic fallback/resolution limits. if self.is_audio_only: # For audio-only playlist, let yt-dlp pick best audio. cmd.extend(["-f", "bestaudio/best"]) logger.debug(f"Playlist mode: using dynamic best audio fallback instead of format_id") else: # If a specific resolution is given, limit to it. Otherwise, select the overall best. # The resolution might be e.g. "1920x1080" or "1080". We want the height. try: if self.resolution and self.resolution != "default": res_str = str(self.resolution) h = min(map(int, res_str.split('x'))) if 'x' in res_str else int(res_str) cmd.extend(["-S", f"res:{h}"]) logger.debug(f"Playlist mode: using resolution limiter -S res:{h}") else: cmd.extend(["-f", "bestvideo+bestaudio/best"]) logger.debug("Playlist mode: using dynamic best quality overall") except ValueError: cmd.extend(["-f", "bestvideo+bestaudio/best"]) logger.debug("Playlist mode: invalid resolution string, using dynamic best quality overall") elif self.format_id: clean_format_id: str = self.format_id.split("-drc")[0] if "-drc" in self.format_id else self.format_id # If the selected format is audio-only, pass it directly. if self.is_audio_only: cmd.extend(["-f", clean_format_id]) logger.debug(f"Using audio-only format selection: {clean_format_id}") # If the selected format already includes an audio track (progressive), no merge needed. elif self.format_has_audio: cmd.extend(["-f", clean_format_id]) logger.debug(f"Using progressive format with bundled audio: {clean_format_id}") else: cmd.extend(["-f", f"{clean_format_id}+bestaudio/best"]) self.expected_phases = 2 # separate video and audio downloads logger.debug(f"Using video-only format merged with best audio: {clean_format_id}+bestaudio/best") else: # If no specific format ID, use resolution-based sorting (-S) res_value: str = self.resolution if self.resolution else "720" # Default to 720p if no resolution specified cmd.extend(["-S", f"res:{res_value}"]) # Force output format if enabled and merging is needed (for video) if self.force_output_format and not self.is_audio_only: if self.format_has_audio: # Progressive format (video with audio) - use remux to convert container cmd.extend(["--remux-video", self.preferred_output_format]) logger.debug(f"Using --remux-video to force progressive format to: {self.preferred_output_format}") else: # Merging video+audio - force merge output format cmd.extend(["--merge-output-format", self.preferred_output_format]) logger.debug(f"Using --merge-output-format to force merged format to: {self.preferred_output_format}") # Force audio format conversion for audio-only downloads if self.is_audio_only and self.force_audio_format: cmd.append("--extract-audio") if self.preferred_audio_format and self.preferred_audio_format != "best": cmd.extend(["--audio-format", self.preferred_audio_format]) logger.debug(f"Using --extract-audio with --audio-format {self.preferred_audio_format} for audio-only download") else: logger.debug("Using --extract-audio with best quality (no conversion) for audio-only download") # Add Audio Normalization if enabled (only applies to audio-only downloads) if self.audio_normalization and self.is_audio_only: # Normalization using FFmpeg filters requires re-encoding the audio stream. # If the user selected "Best (No conversion)", yt-dlp attempts to stream copy (-c:a copy), # which will cause FFmpeg to crash with "Invalid argument". # We fix this by forcing an explicit actual conversion (mp3) if no format was forced. if not self.force_audio_format or self.preferred_audio_format == "best": if "--extract-audio" not in cmd: cmd.append("--extract-audio") cmd.extend(["--audio-format", "mp3"]) logger.debug("Forced audio format to mp3 since normalization requires re-encoding") # Scope the argument specifically to ExtractAudio so it doesn't conflict with other PPs cmd.extend(["--postprocessor-args", "ExtractAudio:-af loudnorm=I=-16:LRA=11:TP=-1.5"]) logger.debug("Added Audio Normalization (--postprocessor-args ExtractAudio:-af loudnorm=...)") # Output template with resolution in filename # Use string concatenation instead of Path.joinpath to avoid Path object issues base_path: str = self.path.as_posix() # Determine the filename part of the template filename_part = self.filename_format if self.filename_format else "%(title)s_%(resolution)s_[%(id)s].%(ext)s" if self.is_playlist: # Create output template with playlist subfolder output_template: str = f"{base_path}/%(playlist_title)s/{filename_part}" else: # For single files, automatically ignore/remove playlist-specific preamble (like "%(playlist_index)s - ") import re filename_part = re.sub(r'%\(playlist_index[^)]*\)[a-zA-Z0-9]*\s*(?:[-_]\s*)?', '', filename_part) output_template: str = f"{base_path}/{filename_part}" cmd.extend(["-o", str(output_template)]) # Add common options cmd.append("--force-overwrites") # Add playlist items if specified if self.is_playlist and self.playlist_items: cmd.extend(["--playlist-items", self.playlist_items]) # Add subtitle options if subtitles are selected if self.subtitle_langs: # Subtitles work with both audio-only and video formats # For audio-only formats, subtitles will be downloaded as separate files cmd.append("--write-subs") # Get language codes from subtitle selections lang_codes: List[str] = [] has_auto_generated = False for sub_selection in self.subtitle_langs: try: # Extract just the language code (e.g., 'en' from 'en - Manual') lang_code = sub_selection.split(" - ")[0] lang_codes.append(lang_code) if "Auto-generated" in sub_selection: has_auto_generated = True except Exception as e: logger.exception(f"Could not parse subtitle selection '{sub_selection}': {e}") if lang_codes: cmd.extend(["--sub-langs", ",".join(lang_codes)]) if has_auto_generated: cmd.append("--write-auto-subs") # Include auto-generated subtitles # Only embed subtitles if merge is enabled if self.merge_subs: cmd.append("--embed-subs") # Add SponsorBlock if enabled if self.enable_sponsorblock and self.sponsorblock_categories: cmd.append("--sponsorblock-remove") cmd.append(",".join(self.sponsorblock_categories)) # Add description saving if enabled if self.save_description: cmd.append("--write-description") # Add chapters embedding if enabled if self.embed_chapters: cmd.append("--embed-chapters") # Add cookies if specified if self.cookie_file: cmd.extend(["--cookies", str(self.cookie_file)]) elif self.browser_cookies: cmd.extend(["--cookies-from-browser", self.browser_cookies]) # Add proxy settings if specified if self.proxy_url: cmd.extend(["--proxy", self.proxy_url]) if self.geo_proxy_url: cmd.extend(["--geo-verification-proxy", self.geo_proxy_url]) # Add rate limit if specified if self.rate_limit: cmd.extend(["-r", self.rate_limit]) # Add download section if specified if self.download_section: cmd.extend(["--download-sections", self.download_section]) # Add force keyframes option if enabled if self.force_keyframes: cmd.append("--force-keyframes-at-cuts") logger.debug(f"Added download section: {self.download_section}, Force keyframes: {self.force_keyframes}") # Add the URL as the final argument if self.is_playlist: cmd.append("--ignore-errors") cmd.append("--no-abort-on-error") cmd.append(self.url) return cmd def run(self) -> None: try: logger.debug("Starting download thread") # Get initial list of subtitle files to compare later self.initial_subtitle_files = set() if self.merge_subs: try: # Scan for existing subtitle files in the directory for file in self.path.rglob("*"): if file.suffix in {".vtt", ".srt"}: self.initial_subtitle_files.add(file) logger.debug(f"Found {len(self.initial_subtitle_files)} existing subtitle files before download") except Exception as e: logger.exception(f"Error scanning for initial subtitle files: {e}") # Use direct CLI command self._run_direct_command() except Exception as e: # Catch errors during setup logger.critical(f"Critical error in download thread: {e}", exc_info=True) self.error_signal.emit(f"Critical error in download thread: {e}") def _run_direct_command(self) -> None: """Run yt-dlp as a direct command line process instead of using Python API.""" try: self.error_lines = [] # Initialize error capture list cmd: List[str] = self._build_yt_dlp_command() cmd_str: str = " ".join(shlex.quote(str(arg)) for arg in cmd) logger.debug(f"Executing command: {cmd_str}") self.status_signal.emit(_("download.starting")) self.progress_signal.emit(0) # Start the process # Extra logic moved to src\utils\ytsage_constants.py # Use start_new_session on Unix to enable process group termination popen_kwargs = { "stdout": subprocess.PIPE, "stderr": subprocess.STDOUT, "bufsize": 1, # Line buffered "encoding": "utf-8", "errors": "replace", } if sys.platform == "win32": popen_kwargs["creationflags"] = SUBPROCESS_CREATIONFLAGS else: # On Unix, start a new session so we can kill the entire process group popen_kwargs["start_new_session"] = True self.process = subprocess.Popen(cmd, **popen_kwargs) # Process output line by line to update progress for line in iter(self.process.stdout.readline, ""): # type: ignore if self.cancelled: # Kill the entire process tree (yt-dlp + ffmpeg children) self._terminate_process_tree(self.process) # Add delay before cleanup to allow file handles to be released time.sleep(2) self.cleanup_partial_files() self.status_signal.emit(_("download.cancelled")) self.finished_signal.emit() return # Wait if paused while self.paused and not self.cancelled: time.sleep(0.1) # Parse the line for download progress and status updates self._parse_output_line(line) # Wait for process to complete return_code: int = self.process.wait() # Special handling for specific errors # return code 127 typically means command not found if return_code == 127: self.error_signal.emit( _("errors.ytdlp_not_found_path") ) return if return_code == 0 or (self.is_playlist and return_code != 0 and self.current_filename is not None): self.progress_signal.emit(100) # Robust file finding: Always search for the most recent file # This handles all post-processing scenarios (merging, remuxing, subtitle embedding, etc.) final_file_found = False try: # First, check if last_file_path exists and is valid if self.last_file_path: last_path = Path(self.last_file_path) if last_path.exists() and last_path.is_file(): # File exists at the tracked path self.current_filename = last_path.name final_file_found = True logger.info(f"Found file at tracked path: {self.last_file_path}") # If not found at tracked path, search for the most recent file if not final_file_found: logger.info("Searching for most recent downloaded file...") potential_files = [] # Search in download directory and subdirectories (for playlists) for ext in MEDIA_EXTENSIONS: potential_files.extend(self.path.glob(f'*{ext}')) # Also check subdirectories (for playlist downloads) potential_files.extend(self.path.glob(f'*/*{ext}')) if potential_files: # Sort by modification time and get the most recent most_recent = max(potential_files, key=lambda p: p.stat().st_mtime) # Verify it was modified recently (within last 30 seconds to account for post-processing) time_since_modification = time.time() - most_recent.stat().st_mtime if time_since_modification < 30: self.last_file_path = str(most_recent) self.current_filename = most_recent.name final_file_found = True logger.info(f"Found most recent file (modified {time_since_modification:.1f}s ago): {self.last_file_path}") else: logger.warning(f"Most recent file is too old ({time_since_modification:.1f}s), might not be the right one") else: logger.warning("No video/audio files found in download directory") except Exception as e: logger.error(f"Error finding final file: {e}", exc_info=True) # Set completion status if return_code != 0: self.status_signal.emit(_("download.completed") + " (with some errors)") else: self.status_signal.emit(_("download.completed")) # Clean up subtitle files if they were merged, with a small delay # to ensure the embedding process has completed if self.merge_subs: # Add a significant delay to ensure ffmpeg has released all file handles # and any post-processing is complete self.status_signal.emit(_("download.completed_cleaning")) time.sleep(3) # Increased delay to 3 seconds self.cleanup_subtitle_files() self.finished_signal.emit() else: # Check if it was cancelled if self.cancelled: self.status_signal.emit(_("download.cancelled")) self.finished_signal.emit() else: # Provide informative error message based on captured output if self.error_lines: # Use the captured error lines (last 2 for context) error_msg = "\n".join(self.error_lines[-2:]) self.error_signal.emit( _("errors.ytdlp_failed", error=error_msg) ) else: # Fallback to generic return code error self.error_signal.emit( _("errors.download_failed_return_code", return_code=return_code) ) # Add delay before cleanup to allow file handles to be released time.sleep(1) self.cleanup_partial_files() except Exception as e: logger.exception(f"Error in direct command: {e}") self.error_signal.emit(_("errors.direct_command_error", error=str(e))) # Add delay before cleanup to allow file handles to be released time.sleep(1) self.cleanup_partial_files() def _parse_output_line(self, line: str) -> None: """Parse yt-dlp command output to update progress and status.""" line = line.strip() # logger.info(f"yt-dlp: {line}") # Log all output - OPTIONALLY UNCOMMENT FOR VERBOSE DEBUG # Capture error lines if "ERROR:" in line: if hasattr(self, 'error_lines'): self.error_lines.append(line) # Extract filename when the destination line appears # Use a slightly more robust regex looking for the start of the line dest_match = re.search(r"^\[download\] Destination:\s*(.*)", line) if dest_match: try: filepath = dest_match.group(1).strip() self.current_filename = Path(filepath).name self.last_file_path = filepath # Store the full path for later cleanup if Path(filepath) not in self.download_files and Path(filepath).suffix.lower() not in SUBTITLE_EXTENSIONS: self._media_phase += 1 self.download_files.add(Path(filepath)) logger.debug(f"Extracted filename: {self.current_filename}") # DEBUG # Check if this is an audio-only download by looking in the previous lines is_audio_download = False # Look for audio format indicators in the current line or preceding output # yt-dlp typically mentions format like "Downloading format 251 - audio only" if " - audio only" in line: is_audio_download = True # Check if the format ID is mentioned earlier in the line format_match = re.search(r"Downloading format (\d+)", line) if format_match: format_id = format_match.group(1) logger.debug(f"Detected format ID: {format_id}") # Format IDs for audio typically have different patterns # (like 140, 251 for audio vs 137, 248 for video) # This is just a heuristic since format IDs can vary # Determine file type based on extension and context ext = Path(self.current_filename).suffix.lower() # Check if this is explicitly an audio stream download if is_audio_download or "Downloading audio" in line: self.status_signal.emit(_("download.downloading_audio")) # Video file extensions with likely video content elif ext in VIDEO_EXTENSIONS: self.status_signal.emit(_("download.downloading_video")) # Audio file extensions elif ext in AUDIO_EXTENSIONS: self.status_signal.emit(_("download.downloading_audio")) # Subtitle file extensions elif ext in SUBTITLE_EXTENSIONS: self.status_signal.emit(_("download.downloading_subtitle")) # Default case else: self.status_signal.emit(_("download.downloading")) except Exception as e: logger.exception(f"Error extracting filename from line '{line}': {e}") self.status_signal.emit(_("download.downloading_fallback")) # Fallback status return # Don't process this line further for speed/ETA # Check for specific download types in the output if "Downloading video" in line: self.status_signal.emit(_("download.downloading_video")) return elif "Downloading audio" in line: self.status_signal.emit(_("download.downloading_audio")) return # Detect subtitle file creation # Look for lines like "[info] Writing video subtitles to: filename.xx.vtt" subtitle_match = re.search( r"(?:Writing|Downloading) (?:video )?subtitles.*?(?:to|:)\s*(.+\.(?:vtt|srt))(?:\s|$)", line, re.IGNORECASE, ) if subtitle_match: subtitle_file = subtitle_match.group(1).strip() # Clean up the path - remove any duplicated directory paths # Sometimes yt-dlp output contains malformed paths like "dir: dir/file" if ":" in subtitle_file and os.name == "nt": # Windows paths # Look for pattern like "C:\path: C:\path\file" and extract the latter colon_parts = subtitle_file.split(": ") if len(colon_parts) > 1: # Take the last part which should be the actual file path subtitle_file = colon_parts[-1].strip() # Show subtitle download message self.status_signal.emit(_("download.downloading_subtitle")) # Store the subtitle file path for later deletion if merging is enabled if self.merge_subs: subtitle_path = Path(subtitle_file) if not subtitle_path.is_absolute(): # If it's a relative path, make it absolute based on current path subtitle_path = self.path.joinpath(subtitle_file) self.subtitle_files.append(str(subtitle_path)) logger.debug(f"Tracking subtitle file for later cleanup: {subtitle_path}") return # Send status updates based on output line content if "Downloading webpage" in line or "Extracting URL" in line: self.status_signal.emit(_("download.fetching_info")) self.progress_signal.emit(0) elif "[download] Destination:" in line: # Extract the destination filename match = re.search(r"Destination: (.+)", line) if match: dest_path = match.group(1).strip() self.current_filename = Path(dest_path).name self.last_file_path = dest_path self.download_files.add(Path(dest_path)) logger.debug(f"Captured destination filename: {self.current_filename}") elif "Downloading API JSON" in line: self.status_signal.emit(_("download.processing_playlist")) self.progress_signal.emit(0) elif "Downloading m3u8 information" in line: self.status_signal.emit(_("download.preparing_streams")) self.progress_signal.emit(0) elif "[download] Downloading video " in line: self.status_signal.emit(_("download.downloading_video")) elif "[download] Downloading audio " in line: self.status_signal.emit(_("download.downloading_audio")) elif "Downloading format" in line: # Try to detect if it's audio or video format if " - audio only" in line: self.status_signal.emit(_("download.downloading_audio")) elif " - video only" in line: self.status_signal.emit(_("download.downloading_video")) else: # Don't emit generic message - format is unclear pass # Look for download percentage percent_match = re.search(r"(\d+\.\d+)%", line) if percent_match: try: percent = float(percent_match.group(1)) # When video and audio download as separate streams, scale each # phase into its share of the bar instead of jumping 0-100 twice if self.expected_phases > 1 and not self.is_playlist: completed = max(0, min(self._media_phase - 1, self.expected_phases - 1)) percent = (completed * 100.0 + percent) / self.expected_phases self.progress_signal.emit(percent) except (ValueError, IndexError): pass # Check for download speed and ETA if "[download]" in line and "%" in line: # Try to extract more detailed status info try: # Look for speed speed_match = re.search(r"at\s+(\d+\.\d+[KMG]iB/s)", line) speed_str = speed_match.group(1) if speed_match else "N/A" # Look for ETA eta_match = re.search(r"ETA\s+(\d+:\d+)", line) eta_str = eta_match.group(1) if eta_match else "N/A" # Simplify status message to only show the speed and ETA status = f"{_('download.speed')}: {speed_str} | {_('download.eta')}: {eta_str}" self.update_details.emit(status) except Exception as e: # If parsing fails, just show basic status (maybe log the error) logger.exception(f"Error parsing download details line: {line} -> {e}") pass # Keep basic status emission below if needed, or emit generic details # Check for post-processing if "[Merger]" in line or "Merging formats" in line: self.status_signal.emit(_("download.merging_formats")) self.progress_signal.emit(95) # Extract the merged output filename merger_match = re.search(r"Merging formats into \"(.+?)\"", line) if merger_match: merged_filepath = merger_match.group(1).strip() self.current_filename = Path(merged_filepath).name self.last_file_path = merged_filepath self.download_files.add(Path(merged_filepath)) logger.debug(f"Updated to merged filename: {self.current_filename}") elif "SponsorBlock" in line: self.status_signal.emit(_("download.removing_sponsor_segments")) self.progress_signal.emit(97) elif "Deleting original file" in line: self.progress_signal.emit(98) elif "has already been downloaded" in line: # File already exists - extract filename match = re.search(r"(.*?) has already been downloaded", line) if match: filename = Path(match.group(1)).name # Determine file type based on extension for existing file message ext = Path(filename).suffix.lower() if ext in VIDEO_EXTENSIONS: self.status_signal.emit(f"⚠️ Video file already exists") elif ext in AUDIO_EXTENSIONS: self.status_signal.emit(f"⚠️ Audio file already exists") elif ext in SUBTITLE_EXTENSIONS: self.status_signal.emit(f"⚠️ Subtitle file already exists") else: self.status_signal.emit(f"⚠️ File already exists") self.file_exists_signal.emit(filename) else: logger.info(f"Could not extract filename from 'already downloaded' line: {line}") self.status_signal.emit(_("download.file_exists")) # Fallback status elif "Finished downloading" in line: self.progress_signal.emit(100) # Show completion message based on file type if self.current_filename: ext = Path(self.current_filename).suffix.lower() # Video file extensions if ext in VIDEO_EXTENSIONS: self.status_signal.emit(_("download.video_completed")) # Audio file extensions elif ext in AUDIO_EXTENSIONS: self.status_signal.emit(_("download.audio_completed")) # Subtitle file extensions elif ext in SUBTITLE_EXTENSIONS: self.status_signal.emit(_("download.subtitle_completed")) # Default case else: self.status_signal.emit(_("download.completed")) else: self.status_signal.emit(_("download.completed")) self.update_details.emit("") # Clear details label on completion def pause(self) -> None: self.paused = True self._signal_process_group(signal.SIGSTOP if sys.platform != "win32" else None) def resume(self) -> None: self.paused = False self._signal_process_group(signal.SIGCONT if sys.platform != "win32" else None) def _signal_process_group(self, sig: Optional[int]) -> None: """Send a signal to yt-dlp's whole process group (yt-dlp + ffmpeg children). On Windows there is no SIGSTOP/SIGCONT; pausing there only stops output consumption, which is a known limitation. """ if sig is None or not self.process: return try: os.killpg(os.getpgid(self.process.pid), sig) except (ProcessLookupError, PermissionError, OSError) as e: logger.debug(f"Could not signal process group: {e}") def cancel(self) -> None: self.cancelled = True # Terminate the subprocess if it's running if self.process: try: self.process.terminate() except Exception: pass