""" Cross-platform system utility finder. Detects and caches paths to system executables (ffprobe, ffmpeg, wkhtmltopdf). Supports: - Linux (Ubuntu, Debian, etc.) - macOS (Intel and Apple Silicon) - Windows WSL Author: Video Query Application """ import os import platform import subprocess import shutil import logging from typing import Optional, Dict, List from functools import lru_cache logger = logging.getLogger('video_query') class SystemUtility: """Find and manage system utility paths across platforms.""" # Platform detection constants PLATFORM_LINUX = 'linux' PLATFORM_MACOS = 'darwin' PLATFORM_WINDOWS = 'windows' # Common paths for ffprobe by platform FFPROBE_PATHS = { PLATFORM_LINUX: [ '/usr/bin/ffprobe', '/usr/local/bin/ffprobe', '/snap/bin/ffprobe' ], PLATFORM_MACOS: [ '/opt/homebrew/bin/ffprobe', # Apple Silicon (M1/M2/M3) '/usr/local/bin/ffprobe', # Intel Mac '/usr/bin/ffprobe' # Fallback ], PLATFORM_WINDOWS: [ 'C:\\Program Files\\ffmpeg\\bin\\ffprobe.exe', 'C:\\ffmpeg\\bin\\ffprobe.exe', 'ffprobe.exe' # Try PATH ] } # Common paths for ffmpeg by platform FFMPEG_PATHS = { PLATFORM_LINUX: [ '/usr/bin/ffmpeg', '/usr/local/bin/ffmpeg', '/snap/bin/ffmpeg' ], PLATFORM_MACOS: [ '/opt/homebrew/bin/ffmpeg', # Apple Silicon '/usr/local/bin/ffmpeg', # Intel Mac '/usr/bin/ffmpeg' ], PLATFORM_WINDOWS: [ 'C:\\Program Files\\ffmpeg\\bin\\ffmpeg.exe', 'C:\\ffmpeg\\bin\\ffmpeg.exe', 'ffmpeg.exe' ] } # Common paths for wkhtmltopdf by platform WKHTMLTOPDF_PATHS = { PLATFORM_LINUX: [ '/usr/bin/wkhtmltopdf', '/usr/local/bin/wkhtmltopdf', '/snap/bin/wkhtmltopdf' ], PLATFORM_MACOS: [ '/opt/homebrew/bin/wkhtmltopdf', # Apple Silicon '/usr/local/bin/wkhtmltopdf', # Intel Mac '/usr/bin/wkhtmltopdf' ], PLATFORM_WINDOWS: [ 'C:\\Program Files\\wkhtmltopdf\\bin\\wkhtmltopdf.exe', 'C:\\wkhtmltopdf\\bin\\wkhtmltopdf.exe', 'wkhtmltopdf.exe' ] } def __init__(self): """Initialize with platform detection and caching.""" self._platform = self._detect_os() self._cache = {} logger.info(f"SystemUtility initialized for platform: {self._platform}") # Log detected architecture for macOS if self._platform == self.PLATFORM_MACOS: arch = platform.machine() logger.info(f"macOS architecture detected: {arch}") def _detect_os(self) -> str: """ Detect operating system. Returns: Platform constant (PLATFORM_LINUX, PLATFORM_MACOS, or PLATFORM_WINDOWS) """ system = platform.system().lower() if 'linux' in system: return self.PLATFORM_LINUX elif 'darwin' in system: return self.PLATFORM_MACOS elif 'windows' in system: return self.PLATFORM_WINDOWS else: logger.warning(f"Unknown platform: {system}, defaulting to Linux") return self.PLATFORM_LINUX @lru_cache(maxsize=10) def find_ffprobe(self) -> str: """ Find ffprobe executable path (cached). Returns: Absolute path to ffprobe executable Raises: FileNotFoundError: If ffprobe cannot be found """ return self._find_executable( name='ffprobe', paths=self.FFPROBE_PATHS[self._platform], install_instructions=self._get_ffprobe_install_instructions() ) @lru_cache(maxsize=10) def find_ffmpeg(self) -> str: """ Find ffmpeg executable path (cached). Returns: Absolute path to ffmpeg executable Raises: FileNotFoundError: If ffmpeg cannot be found """ return self._find_executable( name='ffmpeg', paths=self.FFMPEG_PATHS[self._platform], install_instructions=self._get_ffmpeg_install_instructions() ) @lru_cache(maxsize=10) def find_wkhtmltopdf(self) -> str: """ Find wkhtmltopdf executable path (cached). Returns: Absolute path to wkhtmltopdf executable Raises: FileNotFoundError: If wkhtmltopdf cannot be found """ return self._find_executable( name='wkhtmltopdf', paths=self.WKHTMLTOPDF_PATHS[self._platform], install_instructions=self._get_wkhtmltopdf_install_instructions() ) def _find_executable(self, name: str, paths: List[str], install_instructions: str) -> str: """ Generic executable finder with fallback logic. Search order: 1. Check cache 2. Check predefined platform-specific paths 3. Check PATH environment variable 4. Raise error with installation instructions Args: name: Name of executable (e.g., 'ffprobe') paths: List of paths to check for this platform install_instructions: Installation instructions for error message Returns: Absolute path to executable Raises: FileNotFoundError: If executable cannot be found """ # 1. Check cache if name in self._cache: cached_path = self._cache[name] if os.path.exists(cached_path) and self.verify_executable(cached_path, name): logger.debug(f"Using cached path for {name}: {cached_path}") return cached_path else: logger.warning(f"Cached path for {name} is no longer valid: {cached_path}") del self._cache[name] # 2. Check predefined platform-specific paths logger.info(f"Searching for {name} in platform-specific locations...") for path in paths: if os.path.exists(path): logger.info(f"Found {name} at predefined path: {path}") if self.verify_executable(path, name): logger.info(f"Verified {name} is executable and working") self._cache[name] = path return path else: logger.warning(f"Found {name} at {path} but verification failed") # 3. Check PATH environment variable using shutil.which logger.info(f"Searching for {name} in PATH environment variable...") path_from_env = shutil.which(name) if path_from_env: logger.info(f"Found {name} in PATH: {path_from_env}") if self.verify_executable(path_from_env, name): logger.info(f"Verified {name} from PATH is working") self._cache[name] = path_from_env return path_from_env else: logger.warning(f"Found {name} in PATH but verification failed: {path_from_env}") # 4. Not found - raise error with detailed instructions error_msg = self._format_not_found_error(name, paths, install_instructions) logger.error(error_msg) raise FileNotFoundError(error_msg) def _format_not_found_error(self, name: str, paths: List[str], install_instructions: str) -> str: """ Format detailed error message when executable is not found. Args: name: Name of executable paths: Paths that were searched install_instructions: Installation instructions Returns: Formatted error message """ error_msg = f"\n{'='*80}\n" error_msg += f"ERROR: {name} not found on this system\n" error_msg += f"{'='*80}\n\n" error_msg += f"Platform: {self._platform}\n" error_msg += f"Python: {platform.python_version()}\n" error_msg += f"OS: {platform.platform()}\n\n" error_msg += f"Searched locations:\n" for path in paths: exists = "✓" if os.path.exists(path) else "✗" error_msg += f" {exists} {path}\n" error_msg += f" ✗ PATH environment variable\n\n" error_msg += f"Installation Instructions:\n" error_msg += f"{install_instructions}\n\n" error_msg += f"After installation, restart the application.\n" error_msg += f"{'='*80}\n" return error_msg def verify_executable(self, path: str, name: str) -> bool: """ Verify that executable exists and runs properly. Args: path: Path to executable name: Name of executable (for logging) Returns: True if executable works, False otherwise """ try: # Check if file exists and is executable if not os.path.exists(path): logger.debug(f"Path does not exist: {path}") return False if not os.access(path, os.X_OK): logger.debug(f"Path is not executable: {path}") return False # Try to run with version flag # wkhtmltopdf uses --version, ffmpeg/ffprobe use -version version_flags = ['--version', '-version', '-V'] for flag in version_flags: try: result = subprocess.run( [path, flag], capture_output=True, timeout=5, text=True ) # If returncode is 0 or we got output, it's working if result.returncode == 0 or result.stdout or result.stderr: # Log version info output = result.stdout or result.stderr or '' version_output = output.split('\n')[0] if output else 'unknown' logger.debug(f"{name} verified with {flag}: {version_output[:50]}") return True except subprocess.TimeoutExpired: continue except Exception: continue # If no version flag worked, but file exists and is executable, assume it works # This handles cases where the executable doesn't support version flags logger.debug(f"{name} exists and is executable at {path}, assuming functional") return True except subprocess.TimeoutExpired: logger.warning(f"Timeout while verifying {name} at {path}") # Still return True if file exists and is executable return os.path.exists(path) and os.access(path, os.X_OK) except Exception as e: logger.debug(f"Error verifying {name} at {path}: {str(e)}") # Still return True if file exists and is executable return os.path.exists(path) and os.access(path, os.X_OK) def _get_ffprobe_install_instructions(self) -> str: """Get platform-specific installation instructions for ffprobe.""" if self._platform == self.PLATFORM_LINUX: return ( " Ubuntu/Debian:\n" " sudo apt-get update\n" " sudo apt-get install ffmpeg\n\n" " CentOS/RHEL:\n" " sudo yum install ffmpeg\n\n" " Snap:\n" " sudo snap install ffmpeg" ) elif self._platform == self.PLATFORM_MACOS: return ( " Using Homebrew (recommended):\n" " brew install ffmpeg\n\n" " Note: On Apple Silicon Macs, Homebrew installs to /opt/homebrew/\n" " On Intel Macs, Homebrew installs to /usr/local/" ) else: return ( " Download from: https://ffmpeg.org/download.html\n" " Or use Chocolatey:\n" " choco install ffmpeg" ) def _get_ffmpeg_install_instructions(self) -> str: """Get platform-specific installation instructions for ffmpeg.""" # Same as ffprobe since they come in the same package return self._get_ffprobe_install_instructions() def _get_wkhtmltopdf_install_instructions(self) -> str: """Get platform-specific installation instructions for wkhtmltopdf.""" if self._platform == self.PLATFORM_LINUX: return ( " Ubuntu/Debian:\n" " sudo apt-get update\n" " sudo apt-get install wkhtmltopdf\n\n" " CentOS/RHEL:\n" " sudo yum install wkhtmltopdf\n\n" " Or download from: https://wkhtmltopdf.org/downloads.html" ) elif self._platform == self.PLATFORM_MACOS: return ( " Using Homebrew (recommended):\n" " brew install wkhtmltopdf\n\n" " Or download from: https://wkhtmltopdf.org/downloads.html" ) else: return ( " Download from: https://wkhtmltopdf.org/downloads.html\n" " Or use Chocolatey:\n" " choco install wkhtmltopdf" ) def get_system_info(self) -> Dict: """ Get comprehensive system information for debugging. Returns: Dictionary with system details and executable paths """ info = { 'platform': self._platform, 'platform_name': platform.system(), 'platform_version': platform.version(), 'platform_machine': platform.machine(), 'python_version': platform.python_version(), 'python_implementation': platform.python_implementation(), 'os_details': platform.platform(), 'executables': {} } # Try to find each executable for name, finder in [ ('ffprobe', self.find_ffprobe), ('ffmpeg', self.find_ffmpeg), ('wkhtmltopdf', self.find_wkhtmltopdf) ]: try: path = finder() info['executables'][name] = { 'path': path, 'found': True, 'verified': self.verify_executable(path, name) } except FileNotFoundError: info['executables'][name] = { 'path': None, 'found': False, 'verified': False } return info def clear_cache(self): """Clear the executable path cache. Useful for testing.""" self._cache.clear() # Also clear lru_cache for the find methods self.find_ffprobe.cache_clear() self.find_ffmpeg.cache_clear() self.find_wkhtmltopdf.cache_clear() logger.info("Cleared system utility cache") # Global singleton instance system_utils = SystemUtility() # Convenience functions for direct use def find_ffprobe() -> str: """Find ffprobe executable (convenience function).""" return system_utils.find_ffprobe() def find_ffmpeg() -> str: """Find ffmpeg executable (convenience function).""" return system_utils.find_ffmpeg() def find_wkhtmltopdf() -> str: """Find wkhtmltopdf executable (convenience function).""" return system_utils.find_wkhtmltopdf() def get_system_info() -> Dict: """Get system information (convenience function).""" return system_utils.get_system_info() if __name__ == "__main__": """Test the system utility finder.""" print("="*80) print("System Utility Finder Test") print("="*80) info = system_utils.get_system_info() print(f"\nPlatform: {info['platform_name']} ({info['platform']})") print(f"Machine: {info['platform_machine']}") print(f"Python: {info['python_version']}") print(f"\nExecutables Found:") for name, details in info['executables'].items(): status = "✓" if details['found'] else "✗" verified = "✓" if details['verified'] else "✗" path = details['path'] or "Not found" print(f" {status} {name}: {path} (verified: {verified})") print("\n" + "="*80)