193 lines
9.2 KiB
Python
Executable File
193 lines
9.2 KiB
Python
Executable File
#!/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()
|