Files
SageTube/src/core/ytsage_utils.py
T
Your Name a6a175b3d7 Improve handling of console window in windowed apps
Suppresses console window flicker for PyInstaller windowed applications by redirecting stdout and stderr in main.py. Updates logging setup to avoid console output in windowed mode and adds fallback error logging. Suppresses pkg_resources deprecation warnings to further reduce unwanted console output.
2025-08-27 22:17:13 +03:00

708 lines
27 KiB
Python

import json
import os
import subprocess
import sys
import tempfile
import time
import warnings
from pathlib import Path
# Suppress the pkg_resources deprecation warning to prevent console window flicker
warnings.filterwarnings("ignore", message=".*pkg_resources is deprecated.*", category=UserWarning)
import pkg_resources
import requests
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},
}
# Cache expiry time in seconds (5 minutes)
CACHE_EXPIRY = 300
def get_file_mtime(filepath) -> float:
"""Get file modification time safely."""
try:
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) -> 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"):
return True
# Refresh if path changed
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):
return True
# Refresh if cache expired
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) -> 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,
}
# Save to persistent config
if force_save:
save_version_cache_to_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", {})
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() -> None:
"""Save version cache to config file."""
try:
config = load_config()
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() -> 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 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)
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() -> 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 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)
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) -> 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)
# Refresh FFmpeg
version_info = get_ffmpeg_version_direct()
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() -> str:
"""Get the version of yt-dlp (uses cached version for performance)."""
return get_ytdlp_version_cached()
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) -> 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"
# Extra logic moved to src\utils\ytsage_constants.py
result = subprocess.run(
[yt_dlp_path, "--version"], capture_output=True, text=True, timeout=10, creationflags=SUBPROCESS_CREATIONFLAGS
)
if result.returncode == 0:
return result.stdout.strip()
else:
return "Error getting version"
except Exception as e:
logger.error(f"Error getting yt-dlp version: {e}")
return "Error getting version"
def get_ffmpeg_version_direct() -> str:
"""Get FFmpeg version directly without caching."""
try:
# Extra logic moved to src\utils\ytsage_constants.py
result = subprocess.run(
["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")
if lines:
first_line = lines[0]
# Extract version from something like "ffmpeg version 4.4.2 Copyright..."
if "version" in first_line:
parts = first_line.split()
for i, part in enumerate(parts):
if part == "version" and i + 1 < len(parts):
return parts[i + 1]
return first_line.strip()
return "Unknown version"
else:
return "Not found"
except FileNotFoundError:
# If ffmpeg is not in PATH, try the installation directory
try:
ffmpeg_path = get_ffmpeg_install_path()
if OS_NAME == "Windows":
ffmpeg_exe = Path(ffmpeg_path).joinpath("ffmpeg.exe")
else:
ffmpeg_exe = Path(ffmpeg_path).joinpath("ffmpeg")
if ffmpeg_exe.exists():
result = subprocess.run(
[ffmpeg_exe, "-version"], capture_output=True, text=True, timeout=10, creationflags=SUBPROCESS_CREATIONFLAGS
)
if result.returncode == 0:
lines = result.stdout.split("\n")
if lines:
first_line = lines[0]
if "version" in first_line:
parts = first_line.split()
for i, part in enumerate(parts):
if part == "version" and i + 1 < len(parts):
return parts[i + 1]
return first_line.strip()
return "Unknown version"
return "Not found"
except Exception as e:
logger.error(f"Error getting FFmpeg version from install path: {e}")
return "Not found"
except Exception as e:
logger.error(f"Error getting FFmpeg version: {e}")
return "Error getting version"
# 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 load_config() -> dict:
"""Load the application configuration from file."""
default_config = {
"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 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():
if key not in config:
config[key] = value
return config
except (json.JSONDecodeError, UnicodeError, Exception) as e:
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) -> bool:
"""Save the application configuration to file."""
try:
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() -> 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 OS_NAME == "Windows":
ffmpeg_path = get_ffmpeg_install_path()
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', '')}"
return True
except Exception as e:
logger.error(f"Error updating PATH: {e}")
return False
# For macOS, check common paths
elif OS_NAME == "Darwin":
common_paths = [
"/usr/local/bin/ffmpeg",
"/opt/homebrew/bin/ffmpeg",
"/usr/bin/ffmpeg",
]
for path in common_paths:
if Path(path).exists():
try:
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) -> None:
"""Load saved download path with enhanced error handling."""
try:
if APP_CONFIG_FILE.exists():
try:
with open(APP_CONFIG_FILE, "r", encoding="utf-8") as f:
config = json.load(f)
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:
APP_CONFIG_FILE.unlink(missing_ok=True)
except Exception:
pass
# Fallback to Downloads folder
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) -> bool:
"""Save download path with enhanced error handling."""
try:
# Verify the path is valid and writable
if not Path(path).exists():
try:
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
# Save the config
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() -> 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()
# Extra logic moved to src\utils\ytsage_constants.py
# For binaries downloaded with our app, use direct binary update approach
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
# Extra logic moved to src\utils\ytsage_constants.py
# Download the latest version
try:
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:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
# Make executable on Unix systems
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 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:
logger.error(f"Error replacing yt-dlp binary: {e}")
return False
else:
logger.info(f"Failed to download latest yt-dlp: HTTP {response.status_code}")
return False
except Exception as e:
logger.error(f"Error downloading yt-dlp update: {e}")
return False
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
logger.info(f"Current yt-dlp version: {current_version}")
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)
if response.status_code == 200:
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",
],
capture_output=True,
text=True,
check=False,
creationflags=SUBPROCESS_CREATIONFLAGS,
)
if update_result.returncode == 0:
logger.info("yt-dlp successfully updated")
return True
else:
logger.error(f"Error updating yt-dlp: {update_result.stderr}")
else:
logger.info("yt-dlp is already up to date")
return True
else:
logger.info(f"Failed to get latest version info: HTTP {response.status_code}")
except Exception as e:
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() -> 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):
return False
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":
# Always check on startup if we haven't checked in the last hour
return time_diff > 3600 # 1 hour
elif frequency == "daily":
return time_diff > 86400 # 24 hours
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() -> 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("_", ".")
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()
save_config(config)
return True
else:
logger.info("Auto-update failed")
return False
else:
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()
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() -> 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),
}
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
save_config(config)
return True
except Exception as e:
logger.error(f"Error updating auto-update settings: {e}")
return False
def parse_yt_dlp_error(error_message: str) -> str:
"""
Parse yt-dlp error messages and return user-friendly error messages.
Args:
error_message: The raw error message from yt-dlp
Returns:
str: A user-friendly error message with actionable advice
"""
error_str = str(error_message).lower()
# Private video errors
if any(keyword in error_str for keyword in ['private video', 'login_required', 'sign in if you']):
return ("This is a private video. You can download it by logging into your account using cookies.\n"
"Go to 'Custom Options' → 'Login with Cookies' → 'Extract cookies from browser' to authenticate.")
# Age-restricted content
if any(keyword in error_str for keyword in ['age restricted', 'age-restricted', 'confirm your age']):
return ("This video is age-restricted. You need to be logged in to access it.\n"
"Use 'Custom Options' → 'Login with Cookies' to authenticate with your account.")
# Geo-blocked content
if any(keyword in error_str for keyword in ['not available in your country', 'geo-blocked', 'video is not available', 'not made this video available in your country']):
return ("This video is not available in your region (geo-blocked).\n"
"You may need to use a VPN or the video might be restricted in your country.")
# Removed/deleted videos
if any(keyword in error_str for keyword in ['video unavailable', 'this video has been removed', 'video does not exist']):
return ("This video has been removed or is no longer available.\n"
"The video may have been deleted by the uploader or removed due to policy violations.")
# Live stream errors
if any(keyword in error_str for keyword in ['live stream', 'livestream', 'is live']):
return ("This is a live stream that cannot be downloaded while active.\n"
"Wait for the stream to end, then try downloading the archived version.")
# Playlist errors
if any(keyword in error_str for keyword in ['playlist', 'no entries']):
return ("Unable to access this playlist. It may be private, deleted, or empty.\n"
"Check if the playlist exists and is publicly accessible.")
# Network/connection errors
if any(keyword in error_str for keyword in ['network error', 'connection', 'timeout', 'unable to download']):
return ("Network connection error. Please check your internet connection and try again.\n"
"If the problem persists, the video server might be temporarily unavailable.")
# Invalid URL
if any(keyword in error_str for keyword in ['invalid url', 'unsupported url', 'no video found']):
return ("Invalid or unsupported URL. Please check the link and try again.\n"
"Make sure you're using a valid YouTube, Vimeo, or other supported platform URL.")
# YouTube premium content
if any(keyword in error_str for keyword in ['youtube premium', 'premium', 'members only']):
return ("This content requires YouTube Premium or channel membership.\n"
"You need to be logged in with an account that has access to this content.")
# Copyright/DMCA
if any(keyword in error_str for keyword in ['copyright', 'dmca', 'blocked']):
return ("This video is blocked due to copyright claims.\n"
"The content owner has restricted access to this video.")
# Extraction errors (could be temporary)
if any(keyword in error_str for keyword in ['unable to extract', 'extraction failed']):
return ("Failed to extract video information. This might be a temporary issue.\n"
"Please try again in a few minutes, or check if the video link is correct.")
# Generic fallback with the original error for debugging
return (f"Could not extract video information. Please check your link.\n"
f"Technical details: {error_message}")