"""Main pipeline orchestrator for video processing.""" import logging import os from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from rich.console import Console from rich.table import Table from .execution import FFmpegExecutor from .intelligence import IntelligenceEngine from .models import ProcessingConfig, VideoSegment from .reframing import Reframer logger = logging.getLogger(__name__) console = Console() class VideoProcessor: """Orchestrates the 3-step video processing pipeline.""" def __init__(self, config: ProcessingConfig) -> None: """ Initialize processor with configuration. Args: config: Processing configuration """ self.config = config self.intelligence = IntelligenceEngine() self.reframer = Reframer(debug=config.debug, output_dir=config.output_dir) self.executor = FFmpegExecutor() def process( self, mock_segments: list[VideoSegment] | None = None, output_dir: Path | None = None, ) -> list[Path]: """Execute the complete processing pipeline. Args: mock_segments: Optional pre-computed segments (bypasses Gemini API). Pass user-adjusted segments here to skip analysis. output_dir: Where to write clips. Overrides ``config.output_dir`` for this call only; the config object is not mutated. Defaults to ``config.output_dir`` (``./output``). Returns: List of paths to generated clips, or [] in dry-run mode. """ segments = self.analyze(mock_segments) if self.config.dry_run: self._print_dry_run_summary(segments) return [] return self.reframe_and_render(segments, output_dir=output_dir) def analyze(self, mock_segments: list[VideoSegment] | None = None) -> list[VideoSegment]: """Phase 1: identify segments and snap starts to frame-accurate scene cuts. If *mock_segments* is provided, Gemini is skipped and those segments are used directly (still snapped to scene boundaries). Pass user-adjusted segments here when driving the pipeline from an external UI. Returns: Validated, scene-snapped list of VideoSegment objects ready for reframing. The list is also printed to the console for logging. """ if mock_segments: console.print( "\n[bold yellow]Step 1: Using mock segments (skipping Gemini)...[/bold yellow]" ) segments = mock_segments else: console.print("\n[bold cyan]Step 1: Analyzing video with Gemini...[/bold cyan]") min_dur = self.config.segment_duration[0] if self.config.segment_duration else None max_dur = self.config.segment_duration[1] if self.config.segment_duration else None segments = self.intelligence.extract_segments( self.config.input_video, num_segments=self.config.num_segments, min_duration=min_dur, max_duration=max_dur, extra_videos=self.config.extra_videos or None, video_source=self.config.video_source, ) console.print(f"[green]✓[/green] Found {len(segments)} segments") console.print(" Snapping segment starts to scene boundaries...") snap_min = self.config.segment_duration[0] if self.config.segment_duration else 2.0 segments = self._snap_segment_starts(segments, snap_min) self._print_segments_summary(segments) return segments def reframe_and_render( self, segments: list[VideoSegment], output_dir: Path | None = None, main_video_segments: list[VideoSegment] | None = None, ) -> list[Path]: """Phases 2+3: reframe and encode clips for the given segments. Accepts externally provided or user-adjusted segments — call this after *analyze()* (and any UI-driven timestamp edits) to produce the final clips without re-running Gemini. Args: segments: Segments to reframe and encode. output_dir: Where to write clips. Overrides ``config.output_dir`` for this call only; the config object is not mutated. Defaults to ``config.output_dir`` (``./output``). main_video_segments: Segments from the main video, used as audio sources for clips sourced from extra (UGC) videos. When None and a UGC segment is encountered, falls back to audio from the main video at t=0. Returns: List of paths to the generated MP4 clips. """ out_dir = output_dir or self.config.output_dir out_dir.mkdir(parents=True, exist_ok=True) # Build audio-pairing list for UGC segments (cycle through main video segments) audio_segments: list[VideoSegment] = main_video_segments or [ s for s in segments if s.source_video is None ] # Steps 2 & 3 run as a pipelined producer-consumer: # - Main thread reframes segments sequentially (MediaPipe state is not thread-safe) # - Each crop path is submitted to the execution pool immediately on completion # - Clip generation overlaps with remaining reframing, saving up to min(T_exec, 2×T_reframe) console.print("\n[bold cyan]Steps 2+3: Reframing & generating clips (pipelined)...[/bold cyan]") max_workers = min(len(segments), os.cpu_count() or 1) console.print(f" {len(segments)} clip(s), {max_workers} execution worker(s)") output_paths: list[Path | None] = [None] * len(segments) ugc_idx = 0 # Counter for cycling through audio_segments for UGC clips def _process_one(job: tuple) -> tuple: idx, out_path, cp, audio_src, a_start, a_dur = job renderer = None seg = cp.segment caption_text = seg.caption_text if seg.caption_text is not None else self.config.caption_text caption_style = seg.caption_style if seg.caption_style is not None else self.config.caption_style caption_position = seg.caption_position if seg.caption_position is not None else self.config.caption_position if caption_text: from .captioning import CaptionConfig, CaptionRenderer renderer = CaptionRenderer(CaptionConfig( text=caption_text, style=caption_style.value, position=caption_position.value, frame_width=cp.keyframes[0].width, frame_height=cp.keyframes[0].height, )) source_video = seg.source_video or self.config.input_video self.executor.generate_clip( source_video, out_path, cp, caption_renderer=renderer, audio_source=audio_src, audio_start=a_start, audio_duration=a_dur, ) valid = self.executor.validate_output(out_path) return idx, out_path, valid futures: dict = {} with ThreadPoolExecutor(max_workers=max_workers) as pool: # Step 2: reframe each segment sequentially, submitting clip jobs as they're ready for i, segment in enumerate(segments, 1): console.print(f" [2] Reframing segment {i}/{len(segments)}...") source_video = segment.source_video or self.config.input_video crop_path = self.reframer.generate_crop_path( source_video, segment, self.config.target_aspect_ratio ) crop_path.debug = self.config.debug # Determine audio source for this clip. # UGC clips always override audio from the main video — never let # audio_segments being empty fall through to UGC audio. if segment.source_video is not None: audio_src = self.config.input_video a_dur = segment.end - segment.start if audio_segments: audio_seg = audio_segments[ugc_idx % len(audio_segments)] a_start = audio_seg.start ugc_idx += 1 else: # No paired segments available — use main video from the # same relative position as the UGC segment (best-effort). a_start = segment.start else: audio_src = None a_start = 0.0 a_dur = None start = crop_path.segment.start end = crop_path.segment.end output_path = out_dir / f"clip_{i}_s{start:.1f}-{end:.1f}.mp4" # Submit to pool immediately — worker starts encoding while we reframe the next segment futures[pool.submit(_process_one, (i, output_path, crop_path, audio_src, a_start, a_dur))] = i console.print(f"[green]✓[/green] Reframing complete — waiting for clip(s) to finish") # Step 3: collect results as each clip completes (workers may still be running) for future in as_completed(futures): i, output_path, valid = future.result() status = "[green]✓[/green] Valid" if valid else "[yellow]![/yellow] Warning: may be incomplete" console.print(f" [3] Clip {i} done: {output_path.name} — {status}") output_paths[i - 1] = output_path output_paths = [p for p in output_paths if p is not None] console.print(f"\n[green]✓[/green] Generated {len(output_paths)} clips") return output_paths def _snap_segment_starts( self, segments: list[VideoSegment], min_duration: float, ) -> list[VideoSegment]: """ Snap each segment's start time to the nearest PySceneDetect scene cut. Gemini timestamps have ±0.5–1s precision (it sees video at ~1fps). PySceneDetect is frame-accurate. For each segment we probe a ±2s window around the Gemini start and, if a cut is found, snap to it — provided the resulting segment is still at least min_duration seconds long. """ result = [] for seg in segments: source_video = seg.source_video or self.config.input_video boundary = self.reframer.find_scene_boundary_near(source_video, seg.start) if boundary is not None and abs(boundary - seg.start) > 0.05: if seg.end - boundary >= min_duration: delta = boundary - seg.start console.print( f" [dim]Snap: {seg.start:.2f}s → {boundary:.2f}s " f"({delta:+.2f}s)[/dim]" ) seg = VideoSegment( start=round(boundary, 3), end=seg.end, score=seg.score, description=seg.description, segment_type=seg.segment_type, caption_text=seg.caption_text, caption_style=seg.caption_style, caption_position=seg.caption_position, source_video=seg.source_video, ) result.append(seg) return result def _print_dry_run_summary(self, segments: list[VideoSegment]) -> None: """ Print formatted summary of segments in dry-run mode. Args: segments: List of video segments """ console.print("\n[bold yellow]DRY RUN - Identified Segments:[/bold yellow]\n") table = Table(show_header=True, header_style="bold magenta") table.add_column("Clip", style="cyan", width=6) table.add_column("Time Range", style="white", width=20) table.add_column("Duration", style="green", width=10) table.add_column("Score", style="yellow", width=8) table.add_column("Description", style="white") for i, segment in enumerate(segments, 1): duration = segment.end - segment.start time_range = f"{segment.start:.1f}s - {segment.end:.1f}s" score_str = f"{segment.score:.2f}" table.add_row(str(i), time_range, f"{duration:.1f}s", score_str, segment.description) console.print(table) console.print("\n[dim]Run without --dry-run to generate clips[/dim]\n") def _print_segments_summary(self, segments: list[VideoSegment]) -> None: """ Print formatted summary of detected segments. Args: segments: List of video segments """ console.print("\n[bold cyan]Detected Segments:[/bold cyan]\n") table = Table(show_header=True, header_style="bold magenta") table.add_column("#", style="cyan", width=4) table.add_column("Time Range", style="white", width=20) table.add_column("Duration", style="green", width=10) table.add_column("Score", style="yellow", width=8) table.add_column("Description", style="white") for i, segment in enumerate(segments, 1): duration = segment.end - segment.start time_range = f"{segment.start:.1f}s - {segment.end:.1f}s" score_str = f"{segment.score:.2f}" table.add_row( str(i), time_range, f"{duration:.1f}s", score_str, segment.description ) console.print(table)