Houmeres/SageTube
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity

Refactor/code cleanup (#37)

Browse Source 
* fix imports
remove unused imports
use full import path
sort import (1. Standard Library, 2. Third-Party, 3. Local) in alphabetic.

* - remove: Method 3 from src.core.ytsage_downlader:cleanup_subtitle_file
  - it could delete the subtitle file of other movies if it present in same directory as it scane recursively.

- refactor: migrate from os.path to pathlib.Path for path handling
  - Replaced os.path methods with pathlib.Path to improve readability,
  - avoid repeatation.
  - cross-platform compatibility, and maintain cleaner code.

- improve: enhance code readability
  - Standardized string literals to use double quotes for consistency
  - Removed unnecessary spaces to maintain cleaner formatting
  - Applied code formatting for better readability and maintainability

* - add: ytsage_constants.py file for one place to store all constants.

- imporve: return type hint for function.

- remove: src/gui/ytsage_gui_dialogs.py file to avoid repetation
  - src/gui/dialogs is renamed to src/gui/ytsage_gui_dialogs for same naming convection. (future import will remains same)
  - use of src/gui/ytsage_gui_dialogs/__init__.py to import the dilogs modules.

- change: variable self.parent to self._parent so it does not overwrite the parent()
  - add type hint checking.

* - refactor: QMetaObject.invokeMethod to Signal
  - I encounter error with incokeMethod. Could not solve it.
  - So, Changed it to Signal to match app code language.

- implement: the ytsage_constants.py to code
  - remove: unnecessary logic
  - remove: repetitive code logic.

- update: yt-dlp logic for src\gui\ytsage_gui_dialogs\ytsage_dialogs_update:_update_binary
  - yt-dlp update logic will use `yt-dlp -U`

* refactor: remove unused imports and streamline code formatting across multiple files
This commit is contained in:
Viren Hirpara
2025-08-26 18:57:12 +05:30
committed by GitHub
parent 6652d7f00f
commit 9c13b4b61c
21 changed files with 2805 additions and 2628 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/ytsage_downloader.py
+255 -291
View File
@@ -1,24 +1,34 @@
from PySide6.QtCore import QThread, Signal, QObject, QProcess, QTimer
from .ytsage_logging import logger
import re
import shlex # For safely parsing command arguments
import subprocess # For direct CLI command execution
import time
from pathlib import Path
from PySide6.QtCore import QObject, QThread, Signal
from src.core.ytsage_logging import logger
from src.core.ytsage_yt_dlp import get_yt_dlp_path
from src.utils.ytsage_constants import SUBPROCESS_CREATIONFLAGS
try:
import yt_dlp # Keep yt_dlp import here - only downloader uses it.
import yt_dlp # Keep yt_dlp import here - only downloader uses it.
YT_DLP_AVAILABLE = True
except ImportError:
YT_DLP_AVAILABLE = False
logger.warning("yt-dlp not available at startup, will be downloaded at runtime")
import time
import os
import re
import subprocess # For direct CLI command execution
import shlex # For safely parsing command arguments
import sys # Added to get executable path information
from pathlib import Path
from .ytsage_yt_dlp import get_yt_dlp_path # Import the new yt-dlp path function
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)
@@ -26,18 +36,36 @@ class DownloadThread(QThread):
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
update_details = Signal(str) # New signal for filename, speed, ETA
def __init__(self, url, path, format_id, 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, rate_limit=None, download_section=None, force_keyframes=False):
def __init__(
self,
url,
path,
format_id,
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,
rate_limit=None,
download_section=None,
force_keyframes=False,
) -> None:
super().__init__()
self.url = url
self.path = path
self.path = Path(path)
self.format_id = format_id
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.sponsorblock_categories = sponsorblock_categories if sponsorblock_categories else ["sponsor"]
self.resolution = resolution
self.playlist_items = playlist_items
self.save_description = save_description
@@ -52,205 +80,136 @@ class DownloadThread(QThread):
self.use_direct_command = True # Flag to use direct CLI command instead of Python API
self.last_output_time = time.time()
self.timeout_timer = None
self.current_filename = None # Initialize filename storage
self.last_file_path = None # Initialize full file path storage
self.subtitle_files = [] # Track subtitle files that are created
self.initial_subtitle_files = set() # Track initial subtitle files before download
self.current_filename = None # Initialize filename storage
self.last_file_path = None # Initialize full file path storage
self.subtitle_files = [] # Track subtitle files that are created
self.initial_subtitle_files = set() # Track initial subtitle files before download
def cleanup_partial_files(self):
def cleanup_partial_files(self) -> None:
"""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):
pattern = re.compile(r"\.f\d+\.") # Pattern to match format codes like .f243.
for file_path in self.path.iterdir():
if file_path.suffix == ".part" or pattern.search(file_path.name):
try:
if os.path.isfile(file_path):
os.remove(file_path)
file_path.unlink(missing_ok=True)
except Exception as e:
logger.error(f"Error deleting {filename}: {str(e)}")
logger.error(f"Error deleting {file_path.name}: {str(e)}")
except Exception as e:
self.error_signal.emit(f"Error cleaning partial files: {str(e)}")
def cleanup_subtitle_files(self):
def cleanup_subtitle_files(self) -> None:
"""Delete subtitle files after they have been merged into the video file"""
if not self.merge_subs:
return # Only cleanup if merge_subs is enabled
try:
deleted_count = 0
# Method 1: Delete tracked subtitle files from output messages
if self.subtitle_files:
for subtitle_file in self.subtitle_files:
try:
if os.path.isfile(subtitle_file):
os.remove(subtitle_file)
deleted_count += 1
logger.debug(f"Deleted tracked subtitle file: {os.path.basename(subtitle_file)}")
except Exception as e:
logger.error(f"Error deleting subtitle file {subtitle_file}: {str(e)}")
logger.debug(f"Deleted {deleted_count} of {len(self.subtitle_files)} tracked subtitle files")
# Method 2: Find newly created subtitle files by comparing with initial set
deleted_count = [0, 0]
def safe_delete(path: Path) -> bool:
try:
new_subtitle_files = set()
for root, dirs, files in os.walk(self.path):
for file in files:
if file.endswith('.vtt') or file.endswith('.srt'):
full_path = os.path.join(root, file)
if full_path not in self.initial_subtitle_files:
new_subtitle_files.add(full_path)
if new_subtitle_files:
logger.debug(f"Found {len(new_subtitle_files)} new subtitle files to delete")
for subtitle_file in new_subtitle_files:
try:
if os.path.isfile(subtitle_file):
os.remove(subtitle_file)
deleted_count += 1
logger.debug(f"Deleted new subtitle file: {os.path.basename(subtitle_file)}")
except Exception as e:
logger.error(f"Error deleting new subtitle file {subtitle_file}: {str(e)}")
path.unlink(missing_ok=True)
logger.debug(f"Deleted subtitle file: {path.name}")
return True
except Exception as e:
logger.error(f"Error in finding new subtitle files: {str(e)}") # Method 3: As a last resort, use timestamp-based approach for recently created files
if self.last_file_path and deleted_count == 0:
target_dir = os.path.dirname(self.last_file_path)
# Look for subtitle files created in last 5 minutes
now = time.time()
for filename in os.listdir(target_dir):
if filename.endswith('.vtt') or filename.endswith('.srt'):
file_path = os.path.join(target_dir, filename)
# Check if it was created in the last 5 minutes
file_time = os.path.getctime(file_path)
if now - file_time < 300: # 5 minutes
try:
os.remove(file_path)
deleted_count += 1
logger.debug(f"Deleted subtitle file by timestamp: {filename}")
except Exception as e:
logger.error(f"Error deleting subtitle file {filename}: {str(e)}")
logger.debug(f"Total subtitle files deleted: {deleted_count}")
logger.error(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 ---
new_subtitle_files = {
f for f in Path(self.path).rglob("*") if f.suffix in [".vtt", ".srt"] and f not in self.initial_subtitle_files
}
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.error(f"Error cleaning subtitle files: {str(e)}")
def check_file_exists(self):
def check_file_exists(self) -> bool | None:
"""Check if the file already exists before downloading"""
try:
logger.debug("Starting file existence check")
# Use yt-dlp to get the filename without downloading, suppressing warnings
ydl_opts_check = {
'quiet': True,
'skip_download': True,
'no_warnings': True, # <-- Suppress warnings during check
'ignoreerrors': True, # Also ignore other potential errors during this check
'outtmpl': {'default': os.path.join(self.path, '%(title)s.%(ext)s')},
'format': self.format_id if self.format_id else 'best' # Use selected format or best
"quiet": True,
"skip_download": True,
"no_warnings": True, # <-- Suppress warnings during check
"ignoreerrors": True, # Also ignore other potential errors during this check
"outtmpl": {"default": Path.joinpath(self.path, "%(title)s.%(ext)s")},
"format": (self.format_id if self.format_id else "best"), # Use selected format or best
}
if self.cookie_file:
ydl_opts_check['cookiefile'] = self.cookie_file
ydl_opts_check["cookiefile"] = self.cookie_file
if YT_DLP_AVAILABLE:
with yt_dlp.YoutubeDL(ydl_opts_check) as ydl:
info = ydl.extract_info(self.url, download=False)
# Handle cases where info extraction fails silently
if not info:
logger.debug("Failed to extract info during file existence check. Skipping check.")
return False # Proceed with download attempt
return False # Proceed with download attempt
# Get the title and sanitize it for filename
title = info.get('title', 'video')
title = info.get("title", "video")
# Don't remove colons and other special characters yet
logger.debug(f"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', '')
for format_info in info.get("formats", []):
if format_info.get("format_id") == self.format_id:
resolution = format_info.get("resolution", "")
break
logger.debug(f"Resolution: {resolution}")
else:
logger.debug("yt-dlp not available, skipping file existence check")
return False # Proceed with download attempt
# 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)
logger.debug(f"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
logger.debug(f"Checking file: {filename}, Title match: {title_match}, Resolution match: {resolution_match}")
if title_match and resolution_match:
logger.debug(f"Found matching file: {filename}")
return filename
logger.debug("No matching file found")
return None
except Exception as e:
logger.debug(f"Error checking file existence: {str(e)}")
import traceback
traceback.print_exc()
return None
def _build_yt_dlp_command(self):
def _build_yt_dlp_command(self) -> list:
"""Build the yt-dlp command line with all options for direct execution."""
# Use the new yt-dlp path function from ytsage_yt_dlp module
yt_dlp_path = get_yt_dlp_path()
cmd = [yt_dlp_path]
cmd: list = [yt_dlp_path]
logger.debug(f"Using yt-dlp from: {yt_dlp_path}")
# Format selection strategy - use format ID if provided or fallback to resolution
if self.format_id:
# Strip the -drc suffix if present to fix issues with certain audio formats
clean_format_id = self.format_id.split('-drc')[0] if '-drc' in self.format_id else self.format_id
clean_format_id = self.format_id.split("-drc")[0] if "-drc" in self.format_id else self.format_id
# Check if this is an audio-only format
is_audio_format = False
try:
if YT_DLP_AVAILABLE:
ydl_opts = {
'quiet': True,
'no_warnings': True,
'skip_download': True,
"quiet": True,
"no_warnings": True,
"skip_download": True,
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(self.url, download=False)
for fmt in info.get('formats', []):
if fmt.get('format_id') == clean_format_id:
if fmt.get('vcodec') == 'none' or 'audio only' in fmt.get('format_note', '').lower():
info = ydl.extract_info(self.url, download=False) or {}
for fmt in info.get("formats", []):
if fmt.get("format_id") == clean_format_id:
if fmt.get("vcodec") == "none" or "audio only" in fmt.get("format_note", "").lower():
is_audio_format = True
logger.debug(f"Detected audio-only format for ID: {clean_format_id}")
break
except Exception as e:
logger.debug(f"Error checking if format is audio-only: {e}")
# For audio-only formats, don't try to merge with video
if is_audio_format:
cmd.extend(["-f", clean_format_id])
@@ -258,7 +217,7 @@ class DownloadThread(QThread):
else:
cmd.extend(["-f", f"{clean_format_id}+bestaudio/best"])
logger.debug(f"Using video format selection with audio: {clean_format_id}+bestaudio/best")
# Determine output format based on the selected format ID - only for video formats
if not is_audio_format:
try:
@@ -266,24 +225,24 @@ class DownloadThread(QThread):
logger.debug(f"Getting format information for format ID: {self.format_id} (using: {clean_format_id})")
if YT_DLP_AVAILABLE:
ydl_opts = {
'quiet': True,
'no_warnings': True,
'skip_download': True,
"quiet": True,
"no_warnings": True,
"skip_download": True,
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(self.url, download=False)
info = ydl.extract_info(self.url, download=False) or {}
# Look for the clean format ID first
for fmt in info.get('formats', []):
if fmt.get('format_id') == clean_format_id:
format_ext = fmt.get('ext')
for fmt in info.get("formats", []):
if fmt.get("format_id") == clean_format_id:
format_ext = fmt.get("ext")
break
# If not found, try the original ID as fallback
if not format_ext:
for fmt in info.get('formats', []):
if fmt.get('format_id') == self.format_id:
format_ext = fmt.get('ext')
for fmt in info.get("formats", []):
if fmt.get("format_id") == self.format_id:
format_ext = fmt.get("ext")
break
if format_ext:
logger.debug(f"Detected format extension: {format_ext}")
# Ensure output matches the selected format - only for video formats
@@ -296,139 +255,136 @@ class DownloadThread(QThread):
# If no specific format ID, use resolution-based sorting (-S)
res_value = self.resolution if self.resolution else "720" # Default to 720p if no resolution specified
cmd.extend(["-S", f"res:{res_value}"])
# Output template with resolution in filename
output_template = os.path.join(self.path, '%(title)s_%(resolution)s.%(ext)s')
output_template = Path.joinpath(self.path, "%(title)s_%(resolution)s.%(ext)s")
# Handle playlist directory creation if needed
if self.is_playlist:
# Create output template with playlist subfolder
output_template = os.path.join(self.path, '%(playlist_title)s/%(title)s_%(resolution)s.%(ext)s')
cmd.extend(["-o", output_template])
output_template = self.path.joinpath("%(playlist_title)s/%(title)s_%(resolution)s.%(ext)s")
cmd.extend(["-o", output_template.as_posix()])
# 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 = []
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_code = sub_selection.split(" - ")[0]
lang_codes.append(lang_code)
except Exception as e:
logger.warning(f"Could not parse subtitle selection '{sub_selection}': {e}")
if lang_codes:
cmd.extend(["--sub-langs", ",".join(lang_codes)])
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", self.cookie_file])
# 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
cmd.append(self.url)
return cmd
def run(self):
def run(self) -> None:
try:
logger.debug("Starting download thread")
# First check if file already exists using original method
existing_file = self.check_file_exists()
if existing_file:
logger.debug(f"File exists, emitting signal: {existing_file}")
self.file_exists_signal.emit(existing_file)
return
logger.debug("No existing file found, proceeding with download")
# 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 root, dirs, files in os.walk(self.path):
for file in files:
if file.endswith('.vtt') or file.endswith('.srt'):
self.initial_subtitle_files.add(os.path.join(root, file))
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.warning(f"Error scanning for initial subtitle files: {e}")
if self.use_direct_command:
# Use direct CLI command instead of Python API
self._run_direct_command()
else:
# Original method using Python API - code left for reference
self._run_python_api()
except Exception as e:
# Catch errors during setup
self.error_signal.emit(f"Critical error in download thread: {str(e)}")
import traceback
traceback.print_exc()
def _run_direct_command(self):
def _run_direct_command(self) -> None:
"""Run yt-dlp as a direct command line process instead of using Python API."""
try:
cmd = self._build_yt_dlp_command()
cmd_str = " ".join(shlex.quote(str(arg)) for arg in cmd)
logger.debug(f"Executing command: {cmd_str}")
self.status_signal.emit("🚀 Starting download...")
self.progress_signal.emit(0)
# Start the process
# Add creationflags=subprocess.CREATE_NO_WINDOW to hide console on Windows
creation_flags = 0
if os.name == 'nt': # Only use flag on Windows
creation_flags = subprocess.CREATE_NO_WINDOW
# Extra logic moved to src\utils\ytsage_constants.py
self.process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
@@ -436,37 +392,39 @@ class DownloadThread(QThread):
text=True,
bufsize=1, # Line buffered
universal_newlines=True,
creationflags=creation_flags # Add this flag
creationflags=SUBPROCESS_CREATIONFLAGS,
)
# Process output line by line to update progress
for line in iter(self.process.stdout.readline, ''):
for line in iter(self.process.stdout.readline, ""): # type: ignore
if self.cancelled:
self.process.terminate()
self.cleanup_partial_files()
self.status_signal.emit("Download cancelled")
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 = self.process.wait()
# Special handling for specific errors
# return code 127 typically means command not found
if return_code == 127:
self.error_signal.emit("Error: yt-dlp executable not found. This could be due to improper installation or a PATH issue.")
self.error_signal.emit(
"Error: yt-dlp executable not found. This could be due to improper installation or a PATH issue."
)
return
if return_code == 0:
self.progress_signal.emit(100)
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:
@@ -475,7 +433,7 @@ class DownloadThread(QThread):
self.status_signal.emit("✅ Download completed! Cleaning up...")
time.sleep(3) # Increased delay to 3 seconds
self.cleanup_subtitle_files()
self.finished_signal.emit()
else:
# Check if it was cancelled
@@ -484,217 +442,223 @@ class DownloadThread(QThread):
else:
# Provide more descriptive error message for possible yt-dlp conflicts
if return_code == 1:
self.error_signal.emit(f"Download failed with return code {return_code}. This may be due to a conflict with multiple yt-dlp installations. Try uninstalling any system-installed yt-dlp (e.g. through snap or apt) and restart the application.")
self.error_signal.emit(
f"Download failed with return code {return_code}. This may be due to a conflict with multiple yt-dlp installations. Try uninstalling any system-installed yt-dlp (e.g. through snap or apt) and restart the application."
)
else:
self.error_signal.emit(f"Download failed with return code {return_code}")
self.cleanup_partial_files()
except Exception as e:
self.error_signal.emit(f"Error in direct command: {str(e)}")
self.cleanup_partial_files()
def _parse_output_line(self, line):
def _parse_output_line(self, line) -> 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
# 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)
dest_match = re.search(r"^\[download\] Destination:\s*(.*)", line)
if dest_match:
try:
filepath = dest_match.group(1).strip()
self.current_filename = os.path.basename(filepath)
self.current_filename = Path(filepath).name
self.last_file_path = filepath # Store the full path for later cleanup
logger.debug(f"Extracted filename: {self.current_filename}") # DEBUG
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:
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)
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
# 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 = os.path.splitext(self.current_filename)[1].lower()
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:
if is_audio_download or "Downloading audio" in line:
self.status_signal.emit(f"⏬ Downloading audio...")
# Video file extensions with likely video content
elif ext in ['.mp4', '.webm', '.mkv', '.avi', '.mov', '.flv']:
elif ext in [".mp4", ".webm", ".mkv", ".avi", ".mov", ".flv"]:
self.status_signal.emit(f"⏬ Downloading video...")
# Audio file extensions
elif ext in ['.mp3', '.m4a', '.aac', '.wav', '.ogg', '.opus', '.flac']:
elif ext in [".mp3", ".m4a", ".aac", ".wav", ".ogg", ".opus", ".flac"]:
self.status_signal.emit(f"⏬ Downloading audio...")
# Subtitle file extensions
elif ext in ['.vtt', '.srt', '.ass', '.ssa']:
elif ext in [".vtt", ".srt", ".ass", ".ssa"]:
self.status_signal.emit(f"⏬ Downloading subtitle...")
# Default case
else:
self.status_signal.emit(f"⏬ Downloading...")
except Exception as e:
logger.error(f"Error extracting filename from line '{line}': {e}")
self.status_signal.emit("⚡ Downloading...") # Fallback status
return # Don't process this line further for speed/ETA
self.status_signal.emit("⚡ Downloading...") # 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(f"⏬ Downloading video...")
return
elif "Downloading audio" in line:
self.status_signal.emit(f"⏬ 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))', line, re.IGNORECASE)
subtitle_match = re.search(
r"(?:Writing|Downloading) (?:video )?subtitles.*?(?:to|:)\s*(.*\.(?:vtt|srt))",
line,
re.IGNORECASE,
)
if subtitle_match:
subtitle_file = subtitle_match.group(1).strip()
# Show subtitle download message
self.status_signal.emit(f"⏬ Downloading subtitle...")
# Store the subtitle file path for later deletion if merging is enabled
if self.merge_subs:
if not os.path.isabs(subtitle_file):
if not Path(subtitle_file).is_absolute():
# If it's a relative path, make it absolute based on current path
subtitle_file = os.path.join(self.path, subtitle_file)
subtitle_file = Path.joinpath(self.path, subtitle_file)
self.subtitle_files.append(subtitle_file)
logger.debug(f"Tracking subtitle file for later cleanup: {subtitle_file}")
return
# Send status updates based on output line content
if 'Downloading webpage' in line or 'Extracting URL' in line:
if "Downloading webpage" in line or "Extracting URL" in line:
self.status_signal.emit("🔍 Fetching video information...")
self.progress_signal.emit(0)
elif 'Downloading API JSON' in line:
elif "Downloading API JSON" in line:
self.status_signal.emit("📋 Processing playlist data...")
self.progress_signal.emit(0)
elif 'Downloading m3u8 information' in line:
elif "Downloading m3u8 information" in line:
self.status_signal.emit("🎯 Preparing video streams...")
self.progress_signal.emit(0)
elif '[download] Downloading video ' in line:
elif "[download] Downloading video " in line:
self.status_signal.emit("⏬ Downloading video...")
elif '[download] Downloading audio ' in line:
elif "[download] Downloading audio " in line:
self.status_signal.emit("⏬ Downloading audio...")
elif 'Downloading format' in line:
elif "Downloading format" in line:
# Try to detect if it's audio or video format
if ' - audio only' in line:
if " - audio only" in line:
self.status_signal.emit("⏬ Downloading audio...")
elif ' - video only' in line:
elif " - video only" in line:
self.status_signal.emit("⏬ Downloading video...")
else:
# Don't emit generic message - format is unclear
pass
# Look for download percentage
percent_match = re.search(r'(\d+\.\d+)%', line)
percent_match = re.search(r"(\d+\.\d+)%", line)
if percent_match:
try:
percent = float(percent_match.group(1))
self.progress_signal.emit(percent)
except (ValueError, IndexError):
pass
# Check for download speed and ETA
if '[download]' in line and '%' in line:
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_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_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"Speed: {speed_str} | ETA: {eta_str}"
self.update_details.emit(status)
except Exception as e:
# If parsing fails, just show basic status (maybe log the error)
logger.error(f"Error parsing download details line: {line} -> {e}")
pass # Keep basic status emission below if needed, or emit generic details
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:
if "[Merger]" in line or "Merging formats" in line:
self.status_signal.emit("✨ Post-processing: Merging formats...")
self.progress_signal.emit(95)
elif 'SponsorBlock' in line:
elif "SponsorBlock" in line:
self.status_signal.emit("✨ Post-processing: Removing sponsor segments...")
self.progress_signal.emit(97)
elif 'Deleting original file' in line:
elif "Deleting original file" in line:
self.progress_signal.emit(98)
elif 'has already been downloaded' in line:
elif "has already been downloaded" in line:
# File already exists - extract filename
match = re.search(r'(.*?) has already been downloaded', line)
match = re.search(r"(.*?) has already been downloaded", line)
if match:
filename = os.path.basename(match.group(1))
filename = Path(match.group(1)).name
# Determine file type based on extension for existing file message
ext = os.path.splitext(filename)[1].lower()
if ext in ['.mp4', '.webm', '.mkv', '.avi', '.mov', '.flv']:
ext = Path(filename).suffix.lower()
if ext in [".mp4", ".webm", ".mkv", ".avi", ".mov", ".flv"]:
self.status_signal.emit(f"⚠️ Video file already exists")
elif ext in ['.mp3', '.m4a', '.aac', '.wav', '.ogg', '.opus', '.flac']:
elif ext in [".mp3", ".m4a", ".aac", ".wav", ".ogg", ".opus", ".flac"]:
self.status_signal.emit(f"⚠️ Audio file already exists")
elif ext in ['.vtt', '.srt', '.ass', '.ssa']:
elif ext in [".vtt", ".srt", ".ass", ".ssa"]:
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("⚠️ File already exists") # Fallback status
elif 'Finished downloading' in line:
self.status_signal.emit("⚠️ File already 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 = os.path.splitext(self.current_filename)[1].lower()
ext = Path(self.current_filename).suffix.lower()
# Video file extensions
if ext in ['.mp4', '.webm', '.mkv', '.avi', '.mov', '.flv']:
if ext in [".mp4", ".webm", ".mkv", ".avi", ".mov", ".flv"]:
self.status_signal.emit(f"✅ Video download completed!")
# Audio file extensions
elif ext in ['.mp3', '.m4a', '.aac', '.wav', '.ogg', '.opus', '.flac']:
elif ext in [".mp3", ".m4a", ".aac", ".wav", ".ogg", ".opus", ".flac"]:
self.status_signal.emit(f"✅ Audio download completed!")
# Subtitle file extensions
elif ext in ['.vtt', '.srt', '.ass', '.ssa']:
elif ext in [".vtt", ".srt", ".ass", ".ssa"]:
self.status_signal.emit(f"✅ Subtitle download 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 _run_python_api(self):
self.update_details.emit("") # Clear details label on completion
def _run_python_api(self) -> None:
"""Original download method using Python API - kept for reference."""
# The existing run method code using yt_dlp.YoutubeDL starts here
# This method is no longer used by default
def pause(self):
def pause(self) -> None:
self.paused = True
def resume(self):
def resume(self) -> None:
self.paused = False
def cancel(self):
def cancel(self) -> None:
self.cancelled = True
# Terminate the subprocess if it's running
if self.process:
try:
self.process.terminate()
except Exception:
pass
pass
src/core/ytsage_ffmpeg.py
+153 -124
View File
@@ -1,33 +1,39 @@
import os
import sys
import subprocess
import requests
import shutil
import tempfile
import hashlib
import os
import shutil
import subprocess
import tempfile
from pathlib import Path
from PySide6.QtGui import QIcon
from .ytsage_logging import logger
def check_7zip_installed():
import requests
from src.core.ytsage_logging import logger
from src.utils.ytsage_constants import (
FFMPEG_7Z_DOWNLOAD_URL,
FFMPEG_7Z_SHA256_URL,
FFMPEG_ZIP_DOWNLOAD_URL,
OS_NAME,
SUBPROCESS_CREATIONFLAGS,
)
def check_7zip_installed() -> bool:
"""Check if 7-Zip is installed on Windows."""
try:
subprocess.run(['7z', '--help'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == 'win32' else 0)
subprocess.run(["7z", "--help"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, creationflags=SUBPROCESS_CREATIONFLAGS)
return True
except (subprocess.SubprocessError, FileNotFoundError):
return False
def download_file(url, dest_path, progress_callback=None):
def download_file(url, dest_path, progress_callback=None) -> bool:
"""Download a file from URL to destination path with progress indication."""
try:
response = requests.get(url, stream=True, timeout=30) # Added timeout
response.raise_for_status() # Check for HTTP errors
total_size = int(response.headers.get('content-length', 0))
with open(dest_path, 'wb') as f:
total_size = int(response.headers.get("content-length", 0))
with open(dest_path, "wb") as f:
if total_size == 0:
f.write(response.content)
else:
@@ -43,7 +49,8 @@ def download_file(url, dest_path, progress_callback=None):
logger.info(f"Download error: {str(e)}")
return False
def get_file_sha256(file_path):
def get_file_sha256(file_path) -> str:
"""Calculate SHA-256 hash of a file."""
sha256_hash = hashlib.sha256()
with open(file_path, "rb") as f:
@@ -51,17 +58,18 @@ def get_file_sha256(file_path):
sha256_hash.update(chunk)
return sha256_hash.hexdigest()
def verify_sha256(file_path, expected_hash_url):
def verify_sha256(file_path, expected_hash_url) -> bool:
"""Verify file SHA-256 hash against expected hash from URL."""
try:
# Download the SHA-256 hash
response = requests.get(expected_hash_url, timeout=10)
response.raise_for_status()
expected_hash = response.text.strip().split()[0] # Get just the hash part
# Calculate actual hash
actual_hash = get_file_sha256(file_path)
# Compare hashes
if actual_hash.lower() == expected_hash.lower():
logger.info("SHA-256 verification successful!")
@@ -75,20 +83,23 @@ def verify_sha256(file_path, expected_hash_url):
logger.info(f"⚠️ SHA-256 verification error: {str(e)}")
return False
def get_ffmpeg_install_path():
"""Get the FFmpeg installation path."""
if sys.platform == 'win32':
return os.path.join(os.getenv('LOCALAPPDATA'), 'ffmpeg', 'ffmpeg-7.1.1-full_build', 'bin')
elif sys.platform == 'darwin':
paths = ['/usr/local/bin', '/opt/homebrew/bin', '/usr/bin']
for path in paths:
if os.path.exists(os.path.join(path, 'ffmpeg')):
return path
return '/usr/local/bin' # Default Homebrew path
else:
return '/usr/bin' # Standard Linux path
def get_ffmpeg_path():
def get_ffmpeg_install_path() -> Path:
"""Get the FFmpeg installation path."""
if OS_NAME == "Windows":
return Path(os.getenv("LOCALAPPDATA")) / "ffmpeg" / "ffmpeg-7.1.1-full_build" / "bin" # type: ignore
elif OS_NAME == "Darwin":
paths = ["/usr/local/bin", "/opt/homebrew/bin", "/usr/bin"]
for path in paths:
if Path(path).joinpath("ffmpeg").exists():
return Path(path)
return Path("/usr/local/bin") # Default Homebrew path
else:
return Path("/usr/bin") # Standard Linux path
def get_ffmpeg_path() -> str | Path:
"""
Get the FFmpeg executable path, either from PATH or installation directory.
Returns:
@@ -96,150 +107,159 @@ def get_ffmpeg_path():
"""
try:
# First try to find ffmpeg in PATH using 'where' on Windows or 'which' on Unix
if sys.platform == 'win32':
if OS_NAME == "Windows":
# On Windows, use 'where' command and hide console window
startupinfo = None
if hasattr(subprocess, 'STARTUPINFO'):
startupinfo = subprocess.STARTUPINFO()
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
startupinfo.wShowWindow = 0 # SW_HIDE
# Extra logic moved to src\utils\ytsage_constants.py
result = subprocess.run(
['where', 'ffmpeg'],
capture_output=True,
text=True,
["where", "ffmpeg"],
capture_output=True,
text=True,
check=False,
startupinfo=startupinfo
creationflags=SUBPROCESS_CREATIONFLAGS,
)
if result.returncode == 0 and result.stdout.strip():
ffmpeg_path = result.stdout.strip().split('\n')[0]
ffmpeg_path = result.stdout.strip().split("\n")[0]
return ffmpeg_path
else:
# On Unix systems, use 'which' command
result = subprocess.run(['which', 'ffmpeg'], capture_output=True, text=True, check=False)
result = subprocess.run(["which", "ffmpeg"], capture_output=True, text=True, check=False)
if result.returncode == 0 and result.stdout.strip():
ffmpeg_path = result.stdout.strip()
return ffmpeg_path
except Exception as e:
logger.error(f"Error finding ffmpeg in PATH: {e}")
# If not found in PATH, check the installation directory
ffmpeg_install_path = get_ffmpeg_install_path()
if sys.platform == 'win32':
ffmpeg_exe = os.path.join(ffmpeg_install_path, 'ffmpeg.exe')
if OS_NAME == "Windows":
ffmpeg_exe = Path(ffmpeg_install_path).joinpath("ffmpeg.exe")
else:
ffmpeg_exe = os.path.join(ffmpeg_install_path, 'ffmpeg')
if os.path.exists(ffmpeg_exe):
ffmpeg_exe = Path(ffmpeg_install_path).joinpath("ffmpeg")
if ffmpeg_exe.exists():
return ffmpeg_exe
# Return command name as fallback
return "ffmpeg"
def check_ffmpeg_installed():
def check_ffmpeg_installed() -> bool:
"""Check if FFmpeg is installed and accessible."""
try:
# First try the PATH
result = subprocess.run(['ffmpeg', '-version'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=True,
creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == 'win32' else 0,
timeout=5) # Added timeout
result = subprocess.run(
["ffmpeg", "-version"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=True,
creationflags=SUBPROCESS_CREATIONFLAGS,
timeout=5,
) # Added timeout
return True
except (subprocess.SubprocessError, FileNotFoundError):
# If not in PATH, check the installation directory
ffmpeg_path = get_ffmpeg_install_path()
if sys.platform == 'win32':
ffmpeg_exe = os.path.join(ffmpeg_path, 'ffmpeg.exe')
if OS_NAME == "Windows":
ffmpeg_exe = Path(ffmpeg_path).joinpath("ffmpeg.exe")
else:
ffmpeg_exe = os.path.join(ffmpeg_path, 'ffmpeg')
if os.path.exists(ffmpeg_exe):
ffmpeg_exe = Path(ffmpeg_path).joinpath("ffmpeg")
if ffmpeg_exe.exists():
# Add to PATH if found
os.environ['PATH'] = f"{ffmpeg_path}{os.pathsep}{os.environ.get('PATH', '')}"
os.environ["PATH"] = f"{ffmpeg_path}{os.pathsep}{os.environ.get('PATH', '')}"
return True
return False
except Exception as e:
logger.info(f"FFmpeg check error: {str(e)}")
return False
def install_ffmpeg_windows():
def install_ffmpeg_windows() -> bool:
"""Install FFmpeg on Windows using 7z method primarily, with zip as fallback."""
ffmpeg_path = get_ffmpeg_install_path()
# Check if already installed
if check_ffmpeg_installed():
logger.info("FFmpeg is already installed!")
return True
try:
# Define variables - prioritize 7z version
ffmpeg_7z_url = "https://github.com/GyanD/codexffmpeg/releases/download/7.1.1/ffmpeg-7.1.1-full_build.7z"
ffmpeg_zip_url = "https://github.com/GyanD/codexffmpeg/releases/download/7.1.1/ffmpeg-7.1.1-full_build.zip"
sha256_url = "https://www.gyan.dev/ffmpeg/builds/packages/ffmpeg-7.1.1-full_build.7z.sha256"
extract_dir = os.path.join(os.getenv('LOCALAPPDATA'), 'ffmpeg')
full_build_dir = os.path.join(extract_dir, 'ffmpeg-7.1.1-full_build')
bin_dir = os.path.join(full_build_dir, 'bin')
# ffmpeg variables moved to src\utils\ytsage_constants.py
extract_dir = Path(os.getenv("LOCALAPPDATA")) / "ffmpeg" # type: ignore
full_build_dir = extract_dir / "ffmpeg-7.1.1-full_build"
bin_dir = full_build_dir / "bin"
# Create extraction directory if it doesn't exist
os.makedirs(extract_dir, exist_ok=True)
extract_dir.mkdir(exist_ok=True)
# Try 7z method first (smaller size)
use_7zip = check_7zip_installed()
if use_7zip:
logger.info("Using 7-Zip method (smaller download size)...")
temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.7z').name
temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".7z").name
# Download 7z file
if not download_file(ffmpeg_7z_url, temp_file,
progress_callback=lambda msg: logger.debug(msg)):
if not download_file(
FFMPEG_7Z_DOWNLOAD_URL,
temp_file,
progress_callback=lambda msg: logger.debug(msg),
):
logger.error("Failed to download 7z file, trying zip fallback...")
use_7zip = False
else:
# Verify SHA-256 hash for 7z file
if verify_sha256(temp_file, sha256_url):
if verify_sha256(temp_file, FFMPEG_7Z_SHA256_URL):
logger.info("Extracting FFmpeg components from 7z archive...")
try:
subprocess.run(['7z', 'x', temp_file, f'-o{extract_dir}', '-y'],
creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == 'win32' else 0,
timeout=300) # 5-minute timeout
subprocess.run(
["7z", "x", temp_file, f"-o{extract_dir}", "-y"],
creationflags=SUBPROCESS_CREATIONFLAGS,
timeout=300,
) # 5-minute timeout
except Exception as e:
logger.error(f"7z extraction failed: {str(e)}, trying zip fallback...")
use_7zip = False
else:
logger.error("SHA-256 verification failed for 7z file, trying zip fallback...")
use_7zip = False
# Fallback to zip method if 7z failed or not available
if not use_7zip:
logger.info("Using ZIP method as fallback...")
temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.zip').name
temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".zip").name
# Download zip file
if not download_file(ffmpeg_zip_url, temp_file,
progress_callback=lambda msg: logger.debug(msg)):
if not download_file(
FFMPEG_ZIP_DOWNLOAD_URL,
temp_file,
progress_callback=lambda msg: logger.debug(msg),
):
raise Exception("Failed to download FFmpeg (both 7z and zip methods failed)")
logger.info("Extracting FFmpeg components from zip archive...")
try:
import zipfile
with zipfile.ZipFile(temp_file, 'r') as zip_ref:
with zipfile.ZipFile(temp_file, "r") as zip_ref:
zip_ref.extractall(extract_dir)
except Exception as e:
raise Exception(f"Extraction failed: {str(e)}")
logger.info("Configuring system paths...")
# Add to System Path
user_path = os.environ.get('PATH', '')
if bin_dir not in user_path:
subprocess.run(['setx', 'PATH', f"{user_path};{bin_dir}"],
creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == 'win32' else 0)
os.environ['PATH'] = f"{user_path};{bin_dir}"
user_path = os.environ.get("PATH", "")
if str(bin_dir) not in user_path.split(os.pathsep):
subprocess.run(
["setx", "PATH", f"{user_path};{bin_dir}"],
creationflags=SUBPROCESS_CREATIONFLAGS,
)
os.environ["PATH"] = f"{user_path};{bin_dir}"
# Clean up
try:
os.unlink(temp_file)
Path(temp_file).unlink(missing_ok=True)
except Exception:
pass # Ignore cleanup errors
@@ -254,16 +274,19 @@ def install_ffmpeg_windows():
logger.error(f"Error installing FFmpeg: {str(e)}")
return False
def install_ffmpeg_macos():
def install_ffmpeg_macos() -> bool:
"""Install FFmpeg on macOS using Homebrew."""
try:
# Check if Homebrew is installed
try:
subprocess.run(['brew', '--version'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=True,
timeout=5)
subprocess.run(
["brew", "--version"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=True,
timeout=5,
)
except (subprocess.SubprocessError, FileNotFoundError):
logger.info("Installing Homebrew...")
brew_install_cmd = '/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)"'
@@ -271,56 +294,62 @@ def install_ffmpeg_macos():
# Install FFmpeg
logger.info("Installing FFmpeg...")
subprocess.run(['brew', 'install', 'ffmpeg'], check=True, timeout=300)
subprocess.run(["brew", "install", "ffmpeg"], check=True, timeout=300)
# Verify installation
if not check_ffmpeg_installed():
raise Exception("FFmpeg installation verification failed")
return True
except Exception as e:
logger.error(f"Error installing FFmpeg: {str(e)}")
return False
def install_ffmpeg_linux():
def install_ffmpeg_linux() -> bool:
"""Install FFmpeg on Linux using appropriate package manager."""
try:
# Detect the package manager
if shutil.which('apt'):
if shutil.which("apt"):
# Debian/Ubuntu
subprocess.run(['sudo', 'apt', 'update'], check=True, timeout=60)
subprocess.run(['sudo', 'apt', 'install', '-y', 'ffmpeg'], check=True, timeout=300)
elif shutil.which('dnf'):
subprocess.run(["sudo", "apt", "update"], check=True, timeout=60)
subprocess.run(["sudo", "apt", "install", "-y", "ffmpeg"], check=True, timeout=300)
elif shutil.which("dnf"):
# Fedora
subprocess.run(['sudo', 'dnf', 'install', '-y', 'ffmpeg'], check=True, timeout=300)
elif shutil.which('pacman'):
subprocess.run(["sudo", "dnf", "install", "-y", "ffmpeg"], check=True, timeout=300)
elif shutil.which("pacman"):
# Arch Linux
subprocess.run(['sudo', 'pacman', '-S', '--noconfirm', 'ffmpeg'], check=True, timeout=300)
elif shutil.which('snap'):
subprocess.run(
["sudo", "pacman", "-S", "--noconfirm", "ffmpeg"],
check=True,
timeout=300,
)
elif shutil.which("snap"):
# Universal snap package
subprocess.run(['sudo', 'snap', 'install', 'ffmpeg'], check=True, timeout=300)
subprocess.run(["sudo", "snap", "install", "ffmpeg"], check=True, timeout=300)
else:
raise Exception("No supported package manager found")
# Verify installation
if not check_ffmpeg_installed():
raise Exception("FFmpeg installation verification failed")
return True
except Exception as e:
logger.error(f"Error installing FFmpeg: {str(e)}")
return False
def auto_install_ffmpeg():
def auto_install_ffmpeg() -> bool:
"""Automatically install FFmpeg based on the operating system."""
if sys.platform == 'win32':
if OS_NAME == "Windows":
return install_ffmpeg_windows()
elif sys.platform == 'darwin':
elif OS_NAME == "Darwin":
return install_ffmpeg_macos()
elif sys.platform.startswith('linux'):
elif OS_NAME == "Linux":
return install_ffmpeg_linux()
else:
logger.info(f"Unsupported operating system: {sys.platform}")
return False
logger.info(f"Unsupported operating system: {OS_NAME}")
return False
src/core/ytsage_logging.py
+63 -49
View File
@@ -5,71 +5,82 @@ This module provides centralized logging configuration for the entire YTSage app
It replaces the inefficient print statements with structured logging using loguru.
"""
import os
import sys
from pathlib import Path
from src.utils.ytsage_constants import APP_LOG_DIR
# Try to import loguru, but handle case where it might not be available
try:
from loguru import logger
LOGURU_AVAILABLE = True
except ImportError:
LOGURU_AVAILABLE = False
# Create a dummy logger class that does nothing
class DummyLogger:
def info(self, *args, **kwargs): pass
def debug(self, *args, **kwargs): pass
def warning(self, *args, **kwargs): pass
def error(self, *args, **kwargs): pass
def critical(self, *args, **kwargs): pass
def remove(self, *args, **kwargs): pass
def add(self, *args, **kwargs): pass
def bind(self, *args, **kwargs): return self
def info(self, *args, **kwargs):
pass
def debug(self, *args, **kwargs):
pass
def warning(self, *args, **kwargs):
pass
def error(self, *args, **kwargs):
pass
def critical(self, *args, **kwargs):
pass
def remove(self, *args, **kwargs):
pass
def add(self, *args, **kwargs):
pass
def bind(self, *args, **kwargs):
return self
@property
def _core(self):
class Core:
handlers = []
return Core()
logger = DummyLogger()
def setup_logging():
"""
Configure loguru logging for YTSage application.
Sets up multiple log levels and outputs:
- Console output for INFO and above
- File output for DEBUG and above
- File output for DEBUG and above
- Separate error log file for ERROR and above
"""
if not LOGURU_AVAILABLE:
return logger
# Remove default logger to avoid duplicate output
try:
logger.remove()
except Exception:
pass
# Get the application data directory with fallbacks
try:
if sys.platform == 'win32':
localappdata = os.environ.get('LOCALAPPDATA')
if localappdata:
log_dir = Path(localappdata) / 'YTSage' / 'logs'
else:
# Fallback for PyInstaller or when LOCALAPPDATA is not set
log_dir = Path.home() / 'AppData' / 'Local' / 'YTSage' / 'logs'
elif sys.platform == 'darwin':
log_dir = Path.home() / 'Library' / 'Application Support' / 'YTSage' / 'logs'
else:
log_dir = Path.home() / '.local' / 'share' / 'YTSage' / 'logs'
# logic moved to src\utils\ytsage_constants.py
log_dir = APP_LOG_DIR
except Exception:
# Ultimate fallback - use current directory
log_dir = Path.cwd() / 'logs'
log_dir = Path.cwd() / "logs"
# Create log directory if it doesn't exist
try:
log_dir.mkdir(parents=True, exist_ok=True)
@@ -80,11 +91,11 @@ def setup_logging():
log_dir.mkdir(exist_ok=True)
except Exception:
pass # If we still can't create it, we'll just log to console
# Console handler - INFO and above, with colors
# Check if stdout is available (it might be None in PyInstaller windowed apps)
stdout_available = sys.stdout is not None
if stdout_available:
try:
logger.add(
@@ -92,7 +103,7 @@ def setup_logging():
level="INFO",
format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
colorize=True,
catch=True
catch=True,
)
except Exception:
# Fallback to basic console logging without colors
@@ -101,11 +112,11 @@ def setup_logging():
sys.stdout,
level="INFO",
format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {message}",
catch=True
catch=True,
)
except Exception:
stdout_available = False
# If stdout is not available, try stderr or skip console logging entirely
if not stdout_available:
try:
@@ -114,26 +125,26 @@ def setup_logging():
sys.stderr,
level="WARNING", # Only warnings and errors to stderr
format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {message}",
catch=True
catch=True,
)
except Exception:
# If even stderr fails, we'll rely only on file logging
pass
# Only add file handlers if we successfully created a log directory
if log_dir and log_dir.exists():
try:
# Main log file - DEBUG and above, with rotation
logger.add(
log_dir / "ytsage.log",
level="DEBUG",
level="DEBUG",
format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",
rotation="10 MB", # Rotate when file reaches 10MB
retention="7 days", # Keep logs for 7 days
compression="zip", # Compress old logs
catch=True
catch=True,
)
# Error log file - ERROR and above only
logger.add(
log_dir / "ytsage_errors.log",
@@ -142,12 +153,12 @@ def setup_logging():
rotation="5 MB",
retention="30 days", # Keep error logs longer
compression="zip",
catch=True
catch=True,
)
except Exception as e:
# If file logging fails, just log to console
logger.warning(f"Could not set up file logging: {e}")
# Log startup message if we have any handlers
if logger._core.handlers:
logger.info("YTSage logging system initialized")
@@ -155,12 +166,13 @@ def setup_logging():
logger.debug(f"Log directory: {log_dir}")
else:
logger.warning("File logging disabled - could not create log directory")
# If no handlers were successfully added, add a null handler to prevent errors
if not logger._core.handlers:
# Add a minimal handler that just discards messages
# This prevents loguru from complaining about no handlers
import tempfile
try:
# Try to add a temporary file handler as last resort
temp_log = Path(tempfile.gettempdir()) / "ytsage_temp.log"
@@ -169,17 +181,17 @@ def setup_logging():
# If even that fails, we're in a very restricted environment
# loguru should handle this gracefully with its internal fallbacks
pass
return logger
def get_logger(name: str = None):
def get_logger(name: str | None = None):
"""
Get a logger instance for a specific module.
Args:
name: Name of the module/component requesting the logger
Returns:
Configured logger instance
"""
@@ -191,12 +203,13 @@ def get_logger(name: str = None):
# Initialize logging when module is imported - with maximum safety
_setup_complete = False
def safe_setup():
"""Safely initialize logging with multiple fallback strategies."""
global _setup_complete
if _setup_complete:
return logger
try:
setup_logging()
_setup_complete = True
@@ -207,12 +220,13 @@ def safe_setup():
logger.remove()
except Exception:
pass
# At this point, just ensure we have something that won't crash
_setup_complete = True
return logger
# Try to set up logging, but don't let it crash the module import
try:
safe_setup()
@@ -221,4 +235,4 @@ except Exception:
pass
# Export the main logger for convenience
__all__ = ['logger', 'get_logger', 'setup_logging']
__all__ = ["logger", "get_logger", "setup_logging"]
src/core/ytsage_style.py
+1 -1
View File
@@ -149,4 +149,4 @@ QMessageBox QLabel {
QMessageBox QPushButton {
min-width: 80px;
}
"""
"""
src/core/ytsage_utils.py
+228 -258
View File
@@ -1,191 +1,201 @@
import sys
import os
import json
import os
import subprocess
import sys
import tempfile
import time
from pathlib import Path
import subprocess
import tempfile
import shutil
import pkg_resources
from packaging import version
import requests
from .ytsage_logging import logger
from .ytsage_ffmpeg import check_ffmpeg_installed, get_ffmpeg_install_path, get_ffmpeg_path
from .ytsage_yt_dlp import get_yt_dlp_path # Import the new function to avoid import errors
from packaging import version
from src.core.ytsage_ffmpeg import check_ffmpeg_installed, get_ffmpeg_install_path
from src.core.ytsage_logging import logger
from src.core.ytsage_yt_dlp import get_yt_dlp_path
from src.utils.ytsage_constants import (
APP_CONFIG_FILE,
OS_NAME,
SUBPROCESS_CREATIONFLAGS,
USER_HOME_DIR,
YTDLP_APP_BIN_PATH,
YTDLP_DOWNLOAD_URL,
)
# Cache for version information to avoid delays
_version_cache = {
'ytdlp': {'version': None, 'path': None, 'last_check': 0, 'path_mtime': 0},
'ffmpeg': {'version': None, 'path': None, 'last_check': 0, 'path_mtime': 0}
"ytdlp": {"version": None, "path": None, "last_check": 0, "path_mtime": 0},
"ffmpeg": {"version": None, "path": None, "last_check": 0, "path_mtime": 0},
}
# Cache expiry time in seconds (5 minutes)
CACHE_EXPIRY = 300
def get_file_mtime(filepath):
def get_file_mtime(filepath) -> float:
"""Get file modification time safely."""
try:
if filepath and os.path.exists(filepath):
return os.path.getmtime(filepath)
if filepath and Path(filepath).exists():
return Path(filepath).stat().st_mtime
except Exception:
pass
return 0
def should_refresh_cache(tool_name, current_path):
def should_refresh_cache(tool_name, current_path) -> bool:
"""Determine if cache should be refreshed for a tool."""
cache = _version_cache.get(tool_name, {})
current_time = time.time()
# Always refresh if no cached data
if not cache.get('version'):
if not cache.get("version"):
return True
# Refresh if path changed
if cache.get('path') != current_path:
if cache.get("path") != current_path:
return True
# Refresh if file was modified
current_mtime = get_file_mtime(current_path)
if current_mtime > cache.get('path_mtime', 0):
if current_mtime > cache.get("path_mtime", 0):
return True
# Refresh if cache expired
if current_time - cache.get('last_check', 0) > CACHE_EXPIRY:
if current_time - cache.get("last_check", 0) > CACHE_EXPIRY:
return True
return False
def update_version_cache(tool_name, version_info, path, force_save=False):
def update_version_cache(tool_name, version_info, path, force_save=False) -> None:
"""Update the version cache and optionally save to config."""
current_time = time.time()
current_mtime = get_file_mtime(path)
_version_cache[tool_name] = {
'version': version_info,
'path': path,
'last_check': current_time,
'path_mtime': current_mtime
"version": version_info,
"path": path,
"last_check": current_time,
"path_mtime": current_mtime,
}
# Save to persistent config
if force_save:
save_version_cache_to_config()
def load_version_cache_from_config():
def load_version_cache_from_config() -> None:
"""Load cached version info from config file."""
try:
config = load_config()
cached_versions = config.get('cached_versions', {})
cached_versions = config.get("cached_versions", {})
for tool_name, cache_data in cached_versions.items():
if tool_name in _version_cache:
_version_cache[tool_name].update(cache_data)
except Exception as e:
logger.error(f"Error loading version cache: {e}")
def save_version_cache_to_config():
def save_version_cache_to_config() -> None:
"""Save version cache to config file."""
try:
config = load_config()
config['cached_versions'] = _version_cache.copy()
config["cached_versions"] = _version_cache.copy()
save_config(config)
except Exception as e:
logger.error(f"Error saving version cache: {e}")
def get_ytdlp_version_cached():
def get_ytdlp_version_cached() -> str:
"""Get yt-dlp version with caching support."""
try:
current_path = get_yt_dlp_path()
# Check if we need to refresh cache
if not should_refresh_cache('ytdlp', current_path):
cached_version = _version_cache['ytdlp'].get('version')
if not should_refresh_cache("ytdlp", current_path):
cached_version = _version_cache["ytdlp"].get("version")
if cached_version:
return cached_version
# Get fresh version info
version_info = get_ytdlp_version_direct(current_path)
# Update cache
update_version_cache('ytdlp', version_info, current_path)
update_version_cache("ytdlp", version_info, current_path)
return version_info
except Exception as e:
logger.error(f"Error getting cached yt-dlp version: {e}")
return "Error getting version"
def get_ffmpeg_version_cached():
def get_ffmpeg_version_cached() -> str:
"""Get FFmpeg version with caching support."""
try:
# Try to find ffmpeg path
current_path = "ffmpeg" # Default to system PATH
# Check if we need to refresh cache
if not should_refresh_cache('ffmpeg', current_path):
cached_version = _version_cache['ffmpeg'].get('version')
if not should_refresh_cache("ffmpeg", current_path):
cached_version = _version_cache["ffmpeg"].get("version")
if cached_version:
return cached_version
# Get fresh version info
version_info = get_ffmpeg_version_direct()
# Update cache
update_version_cache('ffmpeg', version_info, current_path)
update_version_cache("ffmpeg", version_info, current_path)
return version_info
except Exception as e:
logger.error(f"Error getting cached FFmpeg version: {e}")
return "Error getting version"
def refresh_version_cache(force=False):
def refresh_version_cache(force=False) -> bool:
"""Manually refresh version cache for both tools."""
try:
# Refresh yt-dlp
current_path = get_yt_dlp_path()
version_info = get_ytdlp_version_direct(current_path)
update_version_cache('ytdlp', version_info, current_path, force_save=True)
update_version_cache("ytdlp", version_info, current_path, force_save=True)
# Refresh FFmpeg
version_info = get_ffmpeg_version_direct()
update_version_cache('ffmpeg', version_info, "ffmpeg", force_save=True)
update_version_cache("ffmpeg", version_info, "ffmpeg", force_save=True)
return True
except Exception as e:
logger.error(f"Error refreshing version cache: {e}")
return False
def get_ytdlp_version():
def get_ytdlp_version() -> str:
"""Get the version of yt-dlp (uses cached version for performance)."""
return get_ytdlp_version_cached()
def get_ffmpeg_version():
def get_ffmpeg_version() -> str:
"""Get the version of FFmpeg (uses cached version for performance)."""
return get_ffmpeg_version_cached()
def get_ytdlp_version_direct(yt_dlp_path=None):
def get_ytdlp_version_direct(yt_dlp_path=None) -> str:
"""Get yt-dlp version directly without caching."""
try:
if yt_dlp_path is None:
yt_dlp_path = get_yt_dlp_path()
if not yt_dlp_path or yt_dlp_path == "yt-dlp":
return "Not found"
# Create startupinfo to hide console on Windows
startupinfo = None
if sys.platform == 'win32' and hasattr(subprocess, 'STARTUPINFO'):
startupinfo = subprocess.STARTUPINFO()
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
startupinfo.wShowWindow = 0 # SW_HIDE
# Extra logic moved to src\utils\ytsage_constants.py
result = subprocess.run(
[yt_dlp_path, '--version'],
capture_output=True,
text=True,
timeout=10,
startupinfo=startupinfo
[yt_dlp_path, "--version"], capture_output=True, text=True, timeout=10, creationflags=SUBPROCESS_CREATIONFLAGS
)
if result.returncode == 0:
return result.stdout.strip()
else:
@@ -194,34 +204,25 @@ def get_ytdlp_version_direct(yt_dlp_path=None):
logger.error(f"Error getting yt-dlp version: {e}")
return "Error getting version"
def get_ffmpeg_version_direct():
def get_ffmpeg_version_direct() -> str:
"""Get FFmpeg version directly without caching."""
try:
# Create startupinfo to hide console on Windows
startupinfo = None
if sys.platform == 'win32' and hasattr(subprocess, 'STARTUPINFO'):
startupinfo = subprocess.STARTUPINFO()
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
startupinfo.wShowWindow = 0 # SW_HIDE
# Extra logic moved to src\utils\ytsage_constants.py
result = subprocess.run(
['ffmpeg', '-version'],
capture_output=True,
text=True,
timeout=10,
startupinfo=startupinfo
["ffmpeg", "-version"], capture_output=True, text=True, timeout=10, creationflags=SUBPROCESS_CREATIONFLAGS
)
if result.returncode == 0:
# Parse the first line to get version info
lines = result.stdout.split('\n')
lines = result.stdout.split("\n")
if lines:
first_line = lines[0]
# Extract version from something like "ffmpeg version 4.4.2 Copyright..."
if 'version' in first_line:
if "version" in first_line:
parts = first_line.split()
for i, part in enumerate(parts):
if part == 'version' and i + 1 < len(parts):
if part == "version" and i + 1 < len(parts):
return parts[i + 1]
return first_line.strip()
return "Unknown version"
@@ -231,28 +232,24 @@ def get_ffmpeg_version_direct():
# If ffmpeg is not in PATH, try the installation directory
try:
ffmpeg_path = get_ffmpeg_install_path()
if sys.platform == 'win32':
ffmpeg_exe = os.path.join(ffmpeg_path, 'ffmpeg.exe')
if OS_NAME == "Windows":
ffmpeg_exe = Path(ffmpeg_path).joinpath("ffmpeg.exe")
else:
ffmpeg_exe = os.path.join(ffmpeg_path, 'ffmpeg')
if os.path.exists(ffmpeg_exe):
ffmpeg_exe = Path(ffmpeg_path).joinpath("ffmpeg")
if ffmpeg_exe.exists():
result = subprocess.run(
[ffmpeg_exe, '-version'],
capture_output=True,
text=True,
timeout=10,
startupinfo=startupinfo
[ffmpeg_exe, "-version"], capture_output=True, text=True, timeout=10, creationflags=SUBPROCESS_CREATIONFLAGS
)
if result.returncode == 0:
lines = result.stdout.split('\n')
lines = result.stdout.split("\n")
if lines:
first_line = lines[0]
if 'version' in first_line:
if "version" in first_line:
parts = first_line.split()
for i, part in enumerate(parts):
if part == 'version' and i + 1 < len(parts):
if part == "version" and i + 1 < len(parts):
return parts[i + 1]
return first_line.strip()
return "Unknown version"
@@ -264,49 +261,32 @@ def get_ffmpeg_version_direct():
logger.error(f"Error getting FFmpeg version: {e}")
return "Error getting version"
def get_app_data_dir():
"""Get the OS-specific application data directory."""
if sys.platform == 'win32':
# Windows: %LOCALAPPDATA%\YTSage\data\
return Path(os.environ.get('LOCALAPPDATA', '')) / 'YTSage' / 'data'
elif sys.platform == 'darwin':
# macOS: ~/Library/Application Support/YTSage/data/
return Path.home() / 'Library' / 'Application Support' / 'YTSage' / 'data'
else:
# Linux: ~/.local/share/YTSage/data/
return Path.home() / '.local' / 'share' / 'YTSage' / 'data'
def get_config_file_path():
"""Get the path to the main configuration file."""
return get_app_data_dir() / 'ytsage_config.json'
# get_app_data_dir() moved to src\utils\ytsage_constants.py
# get_config_file_path() moved to src\utils\ytsage_constants.py
# ensure_app_data_dir() moved to src\utils\ytsage_constants.py
def ensure_app_data_dir():
"""Ensure the application data directory exists."""
data_dir = get_app_data_dir()
data_dir.mkdir(parents=True, exist_ok=True)
return data_dir
def load_config():
def load_config() -> dict:
"""Load the application configuration from file."""
config_file = get_config_file_path()
default_config = {
'download_path': str(Path.home() / 'Downloads'),
'speed_limit_value': None,
'speed_limit_unit_index': 0,
'cookie_file_path': None,
'last_used_cookie_file': None,
'auto_update_ytdlp': True, # Enable auto-update by default
'auto_update_frequency': 'daily', # daily, weekly, or startup
'last_update_check': 0, # timestamp of last check
'cached_versions': {
'ytdlp': {'version': None, 'path': None, 'last_check': 0, 'path_mtime': 0},
'ffmpeg': {'version': None, 'path': None, 'last_check': 0, 'path_mtime': 0}
}
"download_path": str(USER_HOME_DIR / "Downloads"),
"speed_limit_value": None,
"speed_limit_unit_index": 0,
"cookie_file_path": None,
"last_used_cookie_file": None,
"auto_update_ytdlp": True, # Enable auto-update by default
"auto_update_frequency": "daily", # daily, weekly, or startup
"last_update_check": 0, # timestamp of last check
"cached_versions": {
"ytdlp": {"version": None, "path": None, "last_check": 0, "path_mtime": 0},
"ffmpeg": {"version": None, "path": None, "last_check": 0, "path_mtime": 0},
},
}
try:
if config_file.exists():
with open(config_file, 'r', encoding='utf-8') as f:
if APP_CONFIG_FILE.exists():
with open(APP_CONFIG_FILE, "r", encoding="utf-8") as f:
config = json.load(f)
# Merge with defaults to ensure all keys exist
for key, value in default_config.items():
@@ -317,178 +297,160 @@ def load_config():
logger.error(f"Error reading config file: {e}")
# If config file is corrupted, create a new one with defaults
save_config(default_config)
return default_config
def save_config(config):
def save_config(config) -> bool:
"""Save the application configuration to file."""
config_file = get_config_file_path()
try:
# Ensure the config directory exists
ensure_app_data_dir()
with open(config_file, 'w', encoding='utf-8') as f:
with open(APP_CONFIG_FILE, "w", encoding="utf-8") as f:
json.dump(config, f, ensure_ascii=False, indent=2)
return True
except Exception as e:
logger.error(f"Error saving config: {e}")
return False
def check_ffmpeg():
def check_ffmpeg() -> bool:
"""Check if FFmpeg is installed and accessible with enhanced error handling."""
try:
# Use the enhanced FFmpeg check from ytsage_ffmpeg
if check_ffmpeg_installed():
return True
# For Windows, try to add the FFmpeg path to environment
if sys.platform == 'win32':
if OS_NAME == "Windows":
ffmpeg_path = get_ffmpeg_install_path()
if os.path.exists(os.path.join(ffmpeg_path, 'ffmpeg.exe')):
if ffmpeg_path.joinpath("ffmpeg.exe").exists():
try:
# Add to current session PATH
os.environ['PATH'] = f"{ffmpeg_path}{os.pathsep}{os.environ.get('PATH', '')}"
os.environ["PATH"] = f"{ffmpeg_path}{os.pathsep}{os.environ.get('PATH', '')}"
return True
except Exception as e:
logger.error(f"Error updating PATH: {e}")
return False
# For macOS, check common paths
elif sys.platform == 'darwin':
elif OS_NAME == "Darwin":
common_paths = [
'/usr/local/bin/ffmpeg',
'/opt/homebrew/bin/ffmpeg',
'/usr/bin/ffmpeg'
"/usr/local/bin/ffmpeg",
"/opt/homebrew/bin/ffmpeg",
"/usr/bin/ffmpeg",
]
for path in common_paths:
if os.path.exists(path):
if Path(path).exists():
try:
ffmpeg_dir = os.path.dirname(path)
os.environ['PATH'] = f"{ffmpeg_dir}{os.pathsep}{os.environ.get('PATH', '')}"
ffmpeg_dir = Path(path).parent
os.environ["PATH"] = f"{ffmpeg_dir}{os.pathsep}{os.environ.get('PATH', '')}"
return True
except Exception as e:
logger.error(f"Error updating PATH: {e}")
continue
return False
except Exception as e:
logger.error(f"Error checking FFmpeg: {e}")
return False
def load_saved_path(main_window_instance):
def load_saved_path(main_window_instance) -> None:
"""Load saved download path with enhanced error handling."""
config_file = get_config_file_path()
try:
if config_file.exists():
if APP_CONFIG_FILE.exists():
try:
with open(config_file, 'r', encoding='utf-8') as f:
with open(APP_CONFIG_FILE, "r", encoding="utf-8") as f:
config = json.load(f)
saved_path = config.get('download_path', '')
if os.path.exists(saved_path) and os.access(saved_path, os.W_OK):
saved_path = config.get("download_path", "")
if Path(saved_path).exists() and os.access(saved_path, os.W_OK):
main_window_instance.last_path = saved_path
return
except (json.JSONDecodeError, UnicodeError) as e:
logger.error(f"Error reading config file: {e}")
# If config file is corrupted, try to remove it
try:
os.remove(config_file)
APP_CONFIG_FILE.unlink(missing_ok=True)
except Exception:
pass
# Fallback to Downloads folder
downloads_path = str(Path.home() / 'Downloads')
if os.path.exists(downloads_path) and os.access(downloads_path, os.W_OK):
downloads_path = USER_HOME_DIR / "Downloads"
if downloads_path.exists() and os.access(downloads_path, os.W_OK):
main_window_instance.last_path = downloads_path
else:
# Final fallback to temp directory if Downloads is not accessible
main_window_instance.last_path = tempfile.gettempdir()
except Exception as e:
logger.error(f"Error loading saved settings: {e}")
main_window_instance.last_path = tempfile.gettempdir()
def save_path(main_window_instance, path):
def save_path(main_window_instance, path) -> bool:
"""Save download path with enhanced error handling."""
config_file = get_config_file_path()
try:
# Verify the path is valid and writable
if not os.path.exists(path):
if not Path(path).exists():
try:
os.makedirs(path, exist_ok=True)
Path(path).mkdir(exist_ok=True)
except Exception as e:
logger.error(f"Error creating directory: {e}")
return False
if not os.access(path, os.W_OK):
logger.info("Path is not writable")
return False
# Ensure the config directory exists
ensure_app_data_dir()
# Save the config
config = {'download_path': path}
with open(config_file, 'w', encoding='utf-8') as f:
config = {"download_path": path}
with open(APP_CONFIG_FILE, "w", encoding="utf-8") as f:
json.dump(config, f, ensure_ascii=False)
return True
except Exception as e:
logger.error(f"Error saving settings: {e}")
return False
def update_yt_dlp():
def update_yt_dlp() -> bool:
"""Check for yt-dlp updates and update if a newer version is available."""
try:
# Get the yt-dlp path
yt_dlp_path = get_yt_dlp_path()
# Create startupinfo to hide console on Windows
startupinfo = None
if sys.platform == 'win32' and hasattr(subprocess, 'STARTUPINFO'):
startupinfo = subprocess.STARTUPINFO()
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
startupinfo.wShowWindow = 0 # SW_HIDE
# Extra logic moved to src\utils\ytsage_constants.py
# For binaries downloaded with our app, use direct binary update approach
if os.path.dirname(yt_dlp_path) in [
os.path.join(os.environ.get('LOCALAPPDATA', ''), 'YTSage', 'bin'),
os.path.expanduser(os.path.join('~', 'Library', 'Application Support', 'YTSage', 'bin')),
os.path.expanduser(os.path.join('~', '.local', 'share', 'YTSage', 'bin'))
]:
if yt_dlp_path.samefile(YTDLP_APP_BIN_PATH):
# We're using a binary installed by our app, update directly
logger.info(f"Updating yt-dlp binary at {yt_dlp_path}")
# Determine the URL based on OS
if sys.platform == 'win32':
url = "https://github.com/yt-dlp/yt-dlp/releases/latest/download/yt-dlp.exe"
elif sys.platform == 'darwin':
url = "https://github.com/yt-dlp/yt-dlp/releases/latest/download/yt-dlp_macos"
else:
url = "https://github.com/yt-dlp/yt-dlp/releases/latest/download/yt-dlp"
# Extra logic moved to src\utils\ytsage_constants.py
# Download the latest version
try:
response = requests.get(url, stream=True)
response = requests.get(YTDLP_DOWNLOAD_URL, stream=True)
if response.status_code == 200:
# Create a temporary file
temp_file = f"{yt_dlp_path}.new"
with open(temp_file, 'wb') as f:
with open(temp_file, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
# Make executable on Unix systems
if sys.platform != 'win32':
if OS_NAME != "Windows":
os.chmod(temp_file, 0o755)
# Replace the old file with the new one
try:
# On Windows, we need to remove the old file first
if sys.platform == 'win32' and os.path.exists(yt_dlp_path):
os.remove(yt_dlp_path)
os.rename(temp_file, yt_dlp_path)
if OS_NAME == "Windows" and yt_dlp_path.exists():
yt_dlp_path.unlink(missing_ok=True)
Path(temp_file).rename(yt_dlp_path)
logger.info("yt-dlp binary successfully updated")
return True
except Exception as e:
@@ -503,7 +465,7 @@ def update_yt_dlp():
else:
# We're using a system-installed yt-dlp, use pip to update
logger.info("Using pip to update yt-dlp")
# Get current version
try:
current_version = pkg_resources.get_distribution("yt-dlp").version
@@ -511,7 +473,7 @@ def update_yt_dlp():
except pkg_resources.DistributionNotFound:
logger.info("yt-dlp not installed via pip, attempting update anyway")
current_version = "0.0.0" # Assume very old version to force update
# Get the latest version from PyPI JSON API
try:
response = requests.get("https://pypi.org/pypi/yt-dlp/json", timeout=10)
@@ -519,16 +481,23 @@ def update_yt_dlp():
data = response.json()
latest_version = data["info"]["version"]
logger.info(f"Latest available yt-dlp version: {latest_version}")
# Compare versions and update if needed
if version.parse(latest_version) > version.parse(current_version):
logger.info(f"Updating yt-dlp from {current_version} to {latest_version}...")
update_result = subprocess.run(
[sys.executable, "-m", "pip", "install", "--upgrade", "yt-dlp"],
[
sys.executable,
"-m",
"pip",
"install",
"--upgrade",
"yt-dlp",
],
capture_output=True,
text=True,
check=False,
startupinfo=startupinfo
creationflags=SUBPROCESS_CREATIONFLAGS,
)
if update_result.returncode == 0:
logger.info("yt-dlp successfully updated")
@@ -544,75 +513,76 @@ def update_yt_dlp():
logger.error(f"Error checking for yt-dlp updates: {e}")
except Exception as e:
logger.info(f"Unexpected error during yt-dlp update: {e}")
return False
def should_check_for_auto_update():
def should_check_for_auto_update() -> bool:
"""Check if auto-update should be performed based on user settings."""
try:
config = load_config()
# Check if auto-update is enabled
if not config.get('auto_update_ytdlp', False):
if not config.get("auto_update_ytdlp", False):
return False
frequency = config.get('auto_update_frequency', 'daily')
last_check = config.get('last_update_check', 0)
frequency = config.get("auto_update_frequency", "daily")
last_check = config.get("last_update_check", 0)
current_time = time.time()
# Calculate time since last check
time_diff = current_time - last_check
if frequency == 'startup':
if frequency == "startup":
# Always check on startup if we haven't checked in the last hour
return time_diff > 3600 # 1 hour
elif frequency == 'daily':
elif frequency == "daily":
return time_diff > 86400 # 24 hours
elif frequency == 'weekly':
elif frequency == "weekly":
return time_diff > 604800 # 7 days
return False
except Exception as e:
logger.error(f"Error checking auto-update schedule: {e}")
return False
def check_and_update_ytdlp_auto():
def check_and_update_ytdlp_auto() -> bool:
"""Perform automatic yt-dlp update check and update if needed."""
try:
logger.info("Performing automatic yt-dlp update check...")
# Get current version
current_version = get_ytdlp_version()
if "Error" in current_version:
logger.info("Could not determine current yt-dlp version, skipping auto-update")
return False
# Get latest version from PyPI
try:
response = requests.get("https://pypi.org/pypi/yt-dlp/json", timeout=10)
response.raise_for_status()
latest_version = response.json()["info"]["version"]
# Clean up version strings
current_version = current_version.replace('_', '.')
latest_version = latest_version.replace('_', '.')
current_version = current_version.replace("_", ".")
latest_version = latest_version.replace("_", ".")
logger.info(f"Current yt-dlp version: {current_version}")
logger.info(f"Latest yt-dlp version: {latest_version}")
# Compare versions
from packaging import version as version_parser
if version_parser.parse(latest_version) > version_parser.parse(current_version):
logger.info(f"Auto-updating yt-dlp from {current_version} to {latest_version}...")
# Perform the update
if update_yt_dlp():
logger.info("Auto-update completed successfully!")
# Update the last check timestamp
config = load_config()
config['last_update_check'] = time.time()
config["last_update_check"] = time.time()
save_config(config)
return True
else:
@@ -622,40 +592,40 @@ def check_and_update_ytdlp_auto():
logger.info("yt-dlp is already up to date")
# Still update the timestamp even if no update was needed
config = load_config()
config['last_update_check'] = time.time()
config["last_update_check"] = time.time()
save_config(config)
return True
except requests.RequestException as e:
logger.info(f"Network error during auto-update check: {e}")
return False
except Exception as e:
logger.error(f"Error during auto-update check: {e}")
return False
except Exception as e:
logger.info(f"Critical error in auto-update: {e}")
return False
def get_auto_update_settings():
def get_auto_update_settings() -> dict:
"""Get current auto-update settings from config."""
config = load_config()
return {
'enabled': config.get('auto_update_ytdlp', True),
'frequency': config.get('auto_update_frequency', 'daily'),
'last_check': config.get('last_update_check', 0)
"enabled": config.get("auto_update_ytdlp", True),
"frequency": config.get("auto_update_frequency", "daily"),
"last_check": config.get("last_update_check", 0),
}
def update_auto_update_settings(enabled, frequency):
def update_auto_update_settings(enabled, frequency) -> bool:
"""Update auto-update settings in config."""
try:
config = load_config()
config['auto_update_ytdlp'] = enabled
config['auto_update_frequency'] = frequency
config["auto_update_ytdlp"] = enabled
config["auto_update_frequency"] = frequency
save_config(config)
return True
except Exception as e:
logger.error(f"Error updating auto-update settings: {e}")
return False
return False
src/core/ytsage_yt_dlp.py
+189 -234
View File
@@ -1,87 +1,64 @@
import os
import sys
import platform
import shutil
import subprocess
import requests
from pathlib import Path
from PySide6.QtWidgets import (QDialog, QVBoxLayout, QLabel, QPushButton,
QProgressBar, QRadioButton, QHBoxLayout,
QMessageBox, QFileDialog, QWidget)
from PySide6.QtCore import QThread, Signal, Qt
from typing import Optional
import requests
from PySide6.QtCore import Qt, QThread, Signal
from PySide6.QtGui import QIcon
from .ytsage_logging import logger
from PySide6.QtWidgets import (
QDialog,
QFileDialog,
QHBoxLayout,
QLabel,
QMessageBox,
QProgressBar,
QPushButton,
QRadioButton,
QVBoxLayout,
QWidget,
)
# Define binary URLs
YTDLP_URLS = {
"windows": "https://github.com/yt-dlp/yt-dlp/releases/latest/download/yt-dlp.exe",
"macos": "https://github.com/yt-dlp/yt-dlp/releases/latest/download/yt-dlp_macos",
"linux": "https://github.com/yt-dlp/yt-dlp/releases/latest/download/yt-dlp"
}
from src.core.ytsage_logging import logger
from src.utils.ytsage_constants import (
APP_BIN_DIR,
ICON_PATH,
OS_FULL_NAME,
OS_NAME,
SUBPROCESS_CREATIONFLAGS,
YTDLP_APP_BIN_PATH,
YTDLP_DOWNLOAD_URL,
)
# Define installation paths
def get_ytdlp_install_dir():
"""Get the OS-specific yt-dlp installation directory"""
if sys.platform == 'win32':
# Windows: %LOCALAPPDATA%\YTSage\bin\
return os.path.join(os.environ.get('LOCALAPPDATA'), 'YTSage', 'bin')
elif sys.platform == 'darwin':
# macOS: ~/Library/Application Support/YTSage/bin/
return os.path.expanduser(os.path.join('~', 'Library', 'Application Support', 'YTSage', 'bin'))
else:
# Linux: ~/.local/share/YTSage/bin/
return os.path.expanduser(os.path.join('~', '.local', 'share', 'YTSage', 'bin'))
# YTDLP_URLS moved to src\utils\ytsage_constants.py
# get_ytdlp_install_dir() moved to src\utils\ytsage_constants.py
# get_ytdlp_executable_path() moved to src\utils\ytsage_constants.py
# get_os_type() moved to src\utils\ytsage_constants.py
# ensure_install_dir_exists() moved to src\utils\ytsage_constants.py
def get_ytdlp_executable_path():
"""Get the full path to the yt-dlp executable based on OS"""
install_dir = get_ytdlp_install_dir()
if sys.platform == 'win32':
return os.path.join(install_dir, 'yt-dlp.exe')
else:
return os.path.join(install_dir, 'yt-dlp')
def get_os_type():
"""Detect the operating system"""
if sys.platform == 'win32':
return "windows"
elif sys.platform == 'darwin':
return "macos"
else:
return "linux"
def ensure_install_dir_exists():
"""Make sure the installation directory exists"""
install_dir = get_ytdlp_install_dir()
os.makedirs(install_dir, exist_ok=True)
return install_dir
class DownloadYtdlpThread(QThread):
progress_signal = Signal(int)
finished_signal = Signal(bool, str)
def __init__(self, os_type):
def __init__(self):
super().__init__()
self.os_type = os_type
def run(self):
def run(self) -> None:
try:
url = YTDLP_URLS[self.os_type]
install_dir = ensure_install_dir_exists()
if self.os_type == "windows":
exe_path = os.path.join(install_dir, "yt-dlp.exe")
else:
exe_path = os.path.join(install_dir, "yt-dlp")
# Extra logic moved to src\utils\ytsage_constants.py
exe_path = YTDLP_APP_BIN_PATH
# Download with progress reporting
response = requests.get(url, stream=True)
total_size = int(response.headers.get('content-length', 0))
response = requests.get(YTDLP_DOWNLOAD_URL, stream=True)
total_size = int(response.headers.get("content-length", 0))
block_size = 1024 # 1 Kibibyte
if total_size == 0:
self.progress_signal.emit(100)
with open(exe_path, 'wb') as f:
with open(exe_path, "wb") as f:
downloaded = 0
for data in response.iter_content(block_size):
f.write(data)
@@ -89,44 +66,41 @@ class DownloadYtdlpThread(QThread):
if total_size > 0:
progress = int(downloaded / total_size * 100)
self.progress_signal.emit(progress)
# Make executable on macOS and Linux
if self.os_type != "windows":
if OS_NAME != "Windows":
os.chmod(exe_path, 0o755)
self.finished_signal.emit(True, exe_path)
except Exception as e:
self.finished_signal.emit(False, str(e))
class YtdlpSetupDialog(QDialog):
setup_complete = Signal(str) # Signal emitting the path to yt-dlp
def __init__(self, parent=None):
super().__init__(parent)
self.os_type = get_os_type()
self.setWindowTitle("yt-dlp Setup Required")
self.setMinimumWidth(520)
self.setMinimumHeight(350)
self.resize(520, 380)
# Set the window icon to match the main app
if parent and parent.windowIcon():
self.setWindowIcon(parent.windowIcon())
else:
# Try to load the icon directly if parent not available
# Navigate from src/core/ to project root, then to assets/Icon/
current_dir = os.path.dirname(os.path.abspath(__file__)) # core/
src_dir = os.path.dirname(current_dir) # src/
project_root = os.path.dirname(src_dir) # project root
icon_path = os.path.join(project_root, 'assets', 'Icon', 'icon.png')
if os.path.exists(icon_path):
self.setWindowIcon(QIcon(icon_path))
# icon_path logic moved to src\utils\ytsage_constants.py
icon_path = ICON_PATH
if Path.exists(icon_path):
self.setWindowIcon(QIcon(icon_path.as_posix()))
self.init_ui()
# Apply dark theme styling to match app
self.setStyleSheet("""
self.setStyleSheet(
"""
QDialog {
background-color: #15181b;
color: #ffffff;
@@ -190,55 +164,54 @@ class YtdlpSetupDialog(QDialog):
border: 2px solid #c90000;
background: #c90000;
}
""")
def init_ui(self):
"""
)
def init_ui(self) -> None:
layout = QVBoxLayout()
layout.setSpacing(15)
layout.setContentsMargins(25, 25, 25, 25)
# Header title
title_label = QLabel("yt-dlp Setup Required")
title_label.setStyleSheet("font-size: 16px; font-weight: bold; color: #ffffff; padding: 5px 0;")
title_label.setAlignment(Qt.AlignCenter)
title_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
layout.addWidget(title_label)
# Information label with improved styling
if self.os_type == "windows":
os_name = "Windows"
elif self.os_type == "macos":
os_name = "macOS"
else:
os_name = "Linux"
info_label = QLabel(f"YTSage requires yt-dlp to download videos.<br><br>"
f"yt-dlp was not found in the app's local directory. "
f"YTSage needs to set up yt-dlp for your {os_name} system.<br><br>"
f"Please choose an option below:")
info_label.setAlignment(Qt.AlignCenter)
# os_name logic moved to src\utils\ytsage_constants.py
info_label = QLabel(
f"YTSage requires yt-dlp to download videos.<br><br>"
f"yt-dlp was not found in the app's local directory. "
f"YTSage needs to set up yt-dlp for your {OS_FULL_NAME} system.<br><br>"
f"Please choose an option below:"
)
info_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
info_label.setWordWrap(True)
info_label.setStyleSheet("font-size: 13px; color: #cccccc; padding: 5px; line-height: 1.4;")
layout.addWidget(info_label)
# Radio buttons with minimal spacing
option_widget = QWidget()
option_layout = QVBoxLayout(option_widget)
option_layout.setSpacing(8)
option_layout.setContentsMargins(0, 0, 0, 0)
self.auto_radio = QRadioButton("Download automatically (Recommended)")
self.auto_radio.setChecked(True)
self.manual_radio = QRadioButton("Select path manually")
option_layout.addWidget(self.auto_radio)
option_layout.addWidget(self.manual_radio)
layout.addWidget(option_widget)
# Progress bar with proper sizing
self.progress_bar = QProgressBar()
self.progress_bar.setVisible(False)
self.progress_bar.setFixedHeight(20) # Fixed height for consistency
self.progress_bar.setStyleSheet("""
self.progress_bar.setStyleSheet(
"""
QProgressBar {
border: 1px solid #3d3d3d;
border-radius: 8px;
@@ -254,61 +227,62 @@ class YtdlpSetupDialog(QDialog):
border-radius: 6px;
margin: 1px;
}
""")
"""
)
layout.addWidget(self.progress_bar)
# Status label with better spacing
self.status_label = QLabel("")
self.status_label.setAlignment(Qt.AlignCenter)
self.status_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.status_label.setStyleSheet("font-size: 12px; color: #cccccc; padding: 8px 0;")
self.status_label.setWordWrap(True)
layout.addWidget(self.status_label)
# Add stretch to push buttons to bottom
layout.addStretch()
# Button layout with improved spacing
button_layout = QHBoxLayout()
button_layout.setSpacing(15)
button_layout.setContentsMargins(0, 10, 0, 0) # Add top margin for buttons
self.setup_button = QPushButton("Setup yt-dlp")
self.setup_button.clicked.connect(self.setup_ytdlp)
self.cancel_button = QPushButton("Cancel")
self.cancel_button.clicked.connect(self.reject)
button_layout.addWidget(self.setup_button)
button_layout.addWidget(self.cancel_button)
layout.addLayout(button_layout)
self.setLayout(layout)
def setup_ytdlp(self):
def setup_ytdlp(self) -> None:
if self.auto_radio.isChecked():
self.download_ytdlp()
else:
self.select_ytdlp_path()
def download_ytdlp(self):
def download_ytdlp(self) -> None:
self.progress_bar.setVisible(True)
self.progress_bar.setValue(0)
self.status_label.setText("Downloading yt-dlp...")
self.setup_button.setEnabled(False)
self.cancel_button.setEnabled(False)
self.download_thread = DownloadYtdlpThread(self.os_type)
self.download_thread = DownloadYtdlpThread()
self.download_thread.progress_signal.connect(self.update_progress)
self.download_thread.finished_signal.connect(self.download_finished)
self.download_thread.start()
def update_progress(self, value):
def update_progress(self, value) -> None:
self.progress_bar.setValue(value)
def download_finished(self, success, result):
def download_finished(self, success, result) -> None:
self.setup_button.setEnabled(True)
self.cancel_button.setEnabled(True)
if success:
self.status_label.setText("yt-dlp was successfully installed!")
self.setup_complete.emit(result)
@@ -316,12 +290,13 @@ class YtdlpSetupDialog(QDialog):
else:
self.status_label.setText(f"Error: {result}")
error_dialog = QMessageBox(self)
error_dialog.setIcon(QMessageBox.Critical)
error_dialog.setIcon(QMessageBox.Icon.Critical)
error_dialog.setWindowTitle("Download Failed")
error_dialog.setText(f"Failed to download yt-dlp: {result}")
# Set the window icon to match the main dialog
error_dialog.setWindowIcon(self.windowIcon())
error_dialog.setStyleSheet("""
error_dialog.setStyleSheet(
"""
QMessageBox {
background-color: #15181b;
color: #ffffff;
@@ -340,18 +315,20 @@ class YtdlpSetupDialog(QDialog):
QPushButton:hover {
background-color: #a50000;
}
""")
"""
)
error_dialog.exec()
def select_ytdlp_path(self):
if self.os_type == "windows":
def select_ytdlp_path(self) -> None:
if OS_NAME == "Windows":
file_filter = "Executable Files (*.exe)"
else:
file_filter = "All Files (*)"
# Apply style to QFileDialog
file_dialog = QFileDialog(self)
file_dialog.setStyleSheet("""
file_dialog.setStyleSheet(
"""
QFileDialog {
background-color: #15181b;
color: #ffffff;
@@ -371,57 +348,42 @@ class YtdlpSetupDialog(QDialog):
QPushButton:hover {
background-color: #a50000;
}
""")
file_path, _ = file_dialog.getOpenFileName(
self, "Select yt-dlp executable", "", file_filter
"""
)
file_path, _ = file_dialog.getOpenFileName(self, "Select yt-dlp executable", "", file_filter)
if file_path:
logger.debug(f"User selected file: {file_path}")
# Verify the selected file
try:
# Set up startupinfo to hide console window on Windows
startupinfo = None
if sys.platform == 'win32' and hasattr(subprocess, 'STARTUPINFO'):
startupinfo = subprocess.STARTUPINFO()
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
startupinfo.wShowWindow = 0 # SW_HIDE
# Extra logic moved to src\utils\ytsage_constants.py
# Try to run yt-dlp --version
logger.debug(f"Verifying file with --version command")
result = subprocess.run(
[file_path, "--version"],
capture_output=True,
text=True,
check=False,
startupinfo=startupinfo
[file_path, "--version"], capture_output=True, text=True, check=False, creationflags=SUBPROCESS_CREATIONFLAGS
)
logger.debug(f"Version check result: {result.returncode}, Output: {result.stdout.strip()}")
if result.returncode == 0:
# File is valid, copy it to our app's bin directory
try:
# Ensure the bin directory exists
install_dir = ensure_install_dir_exists()
logger.debug(f"Install directory: {install_dir}")
logger.debug(f"Install directory: {APP_BIN_DIR}")
# Determine the target filename based on OS
if self.os_type == "windows":
target_path = os.path.join(install_dir, "yt-dlp.exe")
else:
target_path = os.path.join(install_dir, "yt-dlp")
target_path = YTDLP_APP_BIN_PATH
logger.debug(f"Target path: {target_path}")
# Copy the file
shutil.copy2(file_path, target_path)
logger.debug(f"File copied successfully")
# Set executable permissions on Unix systems
if self.os_type != "windows":
if OS_NAME != "Windows":
os.chmod(target_path, 0o755)
logger.debug(f"Permissions set on Unix system")
# Return the path of the copied file
self.status_label.setText(f"yt-dlp successfully copied to {target_path}")
logger.debug(f"Emitting setup_complete signal with path: {target_path}")
@@ -430,10 +392,11 @@ class YtdlpSetupDialog(QDialog):
except Exception as copy_error:
logger.debug(f"Error copying file: {str(copy_error)}")
error_dialog = QMessageBox(self)
error_dialog.setIcon(QMessageBox.Critical)
error_dialog.setIcon(QMessageBox.Icon.Critical)
error_dialog.setWindowTitle("Setup Error")
error_dialog.setText(f"Error copying yt-dlp to app directory: {str(copy_error)}")
error_dialog.setStyleSheet("""
error_dialog.setStyleSheet(
"""
QMessageBox {
background-color: #15181b;
color: #ffffff;
@@ -452,15 +415,17 @@ class YtdlpSetupDialog(QDialog):
QPushButton:hover {
background-color: #a50000;
}
""")
"""
)
error_dialog.exec()
else:
logger.debug(f"File verification failed with return code: {result.returncode}")
error_dialog = QMessageBox(self)
error_dialog.setIcon(QMessageBox.Warning)
error_dialog.setIcon(QMessageBox.Icon.Warning)
error_dialog.setWindowTitle("Invalid Executable")
error_dialog.setText("The selected file does not appear to be a valid yt-dlp executable.")
error_dialog.setStyleSheet("""
error_dialog.setStyleSheet(
"""
QMessageBox {
background-color: #15181b;
color: #ffffff;
@@ -479,15 +444,17 @@ class YtdlpSetupDialog(QDialog):
QPushButton:hover {
background-color: #a50000;
}
""")
"""
)
error_dialog.exec()
except Exception as e:
logger.debug(f"Exception during verification: {str(e)}")
error_dialog = QMessageBox(self)
error_dialog.setIcon(QMessageBox.Critical)
error_dialog.setIcon(QMessageBox.Icon.Critical)
error_dialog.setWindowTitle("Error")
error_dialog.setText(f"Error verifying yt-dlp executable: {str(e)}")
error_dialog.setStyleSheet("""
error_dialog.setStyleSheet(
"""
QMessageBox {
background-color: #15181b;
color: #ffffff;
@@ -506,63 +473,57 @@ class YtdlpSetupDialog(QDialog):
QPushButton:hover {
background-color: #a50000;
}
""")
"""
)
error_dialog.exec()
def check_ytdlp_binary():
def check_ytdlp_binary() -> Optional[Path]:
"""
Check if yt-dlp binary exists in the expected location.
Returns:
str or None: Path to yt-dlp binary if found, None otherwise
Path or None: Path to yt-dlp binary if found, None otherwise
"""
exe_path = get_ytdlp_executable_path()
if os.path.exists(exe_path):
exe_path = YTDLP_APP_BIN_PATH
if exe_path.exists():
# Make sure it's executable on Unix systems
if sys.platform != 'win32' and not os.access(exe_path, os.X_OK):
if OS_NAME != "Windows" and not os.access(exe_path, os.X_OK):
try:
os.chmod(exe_path, 0o755)
logger.info(f"Fixed permissions on yt-dlp at {exe_path}")
except Exception as e:
logger.warning(f"Could not set executable permissions on {exe_path}: {e}")
return None
return exe_path
# If not found in app directory, check if yt-dlp is available in PATH
try:
# Use subprocess to check if yt-dlp is available
if sys.platform == 'win32':
if OS_NAME == "Windows":
# On Windows, use 'where' command and hide console window
startupinfo = None
if hasattr(subprocess, 'STARTUPINFO'):
startupinfo = subprocess.STARTUPINFO()
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
startupinfo.wShowWindow = 0 # SW_HIDE
# Extra logic moved to src\utils\ytsage_constants.py
result = subprocess.run(
['where', 'yt-dlp'],
capture_output=True,
text=True,
check=False,
startupinfo=startupinfo
["where", "yt-dlp"], capture_output=True, text=True, check=False, creationflags=SUBPROCESS_CREATIONFLAGS
)
if result.returncode == 0 and result.stdout.strip():
yt_dlp_path = result.stdout.strip().split('\n')[0]
yt_dlp_path = result.stdout.strip().split("\n")[0]
logger.info(f"Found yt-dlp in PATH: {yt_dlp_path}")
return yt_dlp_path
return Path(yt_dlp_path)
else:
# On Unix systems, use 'which' command
result = subprocess.run(['which', 'yt-dlp'], capture_output=True, text=True, check=False)
result = subprocess.run(["which", "yt-dlp"], capture_output=True, text=True, check=False)
if result.returncode == 0 and result.stdout.strip():
yt_dlp_path = result.stdout.strip()
logger.info(f"Found yt-dlp in PATH: {yt_dlp_path}")
return yt_dlp_path
return Path(yt_dlp_path)
except Exception as e:
logger.error(f"Error checking for yt-dlp in PATH: {e}")
# We're only interested in our app-specific installation or system PATH
# We're only interested in our app-specific installation or system PATH
return None
def check_ytdlp_installed():
def check_ytdlp_installed() -> bool:
"""
Check if yt-dlp is installed and accessible.
Returns:
@@ -573,19 +534,9 @@ def check_ytdlp_installed():
if ytdlp_path:
# Try to run yt-dlp --version to verify it's working
try:
# Create startupinfo to hide console on Windows
startupinfo = None
if sys.platform == 'win32' and hasattr(subprocess, 'STARTUPINFO'):
startupinfo = subprocess.STARTUPINFO()
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
startupinfo.wShowWindow = 0 # SW_HIDE
# Extra logic moved to src\utils\ytsage_constants.py
result = subprocess.run(
[ytdlp_path, '--version'],
capture_output=True,
text=True,
timeout=5,
startupinfo=startupinfo
[ytdlp_path, "--version"], capture_output=True, text=True, timeout=5, creationflags=SUBPROCESS_CREATIONFLAGS
)
return result.returncode == 0
except Exception:
@@ -594,7 +545,8 @@ def check_ytdlp_installed():
except Exception:
return False
def get_yt_dlp_path():
def get_yt_dlp_path() -> Path:
"""
Get the yt-dlp path, either from the app's bin directory or system PATH.
This replaces the function in ytsage_utils.py.
@@ -606,10 +558,11 @@ def get_yt_dlp_path():
if ytdlp_path:
logger.info(f"Using yt-dlp from: {ytdlp_path}")
return ytdlp_path
# If not found anywhere, fall back to the command name as a last resort
logger.info("yt-dlp not found in app directory or PATH, falling back to command name")
return "yt-dlp"
return "yt-dlp" # type: ignore[return-value]
def setup_ytdlp(parent_widget=None):
"""
@@ -619,33 +572,33 @@ def setup_ytdlp(parent_widget=None):
"""
logger.debug("Starting yt-dlp setup dialog")
dialog = YtdlpSetupDialog(parent_widget)
# Store the setup result from the signal
setup_result = {"path": None}
def on_setup_complete(path):
def on_setup_complete(path) -> None:
logger.debug(f"Received setup_complete signal with path: {path}")
setup_result["path"] = path
# Connect to the setup_complete signal
dialog.setup_complete.connect(on_setup_complete)
# Show the dialog
result = dialog.exec()
logger.debug(f"Dialog result: {result} (Accepted={QDialog.Accepted})")
if result == QDialog.Accepted:
logger.debug(f"Dialog result: {result} (Accepted={QDialog.DialogCode.Accepted})")
if result == QDialog.DialogCode.Accepted:
# First check if we received a path from the signal
if setup_result["path"] and os.path.exists(setup_result["path"]):
if setup_result["path"] and Path.exists(setup_result["path"]):
logger.debug(f"Using path from signal: {setup_result['path']}")
return setup_result["path"]
# Get the expected path for verification as fallback
expected_path = get_ytdlp_executable_path()
expected_path = YTDLP_APP_BIN_PATH
logger.debug(f"Expected yt-dlp path: {expected_path}")
# Verify the path exists after dialog is accepted
if os.path.exists(expected_path):
if Path.exists(expected_path):
logger.debug(f"yt-dlp successfully found at expected path: {expected_path}")
return expected_path
else:
@@ -653,20 +606,21 @@ def setup_ytdlp(parent_widget=None):
# Try to use the get_yt_dlp_path function to find yt-dlp elsewhere
yt_dlp_path = get_yt_dlp_path()
logger.debug(f"Alternate detection result: {yt_dlp_path}")
if yt_dlp_path != "yt-dlp" and os.path.exists(yt_dlp_path):
if yt_dlp_path != "yt-dlp" and Path.exists(yt_dlp_path):
logger.debug(f"yt-dlp found at alternate location: {yt_dlp_path}")
return yt_dlp_path
# Something went wrong, show an error message
logger.debug(f"Setup failed, showing error dialog")
if parent_widget:
error_dialog = QMessageBox(parent_widget)
error_dialog.setIcon(QMessageBox.Warning)
error_dialog.setIcon(QMessageBox.Icon.Warning)
error_dialog.setWindowTitle("Setup Failed")
error_dialog.setText("Failed to set up yt-dlp. Some features may not work correctly.")
# Set the window icon to match the parent
error_dialog.setWindowIcon(parent_widget.windowIcon())
error_dialog.setStyleSheet("""
error_dialog.setStyleSheet(
"""
QMessageBox {
background-color: #15181b;
color: #ffffff;
@@ -685,12 +639,13 @@ def setup_ytdlp(parent_widget=None):
QPushButton:hover {
background-color: #a50000;
}
""")
"""
)
error_dialog.exec()
logger.warning(f"yt-dlp setup failed, path does not exist: {expected_path}")
else:
logger.debug("User cancelled the setup dialog")
# User cancelled or setup failed, return the fallback command
logger.debug("Returning fallback command 'yt-dlp'")
return "yt-dlp"
return "yt-dlp"