"""Gemini API calls for the cut-to-music pipeline. Three structured-output queries are issued concurrently against uploaded video files: 1. _query_chorus() → ChorusWindow (main video only) 2. _query_performance_window() → PerformanceWindow (main video only) 3. _query_visual_peaks() → list[VisualPeakSegment] (one per source video) Each video in the peak-source list is downsampled, uploaded, and queried in parallel. Results are merged and the top-N segments by score are returned. """ from __future__ import annotations import json import tempfile import time from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from google import genai from google.genai import types from .config import settings from .ffmpeg_utils import downsample_video, probe_video from .models import ( ChorusWindow, PerformanceWindow, VideoMetadata, VideoSource, VisualPeakSegment, VisualPeaksResponse, ) class CutToMusicIntelligence: """Gemini API wrapper for dual-stream cut-to-music analysis.""" def __init__(self) -> None: self.client = genai.Client(api_key=settings.gemini_api_key) self.model_name = settings.gemini_model def analyze( self, video_path: Path, num_segments: int = 5, segment_duration: tuple[float, float] | None = None, extra_videos: list[Path] | None = None, video_source: VideoSource = VideoSource.ALL, ) -> tuple[ChorusWindow, list[VisualPeakSegment]]: """Upload videos and run chorus + visual-peak queries concurrently. Chorus and performance-window analysis always run on the main video. Visual-peak queries run on all applicable videos in parallel (determined by video_source). Each returned VisualPeakSegment is tagged with its source_video. The top num_segments peaks by score are returned. Args: video_path: Path to the main (audio source) video. num_segments: Number of peak segments to use (queries ask for N+1 each). segment_duration: Per-clip duration range (min, max) in seconds. None = Gemini decides. extra_videos: Additional video paths to draw visual segments from. video_source: Which videos contribute visual segments. Returns: Tuple of (ChorusWindow, list[VisualPeakSegment]) sorted descending by score. Raises: RuntimeError: If Gemini upload or a required query fails. ValueError: If chorus window is degenerate (< 5s) or no peak videos given. """ extra = extra_videos or [] if video_source == VideoSource.ALL: peak_videos: list[Path] = [video_path] + extra elif video_source == VideoSource.MAIN_ONLY: peak_videos = [video_path] else: # EXTRAS_ONLY if not extra: raise ValueError( "--video-source extras requires at least one --extra-video" ) peak_videos = extra # All videos that must be uploaded (main always included for chorus/window) all_videos: list[Path] = list(dict.fromkeys([video_path] + extra)) uploaded: dict[Path, object] = {} metadata_map: dict[Path, VideoMetadata] = {} temp_paths: list[Path] = [] try: # Phase A: downsample + upload all videos in parallel with ThreadPoolExecutor(max_workers=len(all_videos)) as executor: upload_futures = { executor.submit(self._prepare_and_upload, vp): vp for vp in all_videos } for future in as_completed(upload_futures): vp = upload_futures[future] exc = future.exception() if exc is not None: raise RuntimeError( f"Failed to prepare/upload {vp.name}: {exc}" ) from exc video_file, meta, temp_path = future.result() uploaded[vp] = video_file metadata_map[vp] = meta temp_paths.append(temp_path) main_file = uploaded[video_path] main_meta = metadata_map[video_path] # Phase B: run all Gemini queries concurrently n_workers = 2 + len(peak_videos) with ThreadPoolExecutor(max_workers=n_workers) as executor: chorus_future = executor.submit( self._query_chorus, main_file, main_meta.duration ) window_future = executor.submit( self._query_performance_window, main_file, main_meta.duration ) peak_futures: dict[object, Path] = { executor.submit( self._query_visual_peaks, uploaded[vp], num_segments + 1, metadata_map[vp].duration, segment_duration, ): vp for vp in peak_videos } results: dict = {"peaks_by_video": {}} all_futures = [chorus_future, window_future] + list(peak_futures.keys()) for future in as_completed(all_futures): exc = future.exception() if future is window_future: results["window"] = ( PerformanceWindow( performance_start=0.0, performance_end=main_meta.duration, ) if exc is not None else future.result() ) elif future is chorus_future: if exc is not None: raise RuntimeError(f"Chorus query failed: {exc}") from exc results["chorus"] = future.result() else: vp = peak_futures[future] if exc is not None: raise RuntimeError( f"Visual peaks query failed for {vp.name}: {exc}" ) from exc segments: list[VisualPeakSegment] = future.result() # Tag each segment with its source video results["peaks_by_video"][vp] = [ s.model_copy(update={"source_video": vp}) for s in segments ] chorus: ChorusWindow = results["chorus"] window: PerformanceWindow = results["window"] # Merge peaks; apply performance-window filter only to main-video peaks all_peaks: list[VisualPeakSegment] = [] for vp, segs in results["peaks_by_video"].items(): if vp == video_path: segs = [ p for p in segs if p.start >= window.performance_start and p.end <= window.performance_end ] all_peaks.extend(segs) top_peaks = sorted(all_peaks, key=lambda s: s.score, reverse=True)[:num_segments] if len(top_peaks) < 2: raise RuntimeError( f"Only {len(top_peaks)} usable visual peak segment(s) available after " "filtering. The performance window may be too narrow, or Gemini placed " "most segments outside the detected performance range. " "Try a different video or re-run (Gemini results vary)." ) return chorus, top_peaks finally: for _vp, video_file in uploaded.items(): try: self.client.files.delete(name=video_file.name) except Exception: pass for temp_path in temp_paths: if temp_path.exists(): temp_path.unlink() def analyze_chorus(self, video_path: Path) -> ChorusWindow: """Run only the chorus detection query against the main video. Cheaper than analyze() — uploads a single downsampled video and issues one Gemini query. Use this during the analyze phase so the chorus window is available for preview before the full cut-to-music processing run. Args: video_path: Path to the main (audio source) video. Returns: ChorusWindow with validated start/end timestamps. Raises: RuntimeError: If Gemini upload fails or the chorus query fails. """ video_file, metadata, temp_path = self._prepare_and_upload(video_path) try: return self._query_chorus(video_file, metadata.duration) finally: try: self.client.files.delete(name=video_file.name) except Exception: pass if temp_path.exists(): temp_path.unlink() def _prepare_and_upload( self, video_path: Path ) -> tuple[object, VideoMetadata, Path]: """Downsample and upload one video to the Gemini Files API. Returns: Tuple of (active Gemini file reference, VideoMetadata, temp downsampled path). """ with tempfile.NamedTemporaryFile(suffix="_downsampled.mp4", delete=False) as tmp: temp_path = Path(tmp.name) downsample_video(video_path, temp_path) metadata = probe_video(video_path) video_file = self._upload_and_wait(temp_path) return video_file, metadata, temp_path def _upload_and_wait(self, video_path: Path) -> object: """Upload video to Gemini Files API and poll until ACTIVE. Args: video_path: Downsampled video path to upload. Returns: Active Gemini file reference. Raises: RuntimeError: If file enters FAILED state or times out. """ video_file = self.client.files.upload(file=str(video_path)) timeout = 120 # seconds elapsed = 0 while video_file.state.name == "PROCESSING": if elapsed >= timeout: raise RuntimeError( f"Gemini file upload timed out after {timeout}s " f"(state: {video_file.state.name})" ) time.sleep(1) elapsed += 1 video_file = self.client.files.get(name=video_file.name) if video_file.state.name != "ACTIVE": raise RuntimeError( f"Gemini file processing failed with state: {video_file.state.name}" ) return video_file def _query_chorus(self, video_file: object, video_duration: float) -> ChorusWindow: """Ask Gemini to identify the most energetic chorus/drop section. Requests a 15-20s window. If Gemini returns < 10s, the window is expanded symmetrically (clamped to video bounds) to at least 10s. Args: video_file: Active Gemini file reference. video_duration: Total video duration in seconds (for validation). Returns: ChorusWindow with validated timestamps. """ generation_config = types.GenerateContentConfig( response_mime_type="application/json", response_schema=ChorusWindow, media_resolution=types.MediaResolution.MEDIA_RESOLUTION_LOW, ) prompt = f"""Analyze this music video and identify the single most energetic section to use as background music for a 15-20 second highlight reel. GOAL: Find the main chorus, drop, or hook — the part where the energy peaks, the melody is most recognizable, and the rhythm is strongest. REQUIREMENTS: - Return EXACTLY ONE section with chorus_start and chorus_end (in seconds) - Duration must be 15-20 seconds - Choose the section with the most percussive, rhythmically clear beat — this will be used to sync visual cuts to drum hits - Prefer the first/main chorus if the song has multiple - Avoid slow intros, quiet verses, and fade-outs - chorus_start and chorus_end must be within [0, {video_duration:.1f}] Return the start/end as floating-point seconds and a brief description.""" response = self.client.models.generate_content( model=self.model_name, contents=[video_file, prompt], config=generation_config, ) data = json.loads(response.text) chorus = ChorusWindow(**data) # Validate bounds chorus_start = max(0.0, min(chorus.chorus_start, video_duration - 5.0)) chorus_end = max(chorus_start + 5.0, min(chorus.chorus_end, video_duration)) # Expand to at least 10s window = chorus_end - chorus_start if window < 10.0: expand = (10.0 - window) / 2.0 chorus_start = max(0.0, chorus_start - expand) chorus_end = min(video_duration, chorus_end + expand) if chorus_end - chorus_start < 5.0: raise ValueError( f"Chorus window too short ({chorus_end - chorus_start:.1f}s). " "Try a different video or a video with a clearer song structure." ) return ChorusWindow( chorus_start=chorus_start, chorus_end=chorus_end, description=chorus.description, ) def _query_performance_window( self, video_file: object, video_duration: float, ) -> PerformanceWindow: """Ask Gemini where actual performance footage begins and ends. Used to hard-exclude title cards at the start and closing credits/outro cards at the end before selecting segments. Args: video_file: Active Gemini file reference. video_duration: Total video duration in seconds. Returns: PerformanceWindow with validated start/end timestamps. """ generation_config = types.GenerateContentConfig( response_mime_type="application/json", response_schema=PerformanceWindow, media_resolution=types.MediaResolution.MEDIA_RESOLUTION_LOW, ) prompt = f"""Identify the usable performance window in this music video. TASK: Return two timestamps marking where actual performance footage begins and ends. performance_start: The timestamp (in seconds) when actual performance footage begins. - Return 0.0 if the video starts immediately with the performance - Return a positive value only if there is a title card, "artist – song name" screen, or any introductory graphic that occupies most of the frame before the performance performance_end: The timestamp (in seconds) of the LAST frame of actual performance footage, immediately before any non-performance sequence begins. - Return {video_duration:.1f} if the video ends on performance footage - Return the timestamp BEFORE any closing credits, crew/cast lists, "follow us" screens, social media handles, "out now" full-frame cards, or any outro sequence begin - Be conservative: when uncertain, return an EARLIER (lower) timestamp IMPORTANT: Credits and social screens are common in official music videos even when brief. Err on the side of a lower performance_end rather than a higher one. Total video duration: {video_duration:.1f} seconds""" response = self.client.models.generate_content( model=self.model_name, contents=[video_file, prompt], config=generation_config, ) data = json.loads(response.text) pw = PerformanceWindow(**data) return PerformanceWindow( performance_start=max(0.0, min(pw.performance_start, video_duration)), performance_end=max(pw.performance_start, min(pw.performance_end, video_duration)), ) def _query_visual_peaks( self, video_file: object, num_segments: int, video_duration: float, segment_duration: tuple[float, float] | None = None, ) -> list[VisualPeakSegment]: """Ask Gemini to identify high-energy segments with their visual peak timestamps. Args: video_file: Active Gemini file reference. num_segments: Exactly this many segments to return. video_duration: Total video duration for validation. segment_duration: Per-clip duration range (min, max) in seconds. None = Gemini decides. Returns: List of VisualPeakSegment sorted descending by score. """ generation_config = types.GenerateContentConfig( response_mime_type="application/json", response_schema=VisualPeaksResponse, media_resolution=types.MediaResolution.MEDIA_RESOLUTION_LOW, ) if segment_duration is not None: seg_min, seg_max = segment_duration goal_duration = f"{seg_min:.0f}–{seg_max:.0f} seconds each" req_duration = f"Each segment: {seg_min:.0f}–{seg_max:.0f} seconds long" else: goal_duration = "appropriate lengths at your discretion, maximum 30 seconds each" req_duration = "Each segment: up to 30 seconds (choose the right length for each moment)" prompt = f"""You are identifying {num_segments} short high-impact moments in this music video to be cut together as a beat-synced highlight reel. GOAL: Find {num_segments} segments ({goal_duration}) spread throughout the video that each contain a visually compelling "peak moment" — the single frame of highest visual impact within that segment. THE VISUAL PEAK (visual_peak_timestamp): This is the exact timestamp (in seconds) of the climax frame — e.g.: - A dancer's foot lands on the ground (the impact frame, not the approach) - A camera whip-pan completes (the frame where motion peaks) - A flash or strobe reaches maximum brightness - A performer hits a pose, reaches the apex of a jump, or throws a punch - Any moment of sudden, high-contrast visual change This timestamp will be used to align the video cut to a drum hit. It MUST be within the segment's [start, end] range. REQUIREMENTS: - Return EXACTLY {num_segments} segments - {req_duration} - Segments must NOT overlap - Spread them throughout the video — don't cluster them all in the chorus - visual_peak_timestamp must satisfy: start < visual_peak_timestamp < end - Score 0.0-1.0: how visually impactful is this peak moment? - Total video duration: {video_duration:.1f} seconds EXCLUSIONS — never return these as segments: - Shots dominated by a full-screen title card (song/artist name at video open) - Shots dominated by closing credits or social-handle screens at video end - Any segment where static text is the primary visual element rather than performance NOTE: small or incidental text overlays throughout the video are acceptable. Return {num_segments} segments ordered highest to lowest score.""" response = self.client.models.generate_content( model=self.model_name, contents=[video_file, prompt], config=generation_config, ) data = json.loads(response.text) peaks_response = VisualPeaksResponse(**data) validated = self._validate_visual_peaks(peaks_response.segments, video_duration) return sorted(validated, key=lambda s: s.score, reverse=True) def _validate_visual_peaks( self, segments: list[VisualPeakSegment], video_duration: float, ) -> list[VisualPeakSegment]: """Validate segments and clamp visual_peak_timestamp to [start+0.1, end-0.1]. Removes segments that are: - Out of video bounds - Too short (< 1.0s) - Have inverted start/end Fixes visual_peak_timestamp that falls outside [start, end] by clamping to the midpoint of the segment. Args: segments: Raw segments from Gemini. video_duration: Total duration for bounds checking. Returns: List of valid segments. """ valid: list[VisualPeakSegment] = [] for seg in segments: # Basic bounds if seg.start < 0 or seg.end > video_duration + 0.5: continue if seg.start >= seg.end: continue if (seg.end - seg.start) < 1.0: continue # Clamp visual peak to [start+0.1, end-0.1] peak = seg.visual_peak_timestamp if peak <= seg.start or peak >= seg.end: peak = (seg.start + seg.end) / 2.0 valid.append( VisualPeakSegment( start=seg.start, end=min(seg.end, video_duration), visual_peak_timestamp=peak, score=seg.score, description=seg.description, source_video=seg.source_video, ) ) return valid