import subprocess import os from pathlib import Path from typing import Callable, Dict, Any import shutil import traceback # Find FFmpeg ffmpeg_path = shutil.which("ffmpeg") or shutil.which("ffmpeg.exe") if not ffmpeg_path: common_paths = [ r"C:\ffmpeg\bin\ffmpeg.exe", r"C:\Program Files\ffmpeg\bin\ffmpeg.exe", ] for path in common_paths: if os.path.exists(path): ffmpeg_path = path break class VideoConverter: SUPPORTED_FORMATS = { ".mp4", ".avi", ".mkv", ".mov", ".webm", ".gif" } def __init__(self): self.ffmpeg_available = self._check_ffmpeg() def _check_ffmpeg(self): if not ffmpeg_path or not os.path.exists(ffmpeg_path): print("⚠ FFmpeg not found") return False return True def convert( self, input_path: str, output_path: str, options: Dict[str, Any] | None = None, progress_callback: Callable = None ) -> bool: options = options or {} try: if not self.ffmpeg_available: print("FFmpeg not available") return False self._update_progress(progress_callback, 10) # Validate input if not os.path.exists(input_path): raise FileNotFoundError(input_path) input_ext = Path(input_path).suffix.lower() output_ext = Path(output_path).suffix.lower() if output_ext not in self.SUPPORTED_FORMATS: raise ValueError( f"Unsupported format: {output_ext}" ) # Create output directory Path(output_path).parent.mkdir( parents=True, exist_ok=True ) self._update_progress(progress_callback, 20) # Quality settings quality_settings = options.get( "quality_settings", {} ) crf_value = str( quality_settings.get("video", "23") ) # Base command cmd = [ ffmpeg_path, "-y", "-i", input_path ] # Video filters video_filters = [] # AI enhancement if options.get("ai_enhancement", False): video_filters.append( "unsharp=5:5:1.0:5:5:0.0" ) # Format settings if output_ext == ".mp4": cmd.extend([ "-c:v", "libx264", "-preset", "medium", "-crf", crf_value, "-c:a", "aac", "-b:a", "128k" ]) elif output_ext == ".avi": cmd.extend([ "-c:v", "libxvid", "-q:v", "4", "-c:a", "mp3", "-b:a", "192k" ]) elif output_ext == ".mkv": cmd.extend([ "-c:v", "libx264", "-crf", crf_value, "-c:a", "aac" ]) elif output_ext == ".mov": cmd.extend([ "-c:v", "libx264", "-pix_fmt", "yuv420p", "-c:a", "aac" ]) elif output_ext == ".webm": cmd.extend([ "-c:v", "libvpx-vp9", "-crf", crf_value, "-b:v", "0", "-c:a", "libopus" ]) elif output_ext == ".gif": video_filters.append( "fps=10,scale=320:-1:flags=lanczos" ) cmd.extend([ "-loop", "0" ]) # Apply filters safely if video_filters: cmd.extend([ "-vf", ",".join(video_filters) ]) self._update_progress(progress_callback, 50) # Output path cmd.append(output_path) # Run ffmpeg result = subprocess.run( cmd, capture_output=True, text=True, timeout=600 ) self._update_progress(progress_callback, 90) if result.returncode != 0: print("FFmpeg Error:") print(result.stderr[:500]) return False self._update_progress(progress_callback, 100) print( f"✓ Converted: " f"{os.path.basename(input_path)} " f"→ {output_ext}" ) return True except subprocess.TimeoutExpired: print("Video conversion timed out") return False except Exception as e: print( f"Video conversion error: {e}" ) traceback.print_exc() return False def _update_progress( self, callback, value ): try: if callback: callback(value) except Exception as e: print( f"Progress callback error: {e}" )