470 lines
16 KiB
Python
470 lines
16 KiB
Python
"""
|
|
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)
|