from PySide6.QtCore import QThread, Signal, QObject import yt_dlp # Keep yt_dlp import here - only downloader uses it. import time import os import re from pathlib import Path class SignalManager(QObject): update_formats = Signal(list) update_status = Signal(str) update_progress = Signal(float) 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 def __init__(self, url, path, format_id, subtitle_lang=None, is_playlist=False, merge_subs=False, enable_sponsorblock=False, resolution='', playlist_items=None): super().__init__() self.url = url self.path = path self.format_id = format_id self.subtitle_lang = subtitle_lang self.is_playlist = is_playlist self.merge_subs = merge_subs self.enable_sponsorblock = enable_sponsorblock self.resolution = resolution self.playlist_items = playlist_items self.paused = False self.cancelled = False def cleanup_partial_files(self): """Delete any partial files including .part and unmerged format-specific files""" try: pattern = re.compile(r'\.f\d+\.') # Pattern to match format codes like .f243. for filename in os.listdir(self.path): file_path = os.path.join(self.path, filename) if filename.endswith('.part') or pattern.search(filename): try: if os.path.isfile(file_path): os.remove(file_path) except Exception as e: print(f"Error deleting {filename}: {str(e)}") except Exception as e: self.error_signal.emit(f"Error cleaning partial files: {str(e)}") def check_file_exists(self): """Check if the file already exists before downloading""" try: print("DEBUG: Starting file existence check") # Use yt-dlp to get the filename without downloading with yt_dlp.YoutubeDL({'quiet': True, 'skip_download': True}) as ydl: info = ydl.extract_info(self.url, download=False) # Get the title and sanitize it for filename title = info.get('title', 'video') # Don't remove colons and other special characters yet print(f"DEBUG: Original video title: {title}") # Get resolution for better matching resolution = "" for format_info in info.get('formats', []): if format_info.get('format_id') == self.format_id: resolution = format_info.get('resolution', '') break print(f"DEBUG: Resolution: {resolution}") # Create the expected filename (more specific) if self.is_playlist and info.get('playlist_title'): playlist_title = re.sub(r'[\\/*?"<>|]', "", info.get('playlist_title', '')).strip() base_path = os.path.join(self.path, playlist_title) else: base_path = self.path # Normalize the path to use consistent separators base_path = os.path.normpath(base_path) print(f"DEBUG: Base path: {base_path}") # Instead of trying to predict the exact filename, scan the directory # and look for files that contain both the title and resolution if os.path.exists(base_path): for filename in os.listdir(base_path): if filename.endswith('.mp4'): # Check if both title parts and resolution are in the filename title_words = title.lower().split() filename_lower = filename.lower() # Check if most title words are in the filename title_match = all(word in filename_lower for word in title_words[:3]) resolution_match = resolution.lower() in filename_lower print(f"DEBUG: Checking file: {filename}, Title match: {title_match}, Resolution match: {resolution_match}") if title_match and resolution_match: print(f"DEBUG: Found matching file: {filename}") return filename print("DEBUG: No matching file found") return None except Exception as e: print(f"DEBUG: Error checking file existence: {str(e)}") import traceback traceback.print_exc() return None def run(self): try: print("DEBUG: Starting download thread") # First check if file already exists existing_file = self.check_file_exists() if existing_file: print(f"DEBUG: File exists, emitting signal: {existing_file}") self.file_exists_signal.emit(existing_file) return print("DEBUG: No existing file found, proceeding with download") class DebugLogger: def debug(self, msg): # Print all debug messages to help diagnose issues print(f"YT-DLP DEBUG: {msg}") # Check for file exists message - look for both patterns if "already exists" in msg or "has already been downloaded" in msg: print(f"FILE EXISTS DETECTED: {msg}") # Try to extract the filename import re match = re.search(r'File (.*?) already exists', msg) if not match: match = re.search(r'(.*?) has already been downloaded', msg) if match: filename = os.path.basename(match.group(1)) self.thread.file_exists_signal.emit(filename) raise Exception("FileExistsError") # Add detection of post-processing messages if "Downloading" in msg: self.thread.status_signal.emit("⚡ Downloading video...") elif "Post-process" in msg or "Sponsorblock" in msg: self.thread.status_signal.emit("✨ Post-processing: Removing sponsor segments...") self.thread.progress_signal.emit(99) # Keep progress bar at 99% elif any(x in msg.lower() for x in ['downloading webpage', 'downloading api']): self.thread.status_signal.emit("🔍 Fetching video information...") self.thread.progress_signal.emit(0) elif 'extracting' in msg.lower(): self.thread.status_signal.emit("📦 Extracting video data...") self.thread.progress_signal.emit(0) elif 'downloading m3u8' in msg.lower(): self.thread.status_signal.emit("🎯 Preparing video streams...") self.thread.progress_signal.emit(0) def warning(self, msg): print(f"YT-DLP WARNING: {msg}") self.thread.status_signal.emit(f"⚠️ Warning: {msg}") # Also check for file exists in warnings if "already exists" in msg or "has already been downloaded" in msg: print(f"FILE EXISTS DETECTED IN WARNING: {msg}") import re match = re.search(r'File (.*?) already exists', msg) if not match: match = re.search(r'(.*?) has already been downloaded', msg) if match: filename = os.path.basename(match.group(1)) self.thread.file_exists_signal.emit(filename) raise Exception("FileExistsError") def error(self, msg): print(f"YT-DLP ERROR: {msg}") self.thread.status_signal.emit(f"❌ Error: {msg}") # Also check for file exists in errors if "already exists" in msg or "has already been downloaded" in msg: print(f"FILE EXISTS DETECTED IN ERROR: {msg}") import re match = re.search(r'File (.*?) already exists', msg) if not match: match = re.search(r'(.*?) has already been downloaded', msg) if match: filename = os.path.basename(match.group(1)) self.thread.file_exists_signal.emit(filename) raise Exception("FileExistsError") def __init__(self, thread): self.thread = thread def progress_hook(d): if self.cancelled: raise Exception("Download cancelled by user") if d['status'] == 'downloading': while self.paused and not self.cancelled: time.sleep(0.1) continue try: downloaded_bytes = d.get('downloaded_bytes', 0) total_bytes = d.get('total_bytes', 0) or d.get('total_bytes_estimate', 0) if total_bytes: progress = (downloaded_bytes / total_bytes) * 100 self.progress_signal.emit(progress) speed = d.get('speed', 0) if speed: speed_str = f"{speed/1024/1024:.1f} MB/s" else: speed_str = "N/A" eta = d.get('eta', 0) if eta: eta_str = f"{eta//60}:{eta%60:02d}" else: eta_str = "N/A" filename = os.path.basename(d.get('filename', '')) status = f"Speed: {speed_str} | ETA: {eta_str} | File: {filename}" self.status_signal.emit(status) except Exception as e: self.status_signal.emit("⚡ Downloading...") elif d['status'] == 'finished': if self.enable_sponsorblock: self.progress_signal.emit(99) self.status_signal.emit("✨ Post-processing: Removing sponsor segments...") else: self.progress_signal.emit(100) self.status_signal.emit("✅ Download completed!") # Get the extension from the format_id with yt_dlp.YoutubeDL({'quiet': True}) as ydl: try: info = ydl.extract_info(self.url, download=False) selected_format = next( f for f in info['formats'] if str(f.get('format_id', '')) == self.format_id ) output_ext = selected_format.get('ext', 'mp4') except Exception as e: self.error_signal.emit(f"Failed to get video information: {str(e)}") return # Base yt-dlp options with resolution in filename output_template = '%(title)s_%(resolution)s.%(ext)s' if self.is_playlist: output_template = '%(playlist_title)s/%(title)s_%(resolution)s.%(ext)s' ydl_opts = { 'format': f'{self.format_id}+bestaudio/best', 'outtmpl': os.path.join(self.path, output_template), 'progress_hooks': [progress_hook], 'merge_output_format': 'mp4', 'logger': DebugLogger(self), 'postprocessors': [{ 'key': 'FFmpegVideoConvertor', 'preferedformat': 'mp4' }], 'force_overwrites': True } # Add subtitle options if selected if self.subtitle_lang: lang_code = self.subtitle_lang.split(' - ')[0] is_auto = 'Auto-generated' in self.subtitle_lang ydl_opts.update({ 'writesubtitles': True, 'subtitleslangs': [lang_code], 'writeautomaticsub': True, 'skip_manual_subs': is_auto, 'skip_auto_subs': not is_auto, 'embedsubtitles': self.merge_subs, }) if self.merge_subs: ydl_opts['postprocessors'].extend([ { 'key': 'FFmpegSubtitlesConvertor', 'format': 'srt', }, { 'key': 'FFmpegEmbedSubtitle', 'already_have_subtitle': False, } ]) # Add SponsorBlock options if enabled if self.enable_sponsorblock: ydl_opts['postprocessors'].extend([{ 'key': 'SponsorBlock', 'categories': ['sponsor'], 'api': 'https://sponsor.ajay.app' }, { 'key': 'ModifyChapters', 'remove_sponsor_segments': ['sponsor'], 'sponsorblock_chapter_title': '[SponsorBlock]', 'force_keyframes': False }]) # Add playlist items if specified if self.playlist_items: ydl_opts['playlist_items'] = self.playlist_items try: # Download the video with yt_dlp.YoutubeDL(ydl_opts) as ydl: ydl.download([self.url]) if not self.cancelled: self.finished_signal.emit() # Clean up subtitle files after successful download if self.merge_subs: for filename in os.listdir(self.path): if filename.lower().endswith(('.vtt', '.srt', '.ass')): try: os.remove(os.path.join(self.path, filename)) except Exception as e: self.error_signal.emit(f"Error deleting subtitle file: {str(e)}") except Exception as e: if str(e) == "Download cancelled by user": self.cleanup_partial_files() self.error_signal.emit("Download cancelled") else: self.error_signal.emit(f"Download failed: {str(e)}") except Exception as e: self.error_signal.emit(f"Critical error: {str(e)}")