"""FFmpeg execution for clip generation.""" import subprocess from pathlib import Path from typing import TYPE_CHECKING from .config import settings from .frame_processor import process_segment_frame_by_frame from .models import CropKeyframe, CropPath if TYPE_CHECKING: from .captioning import CaptionRenderer class FFmpegExecutor: """Handles video clip generation using FFmpeg.""" def _split_keyframes_by_scene(self, keyframes: list[CropKeyframe]) -> list[list[CropKeyframe]]: """ Split keyframes into groups at scene boundaries. Returns: List of keyframe groups, one per scene """ if not keyframes: return [] scenes: list[list[CropKeyframe]] = [[]] for kf in keyframes: if kf.scene_change and scenes[-1]: # New scene (but not the first keyframe) scenes.append([kf]) else: scenes[-1].append(kf) return [s for s in scenes if s] # Remove empty scenes def _get_overlay_output_path(self, cropped_output_path: Path) -> Path: """Generate output path for overlay debug video. Example: clip_1_s134.0-145.0.mp4 → clip_1_s134.0-145.0_overlay.mp4 Args: cropped_output_path: Path to the cropped video output Returns: Path for the overlay debug video """ stem = cropped_output_path.stem suffix = cropped_output_path.suffix return cropped_output_path.parent / f"{stem}_overlay{suffix}" def generate_clip( self, input_path: Path, output_path: Path, crop_path: CropPath, caption_renderer: "CaptionRenderer | None" = None, audio_source: Path | None = None, audio_start: float = 0.0, audio_duration: float | None = None, ) -> Path: """Generate a video clip with dynamic cropping. Uses frame-by-frame processing for smooth interpolation between keyframes, avoiding FFmpeg expression bugs. This is the same approach used by professional tools like Google's AutoFlip. Args: input_path: Path to input video (visual source) output_path: Path for output clip crop_path: CropPath with segment and keyframes audio_source: Optional separate audio source path. When provided (e.g. for UGC clips), audio is extracted from this file at [audio_start, audio_start + clip_duration] instead of from input_path. audio_start: Start offset in audio_source (seconds). audio_duration: Length of audio to use (seconds); None = match clip duration. Returns: Path to generated clip Raises: subprocess.CalledProcessError: If FFmpeg fails """ import sys if not crop_path.keyframes: raise ValueError( f"CropPath has no keyframes for segment " f"{crop_path.segment.start:.1f}s–{crop_path.segment.end:.1f}s. " "This is a reframing bug; reframing.py should always emit at least one keyframe." ) # Use frame-by-frame processing for smooth crops with multiple keyframes, # or whenever a caption overlay is requested (the FFmpeg-only fast path # has no Python frame hook), or when a separate audio source is required. use_frame_by_frame = ( len(crop_path.keyframes) > 1 or caption_renderer is not None or audio_source is not None ) if use_frame_by_frame: result_path = process_segment_frame_by_frame( input_path, output_path, crop_path, caption_renderer=caption_renderer, audio_source_path=audio_source, audio_start=audio_start, audio_duration=audio_duration, ) else: # Fast path for single keyframe (static crop) - use simple FFmpeg segment = crop_path.segment keyframe = crop_path.keyframes[0] # Build simple static crop expression crop_expr = f"crop=w={keyframe.width}:h={keyframe.height}:x={keyframe.x}:y={keyframe.y}" # Calculate duration duration = segment.end - segment.start # Build FFmpeg command cmd = [ str(settings.ffmpeg_path), "-i", str(input_path), "-ss", str(segment.start), "-t", str(duration), "-vf", crop_expr, "-c:a", "copy", # Preserve audio unchanged "-preset", "fast", "-y", str(output_path), ] print(f"\n{'='*60}", file=sys.stderr) print(f"STATIC CROP (Fast Path)", file=sys.stderr) print(f"{'='*60}", file=sys.stderr) print(f"Crop: x={keyframe.x}, y={keyframe.y}, " f"w={keyframe.width}, h={keyframe.height}", file=sys.stderr) print(f"Duration: {duration:.2f}s", file=sys.stderr) print(f"{'='*60}\n", file=sys.stderr) # Execute FFmpeg result = subprocess.run(cmd, capture_output=True, text=True, check=False) if result.returncode != 0: print(f"\nFFmpeg stderr:\n{result.stderr}\n") raise subprocess.CalledProcessError( result.returncode, cmd, result.stdout, result.stderr ) # Verify output was created if not output_path.exists(): raise RuntimeError(f"Failed to create output file: {output_path}") result_path = output_path # Generate overlay debug video if debug enabled if crop_path.debug: overlay_output_path = self._get_overlay_output_path(output_path) print(f"\nGenerating debug overlay video: {overlay_output_path.name}", file=sys.stderr) from .frame_processor import process_segment_with_overlay process_segment_with_overlay(input_path, overlay_output_path, crop_path) print(f"āœ“ Debug overlay saved: {overlay_output_path}\n", file=sys.stderr) return result_path def validate_output(self, output_path: Path) -> bool: """ Validate that output clip has both video and audio streams. Args: output_path: Path to output clip Returns: True if valid, False otherwise """ import json cmd = [ str(settings.ffprobe_path), "-v", "error", "-show_entries", "stream=codec_type", "-of", "json", str(output_path), ] try: result = subprocess.run(cmd, capture_output=True, text=True, check=True) data = json.loads(result.stdout) streams = data.get("streams", []) has_video = any(s.get("codec_type") == "video" for s in streams) has_audio = any(s.get("codec_type") == "audio" for s in streams) return has_video and has_audio except Exception: return False