""" Utility functions for video processing. Provides FFmpeg wrappers and common operations used by all strategies. """ import json import os import subprocess from pathlib import Path def get_video_duration(video_path: str) -> float: """ Get video duration in seconds using ffprobe. Args: video_path: Path to video file Returns: Duration in seconds Raises: RuntimeError: If ffprobe fails or video is invalid """ cmd = [ "ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "json", video_path ] try: result = subprocess.run( cmd, capture_output=True, text=True, check=True ) data = json.loads(result.stdout) return float(data["format"]["duration"]) except subprocess.CalledProcessError as e: raise RuntimeError(f"Failed to get video duration: {e.stderr}") except (KeyError, ValueError) as e: raise RuntimeError(f"Invalid ffprobe output: {e}") def extract_audio( video_path: str, output_path: str | None = None, sample_rate: int = 16000, codec: str = "pcm_s16le" ) -> str: """ Extract audio from video file using FFmpeg. Args: video_path: Path to input video output_path: Path for output audio (default: same name with .wav) sample_rate: Audio sample rate in Hz (16000 recommended for Whisper) codec: Audio codec (pcm_s16le for WAV) Returns: Path to extracted audio file Raises: RuntimeError: If FFmpeg fails """ if output_path is None: video_name = Path(video_path).stem output_path = str(Path(video_path).parent / f"{video_name}_audio.wav") cmd = [ "ffmpeg", "-i", video_path, "-vn", # No video "-acodec", codec, "-ar", str(sample_rate), "-ac", "1", # Mono "-y", # Overwrite output file output_path ] try: subprocess.run( cmd, capture_output=True, text=True, check=True ) return output_path except subprocess.CalledProcessError as e: raise RuntimeError(f"Failed to extract audio: {e.stderr}") def slice_video( video_path: str, start_time: float, end_time: float, output_path: str | None = None ) -> str: """ Slice a video segment using FFmpeg stream copy (fast, no re-encoding). Args: video_path: Path to input video start_time: Start time in seconds end_time: End time in seconds output_path: Path for output segment Returns: Path to output segment Raises: RuntimeError: If FFmpeg fails """ if output_path is None: video_name = Path(video_path).stem output_path = str( Path(video_path).parent / f"{video_name}_{start_time:.1f}-{end_time:.1f}.mp4" ) duration = end_time - start_time cmd = [ "ffmpeg", "-ss", str(start_time), "-i", video_path, "-t", str(duration), "-c", "copy", # Stream copy (no re-encoding) "-y", output_path ] try: subprocess.run( cmd, capture_output=True, text=True, check=True ) return output_path except subprocess.CalledProcessError as e: raise RuntimeError(f"Failed to slice video: {e.stderr}") def check_ffmpeg_installed() -> bool: """Check if FFmpeg and FFprobe are available in PATH.""" try: subprocess.run( ["ffmpeg", "-version"], capture_output=True, check=True ) subprocess.run( ["ffprobe", "-version"], capture_output=True, check=True ) return True except (subprocess.CalledProcessError, FileNotFoundError): return False