scene_cutter.py

#!/usr/bin/env python3
# /// script
# dependencies = [
#   "scenedetect[opencv]",
# ]
# ///

import os
import csv
import argparse
import sys
from typing import List, Tuple
from scenedetect import detect, ContentDetector, AdaptiveDetector, split_video_ffmpeg, VideoManager, FrameTimecode

class SceneCutterWrapper:
    def __init__(self, input_video: str, output_dir: str):
        self.input_video = input_video
        self.output_dir = output_dir
        self.filename = os.path.basename(input_video)
        self.file_base, _ = os.path.splitext(self.filename)

        if not os.path.exists(self.output_dir):
            os.makedirs(self.output_dir)

    def _get_cache_filename(self, method_name: str) -> str:
        """Generates the cache filename based on video name and detection method."""
        return os.path.join(self.output_dir, f"{self.file_base}_scenes_{method_name}.csv")

    def _load_scenes_from_cache(self, cache_file: str, framerate: float) -> List[Tuple[FrameTimecode, FrameTimecode]]:
        """Loads scenes from a CSV file and converts them to FrameTimecode objects."""
        scenes = []
        try:
            with open(cache_file, 'r', newline='') as f:
                reader = csv.DictReader(f)
                for row in reader:
                    start = FrameTimecode(row['Start Time'], fps=framerate)
                    end = FrameTimecode(row['End Time'], fps=framerate)
                    scenes.append((start, end))
            print(f"[INFO] Loaded {len(scenes)} scenes from cache: {cache_file}")
            return scenes
        except FileNotFoundError:
            return None
        except Exception as e:
            print(f"[ERROR] Could not load cache: {e}")
            return None

    def _save_scenes_to_cache(self, scenes: List, cache_file: str):
        """Saves detected scenes to a user-editable CSV."""
        with open(cache_file, 'w', newline='') as f:
            writer = csv.writer(f)
            writer.writerow(['Scene Number', 'Start Time', 'End Time', 'Start Frame', 'End Frame'])

            for i, scene in enumerate(scenes):
                start, end = scene
                writer.writerow([
                    i + 1,
                    start.get_timecode(),
                    end.get_timecode(),
                    start.get_frames(),
                    end.get_frames()
                ])
        print(f"[INFO] Scenes saved to cache: {cache_file}")

    def detect_scenes(self, method='content', threshold=27.0):
        """Detects scenes or loads them from cache."""
        cache_file = self._get_cache_filename(method)

        video = VideoManager([self.input_video])
        framerate = video.get_framerate()
        video.release()

        cached_scenes = self._load_scenes_from_cache(cache_file, framerate)
        if cached_scenes:
            return cached_scenes

        print(f"[INFO] Detecting scenes using {method} (threshold: {threshold})...")
        detector = ContentDetector(threshold=threshold) if method == 'content' else AdaptiveDetector(adaptive_threshold=threshold)

        scenes = detect(self.input_video, detector)
        self._save_scenes_to_cache(scenes, cache_file)

        return scenes

    def parse_selection(self, selection_str: str, total_scenes: int) -> List[int]:
        """Parses a selection string like "1,3-5,8" into a list of 0-based indices."""
        if not selection_str:
            return list(range(total_scenes))

        selected_indices = set()
        parts = selection_str.split(',')
        for part in parts:
            part = part.strip()
            if not part: continue
            if '-' in part:
                try:
                    # Check for ranges like 1-10
                    range_parts = part.split('-')
                    if len(range_parts) == 2:
                        start = int(range_parts[0])
                        end = int(range_parts[1])
                        for i in range(start - 1, end):
                            if 0 <= i < total_scenes: selected_indices.add(i)
                except ValueError: pass
            else:
                try:
                    idx = int(part) - 1
                    if 0 <= idx < total_scenes: selected_indices.add(idx)
                except ValueError: pass

        return sorted(list(selected_indices))

    def process_video(self, scenes, selection=None, mode='normal', scale=None, scene_mode='selective'):
        """Splits the video based on selected scenes and processing mode."""
        if not scenes:
            return

        indices = self.parse_selection(selection, len(scenes))
        if not indices:
            print("[WARN] No scenes selected.")
            return

        final_scenes = []
        if scene_mode == 'selective':
            final_scenes = [scenes[i] for i in indices]
        else:
            for i in range(len(indices)):
                current_idx = indices[i]
                start_time = scenes[current_idx][0]
                if i + 1 < len(indices):
                    next_selected_idx = indices[i + 1]
                    end_time = scenes[next_selected_idx][0]
                else:
                    end_time = scenes[-1][1]
                final_scenes.append((start_time, end_time))

        print(f"[INFO] Processing {len(final_scenes)} clips in '{scene_mode}' mode (Encoder: {mode})...")

        if mode == 'dry-run':
            for i, (s, e) in enumerate(final_scenes):
                print(f"Clip {i+1}: {s.get_timecode()} -> {e.get_timecode()}")
            return

        # Prepare overrides
        arg_list = []
        if mode == 'copy':
            arg_list = ["-c:v", "copy", "-c:a", "copy"]
        elif mode == 'prototype':
            arg_list = ["-c:v", "libx264", "-preset", "ultrafast", "-crf", "28", "-c:a", "aac"]
        else:
            arg_list = ["-c:v", "libx264", "-preset", "fast", "-crf", "22", "-c:a", "aac"]

        if scale and mode != 'copy':
            # Force even dimensions using -2 instead of -1 if necessary,
            # and ensure the filter is added to the list.
            clean_scale = scale.replace("-1", "-2")
            arg_list += ["-vf", f"scale={clean_scale}"]

        # scenedetect's arg_override expects a single string
        arg_override = " ".join(arg_list)

        try:
            split_video_ffmpeg(
                self.input_video,
                final_scenes,
                output_file_template=os.path.join(self.output_dir, "$VIDEO_NAME-Clip-$SCENE_NUMBER.mp4"),
                show_progress=True,
                arg_override=arg_override
            )
        except Exception as e:
            print(f"[ERROR] Failed to split: {e}")

def main():
    parser = argparse.ArgumentParser(description="PySceneDetect Wrapper with Sequence/Selective modes.")
    parser.add_argument("input", help="Path to input video")
    parser.add_argument("-o", "--output", default="output", help="Output directory")
    parser.add_argument("--method", default="content", choices=["content", "adaptive"])
    parser.add_argument("--threshold", type=float, default=27.0)
    parser.add_argument("--select", type=str, help="Select scenes (e.g., '1,3-5')")
    parser.add_argument("--mode", default="normal", choices=["normal", "prototype", "copy", "dry-run"])
    parser.add_argument("--scene-mode", default="selective", choices=["selective", "sequence"])
    parser.add_argument("--scale", type=str, help="FFmpeg scale (e.g., '-1:540')")
    parser.add_argument("--no-split", action="store_true")

    args = parser.parse_args()

    wrapper = SceneCutterWrapper(args.input, args.output)
    scenes = wrapper.detect_scenes(method=args.method, threshold=args.threshold)

    if not args.no_split:
        wrapper.process_video(scenes, selection=args.select, mode=args.mode, scale=args.scale, scene_mode=args.scene_mode)

if __name__ == "__main__":
    main()