#!/usr/bin/env python3 # /// script # dependencies = [] # /// import argparse from bisect import bisect_left, bisect_right import json import logging import math import re import shutil import subprocess import tempfile from pathlib import Path def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Shorten still sections using FFmpeg stream copying, without re-encoding.", epilog="Cuts follow video keyframes and audio packets. Still sections may remain longer than requested. " "Audio during removed time is also removed, including any narration.", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) parser.add_argument("input", type=Path, help="Input video") parser.add_argument("--output", type=Path, help="Output video (default: INPUT.short with the original extension)") parser.add_argument("--min-still", type=float, default=2.0, help="Minimum still section duration to detect, in seconds") parser.add_argument("--keep-still", type=float, default=0.5, help="Target seconds to retain per still section; keyframe alignment may retain more") parser.add_argument("--noise", type=float, default=0.001, help="Frame difference tolerance, from 0 to 1; higher values ignore more movement") parser.add_argument("--dry-run", action="store_true", help="Report cuts without writing output") parser.add_argument("--overwrite", action="store_true", help="Replace an existing output file") arguments = parser.parse_args() if not math.isfinite(arguments.min_still) or arguments.min_still <= 0: parser.error("--min-still must be positive and finite") if not math.isfinite(arguments.keep_still) or arguments.keep_still <= 0: parser.error("--keep-still must be positive and finite") if not 0 <= arguments.noise <= 1: parser.error("--noise must be between 0 and 1") return arguments def probe_video(path: Path) -> tuple[float, float]: result = subprocess.run( args=["ffprobe", "-v", "error", "-show_streams", "-show_format", "-of", "json", str(path)], check=True, capture_output=True, text=True, ) metadata = json.loads(result.stdout) video = next((stream for stream in metadata["streams"] if stream["codec_type"] == "video"), None) if video is None: raise ValueError("Input has no video stream") duration = float(video.get("duration") or metadata["format"].get("duration", 0)) if not math.isfinite(duration) or duration <= 0: raise ValueError("Cannot determine video duration") start_time = float(video.get("start_time") or metadata["format"].get("start_time", 0)) return duration, start_time def probe_keyframes(path: Path, start_time: float) -> list[tuple[float, float]]: result = subprocess.run( args=["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_packets", "-show_entries", "packet=pts_time,dts_time,flags", "-of", "csv=p=0", str(path)], check=True, capture_output=True, text=True, ) keyframes = [] for line in result.stdout.splitlines(): fields = line.split(",") if len(fields) >= 3 and "K" in fields[2] and fields[0] != "N/A": presentation_time = float(fields[0]) - start_time decode_time = float(fields[1]) - start_time if fields[1] != "N/A" else presentation_time keyframes.append((max(0.0, presentation_time), decode_time)) if not keyframes: raise ValueError("No video keyframes found; cannot safely trim by stream copying") return sorted(keyframes) def parse_freezes(log: str, duration: float) -> list[tuple[float, float]]: freezes = [] start = None for kind, timestamp in re.findall(r"lavfi\.freezedetect\.freeze_(start|end):\s*([-+\d.eE]+)", log): timestamp = min(duration, max(0.0, float(timestamp))) if kind == "start": start = timestamp elif start is not None: if timestamp > start: freezes.append((start, timestamp)) start = None if start is not None and duration > start: freezes.append((start, duration)) return freezes def detect_freezes(path: Path, duration: float, min_still: float, noise: float) -> list[tuple[float, float]]: logging.info("Detecting still sections...") result = subprocess.run( args=[ "ffmpeg", "-hide_banner", "-nostdin", "-nostats", "-i", str(path), "-map", "0:v:0", "-vf", f"setpts=PTS-STARTPTS,freezedetect=noise={noise}:duration={min_still}", "-an", "-f", "null", "-", ], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, ) return parse_freezes(log=result.stderr, duration=duration) def plan_segments(duration: float, freezes: list[tuple[float, float]], keep_still: float, keyframes: list[float]) -> list[tuple[float, float]]: segments = [] cursor = 0.0 for start, end in freezes: if end - start <= keep_still: continue first = bisect_left(keyframes, start + keep_still / 2) last = bisect_right(keyframes, end - keep_still / 2) - 1 if first >= len(keyframes) or last < 0 or keyframes[first] >= keyframes[last]: logging.info(f"Still section {start:.2f}–{end:.2f}s: no removable span between keyframes") continue cut_start = keyframes[first] cut_end = keyframes[last] logging.info(f"Still section {start:.2f}–{end:.2f}s: retain {end - start - (cut_end - cut_start):.2f}s") if cut_start > cursor: segments.append((cursor, cut_start)) cursor = cut_end if cursor < duration: segments.append((cursor, duration)) return segments def copy_video(source: Path, output: Path, segments: list[tuple[float, float]], duration: float, start_time: float, keyframes: list[tuple[float, float]], overwrite: bool) -> None: decode_times = dict(keyframes) with tempfile.TemporaryDirectory() as temporary_directory: playlist = Path(temporary_directory) / "segments.ffconcat" local_source = Path(temporary_directory) / f"source{source.suffix}" local_source.symlink_to(target=source) entries = ["ffconcat version 1.0"] for start, end in segments: entries.extend([f"file '{local_source.name}'", f"inpoint {start + start_time:.6f}"]) if end < duration: # The concat demuxer's outpoint uses DTS, while inpoint uses PTS. entries.append(f"outpoint {decode_times[end] + start_time:.6f}") entries.append(f"duration {end - start:.6f}") playlist.write_text(data="\n".join(entries) + "\n") command = [ "ffmpeg", "-hide_banner", "-nostdin", "-loglevel", "warning", "-stats", "-y" if overwrite else "-n", "-f", "concat", "-safe", "0", "-auto_convert", "0", "-i", str(playlist), "-map", "0:v:0", "-map", "0:a?", "-c", "copy", ] if output.suffix.lower() in {".mp4", ".mov"}: command.extend(["-movflags", "+faststart"]) command.append(str(output)) subprocess.run(args=command, check=True) def main() -> None: logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s") arguments = parse_args() try: for executable in ("ffmpeg", "ffprobe"): if shutil.which(cmd=executable) is None: raise ValueError(f"{executable} must be installed and available on PATH") source = arguments.input.resolve(strict=True) output = (arguments.output or source.with_name(f"{source.stem}.short{source.suffix or '.mkv'}")).resolve() if source == output: raise ValueError("Input and output must be different files") if output.exists() and source.samefile(output): raise ValueError("Input and output must be different files") if not arguments.dry_run and output.exists() and not arguments.overwrite: raise ValueError(f"Output exists: {output}; use --overwrite to replace it") duration, start_time = probe_video(path=source) freezes = detect_freezes(path=source, duration=duration, min_still=arguments.min_still, noise=arguments.noise) keyframes = probe_keyframes(path=source, start_time=start_time) segments = plan_segments(duration=duration, freezes=freezes, keep_still=arguments.keep_still, keyframes=[time for time, _ in keyframes]) retained = sum(end - start for start, end in segments) logging.info(f"{duration:.2f}s → {retained:.2f}s; removing {duration - retained:.2f}s ({(1 - retained / duration):.1%})") if arguments.dry_run: return copy_video(source=source, output=output, segments=segments, duration=duration, start_time=start_time, keyframes=keyframes, overwrite=arguments.overwrite) logging.info(f"Saved {output}") except (OSError, ValueError, subprocess.CalledProcessError) as error: logging.error(str(error)) if isinstance(error, subprocess.CalledProcessError) and error.stderr: logging.error(error.stderr) raise SystemExit(1) from error if __name__ == "__main__": main()