Remove yt-dlp Python package usage; use binary only

Refactored all modules to remove dependency on the yt-dlp Python package, switching to subprocess calls to the yt-dlp binary for all extraction and update operations. Update logic now exclusively manages the app's own yt-dlp binary and no longer supports pip-based updates. Simplified format/audio detection and thumbnail extraction to use CLI heuristics and subprocess output. This improves reliability and avoids Python package conflicts.
This commit is contained in:
oop7
2025-10-25 19:49:46 +03:00
parent aab5fd9549
commit 41d35060c1
5 changed files with 93 additions and 614 deletions
+9 -111
View File
@@ -15,14 +15,6 @@ from src.utils.ytsage_logger import logger
# Shorthand for localization # Shorthand for localization
_ = LocalizationManager.get_text _ = LocalizationManager.get_text
try:
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")
class SignalManager(QObject): class SignalManager(QObject):
update_formats = Signal(list) update_formats = Signal(list)
@@ -162,63 +154,11 @@ class DownloadThread(QThread):
def check_file_exists(self) -> bool | None: def check_file_exists(self) -> bool | None:
"""Check if the file already exists before downloading""" """Check if the file already exists before downloading"""
try: # This method is kept for backwards compatibility but always returns False
logger.debug("Starting file existence check") # to proceed with download. File existence is now handled by yt-dlp CLI itself.
# Use yt-dlp to get the filename without downloading, suppressing warnings logger.debug("Skipping file existence check (handled by yt-dlp)")
ydl_opts_check = {
"logger": logger, # passed app logger
"quiet": True,
"skip_download": True,
"no_warnings": True, # <-- Suppress warnings during check
"ignoreerrors": True, # Also ignore other potential errors during this check
"outtmpl": {"default": str(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"] = str(self.cookie_file)
elif self.browser_cookies:
ydl_opts_check["cookiesfrombrowser"] = (
self.browser_cookies.split(":")[0],
self.browser_cookies.split(":")[1] if ":" in self.browser_cookies else None,
)
# Add proxy settings if specified
if self.proxy_url:
ydl_opts_check["proxy"] = self.proxy_url
if self.geo_proxy_url:
ydl_opts_check["geo_verification_proxy"] = self.geo_proxy_url
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")
# 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", "")
break
logger.debug(f"Resolution: {resolution}")
else:
logger.debug("yt-dlp not available, skipping file existence check")
return False # Proceed with download attempt
except Exception as e:
logger.exception(f"Error checking file existence: {e}")
return None
def _build_yt_dlp_command(self) -> list: def _build_yt_dlp_command(self) -> list:
"""Build the yt-dlp command line with all options for direct execution.""" """Build the yt-dlp command line with all options for direct execution."""
# Use the new yt-dlp path function from ytsage_yt_dlp module # Use the new yt-dlp path function from ytsage_yt_dlp module
@@ -231,26 +171,9 @@ class DownloadThread(QThread):
# Strip the -drc suffix if present to fix issues with certain audio formats # 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 # Use a simple heuristic: if format_id contains 'audio' or ends with 'a', treat as audio-only
is_audio_format = False # This avoids the need for Python API metadata extraction
try: is_audio_format = "audio" in clean_format_id.lower() or clean_format_id.endswith("a")
if YT_DLP_AVAILABLE:
ydl_opts = {
"logger": logger,
"quiet": True,
"no_warnings": True,
"skip_download": True,
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
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.exception(f"Error checking if format is audio-only: {e}")
# For audio-only formats, don't try to merge with video # For audio-only formats, don't try to merge with video
if is_audio_format: if is_audio_format:
@@ -261,35 +184,10 @@ class DownloadThread(QThread):
logger.debug(f"Using video format selection with audio: {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 # Determine output format based on the selected format ID - only for video formats
# Note: This is a best-effort approach without Python API metadata
if not is_audio_format: if not is_audio_format:
try: # Let yt-dlp handle format detection automatically via CLI
format_ext = None logger.debug("Letting yt-dlp CLI determine output format automatically")
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, "logger": logger}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
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")
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")
break
if format_ext:
logger.debug(f"Detected format extension: {format_ext}")
# Ensure output matches the selected format - only for video formats
cmd.extend(["--merge-output-format", format_ext])
except Exception as e:
logger.exception(f"Error detecting format extension: {e}")
# If we can't determine the format, don't specify merge-output-format
pass
else: else:
# If no specific format ID, use resolution-based sorting (-S) # 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 res_value = self.resolution if self.resolution else "720" # Default to 720p if no resolution specified
+6 -27
View File
@@ -480,9 +480,10 @@ class YtdlpSetupDialog(QDialog):
def check_ytdlp_binary() -> Optional[Path]: def check_ytdlp_binary() -> Optional[Path]:
""" """
Check if yt-dlp binary exists in the expected location. Check if yt-dlp binary exists in the app's bin directory ONLY.
We now ignore system PATH and only use our managed binary.
Returns: Returns:
Path or None: Path to yt-dlp binary if found, None otherwise Path or None: Path to yt-dlp binary if found in app bin, None otherwise
""" """
exe_path = YTDLP_APP_BIN_PATH exe_path = YTDLP_APP_BIN_PATH
if exe_path.exists(): if exe_path.exists():
@@ -493,33 +494,11 @@ def check_ytdlp_binary() -> Optional[Path]:
logger.info(f"Fixed permissions on yt-dlp at {exe_path}") logger.info(f"Fixed permissions on yt-dlp at {exe_path}")
except Exception as e: except Exception as e:
logger.exception(f"Could not set executable permissions on {exe_path}: {e}") logger.exception(f"Could not set executable permissions on {exe_path}: {e}")
logger.info(f"Found yt-dlp in app bin directory: {exe_path}")
return exe_path return exe_path
# If not found in app directory, check if yt-dlp is available in PATH # Binary not found in app directory - return None to trigger setup
try: logger.warning(f"yt-dlp binary not found in app bin directory: {exe_path}")
# Use subprocess to check if yt-dlp is available
if OS_NAME == "Windows":
# On Windows, use 'where' command and hide console window
# Extra logic moved to src\utils\ytsage_constants.py
result = subprocess.run(
["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]
logger.info(f"Found yt-dlp in PATH: {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)
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 Path(yt_dlp_path)
except Exception as e:
logger.exception(f"Error checking for yt-dlp in PATH: {e}")
# We're only interested in our app-specific installation or system PATH
return None return None
@@ -1,13 +1,12 @@
""" """
Update-related dialogs and threads for YTSage application. Update-related dialogs and threads for YTSage application.
Contains dialogs and background threads for checking and performing yt-dlp updates. Contains dialogs and background threads for checking and performing yt-dlp binary updates.
Note: This module only handles binary updates. Python package updates have been removed.
""" """
import os import os
import subprocess import subprocess
import sys
import time import time
from importlib.metadata import PackageNotFoundError
from pathlib import Path from pathlib import Path
import requests import requests
@@ -25,30 +24,6 @@ _ = LocalizationManager.get_text
from src.utils.ytsage_localization import _ from src.utils.ytsage_localization import _
from src.utils.ytsage_logger import logger from src.utils.ytsage_logger import logger
try:
from importlib.metadata import PackageNotFoundError as ImportlibPackageNotFoundError
from importlib.metadata import version as importlib_version
def get_version(package_name: str) -> str:
return importlib_version(package_name)
PackageNotFoundError = ImportlibPackageNotFoundError
except ImportError:
# Fallback for older Python versions
import pkg_resources
def get_version(package_name: str) -> str:
return pkg_resources.get_distribution(package_name).version
PackageNotFoundError = pkg_resources.DistributionNotFound
try:
import yt_dlp
YT_DLP_AVAILABLE = True
except ImportError:
YT_DLP_AVAILABLE = False
class VersionCheckThread(QThread): class VersionCheckThread(QThread):
finished = Signal(str, str, str) # current_version, latest_version, error_message finished = Signal(str, str, str) # current_version, latest_version, error_message
@@ -73,31 +48,20 @@ class VersionCheckThread(QThread):
) )
if result.returncode == 0: if result.returncode == 0:
current_version = result.stdout.strip() current_version = result.stdout.strip()
else: # Try fallback if command failed
if YT_DLP_AVAILABLE:
current_version = yt_dlp.version.__version__ # type: ignore[reportAttributeAccessIssue]
else: else:
error_message = "yt-dlp not available." error_message = "yt-dlp binary not accessible."
self.finished.emit(current_version, latest_version, error_message) self.finished.emit(current_version, latest_version, error_message)
return return
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
# Try fallback if timeout error_message = "yt-dlp version check timed out."
if YT_DLP_AVAILABLE:
current_version = yt_dlp.version.__version__ # type: ignore[reportAttributeAccessIssue]
else:
error_message = "yt-dlp version check timed out and package not found."
self.finished.emit(current_version, latest_version, error_message) self.finished.emit(current_version, latest_version, error_message)
return return
except Exception: except Exception as e:
# Fallback to importing yt_dlp package directly if subprocess fails error_message = f"yt-dlp not found or accessible: {e}"
if YT_DLP_AVAILABLE:
current_version = yt_dlp.version.__version__ # type: ignore[reportAttributeAccessIssue]
else:
error_message = "yt-dlp not found or accessible."
self.finished.emit(current_version, latest_version, error_message) self.finished.emit(current_version, latest_version, error_message)
return return
# Get latest version from PyPI # Get latest version from PyPI (yt-dlp releases are also published to PyPI)
response = requests.get("https://pypi.org/pypi/yt-dlp/json", timeout=10) response = requests.get("https://pypi.org/pypi/yt-dlp/json", timeout=10)
response.raise_for_status() response.raise_for_status()
latest_version = response.json()["info"]["version"] latest_version = response.json()["info"]["version"]
@@ -135,33 +99,11 @@ class UpdateThread(QThread):
self.update_finished.emit(False, _('update.error_getting_path', error=e)) self.update_finished.emit(False, _('update.error_getting_path', error=e))
return return
# Extra logic moved to src\utils\ytsage_constants.py
self.update_progress.emit(20) self.update_progress.emit(20)
# Extra logic moved to src\utils\ytsage_constants.py # Update the binary (no more pip-based updates)
# Check if this is an app-managed binary by comparing paths safely
is_app_managed = False
try:
# Only compare if both files exist
if yt_dlp_path.exists() and YTDLP_APP_BIN_PATH.exists():
is_app_managed = yt_dlp_path.samefile(YTDLP_APP_BIN_PATH)
elif str(yt_dlp_path) == str(YTDLP_APP_BIN_PATH):
# If paths are identical as strings, consider it app-managed
is_app_managed = True
else:
# If app binary doesn't exist, this is definitely not app-managed
is_app_managed = False
except (OSError, IOError) as e:
logger.debug(f"Error comparing paths: {e}")
is_app_managed = False
if is_app_managed:
self.update_status.emit(_('update.updating_binary')) self.update_status.emit(_('update.updating_binary'))
success = self._update_binary(yt_dlp_path) success = self._update_binary(yt_dlp_path)
else:
self.update_status.emit(_('update.updating_pip'))
success = self._update_via_pip()
if success: if success:
self.update_progress.emit(100) self.update_progress.emit(100)
@@ -219,77 +161,6 @@ class UpdateThread(QThread):
self.update_status.emit(_('update.unexpected_error', error=e)) self.update_status.emit(_('update.unexpected_error', error=e))
return False return False
def _update_via_pip(self) -> bool:
"""Update yt-dlp via pip."""
try:
self.update_status.emit(_('update.checking_pip'))
self.update_progress.emit(30)
# Get current version
try:
current_version = get_version("yt-dlp")
self.update_status.emit(_('update.current_version', version=current_version))
except PackageNotFoundError:
self.update_status.emit(_('update.not_found_pip'))
current_version = "0.0.0"
self.update_progress.emit(40)
# Get the latest version from PyPI
self.update_status.emit(_('update.checking_latest'))
response = requests.get("https://pypi.org/pypi/yt-dlp/json", timeout=10)
if response.status_code != 200:
self.update_status.emit(_('update.failed_check_updates'))
return False
data = response.json()
latest_version = data["info"]["version"]
self.update_status.emit(_('update.latest_version', version=latest_version))
self.update_progress.emit(50)
# Compare versions
if version.parse(latest_version) > version.parse(current_version):
self.update_status.emit(_('update.updating_from_to', current=current_version, latest=latest_version))
self.update_progress.emit(60)
try:
# Run pip update with timeout
self.update_status.emit(_('update.running_pip_install'))
update_result = subprocess.run(
[sys.executable, "-m", "pip", "install", "--upgrade", "yt-dlp"],
capture_output=True,
text=True,
check=False,
timeout=300, # 5 minute timeout for pip install
creationflags=SUBPROCESS_CREATIONFLAGS,
)
self.update_progress.emit(85)
if update_result.returncode == 0:
self.update_status.emit(_('update.pip_completed'))
self.update_progress.emit(95)
return True
else:
self.update_status.emit(_('update.pip_failed', error=update_result.stderr))
return False
except subprocess.TimeoutExpired:
self.update_status.emit(_("update.pip_timeout"))
return False
except Exception as e:
self.update_status.emit(_('update.error_pip_update', error=e))
return False
else:
self.update_status.emit(_("update.already_up_to_date"))
self.update_progress.emit(95)
return True
except Exception as e:
self.update_status.emit(_('update.pip_update_failed', error=e))
return False
class YTDLPUpdateDialog(QDialog): class YTDLPUpdateDialog(QDialog):
def __init__(self, parent=None) -> None: def __init__(self, parent=None) -> None:
@@ -569,35 +440,14 @@ class AutoUpdateThread(QThread):
self.update_finished.emit(False, f"Critical error: {e}") self.update_finished.emit(False, f"Critical error: {e}")
def _perform_update(self) -> bool: def _perform_update(self) -> bool:
"""Perform the actual update using similar logic to UpdateThread but without UI feedback.""" """Perform the actual binary update."""
try: try:
# Get the yt-dlp path # Get the yt-dlp path
yt_dlp_path = get_yt_dlp_path() yt_dlp_path = get_yt_dlp_path()
# Extra logic moved to src\utils\ytsage_constants.py # Always update the binary (no more pip-based updates)
logger.info("AutoUpdateThread: Updating yt-dlp binary...")
# Check if this is an app-managed binary by comparing paths safely
is_app_managed = False
try:
# Only compare if both files exist
if yt_dlp_path.exists() and YTDLP_APP_BIN_PATH.exists():
is_app_managed = yt_dlp_path.samefile(YTDLP_APP_BIN_PATH)
elif str(yt_dlp_path) == str(YTDLP_APP_BIN_PATH):
# If paths are identical as strings, consider it app-managed
is_app_managed = True
else:
# If app binary doesn't exist, this is definitely not app-managed
is_app_managed = False
except (OSError, IOError) as e:
logger.debug(f"AutoUpdateThread: Error comparing paths: {e}")
is_app_managed = False
if is_app_managed:
logger.info("AutoUpdateThread: Updating app-managed yt-dlp binary...")
return self._update_binary(yt_dlp_path) return self._update_binary(yt_dlp_path)
else:
logger.info("AutoUpdateThread: Updating system yt-dlp via pip...")
return self._update_via_pip()
except Exception as e: except Exception as e:
logger.exception(f"AutoUpdateThread: Error in _perform_update: {e}") logger.exception(f"AutoUpdateThread: Error in _perform_update: {e}")
@@ -636,58 +486,3 @@ class AutoUpdateThread(QThread):
except Exception as e: except Exception as e:
logger.exception(f"AutoUpdateThread: Unexpected error during update: {e}") logger.exception(f"AutoUpdateThread: Unexpected error during update: {e}")
return False return False
def _update_via_pip(self) -> bool:
"""Update yt-dlp via pip (silent version)."""
try:
logger.info("AutoUpdateThread: Checking current pip installation...")
# Get current version
try:
current_version = get_version("yt-dlp")
logger.info(f"AutoUpdateThread: Current version: {current_version}")
except PackageNotFoundError:
logger.warning("AutoUpdateThread: yt-dlp not found via pip, attempting installation...")
current_version = "0.0.0"
# Get the latest version from PyPI
logger.info("AutoUpdateThread: Checking for latest version...")
response = requests.get("https://pypi.org/pypi/yt-dlp/json", timeout=10)
if response.status_code != 200:
logger.error("AutoUpdateThread: Failed to check for updates")
return False
data = response.json()
latest_version = data["info"]["version"]
logger.info(f"AutoUpdateThread: Latest version: {latest_version}")
# Compare versions
if version.parse(latest_version) > version.parse(current_version):
logger.info(f"AutoUpdateThread: Updating from {current_version} to {latest_version}...")
# Extra logic moved to src\utils\ytsage_constants.py
# Run pip update
logger.info("AutoUpdateThread: Running pip install --upgrade...")
update_result = subprocess.run(
[sys.executable, "-m", "pip", "install", "--upgrade", "yt-dlp"],
capture_output=True,
text=True,
check=False,
creationflags=SUBPROCESS_CREATIONFLAGS,
)
if update_result.returncode == 0:
logger.info("AutoUpdateThread: Pip update completed successfully!")
return True
else:
logger.error(f"AutoUpdateThread: Pip update failed: {update_result.stderr}")
return False
else:
logger.info("AutoUpdateThread: yt-dlp is already up to date!")
return True
except Exception as e:
logger.exception(f"AutoUpdateThread: Pip update failed: {e}")
return False
+9 -213
View File
@@ -48,13 +48,15 @@ from src.utils.ytsage_logger import logger
from src.utils.ytsage_config_manager import ConfigManager from src.utils.ytsage_config_manager import ConfigManager
from src.utils.ytsage_localization import LocalizationManager, _ from src.utils.ytsage_localization import LocalizationManager, _
try: # Note: yt-dlp Python package removed - using binary-only approach
import yt_dlp # DownloadError and ExtractorError definitions kept for compatibility
from yt_dlp.utils import DownloadError, ExtractorError class DownloadError(Exception):
"""Compatibility class for yt-dlp DownloadError"""
pass
YT_DLP_AVAILABLE = True class ExtractorError(Exception):
except ImportError: """Compatibility class for yt-dlp ExtractorError"""
YT_DLP_AVAILABLE = False pass
class YTSageApp(QMainWindow, FormatTableMixin, VideoInfoMixin): # Inherit from mixins class YTSageApp(QMainWindow, FormatTableMixin, VideoInfoMixin): # Inherit from mixins
@@ -65,10 +67,6 @@ class YTSageApp(QMainWindow, FormatTableMixin, VideoInfoMixin): # Inherit from
saved_language = ConfigManager.get("language") or "en" saved_language = ConfigManager.get("language") or "en"
LocalizationManager.initialize(saved_language) LocalizationManager.initialize(saved_language)
# Log startup warnings for missing dependencies
if not YT_DLP_AVAILABLE:
logger.warning("yt-dlp not available at startup, will be downloaded at runtime")
# Check for FFmpeg before proceeding # Check for FFmpeg before proceeding
if not check_ffmpeg(): if not check_ffmpeg():
self.show_ffmpeg_dialog() self.show_ffmpeg_dialog()
@@ -722,210 +720,8 @@ class YTSageApp(QMainWindow, FormatTableMixin, VideoInfoMixin): # Inherit from
playlist_id = url.split("list=")[1].split("&")[0] playlist_id = url.split("list=")[1].split("&")[0]
url = f"https://www.youtube.com/playlist?list={playlist_id}" url = f"https://www.youtube.com/playlist?list={playlist_id}"
# Check if yt-dlp Python module is available # Always use subprocess to call yt-dlp binary (Python package removed)
if not YT_DLP_AVAILABLE:
# Use subprocess to call yt-dlp executable
self._analyze_url_with_subprocess(url) self._analyze_url_with_subprocess(url)
return
# Initial extraction with basic options - suppress warnings here too
ydl_opts = {
"logger": logger,
"quiet": False,
"no_warnings": True, # <-- Suppress warnings for initial check
"extract_flat": True,
"force_generic_extractor": False,
"ignoreerrors": False, # Set to False to catch errors properly
"no_color": True,
"verbose": True,
"cookiefile": None,
}
# Add cookies argument if cookie file path is set
if self.cookie_file_path:
ydl_opts["cookiefile"] = str(self.cookie_file_path)
elif self.browser_cookies_option:
ydl_opts["cookiesfrombrowser"] = (
self.browser_cookies_option.split(":")[0],
self.browser_cookies_option.split(":")[1] if ":" in self.browser_cookies_option else None,
)
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
try:
basic_info = ydl.extract_info(url, download=False)
if not basic_info:
logger.error("Could not extract basic video information")
self.signals.update_status.emit(
_("main_ui.error_extract_info")
)
# Hide playlist UI on error
self.signals.playlist_info_label_visible.emit(False)
self.signals.playlist_select_btn_visible.emit(False)
return
except Exception as e:
logger.exception(f"First extraction failed: {e}")
self.signals.playlist_info_label_visible.emit(False)
self.signals.playlist_select_btn_visible.emit(False)
user_friendly_error = parse_yt_dlp_error(str(e))
self.signals.update_status.emit(user_friendly_error)
return
self.signals.update_status.emit(_("main_ui.analyzing_extracting_detailed"))
# Configure options for detailed extraction (keep other options)
# Add no_warnings here as well, as this is where detailed info is fetched
ydl_opts_detail = {
"logger": logger,
"extract_flat": False,
"format": None,
"writesubtitles": True,
"allsubtitles": True,
"writeautomaticsub": True,
"playliststart": 1,
"playlistend": 1,
"youtube_include_dash_manifest": True,
"youtube_include_hls_manifest": True,
"no_warnings": True, # <-- Add flag here for detailed extraction
}
# Add cookies argument if cookie file path is set
if self.cookie_file_path:
ydl_opts_detail["cookiefile"] = str(self.cookie_file_path)
elif self.browser_cookies_option:
ydl_opts_detail["cookiesfrombrowser"] = (
self.browser_cookies_option.split(":")[0],
self.browser_cookies_option.split(":")[1] if ":" in self.browser_cookies_option else None,
)
# Use a separate options dict for the detailed extraction
with yt_dlp.YoutubeDL(ydl_opts_detail) as ydl_detail:
try:
self.signals.update_status.emit(_("main_ui.analyzing_processing_video"))
if basic_info.get("_type") == "playlist":
self.is_playlist = True
self.playlist_info = basic_info
self.selected_playlist_items = None # Reset selection for new playlist
self.playlist_entries = [entry for entry in basic_info.get("entries", []) if entry] # Store entries
# Ensure there are entries before proceeding
if not self.playlist_entries:
logger.error("Playlist contains no valid videos.")
self.signals.update_status.emit(_("errors.playlist_no_videos"))
self.signals.playlist_info_label_visible.emit(False)
self.signals.playlist_select_btn_visible.emit(False)
return
# Extract detailed info for the FIRST video in the playlist
# This provides formats/subs for the UI, assuming consistency
first_video_url = self.playlist_entries[0].get("url")
if not first_video_url:
logger.error("Could not get URL for the first playlist video.")
self.signals.update_status.emit(_("errors.playlist_no_url"))
self.signals.playlist_info_label_visible.emit(False)
self.signals.playlist_select_btn_visible.emit(False)
return
try:
# Use the ydl_detail instance with no_warnings
self.video_info = ydl_detail.extract_info(first_video_url, download=False)
except Exception as e:
logger.exception(f"Failed to extract info for the first playlist video: {e}")
self.signals.playlist_info_label_visible.emit(False)
self.signals.playlist_select_btn_visible.emit(False)
user_friendly_error = parse_yt_dlp_error(str(e))
self.signals.update_status.emit(user_friendly_error)
return
# Update playlist info label text (remains the same)
playlist_text = _("playlist.display_format",
title=basic_info.get('title', _('playlist.unknown')),
count=len(self.playlist_entries))
# update signal method from QMetaObject.invokeMethod to signals
self.signals.playlist_info_label_text.emit(playlist_text)
self.signals.playlist_info_label_visible.emit(True)
# Show playlist selection BUTTON
# update signal method from QMetaObject.invokeMethod to signals
self.signals.playlist_select_btn_text.emit(_("main_ui.select_videos_all"))
self.signals.playlist_select_btn_visible.emit(True)
else: # Single video
self.is_playlist = False
# Use ydl_detail instance here too for consistency
self.video_info = ydl_detail.extract_info(url, download=False)
self.playlist_entries = [] # Clear entries
self.selected_playlist_items = None # Clear selection
# Hide playlist info label and button
# update signal method from QMetaObject.invokeMethod to signals
self.signals.playlist_info_label_visible.emit(False)
self.signals.playlist_select_btn_visible.emit(False)
# Verify we have format information
if not self.video_info or "formats" not in self.video_info:
logger.debug(f"video_info keys: {self.video_info.keys() if self.video_info else 'None'}")
self.signals.update_status.emit(_("main_ui.error_no_format_info"))
self.signals.playlist_info_label_visible.emit(False)
self.signals.playlist_select_btn_visible.emit(False)
return
self.signals.update_status.emit(_("main_ui.analyzing_processing_formats"))
self.all_formats = self.video_info["formats"]
# Update UI
self.update_video_info(self.video_info)
# Update thumbnail
self.signals.update_status.emit(_("main_ui.analyzing_loading_thumbnail"))
# Try to get thumbnail from playlist info first
# Fallback to video thumbnail if playlist thumbnail not found or not a playlist
thumbnail_url = (self.playlist_info or {}).get("thumbnail") or (self.video_info or {}).get("thumbnail")
self.download_thumbnail(thumbnail_url)
# Save thumbnail if enabled - use the stored VIDEO URL
if self.save_thumbnail:
self.download_thumbnail_file(
self.video_url, self.path_input.text() # type: ignore[reportAttributeAccessIssue]
)
# --- Subtitle Handling ---
self.signals.update_status.emit(_("main_ui.analyzing_processing_subtitles"))
# Clear previous selections when analyzing a new video
self.selected_subtitles = []
self.available_subtitles = self.video_info.get("subtitles", {})
self.available_automatic_subtitles = self.video_info.get("automatic_captions", {})
# Update the UI elements related to subtitle selection state
# update signal method from QMetaObject.invokeMethod to signals
self.signals.selected_subs_label_text.emit(_("main_ui.subtitles_selected", count=0))
# QMetaObject.invokeMethod(
# self.subtitle_select_btn,
# b"setProperty",
# Qt.ConnectionType.QueuedConnection,
# Q_ARG(str, b"subtitlesSelected"),
# Q_ARG(bool, False),
# ) # <-- COMMENT OUT THIS LINE
# REMOVE the merge_subs_checkbox update call from here
# QMetaObject.invokeMethod(
# self.merge_subs_checkbox,
# b"setEnabled",
# Qt.ConnectionType.QueuedConnection,
# Q_ARG(bool, False),
# )
# Update format table
self.signals.update_status.emit(_("main_ui.analyzing_updating_table"))
self.video_button.setChecked(True)
self.audio_button.setChecked(False)
self.filter_formats()
self.signals.update_status.emit(_("main_ui.analysis_complete"))
except Exception as e:
logger.exception(f"Detailed extraction failed: {e}")
self.signals.playlist_info_label_visible.emit(False)
self.signals.playlist_select_btn_visible.emit(False)
# Use the error parser for other extraction errors too
user_friendly_error = parse_yt_dlp_error(str(e))
self.signals.update_status.emit(user_friendly_error)
return
except Exception as e: except Exception as e:
logger.exception(f"Error in analysis: {e}") logger.exception(f"Error in analysis: {e}")
+23 -12
View File
@@ -385,26 +385,36 @@ class VideoInfoMixin:
return False return False
try: try:
# Import yt_dlp locally to avoid import errors when yt-dlp is not installed # Note: yt_dlp Python package removed - this feature now uses subprocess
from yt_dlp import YoutubeDL # Extract thumbnail info using yt-dlp CLI instead
from src.core.ytsage_yt_dlp import get_yt_dlp_path
from src.utils.ytsage_constants import SUBPROCESS_CREATIONFLAGS
import subprocess
import json
logger.debug(f"Attempting to save thumbnail for URL: {video_url}") logger.debug(f"Attempting to save thumbnail for URL: {video_url}")
ydl_opts = { ytdlp_path = get_yt_dlp_path()
"logger": logger,
"quiet": True,
"skip_download": True,
"force_generic_extractor": False,
"no_warnings": True,
"extract_flat": False,
}
with YoutubeDL(ydl_opts) as ydl: # Use yt-dlp CLI to extract thumbnail info
info = ydl.extract_info(video_url, download=False) result = subprocess.run(
[ytdlp_path, "--dump-json", "--skip-download", video_url],
capture_output=True,
text=True,
timeout=30,
creationflags=SUBPROCESS_CREATIONFLAGS,
)
if result.returncode != 0:
logger.error(f"Failed to extract video info: {result.stderr}")
return False
info = json.loads(result.stdout)
thumbnails = info.get("thumbnails", []) thumbnails = info.get("thumbnails", [])
if not thumbnails: if not thumbnails:
logger.info("No thumbnails available") logger.info("No thumbnails available")
return False
thumbnail_url = max( thumbnail_url = max(
thumbnails, thumbnails,
@@ -413,6 +423,7 @@ class VideoInfoMixin:
if not thumbnail_url: if not thumbnail_url:
logger.info("Failed to extract thumbnail URL") logger.info("Failed to extract thumbnail URL")
return False
# Download using requests # Download using requests
response = requests.get(thumbnail_url) response = requests.get(thumbnail_url)