""" FFmpeg utilities for audio-driven video assembly. This module provides production-grade FFmpeg filter building and execution for synchronizing video duration with actual audio duration. Key strategies: 1. Audio is the source of truth (measured with pydub, stored in DB) 2. Video clips are adjusted to match audio duration 3. Image clips are looped to match audio duration 4. atempo filter used for audio speed adjustment (pitch-preserving) 5. setpts used for video speed adjustment (frame-skipping) """ from __future__ import annotations import asyncio import logging from dataclasses import dataclass from pathlib import Path from typing import Optional logger = logging.getLogger(__name__) @dataclass class ClipDurationSpec: """Specification for a clip's timing and duration adjustment.""" clip_id: str file_path: Path is_image: bool is_video: bool original_duration: float target_duration: float @property def speed_factor(self) -> float: """Factor to scale playback speed: >1.0 = faster, <1.0 = slower.""" if self.original_duration <= 0: return 1.0 return self.original_duration / self.target_duration @property def needs_adjustment(self) -> bool: """True if clip needs speed/duration adjustment.""" return abs(self.speed_factor - 1.0) > 0.05 @dataclass class AudioClipSpec: """Specification for audio clip with optional speed adjustment.""" clip_id: str file_path: Path original_duration: float target_duration: float @property def atempo_value(self) -> float: """FFmpeg atempo value: 0.5 to 2.0 (1.0 = no change).""" if self.original_duration <= 0: return 1.0 factor = self.original_duration / self.target_duration return max(0.5, min(2.0, factor)) @property def needs_adjustment(self) -> bool: """True if audio needs tempo adjustment.""" return abs(self.atempo_value - 1.0) > 0.01 class FFmpegFilterBuilder: """Builds complex FFmpeg filter chains for audio-driven assembly.""" def __init__(self, width: int = 1080, height: int = 1920, fps: int = 30): self.width = width self.height = height self.fps = fps self.filter_parts = [] self.input_idx = 0 self.video_labels = [] self.audio_labels = [] def add_video_clip(self, clip_spec: ClipDurationSpec, ken_burns: Optional[dict] = None) -> str: """Add a video or image clip with automatic duration adjustment.""" label = f"v{self.input_idx}" input_id = self.input_idx self.input_idx += 1 if clip_spec.is_image: if ken_burns and ken_burns.get("enabled"): self._add_ken_burns_image(input_id, clip_spec, ken_burns, label) else: if clip_spec.needs_adjustment: self._add_looped_image_with_speed(input_id, clip_spec, label) else: self._add_looped_image(input_id, clip_spec, label) else: if clip_spec.needs_adjustment: self._add_speed_adjusted_video(input_id, clip_spec, label) else: self._add_standard_video(input_id, label) self.video_labels.append(label) return label def _add_looped_image(self, input_id: int, spec: ClipDurationSpec, label: str): """Loop image for target duration without speed adjustment.""" frames = int(spec.target_duration * self.fps) self.filter_parts.append( f"[{input_id}:v]scale={self.width}:{self.height}:force_original_aspect_ratio=decrease," f"pad={self.width}:{self.height}:(ow-iw)/2:(oh-ih)/2:black," f"fps={self.fps},loop=loop={frames}:size=1:start=0[{label}]" ) def _add_looped_image_with_speed(self, input_id: int, spec: ClipDurationSpec, label: str): """Loop image and adjust playback speed to match target duration.""" speed = spec.speed_factor frames = int(spec.original_duration * self.fps) self.filter_parts.append( f"[{input_id}:v]scale={self.width}:{self.height}:force_original_aspect_ratio=decrease," f"pad={self.width}:{self.height}:(ow-iw)/2:(oh-ih)/2:black," f"fps={self.fps},loop=loop={frames}:size=1:start=0," f"setpts='N/(FRAME_RATE*{speed:.3f})'/TB[{label}]" ) def _add_ken_burns_image(self, input_id: int, spec: ClipDurationSpec, kb: dict, label: str): """Image with Ken Burns effect, duration controlled by zoompan.""" frames = int(spec.target_duration * self.fps) s_scale = kb.get("startScale", 1.0) e_scale = kb.get("endScale", 1.15) s_x = kb.get("startX", 0.5) s_y = kb.get("startY", 0.5) e_x = kb.get("endX", 0.5) e_y = kb.get("endY", 0.5) self.filter_parts.append( f"[{input_id}:v]scale={self.width*2}:{self.height*2}," f"zoompan=z='min({s_scale}+({e_scale}-{s_scale})*on/{frames},{e_scale})'" f":x='iw*({s_x}+({e_x}-{s_x})*on/{frames})-iw/zoom/2'" f":y='ih*({s_y}+({e_y}-{s_y})*on/{frames})-ih/zoom/2'" f":d={frames}:s={self.width}x{self.height}:fps={self.fps}," f"setsar=1[{label}]" ) def _add_speed_adjusted_video(self, input_id: int, spec: ClipDurationSpec, label: str): """Video clip with speed adjustment to match target duration.""" speed = spec.speed_factor self.filter_parts.append( f"[{input_id}:v]scale={self.width}:{self.height}:force_original_aspect_ratio=decrease," f"pad={self.width}:{self.height}:(ow-iw)/2:(oh-ih)/2:black," f"setpts='N/(FRAME_RATE*{speed:.3f})'/TB," f"fps={self.fps}[{label}]" ) def _add_standard_video(self, input_id: int, label: str): """Standard video scaling without speed adjustment.""" self.filter_parts.append( f"[{input_id}:v]scale={self.width}:{self.height}:force_original_aspect_ratio=decrease," f"pad={self.width}:{self.height}:(ow-iw)/2:(oh-ih)/2:black," f"fps={self.fps}[{label}]" ) def build(self) -> str: """Return the complete filter_complex string.""" return ";".join(self.filter_parts) async def execute_ffmpeg(cmd: list[str], timeout: int = 600) -> tuple[bool, str, str]: """Execute FFmpeg command asynchronously. Returns: (success, stdout, stderr)""" try: proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) return proc.returncode == 0, stdout.decode(), stderr.decode() except asyncio.TimeoutError: try: proc.kill() except Exception: pass return False, "", "FFmpeg timeout exceeded" except Exception as e: return False, "", str(e)