AI-Talent-Force/dev_caio
0
1"""2ShortSmith v2 - Clip Extractor Module3 4Final clip extraction and output generation.5Handles cutting clips at precise timestamps with various output options.6"""7 8from pathlib import Path9from typing import List, Optional, Tuple10from dataclasses import dataclass, field11import shutil12 13from utils.logger import get_logger, LogTimer14from utils.helpers import (15 VideoProcessingError,16 ensure_dir,17 format_timestamp,18 get_unique_filename,19)20from config import get_config, ProcessingConfig21from core.video_processor import VideoProcessor, VideoMetadata22 23logger = get_logger("core.clip_extractor")24 25 26@dataclass27class ExtractedClip:28 """Represents an extracted video clip."""29 clip_path: Path # Path to the clip file30 start_time: float # Start timestamp in source video31 end_time: float # End timestamp in source video32 hype_score: float # Normalized hype score (0-1)33 rank: int # Rank among all clips (1 = best)34 thumbnail_path: Optional[Path] = None # Path to thumbnail35 36 # Metadata37 source_video: Optional[Path] = None38 person_detected: bool = False39 person_screen_time: float = 0.0 # Percentage of clip with target person40 41 # Additional scores42 visual_score: float = 0.043 audio_score: float = 0.044 motion_score: float = 0.045 46 @property47 def duration(self) -> float:48 """Clip duration in seconds."""49 return self.end_time - self.start_time50 51 @property52 def time_range(self) -> str:53 """Human-readable time range."""54 return f"{format_timestamp(self.start_time)} - {format_timestamp(self.end_time)}"55 56 def to_dict(self) -> dict:57 """Convert to dictionary for JSON serialization."""58 return {59 "clip_path": str(self.clip_path),60 "start_time": self.start_time,61 "end_time": self.end_time,62 "duration": self.duration,63 "hype_score": round(self.hype_score, 4),64 "rank": self.rank,65 "time_range": self.time_range,66 "visual_score": round(self.visual_score, 4),67 "audio_score": round(self.audio_score, 4),68 "motion_score": round(self.motion_score, 4),69 "person_detected": self.person_detected,70 "person_screen_time": round(self.person_screen_time, 4),71 }72 73 74@dataclass75class ClipCandidate:76 """A candidate segment for clip extraction."""77 start_time: float78 end_time: float79 hype_score: float80 visual_score: float = 0.081 audio_score: float = 0.082 motion_score: float = 0.083 person_score: float = 0.0 # Target person visibility84 85 @property86 def duration(self) -> float:87 return self.end_time - self.start_time88 89 90class ClipExtractor:91 """92 Extracts final clips from video based on hype scores.93 94 Handles:95 - Selecting top segments based on scores96 - Enforcing diversity (minimum gap between clips)97 - Adjusting clip boundaries to scene cuts98 - Generating thumbnails99 """100 101 def __init__(102 self,103 video_processor: VideoProcessor,104 config: Optional[ProcessingConfig] = None,105 ):106 """107 Initialize clip extractor.108 109 Args:110 video_processor: VideoProcessor instance for clip cutting111 config: Processing configuration (uses default if None)112 """113 self.video_processor = video_processor114 self.config = config or get_config().processing115 116 logger.info(117 f"ClipExtractor initialized (duration={self.config.min_clip_duration}-"118 f"{self.config.max_clip_duration}s, gap={self.config.min_gap_between_clips}s)"119 )120 121 def select_clips(122 self,123 candidates: List[ClipCandidate],124 num_clips: int,125 enforce_diversity: bool = True,126 ) -> List[ClipCandidate]:127 """128 Select top clips from candidates.129 130 Args:131 candidates: List of clip candidates with scores132 num_clips: Number of clips to select133 enforce_diversity: Enforce minimum gap between clips134 135 Returns:136 List of selected ClipCandidate objects137 """138 if not candidates:139 logger.warning("No candidates provided for selection")140 return []141 142 # Sort by hype score143 sorted_candidates = sorted(144 candidates, key=lambda c: c.hype_score, reverse=True145 )146 147 if not enforce_diversity:148 return sorted_candidates[:num_clips]149 150 # Select with diversity constraint151 selected = []152 min_gap = self.config.min_gap_between_clips153 154 for candidate in sorted_candidates:155 if len(selected) >= num_clips:156 break157 158 # Check if this candidate is far enough from existing selections159 is_diverse = True160 for existing in selected:161 # Calculate gap between clip starts162 gap = abs(candidate.start_time - existing.start_time)163 if gap < min_gap:164 is_diverse = False165 break166 167 if is_diverse:168 selected.append(candidate)169 170 # If we couldn't get enough with diversity, relax constraint171 if len(selected) < num_clips:172 logger.warning(173 f"Only {len(selected)} diverse clips found, "174 f"relaxing diversity constraint"175 )176 for candidate in sorted_candidates:177 if candidate not in selected:178 selected.append(candidate)179 if len(selected) >= num_clips:180 break181 182 logger.info(f"Selected {len(selected)} clips from {len(candidates)} candidates")183 return selected184 185 def adjust_to_scene_boundaries(186 self,187 candidates: List[ClipCandidate],188 scene_boundaries: List[float],189 tolerance: float = 1.0,190 ) -> List[ClipCandidate]:191 """192 Adjust clip boundaries to align with scene cuts.193 194 Args:195 candidates: List of clip candidates196 scene_boundaries: List of scene boundary timestamps197 tolerance: Maximum adjustment in seconds198 199 Returns:200 List of adjusted ClipCandidate objects201 """202 if not scene_boundaries:203 return candidates204 205 adjusted = []206 207 for candidate in candidates:208 new_start = candidate.start_time209 new_end = candidate.end_time210 211 # Find nearest scene boundary for start212 for boundary in scene_boundaries:213 if abs(boundary - candidate.start_time) < tolerance:214 new_start = boundary215 break216 217 # Find nearest scene boundary for end218 for boundary in scene_boundaries:219 if abs(boundary - candidate.end_time) < tolerance:220 new_end = boundary221 break222 223 # Ensure minimum duration224 if new_end - new_start < self.config.min_clip_duration:225 # Keep original boundaries226 new_start = candidate.start_time227 new_end = candidate.end_time228 229 adjusted.append(ClipCandidate(230 start_time=new_start,231 end_time=new_end,232 hype_score=candidate.hype_score,233 visual_score=candidate.visual_score,234 audio_score=candidate.audio_score,235 motion_score=candidate.motion_score,236 person_score=candidate.person_score,237 ))238 239 return adjusted240 241 def extract_clips(242 self,243 video_path: str | Path,244 output_dir: str | Path,245 candidates: List[ClipCandidate],246 num_clips: Optional[int] = None,247 generate_thumbnails: bool = True,248 reencode: bool = False,249 ) -> List[ExtractedClip]:250 """251 Extract clips from video.252 253 Args:254 video_path: Path to source video255 output_dir: Directory for output clips256 candidates: List of clip candidates257 num_clips: Number of clips to extract (None = use config default)258 generate_thumbnails: Whether to generate thumbnails259 reencode: Whether to re-encode clips (slower but precise)260 261 Returns:262 List of ExtractedClip objects263 """264 video_path = Path(video_path)265 output_dir = ensure_dir(output_dir)266 num_clips = num_clips or self.config.default_num_clips267 268 with LogTimer(logger, f"Extracting {num_clips} clips"):269 # Select top clips270 selected = self.select_clips(candidates, num_clips)271 272 if not selected:273 logger.warning("No clips to extract")274 return []275 276 # Extract each clip277 clips = []278 279 for rank, candidate in enumerate(selected, 1):280 try:281 clip = self._extract_single_clip(282 video_path=video_path,283 output_dir=output_dir,284 candidate=candidate,285 rank=rank,286 generate_thumbnail=generate_thumbnails,287 reencode=reencode,288 )289 clips.append(clip)290 291 except Exception as e:292 logger.error(f"Failed to extract clip {rank}: {e}")293 294 logger.info(f"Successfully extracted {len(clips)} clips")295 return clips296 297 def _extract_single_clip(298 self,299 video_path: Path,300 output_dir: Path,301 candidate: ClipCandidate,302 rank: int,303 generate_thumbnail: bool,304 reencode: bool,305 ) -> ExtractedClip:306 """Extract a single clip."""307 # Generate output filename308 clip_filename = f"clip_{rank:02d}_{format_timestamp(candidate.start_time).replace(':', '-')}.mp4"309 clip_path = output_dir / clip_filename310 311 # Cut the clip312 self.video_processor.cut_clip(313 video_path=video_path,314 output_path=clip_path,315 start_time=candidate.start_time,316 end_time=candidate.end_time,317 reencode=reencode,318 )319 320 # Generate thumbnail321 thumbnail_path = None322 if generate_thumbnail:323 try:324 thumb_filename = f"thumb_{rank:02d}.jpg"325 thumbnail_path = output_dir / "thumbnails" / thumb_filename326 thumbnail_path.parent.mkdir(exist_ok=True)327 328 # Thumbnail at 1/3 into the clip329 thumb_time = candidate.start_time + (candidate.duration / 3)330 self.video_processor.generate_thumbnail(331 video_path=video_path,332 output_path=thumbnail_path,333 timestamp=thumb_time,334 )335 except Exception as e:336 logger.warning(f"Failed to generate thumbnail for clip {rank}: {e}")337 thumbnail_path = None338 339 return ExtractedClip(340 clip_path=clip_path,341 start_time=candidate.start_time,342 end_time=candidate.end_time,343 hype_score=candidate.hype_score,344 rank=rank,345 thumbnail_path=thumbnail_path,346 source_video=video_path,347 visual_score=candidate.visual_score,348 audio_score=candidate.audio_score,349 motion_score=candidate.motion_score,350 person_detected=candidate.person_score > 0,351 person_screen_time=candidate.person_score,352 )353 354 def create_fallback_clips(355 self,356 video_path: str | Path,357 output_dir: str | Path,358 duration: float,359 num_clips: int,360 ) -> List[ExtractedClip]:361 """362 Create uniformly distributed clips when no highlights are detected.363 364 Args:365 video_path: Path to source video366 output_dir: Directory for output clips367 duration: Video duration in seconds368 num_clips: Number of clips to create369 370 Returns:371 List of fallback ExtractedClip objects372 """373 logger.warning("Creating fallback clips (no highlights detected)")374 375 clip_duration = self.config.default_clip_duration376 total_clip_time = clip_duration * num_clips377 378 if total_clip_time >= duration:379 # Video too short, adjust380 clip_duration = max(381 self.config.min_clip_duration,382 duration / (num_clips + 1)383 )384 385 # Calculate evenly spaced start times386 gap = (duration - clip_duration * num_clips) / (num_clips + 1)387 candidates = []388 389 for i in range(num_clips):390 start = gap + i * (clip_duration + gap)391 end = start + clip_duration392 393 candidates.append(ClipCandidate(394 start_time=start,395 end_time=min(end, duration),396 hype_score=0.5, # Neutral score397 ))398 399 return self.extract_clips(400 video_path=video_path,401 output_dir=output_dir,402 candidates=candidates,403 num_clips=num_clips,404 )405 406 def merge_adjacent_candidates(407 self,408 candidates: List[ClipCandidate],409 max_gap: float = 2.0,410 max_duration: Optional[float] = None,411 ) -> List[ClipCandidate]:412 """413 Merge adjacent high-scoring candidates into longer clips.414 415 Args:416 candidates: List of clip candidates417 max_gap: Maximum gap between candidates to merge418 max_duration: Maximum merged clip duration419 420 Returns:421 List of merged ClipCandidate objects422 """423 max_duration = max_duration or self.config.max_clip_duration424 425 if not candidates:426 return []427 428 # Sort by start time429 sorted_candidates = sorted(candidates, key=lambda c: c.start_time)430 merged = []431 current = sorted_candidates[0]432 433 for candidate in sorted_candidates[1:]:434 gap = candidate.start_time - current.end_time435 potential_duration = candidate.end_time - current.start_time436 437 if gap <= max_gap and potential_duration <= max_duration:438 # Merge439 current = ClipCandidate(440 start_time=current.start_time,441 end_time=candidate.end_time,442 hype_score=max(current.hype_score, candidate.hype_score),443 visual_score=max(current.visual_score, candidate.visual_score),444 audio_score=max(current.audio_score, candidate.audio_score),445 motion_score=max(current.motion_score, candidate.motion_score),446 person_score=max(current.person_score, candidate.person_score),447 )448 else:449 merged.append(current)450 current = candidate451 452 merged.append(current)453 return merged454 455 456# Export public interface457__all__ = ["ClipExtractor", "ExtractedClip", "ClipCandidate"]458