AI-Talent-Force/dev_caio

1"""2ShortSmith v2 - Scene Detector Module3 4PySceneDetect integration for detecting scene/shot boundaries in videos.5Uses content-aware detection to find cuts, fades, and transitions.6"""7 8from pathlib import Path9from typing import List, Optional, Tuple10from dataclasses import dataclass11 12from utils.logger import get_logger, LogTimer13from utils.helpers import VideoProcessingError14from config import get_config15 16logger = get_logger("core.scene_detector")17 18 19@dataclass20class Scene:21    """Represents a detected scene/shot in the video."""22    start_time: float  # Start timestamp in seconds23    end_time: float    # End timestamp in seconds24    start_frame: int   # Start frame number25    end_frame: int     # End frame number26 27    @property28    def duration(self) -> float:29        """Scene duration in seconds."""30        return self.end_time - self.start_time31 32    @property33    def frame_count(self) -> int:34        """Number of frames in scene."""35        return self.end_frame - self.start_frame36 37    @property38    def midpoint(self) -> float:39        """Midpoint timestamp of the scene."""40        return (self.start_time + self.end_time) / 241 42    def contains_timestamp(self, timestamp: float) -> bool:43        """Check if timestamp falls within this scene."""44        return self.start_time <= timestamp < self.end_time45 46    def overlaps_with(self, other: "Scene") -> bool:47        """Check if this scene overlaps with another."""48        return not (self.end_time <= other.start_time or other.end_time <= self.start_time)49 50    def __repr__(self) -> str:51        return f"Scene({self.start_time:.2f}s - {self.end_time:.2f}s, {self.duration:.2f}s)"52 53 54class SceneDetector:55    """56    Scene boundary detector using PySceneDetect.57 58    Supports multiple detection modes:59    - Content-aware: Detects cuts based on color histogram changes60    - Adaptive: Uses rolling average for more robust detection61    - Threshold: Simple luminance-based detection (for fades)62    """63 64    def __init__(65        self,66        threshold: float = 27.0,67        min_scene_length: float = 0.5,68        adaptive_threshold: bool = True,69    ):70        """71        Initialize scene detector.72 73        Args:74            threshold: Detection sensitivity (lower = more sensitive)75            min_scene_length: Minimum scene duration in seconds76            adaptive_threshold: Use adaptive threshold for varying content77 78        Raises:79            ImportError: If PySceneDetect is not installed80        """81        self.threshold = threshold82        self.min_scene_length = min_scene_length83        self.adaptive_threshold = adaptive_threshold84 85        # Verify PySceneDetect is available86        self._verify_dependencies()87 88        logger.info(89            f"SceneDetector initialized (threshold={threshold}, "90            f"min_length={min_scene_length}s, adaptive={adaptive_threshold})"91        )92 93    def _verify_dependencies(self) -> None:94        """Verify that PySceneDetect is installed."""95        try:96            import scenedetect97            self._scenedetect = scenedetect98        except ImportError as e:99            raise ImportError(100                "PySceneDetect is required for scene detection. "101                "Install with: pip install scenedetect[opencv]"102            ) from e103 104    def detect_scenes(105        self,106        video_path: str | Path,107        start_time: Optional[float] = None,108        end_time: Optional[float] = None,109    ) -> List[Scene]:110        """111        Detect scene boundaries in a video.112 113        Args:114            video_path: Path to the video file115            start_time: Start analysis at this timestamp (seconds)116            end_time: End analysis at this timestamp (seconds)117 118        Returns:119            List of detected Scene objects120 121        Raises:122            VideoProcessingError: If scene detection fails123        """124        from scenedetect import open_video, SceneManager125        from scenedetect.detectors import ContentDetector, AdaptiveDetector126 127        video_path = Path(video_path)128 129        if not video_path.exists():130            raise VideoProcessingError(f"Video file not found: {video_path}")131 132        with LogTimer(logger, f"Detecting scenes in {video_path.name}"):133            try:134                # Open video135                video = open_video(str(video_path))136 137                # Set up scene manager138                scene_manager = SceneManager()139 140                # Choose detector141                if self.adaptive_threshold:142                    detector = AdaptiveDetector(143                        adaptive_threshold=self.threshold,144                        min_scene_len=int(self.min_scene_length * video.frame_rate),145                    )146                else:147                    detector = ContentDetector(148                        threshold=self.threshold,149                        min_scene_len=int(self.min_scene_length * video.frame_rate),150                    )151 152                scene_manager.add_detector(detector)153 154                # Set time range if specified155                if start_time is not None:156                    start_frame = int(start_time * video.frame_rate)157                    video.seek(start_frame)158                else:159                    start_frame = 0160 161                if end_time is not None:162                    duration_frames = int((end_time - (start_time or 0)) * video.frame_rate)163                else:164                    duration_frames = None165 166                # Detect scenes167                scene_manager.detect_scenes(video, frame_skip=0, end_time=duration_frames)168 169                # Get scene list170                scene_list = scene_manager.get_scene_list()171 172                # Convert to Scene objects173                scenes = []174                for scene_start, scene_end in scene_list:175                    scene = Scene(176                        start_time=scene_start.get_seconds(),177                        end_time=scene_end.get_seconds(),178                        start_frame=scene_start.get_frames(),179                        end_frame=scene_end.get_frames(),180                    )181                    scenes.append(scene)182 183                logger.info(f"Detected {len(scenes)} scenes")184 185                # If no scenes detected, create a single scene for entire video186                if not scenes:187                    logger.warning("No scene cuts detected, treating as single scene")188                    video_duration = video.duration.get_seconds()189                    scenes = [Scene(190                        start_time=0,191                        end_time=video_duration,192                        start_frame=0,193                        end_frame=int(video_duration * video.frame_rate),194                    )]195 196                return scenes197 198            except Exception as e:199                logger.error(f"Scene detection failed: {e}")200                raise VideoProcessingError(f"Scene detection failed: {e}") from e201 202    def detect_scene_boundaries(203        self,204        video_path: str | Path,205    ) -> List[float]:206        """207        Get just the scene boundary timestamps.208 209        Args:210            video_path: Path to the video file211 212        Returns:213            List of timestamps where scene changes occur214        """215        scenes = self.detect_scenes(video_path)216        boundaries = [0.0]  # Start of video217 218        for scene in scenes:219            if scene.start_time > 0:220                boundaries.append(scene.start_time)221 222        # Remove duplicates and sort223        return sorted(set(boundaries))224 225    def get_scene_at_timestamp(226        self,227        scenes: List[Scene],228        timestamp: float,229    ) -> Optional[Scene]:230        """231        Find the scene containing a specific timestamp.232 233        Args:234            scenes: List of detected scenes235            timestamp: Timestamp to search for236 237        Returns:238            Scene containing the timestamp, or None if not found239        """240        for scene in scenes:241            if scene.contains_timestamp(timestamp):242                return scene243        return None244 245    def get_scenes_in_range(246        self,247        scenes: List[Scene],248        start_time: float,249        end_time: float,250    ) -> List[Scene]:251        """252        Get all scenes that overlap with a time range.253 254        Args:255            scenes: List of detected scenes256            start_time: Range start257            end_time: Range end258 259        Returns:260            List of overlapping scenes261        """262        range_scene = Scene(263            start_time=start_time,264            end_time=end_time,265            start_frame=0,266            end_frame=0,267        )268 269        return [s for s in scenes if s.overlaps_with(range_scene)]270 271    def merge_short_scenes(272        self,273        scenes: List[Scene],274        min_duration: float = 2.0,275    ) -> List[Scene]:276        """277        Merge scenes that are shorter than minimum duration.278 279        Args:280            scenes: List of scenes to process281            min_duration: Minimum scene duration in seconds282 283        Returns:284            List of merged scenes285        """286        if not scenes:287            return []288 289        merged = []290        current = scenes[0]291 292        for scene in scenes[1:]:293            if current.duration < min_duration:294                # Merge with next scene295                current = Scene(296                    start_time=current.start_time,297                    end_time=scene.end_time,298                    start_frame=current.start_frame,299                    end_frame=scene.end_frame,300                )301            else:302                merged.append(current)303                current = scene304 305        merged.append(current)306 307        logger.debug(f"Merged {len(scenes)} scenes into {len(merged)}")308        return merged309 310    def split_long_scenes(311        self,312        scenes: List[Scene],313        max_duration: float = 30.0,314        video_fps: float = 30.0,315    ) -> List[Scene]:316        """317        Split scenes that are longer than maximum duration.318 319        Args:320            scenes: List of scenes to process321            max_duration: Maximum scene duration in seconds322            video_fps: Video frame rate for frame calculations323 324        Returns:325            List of scenes with long ones split326        """327        result = []328 329        for scene in scenes:330            if scene.duration <= max_duration:331                result.append(scene)332            else:333                # Split into chunks334                num_chunks = int(scene.duration / max_duration) + 1335                chunk_duration = scene.duration / num_chunks336 337                for i in range(num_chunks):338                    start = scene.start_time + (i * chunk_duration)339                    end = min(scene.start_time + ((i + 1) * chunk_duration), scene.end_time)340 341                    result.append(Scene(342                        start_time=start,343                        end_time=end,344                        start_frame=int(start * video_fps),345                        end_frame=int(end * video_fps),346                    ))347 348        logger.debug(f"Split {len(scenes)} scenes into {len(result)}")349        return result350 351 352# Export public interface353__all__ = ["SceneDetector", "Scene"]354