Spaces:
Runtime error
Runtime error
| """Scene detection using PySceneDetect.""" | |
| from pathlib import Path | |
| from typing import Optional | |
| from loguru import logger | |
| def detect_scenes( | |
| video_path: Path, | |
| threshold: float = 27.0, | |
| min_scene_len_sec: float = 2.0, | |
| ) -> list[dict]: | |
| """Detect scene cuts and return list of scenes with timestamps. | |
| Returns: | |
| [{"start": float, "end": float, "duration": float}, ...] | |
| """ | |
| try: | |
| from scenedetect import open_video, SceneManager | |
| from scenedetect.detectors import ContentDetector | |
| video = open_video(str(video_path)) | |
| scene_manager = SceneManager() | |
| scene_manager.add_detector(ContentDetector(threshold=threshold)) | |
| logger.info(f"Running scene detection on: {video_path.name}") | |
| scene_manager.detect_scenes(video, show_progress=False) | |
| scene_list = scene_manager.get_scene_list() | |
| scenes = [] | |
| for start_tc, end_tc in scene_list: | |
| start = start_tc.get_seconds() | |
| end = end_tc.get_seconds() | |
| duration = end - start | |
| if duration >= min_scene_len_sec: | |
| scenes.append({"start": start, "end": end, "duration": duration}) | |
| logger.info(f"Detected {len(scenes)} scenes") | |
| if not scenes: | |
| logger.warning("0 scenes from ContentDetector — using fixed-interval fallback") | |
| return _fixed_interval_scenes(video_path, interval_sec=8.0) | |
| return scenes | |
| except ImportError: | |
| logger.warning("scenedetect not installed, using fixed-interval fallback") | |
| return _fixed_interval_scenes(video_path, interval_sec=5.0) | |
| except Exception as e: | |
| logger.error(f"Scene detection failed: {e}") | |
| return _fixed_interval_scenes(video_path, interval_sec=5.0) | |
| def _fixed_interval_scenes(video_path: Path, interval_sec: float = 5.0) -> list[dict]: | |
| """Fallback: split video into fixed-interval scenes.""" | |
| import subprocess | |
| result = subprocess.run( | |
| ["ffprobe", "-v", "error", "-show_entries", "format=duration", | |
| "-of", "default=noprint_wrappers=1:nokey=1", str(video_path)], | |
| capture_output=True, text=True | |
| ) | |
| try: | |
| total = float(result.stdout.strip()) | |
| except ValueError: | |
| total = 300.0 | |
| scenes = [] | |
| t = 0.0 | |
| while t < total: | |
| end = min(t + interval_sec, total) | |
| scenes.append({"start": t, "end": end, "duration": end - t}) | |
| t = end | |
| return scenes | |
| def sample_frames( | |
| video_path: Path, | |
| scenes: list[dict], | |
| output_dir: Path, | |
| frames_per_scene: int = 3, | |
| ) -> list[dict]: | |
| """Extract representative frames from each scene for vision analysis. | |
| Returns scenes with added 'frame_paths' key. | |
| """ | |
| import subprocess | |
| output_dir.mkdir(parents=True, exist_ok=True) | |
| result_scenes = [] | |
| for i, scene in enumerate(scenes): | |
| mid = scene["start"] + scene["duration"] / 2 | |
| frame_paths = [] | |
| # Sample frames at start, middle, end of scene | |
| timestamps = [ | |
| scene["start"] + scene["duration"] * 0.2, | |
| mid, | |
| scene["start"] + scene["duration"] * 0.8, | |
| ][:frames_per_scene] | |
| for j, ts in enumerate(timestamps): | |
| frame_path = output_dir / f"scene_{i:04d}_frame_{j}.jpg" | |
| cmd = [ | |
| "ffmpeg", "-y", "-ss", str(ts), "-i", str(video_path), | |
| "-vframes", "1", "-q:v", "2", "-vf", "scale=640:-1", | |
| str(frame_path) | |
| ] | |
| subprocess.run(cmd, capture_output=True) | |
| if frame_path.exists(): | |
| frame_paths.append(str(frame_path)) | |
| result_scenes.append({**scene, "index": i, "frame_paths": frame_paths}) | |
| return result_scenes | |