"""Data models for the Fansifter Video Clipper.""" import math from enum import Enum from pathlib import Path from pydantic import BaseModel, Field, field_validator class CaptionStyle(str, Enum): """Visual preset for caption overlay text.""" VIRAL_HOOK = "viral_hook" AESTHETIC_LABEL = "aesthetic_label" MODERN_MINIMAL = "modern_minimal" TIKTOK = "tiktok" class CaptionPosition(str, Enum): """Vertical placement of caption overlay.""" TOP = "top" BOTTOM = "bottom" CENTER = "center" class VideoSegment(BaseModel): """Represents a video segment identified by Gemini API.""" start: float = Field(..., description="Start timestamp in seconds") end: float = Field(..., description="End timestamp in seconds") score: float = Field( ..., description=( "Trend/viral potential score from 0.0 to 1.0: how likely this clip is to make " "a viewer recognize the track, engage with it, or use it as a TikTok/Instagram sound" ), ) description: str = Field(..., description="Description of the segment") segment_type: str = Field( default="unknown", description=( "Musical section type — one of: 'chorus', 'hook', 'pre_chorus', 'verse', " "'bridge', 'intro', 'outro', 'instrumental'" ), ) caption_text: str | None = Field( default=None, description="Per-segment caption override (None = use ProcessingConfig default)", ) caption_style: CaptionStyle | None = Field( default=None, description="Per-segment caption style override" ) caption_position: CaptionPosition | None = Field( default=None, description="Per-segment caption position override" ) source_video: Path | None = Field( default=None, description="Path to the source video this segment was identified in (None = main input video)", ) class SegmentsResponse(BaseModel): """Container for Gemini API response with video segments.""" segments: list[VideoSegment] = Field(..., description="List of identified segments") class Detection(BaseModel): """Raw detection result with timestamp.""" time: float = Field(..., description="Time in seconds") center_x: int | None = Field(None, description="Subject X position (None if not detected)") center_y: int | None = Field(None, description="Subject Y position (None if not detected)") confidence: float = Field(..., ge=0.0, le=1.0, description="Detection confidence (0.0-1.0)") detection_method: str = Field(..., description="Method used (e.g., 'mediapipe_face')") class CropKeyframe(BaseModel): """Represents a single crop keyframe.""" time: float = Field(..., description="Time in seconds") x: int = Field(..., ge=0, description="X coordinate of crop window") y: int = Field(..., ge=0, description="Y coordinate of crop window") width: int = Field(..., gt=0, description="Width of crop window") height: int = Field(..., gt=0, description="Height of crop window") confidence: float = Field(default=0.0, ge=0.0, le=1.0, description="Detection confidence score") detection_method: str = Field(default="unknown", description="Detection method used") scene_change: bool = Field( default=False, description="True if this keyframe starts a new scene" ) class CropPath(BaseModel): """Represents a crop path over time for a segment.""" segment: VideoSegment = Field(..., description="The video segment") keyframes: list[CropKeyframe] = Field(..., description="List of crop keyframes") debug: bool = Field( default=False, description="If True, generate debug overlay video showing crop rectangles" ) class VideoMetadata(BaseModel): """Video file metadata.""" width: int = Field(..., gt=0, description="Video width in pixels") height: int = Field(..., gt=0, description="Video height in pixels") duration: float = Field(..., gt=0.0, description="Video duration in seconds") fps: float = Field(..., gt=0.0, description="Frames per second") codec: str = Field(..., description="Video codec") aspect_ratio: float = Field(..., gt=0.0, description="Width / Height ratio") class VideoSource(str, Enum): """Which input videos contribute visual segments for cut-to-music.""" ALL = "all" MAIN_ONLY = "main" EXTRAS_ONLY = "extras" class ProcessingConfig(BaseModel): """Configuration for the video processing pipeline.""" input_video: Path = Field(..., description="Path to input video file") output_dir: Path = Field( default=Path("./output"), description="Output directory for clips (relative to CWD of the calling process)", ) target_aspect_ratio: tuple[int, int] = Field( default=(9, 16), description="Target aspect ratio (width, height)" ) segment_duration: tuple[float, float] | None = Field( default=None, description="Min and max segment duration in seconds (None = Gemini decides, max 30s)", ) num_segments: int = Field(default=3, ge=1, description="Number of clips to generate") dry_run: bool = Field(default=False, description="Only analyze, don't process") debug: bool = Field(default=False, description="Enable debug mode with visualization") downsample_max_height: int = Field( default=720, gt=0, description="Max height for downsampling before Gemini API" ) downsample_max_fps: int = Field( default=24, gt=0, description="Max FPS for downsampling before Gemini API" ) caption_text: str | None = Field( default=None, description="Static text phrase to overlay on every frame (None = no caption)", ) caption_style: CaptionStyle = Field( default=CaptionStyle.VIRAL_HOOK, description="Visual style preset for the caption overlay", ) caption_position: CaptionPosition = Field( default=CaptionPosition.BOTTOM, description="Vertical placement of caption: top or bottom", ) extra_videos: list[Path] = Field( default_factory=list, description="Additional video files to draw segments from (UGC / artist socials)", ) video_source: VideoSource = Field( default=VideoSource.ALL, description="Which videos contribute segments: all, main, or extras", ) @field_validator("caption_text") @classmethod def _strip_empty_caption(cls, v: str | None) -> str | None: if v is not None: v = v.strip() return v or None class Config: """Pydantic config.""" arbitrary_types_allowed = True # --------------------------------------------------------------------------- # Cut-to-music pipeline models # --------------------------------------------------------------------------- class PerformanceWindow(BaseModel): """Usable performance window — excludes title cards at start and credits at end.""" performance_start: float = Field( ..., description=( "Timestamp (seconds) where actual performance footage begins, " "after any title/artist-name intro cards" ), ) performance_end: float = Field( ..., description=( "Timestamp (seconds) of the last frame of actual performance footage, " "before any closing credits, social-handle screens, or outro cards" ), ) class ChorusWindow(BaseModel): """The most energetic chorus/drop section identified by Gemini.""" chorus_start: float = Field(..., description="Start of chorus in source video (seconds)") chorus_end: float = Field(..., description="End of chorus in source video (seconds)") description: str = Field(default="", description="Gemini description of this section") class VisualPeakSegment(BaseModel): """A video segment with its identified climax (visual peak) timestamp.""" start: float = Field(..., description="Segment start in source video (seconds)") end: float = Field(..., description="Segment end in source video (seconds)") visual_peak_timestamp: float = Field( ..., description=( "Timestamp of the climax moment within this segment (seconds) — " "the single frame of highest visual impact (foot-hit, whip-pan, flash peak)" ), ) score: float = Field(..., ge=0.0, le=1.0, description="Energy/impact score 0-1") description: str = Field(default="", description="Description of the peak moment") source_video: Path | None = Field( default=None, description="Source video path (set after Gemini query, not part of schema)" ) class VisualPeaksResponse(BaseModel): """Container for Gemini API response with visual peak segments.""" segments: list[VisualPeakSegment] = Field(..., description="List of identified segments") class SyncedClip(BaseModel): """A single clip's mapping between source video and output timeline.""" source_segment: VisualPeakSegment = Field(..., description="Original Gemini-identified segment") audio_onset_time: float = Field( ..., description="Absolute time of the paired audio onset in source audio (seconds)" ) source_start: float = Field(..., description="Where to cut from in source video (seconds)") source_end: float = Field(..., description="Where to cut to in source video (seconds)") output_position: float = Field( ..., description="Position in output timeline (seconds from 0) = onset_N - onset_0" ) speed_factor: float = Field( ..., description=( "Time-warp factor: 1.0 = no change, <1.0 = slow down, >1.0 = speed up. " "Only applied when within ±speed_factor_tolerance of 1.0." ), ) duration: float = Field(..., description="Source clip duration (seconds, pre-warp)") slot_duration: float = Field( ..., description=( "Intended output duration = one inter-onset gap (seconds). " "Used by assembler for total audio extraction length. " "Equals gap_before + gap_after and is stable across scene snapping." ), ) class CutToMusicConfig(BaseModel): """Configuration for the cut-to-music pipeline.""" input_video: Path = Field(..., description="Path to input video file") output_dir: Path = Field( default=Path("./output"), description="Output directory for the assembled reel (relative to CWD of the calling process)", ) extra_videos: list[Path] = Field( default_factory=list, description="Additional video files/URLs to draw visual segments from", ) video_source: VideoSource = Field( default=VideoSource.ALL, description="Which videos contribute visual segments: all, main, or extras", ) total_duration: tuple[float, float] | None = Field( default=None, description="Target total output duration range in seconds (min, max); None = Gemini decides", ) segment_duration: tuple[float, float] | None = Field( default=None, description="Per-clip duration range in seconds (min, max); None = Gemini decides, max 30s", ) target_aspect_ratio: tuple[int, int] = Field( default=(9, 16), description="Target aspect ratio (width, height)" ) debug: bool = Field(default=False, description="Enable debug mode with visualization") caption_text: str | None = Field( default=None, description="Static text phrase to overlay on every frame" ) caption_style: CaptionStyle = Field( default=CaptionStyle.VIRAL_HOOK, description="Visual style preset for the caption overlay", ) caption_position: CaptionPosition = Field( default=CaptionPosition.BOTTOM, description="Vertical placement of caption: top or bottom", ) speed_factor_tolerance: float = Field( default=0.1, description="Max deviation from 1.0 before using crop-only mode instead of time-warp", ) @property def num_segments(self) -> int: """Number of clips in the final output, derived from duration ranges.""" avg_total = (self.total_duration[0] + self.total_duration[1]) / 2 if self.total_duration else 14.5 avg_segment = (self.segment_duration[0] + self.segment_duration[1]) / 2 if self.segment_duration else 4.0 return max(2, math.ceil(avg_total / avg_segment)) @property def min_onset_spacing(self) -> float: """Minimum seconds between onsets, equal to the lower segment duration bound.""" return self.segment_duration[0] if self.segment_duration else 2.0 class Config: """Pydantic config.""" arbitrary_types_allowed = True