Jaroslav Beneš
b40deb0d5f
setx truncates values at 1024 characters, and the old code fed it the merged process PATH (system + user), permanently duplicating every system entry into the user hive and silently dropping anything past the limit. Read and rewrite only the HKCU Environment Path value with winreg, preserving REG_EXPAND_SZ, and broadcast WM_SETTINGCHANGE. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
474 lines
18 KiB
Python
474 lines
18 KiB
Python
import hashlib
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
|
|
from ..utils.ytsage_logger import logger
|
|
from ..utils.ytsage_constants import (
|
|
FFMPEG_7Z_DOWNLOAD_URL,
|
|
FFMPEG_7Z_SHA256_URL,
|
|
FFMPEG_ZIP_DOWNLOAD_URL,
|
|
FFMPEG_ZIP_SHA256_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_CREATIONFLAGS)
|
|
return True
|
|
except (subprocess.SubprocessError, FileNotFoundError):
|
|
return False
|
|
|
|
|
|
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:
|
|
if total_size == 0:
|
|
f.write(response.content)
|
|
else:
|
|
downloaded = 0
|
|
for data in response.iter_content(chunk_size=8192):
|
|
downloaded += len(data)
|
|
f.write(data)
|
|
if progress_callback:
|
|
progress = int((downloaded / total_size) * 100)
|
|
progress_callback(f"⚡ Downloading FFmpeg components... {progress}%")
|
|
return True
|
|
except requests.RequestException as e:
|
|
logger.info(f"Download error: {e}")
|
|
return False
|
|
|
|
|
|
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:
|
|
for chunk in iter(lambda: f.read(4096), b""):
|
|
sha256_hash.update(chunk)
|
|
return sha256_hash.hexdigest()
|
|
|
|
|
|
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!")
|
|
return True
|
|
else:
|
|
logger.error(f"SHA-256 verification failed!")
|
|
logger.info(f"Expected: {expected_hash}")
|
|
logger.info(f"Actual: {actual_hash}")
|
|
return False
|
|
except Exception as e:
|
|
logger.info(f"⚠️ SHA-256 verification error: {e}")
|
|
return False
|
|
|
|
|
|
def get_ffmpeg_install_path() -> Path:
|
|
"""
|
|
Get the FFmpeg installation path.
|
|
For Windows, tries to find the latest essentials build dynamically.
|
|
"""
|
|
if OS_NAME == "Windows":
|
|
ffmpeg_base = Path(os.getenv("LOCALAPPDATA")) / "ffmpeg" # type: ignore
|
|
|
|
# If the directory exists, look for any ffmpeg-*-essentials_build folder
|
|
if ffmpeg_base.exists():
|
|
# Find all directories matching the pattern
|
|
essentials_dirs = list(ffmpeg_base.glob("ffmpeg-*-essentials_build"))
|
|
if essentials_dirs:
|
|
# Sort by name (which includes version) and take the latest
|
|
latest_dir = sorted(essentials_dirs, reverse=True)[0]
|
|
bin_dir = latest_dir / "bin"
|
|
if bin_dir.exists():
|
|
return bin_dir
|
|
|
|
# Fallback: return default path (even if it doesn't exist yet)
|
|
return ffmpeg_base / "ffmpeg-essentials_build" / "bin"
|
|
|
|
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:
|
|
str: Path to FFmpeg executable or 'ffmpeg' if found in PATH but path unknown
|
|
"""
|
|
try:
|
|
# First try to find ffmpeg in PATH using 'where' on Windows or 'which' on Unix
|
|
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", "ffmpeg"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
creationflags=SUBPROCESS_CREATIONFLAGS,
|
|
)
|
|
if result.returncode == 0 and result.stdout.strip():
|
|
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)
|
|
if result.returncode == 0 and result.stdout.strip():
|
|
ffmpeg_path = result.stdout.strip()
|
|
return ffmpeg_path
|
|
except Exception as e:
|
|
logger.exception(f"Error finding ffmpeg in PATH: {e}")
|
|
|
|
# If not found in PATH, check the installation directory
|
|
ffmpeg_install_path = get_ffmpeg_install_path()
|
|
if OS_NAME == "Windows":
|
|
ffmpeg_exe = Path(ffmpeg_install_path).joinpath("ffmpeg.exe")
|
|
else:
|
|
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() -> 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_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 OS_NAME == "Windows":
|
|
ffmpeg_exe = Path(ffmpeg_path).joinpath("ffmpeg.exe")
|
|
else:
|
|
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', '')}"
|
|
return True
|
|
return False
|
|
except Exception as e:
|
|
logger.info(f"FFmpeg check error: {e}")
|
|
return False
|
|
|
|
|
|
def install_ffmpeg_windows(progress_callback=None) -> bool:
|
|
"""Install FFmpeg on Windows using essentials build with 7z method primarily, with zip as fallback."""
|
|
# Check if already installed
|
|
if check_ffmpeg_installed():
|
|
logger.info("FFmpeg is already installed!")
|
|
if progress_callback:
|
|
progress_callback("✅ FFmpeg is already installed!")
|
|
return True
|
|
|
|
try:
|
|
# Define variables for essentials build
|
|
extract_dir = Path(os.getenv("LOCALAPPDATA")) / "ffmpeg" # type: ignore
|
|
|
|
# Create extraction directory if it doesn't exist
|
|
extract_dir.mkdir(exist_ok=True)
|
|
|
|
# Try 7z method first (smaller size)
|
|
use_7zip = check_7zip_installed()
|
|
success = False
|
|
|
|
if use_7zip:
|
|
logger.info("Using 7-Zip method (smaller download size)...")
|
|
if progress_callback:
|
|
progress_callback("⚡ Using 7-Zip method...")
|
|
temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".7z").name
|
|
|
|
# Download 7z file
|
|
if progress_callback:
|
|
progress_callback("⬇ Downloading FFmpeg (7z)...")
|
|
if download_file(
|
|
FFMPEG_7Z_DOWNLOAD_URL,
|
|
temp_file,
|
|
progress_callback=progress_callback,
|
|
):
|
|
# Verify SHA-256 hash for 7z file
|
|
if progress_callback:
|
|
progress_callback("🔐 Verifying download integrity...")
|
|
if verify_sha256(temp_file, FFMPEG_7Z_SHA256_URL):
|
|
logger.info("Extracting FFmpeg components from 7z archive...")
|
|
if progress_callback:
|
|
progress_callback("⚙ Extracting FFmpeg...")
|
|
try:
|
|
subprocess.run(
|
|
["7z", "x", temp_file, f"-o{extract_dir}", "-y"],
|
|
creationflags=SUBPROCESS_CREATIONFLAGS,
|
|
timeout=300,
|
|
check=True,
|
|
)
|
|
success = True
|
|
except Exception as e:
|
|
logger.exception(f"7z extraction failed: {e}, trying zip fallback...")
|
|
if progress_callback:
|
|
progress_callback("❌ 7z extraction failed, trying zip fallback...")
|
|
else:
|
|
logger.error("SHA-256 verification failed for 7z file, trying zip fallback...")
|
|
if progress_callback:
|
|
progress_callback("❌ SHA-256 verification failed, trying zip fallback...")
|
|
|
|
# Clean up temp file
|
|
try:
|
|
Path(temp_file).unlink(missing_ok=True)
|
|
except Exception:
|
|
pass
|
|
|
|
# Fallback to zip method if 7z failed or not available
|
|
if not success:
|
|
logger.info("Using ZIP method as fallback...")
|
|
if progress_callback:
|
|
progress_callback("📦 Using ZIP method...")
|
|
temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".zip").name
|
|
|
|
# Download zip file
|
|
if progress_callback:
|
|
progress_callback("⬇ Downloading FFmpeg (zip)...")
|
|
if not download_file(
|
|
FFMPEG_ZIP_DOWNLOAD_URL,
|
|
temp_file,
|
|
progress_callback=progress_callback,
|
|
):
|
|
logger.error("Failed to download FFmpeg (both 7z and zip methods failed)")
|
|
if progress_callback:
|
|
progress_callback("❌ Failed to download FFmpeg")
|
|
return False
|
|
|
|
# Verify SHA-256 hash for zip
|
|
if progress_callback:
|
|
progress_callback("🔐 Verifying download integrity...")
|
|
if not verify_sha256(temp_file, FFMPEG_ZIP_SHA256_URL):
|
|
logger.warning("SHA-256 verification failed for zip file, proceeding anyway...")
|
|
if progress_callback:
|
|
progress_callback("⚠️ SHA-256 verification failed, proceeding anyway...")
|
|
|
|
logger.info("Extracting FFmpeg components from zip archive...")
|
|
if progress_callback:
|
|
progress_callback("⚙ Extracting FFmpeg...")
|
|
try:
|
|
import zipfile
|
|
with zipfile.ZipFile(temp_file, "r") as zip_ref:
|
|
zip_ref.extractall(extract_dir)
|
|
success = True
|
|
except Exception as e:
|
|
logger.exception(f"Extraction failed: {e}")
|
|
if progress_callback:
|
|
progress_callback(f"❌ Extraction failed: {e}")
|
|
return False
|
|
finally:
|
|
# Clean up temp file
|
|
try:
|
|
Path(temp_file).unlink(missing_ok=True)
|
|
except Exception:
|
|
pass
|
|
|
|
if not success:
|
|
logger.error("Both 7z and zip methods failed")
|
|
if progress_callback:
|
|
progress_callback("❌ Installation failed")
|
|
return False
|
|
|
|
# Find the bin directory in the extracted essentials build
|
|
logger.info("Locating FFmpeg binaries...")
|
|
if progress_callback:
|
|
progress_callback("🔍 Locating FFmpeg binaries...")
|
|
|
|
bin_dir = None
|
|
# Look for any directory matching ffmpeg-*-essentials_build pattern
|
|
for item in extract_dir.iterdir():
|
|
if item.is_dir() and "essentials_build" in item.name.lower():
|
|
potential_bin = item / "bin"
|
|
if potential_bin.exists():
|
|
bin_dir = potential_bin
|
|
logger.info(f"Found FFmpeg bin directory: {bin_dir}")
|
|
break
|
|
|
|
if not bin_dir:
|
|
logger.error("Could not locate FFmpeg bin directory")
|
|
if progress_callback:
|
|
progress_callback("❌ Could not locate FFmpeg bin directory")
|
|
return False
|
|
|
|
logger.info("Configuring system paths...")
|
|
if progress_callback:
|
|
progress_callback("🔧 Configuring system paths...")
|
|
|
|
# Persist to the user PATH via the registry. setx must not be used here:
|
|
# it truncates values at 1024 characters, and os.environ["PATH"] is the
|
|
# merged system+user PATH, so writing it back would permanently duplicate
|
|
# every system entry into the user hive.
|
|
try:
|
|
import winreg
|
|
|
|
with winreg.OpenKey(winreg.HKEY_CURRENT_USER, "Environment", 0, winreg.KEY_READ | winreg.KEY_WRITE) as key:
|
|
try:
|
|
stored_path, value_type = winreg.QueryValueEx(key, "Path")
|
|
except FileNotFoundError:
|
|
stored_path, value_type = "", winreg.REG_EXPAND_SZ
|
|
user_parts = [p for p in stored_path.split(os.pathsep) if p]
|
|
# Drop stale ffmpeg entries we previously added, then prepend the new one
|
|
user_parts = [p for p in user_parts if "ffmpeg" not in p.lower() or p == str(bin_dir)]
|
|
if str(bin_dir) not in user_parts:
|
|
user_parts.insert(0, str(bin_dir))
|
|
winreg.SetValueEx(key, "Path", 0, value_type, os.pathsep.join(user_parts))
|
|
|
|
# Broadcast the change so new shells pick it up without relogin
|
|
import ctypes
|
|
|
|
ctypes.windll.user32.SendMessageTimeoutW(0xFFFF, 0x001A, 0, "Environment", 0x0002, 5000, None)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to update PATH permanently: {e}")
|
|
|
|
# Update for the current session regardless
|
|
if str(bin_dir) not in os.environ.get("PATH", "").split(os.pathsep):
|
|
os.environ["PATH"] = str(bin_dir) + os.pathsep + os.environ.get("PATH", "")
|
|
|
|
# Verify installation
|
|
if progress_callback:
|
|
progress_callback("✅ Verifying installation...")
|
|
if not check_ffmpeg_installed():
|
|
logger.error("FFmpeg installation verification failed")
|
|
if progress_callback:
|
|
progress_callback("⚠️ Installation completed but verification failed")
|
|
return True # Still return True as files were extracted
|
|
|
|
logger.info("FFmpeg installation completed successfully!")
|
|
if progress_callback:
|
|
progress_callback("✅ FFmpeg installation completed successfully!")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.exception(f"Error installing FFmpeg: {e}")
|
|
if progress_callback:
|
|
progress_callback(f"❌ Error installing FFmpeg: {e}")
|
|
return False
|
|
|
|
|
|
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,
|
|
)
|
|
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)"'
|
|
subprocess.run(brew_install_cmd, shell=True, check=True, timeout=300)
|
|
|
|
# Install FFmpeg
|
|
logger.info("Installing FFmpeg...")
|
|
subprocess.run(["brew", "install", "ffmpeg"], check=True, timeout=300)
|
|
|
|
# Verify installation
|
|
if not check_ffmpeg_installed():
|
|
logger.error("FFmpeg installation verification failed")
|
|
return False
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.exception(f"Error installing FFmpeg: {e}")
|
|
return False
|
|
|
|
|
|
def install_ffmpeg_linux() -> bool:
|
|
"""Install FFmpeg on Linux using appropriate package manager."""
|
|
try:
|
|
# Detect the package manager
|
|
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"):
|
|
# Fedora
|
|
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"):
|
|
# Universal snap package
|
|
subprocess.run(["sudo", "snap", "install", "ffmpeg"], check=True, timeout=300)
|
|
else:
|
|
logger.error("No supported package manager found")
|
|
return False
|
|
|
|
# Verify installation
|
|
if not check_ffmpeg_installed():
|
|
logger.error("FFmpeg installation verification failed")
|
|
return False
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.exception(f"Error installing FFmpeg: {e}")
|
|
return False
|
|
|
|
|
|
def auto_install_ffmpeg(progress_callback=None) -> bool:
|
|
"""Automatically install FFmpeg based on the operating system."""
|
|
if OS_NAME == "Windows":
|
|
return install_ffmpeg_windows(progress_callback=progress_callback)
|
|
elif OS_NAME == "Darwin":
|
|
return install_ffmpeg_macos()
|
|
elif OS_NAME == "Linux":
|
|
return install_ffmpeg_linux()
|
|
else:
|
|
logger.info(f"Unsupported operating system: {OS_NAME}")
|
|
if progress_callback:
|
|
progress_callback(f"❌ Unsupported operating system: {OS_NAME}")
|
|
return False
|