Auto-merge same-resolution video files with ffmpeg and Python

$ python3 merge.py /path/to/videos

=== DRY RUN MODE ===
Found 3 group(s) capable of being concatenated.

Group: 1920x1080 (h264) | Will merge 138 files together | Total duration: 02:15:30 | Total size: 12.3 GB:
  - video1.mp4 (00:01:23, 85.2 MB)
  - video2.mp4 (00:00:45, 42.1 MB)
  ...
  => /usr/bin/python3 merge.py --res 1920x1080 --codec h264 --run

Group: 2160x3840 (av1) | Will merge 59 files together | Total duration: 00:14:25 | Total size: 1.4 GB:
  - clip1.mp4 (00:00:08, 14.1 MB)
  - clip2.mp4 (00:00:09, 14.4 MB)
  ...
  => /usr/bin/python3 merge.py --res 2160x3840 --codec av1 --run

python3 merge.py --res 2160x3840 --codec av1 --run -D

# -skip_frame nokey makes ffmpeg ignore all non-keyframes (very fast)
# -f framemd5 outputs MD5 hashes of raw pixel data for each keyframe
ffmpeg -skip_frame nokey -i video.mp4 -map 0:v:0 -f framemd5 -

#!/usr/bin/env python3
import os
import sys
import argparse
import subprocess
import json
import shutil
from collections import defaultdict
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor

def parse_arguments():
    parser = argparse.ArgumentParser(
        description="Find and concatenate video files with the same resolution and codec without re-encoding.",
        epilog="By default, the script acts as a dry run and only finds groups. Use --run to execute."
    )
    
    # Target directory
    parser.add_argument("dir", nargs="?", default=".", help="Target directory (default: current directory)")
    
    # CLI Filters
    parser.add_argument("-e", "--ext", action="append", help="Filter: File extension to process (e.g., .mp4). Can be used multiple times.")
    parser.add_argument("--min-files", type=int, default=2, help="Filter: Minimum number of files in a group to be considered (default: 2)")
    parser.add_argument("--min-size-mb", type=float, default=0, help="Filter: Minimum file size in MB")
    parser.add_argument("--max-size-mb", type=float, default=0, help="Filter: Maximum file size in MB (0 = unlimited)")
    parser.add_argument("--contains", type=str, default="", help="Filter: Filename must contain this specific string")
    parser.add_argument("--res", type=str, default="", help="Filter: Target resolution as WxH (e.g., 2160x3840). Only this group will be processed.")
    parser.add_argument("--codec", type=str, default="", help="Filter: Target codec (e.g., av1). Only this group will be processed.")
    parser.add_argument("--max-duration", type=str, default="", help="Filter: Skip files longer than this duration (hh:mm:ss or mm:ss or seconds). E.g., 00:01:00 skips files over 1 minute.")
    parser.add_argument("--sort", type=str, default="count", choices=["count", "duration", "size", "count-asc", "duration-asc", "size-asc"],
                        help="Sort groups by: count (most files, default), duration (longest total), size (largest total), or -asc variants.")
    
    # Actions
    parser.add_argument("-N", "--dry-run", action="store_true", help="Dry-run mode (default behavior). Only find and show groups.")
    parser.add_argument("--run", action="store_true", help="Execute the concatenation and verification (overrides default dry-run).")
    parser.add_argument("-D", "--delete", action="store_true", help="Delete source files ONLY AFTER successful concatenation and keyframe verification.")
    parser.add_argument("--reencode", action="store_true", help="Re-encode during concatenation (slower but lossless joins). Uses concat demuxer with re-encoding instead of stream copy.")
    parser.add_argument("--av1", action="store_true", help="Force SVT-AV1 re-encoding (CRF 24, preset 8). Overrides --vcodec/--crf/--preset.")
    parser.add_argument("--h264", action="store_true", help="Force H.264 NVENC hardware re-encoding (CQ 20, preset p4). Overrides --vcodec/--crf/--preset.")
    parser.add_argument("--vcodec", type=str, default="h264_nvenc", help="Video codec for re-encoding (default: h264_nvenc hardware). E.g., libx264, libx265, libsvtav1, h264_vaapi, h264_qsv.")
    parser.add_argument("--crf", type=str, default="20", help="CRF/quality value for re-encoding (default: 20). For NVENC, this maps to -cq.")
    parser.add_argument("--preset", type=str, default="p4", help="Encoder preset. NVENC: p1-p7 (default: p4). libsvtav1: 0-13. libx264: medium/slow/etc.")
    parser.add_argument("--tolerance", type=float, default=0.1, help="Minimum duration verification tolerance in seconds (default: 0.1). Actual tolerance is max(this, 0.01%% of source duration).")
    parser.add_argument("--strict", action="store_true", help="Use strict keyframe hash verification instead of duration-based (for stream copy mode).")
    parser.add_argument("-j", "--parallel", type=int, default=4, help="Number of parallel ffprobe/ffmpeg processes for metadata probing and timestamp normalization (default: 4).")

    args = parser.parse_args()

    # --av1 forces SVT-AV1 mode
    if args.av1:
        args.vcodec = "libsvtav1"
        args.crf = "24"
        args.preset = "8"

    # --h264 forces H.264 NVENC hardware mode
    if args.h264:
        args.vcodec = "h264_nvenc"
        args.crf = "20"
        args.preset = "p4"
        args.reencode = True

    # Parse --max-duration into seconds
    args.max_duration_s = 0.0
    if args.max_duration:
        try:
            args.max_duration_s = parse_duration(args.max_duration)
        except ValueError as e:
            print(f"Error: {e}", file=sys.stderr)
            sys.exit(1)

    # Default to dry-run if --run is not specified
    if not args.run:
        args.dry_run = True

    # Normalize extensions
    if not args.ext:
        args.ext = ['.mp4']
    else:
        args.ext = [e if e.startswith('.') else f'.{e}' for e in args.ext]
    args.ext = [e.lower() for e in args.ext]

    return args

def check_dependencies():
    if not shutil.which("ffmpeg") or not shutil.which("ffprobe"):
        print("Error: 'ffmpeg' and 'ffprobe' must be installed and available in your PATH.", file=sys.stderr)
        sys.exit(1)

def get_video_metadata(filepath):
    """Uses ffprobe to extract width, height, and codec of the primary video stream."""
    cmd = [
        "ffprobe", "-v", "error", "-select_streams", "v:0",
        "-show_entries", "stream=width,height,codec_name",
        "-of", "json", str(filepath)
    ]
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, check=True)
        data = json.loads(result.stdout)
        streams = data.get("streams", [])
        if streams:
            s = streams[0]
            return s.get("width"), s.get("height"), s.get("codec_name")
    except Exception as e:
        print(f"Error reading metadata for {filepath}: {e}", file=sys.stderr)
    return None, None, None

def probe_file(filepath):
    """Combined ffprobe call: returns (filepath, (w, h, codec), duration, size_bytes) in a single subprocess."""
    cmd = [
        "ffprobe", "-v", "error",
        "-select_streams", "v:0",
        "-show_entries", "stream=width,height,codec_name",
        "-show_entries", "format=duration",
        "-of", "json", str(filepath)
    ]
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, check=True)
        data = json.loads(result.stdout)
        streams = data.get("streams", [])
        w, h, codec = None, None, None
        if streams:
            s = streams[0]
            w, h, codec = s.get("width"), s.get("height"), s.get("codec_name")
        duration = None
        fmt = data.get("format", {})
        if fmt and "duration" in fmt:
            try:
                duration = float(fmt["duration"])
            except (ValueError, TypeError):
                pass
        size = filepath.stat().st_size
        return filepath, (w, h, codec), duration, size
    except Exception as e:
        print(f"Error reading metadata for {filepath}: {e}", file=sys.stderr)
        try:
            size = filepath.stat().st_size
        except OSError:
            size = 0
        return filepath, (None, None, None), None, size

def get_duration(filepath):
    """Uses ffprobe to get the duration of a file in seconds."""
    cmd = [
        "ffprobe", "-v", "error",
        "-show_entries", "format=duration",
        "-of", "default=noprint_wrappers=1:nokey=1",
        str(filepath)
    ]
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, check=True)
        return float(result.stdout.strip())
    except Exception:
        return None

def format_duration(seconds):
    """Format seconds as hh:mm:ss."""
    if seconds is None:
        return "?"
    h = int(seconds // 3600)
    m = int((seconds % 3600) // 60)
    s = int(seconds % 60)
    return f"{h:02d}:{m:02d}:{s:02d}"

def format_size(num_bytes):
    """Format bytes as human-readable size (e.g. 1.5 GB)."""
    if num_bytes is None:
        return "?"
    for unit in ["B", "KB", "MB", "GB", "TB"]:
        if num_bytes < 1024:
            return f"{num_bytes:.1f} {unit}"
        num_bytes /= 1024
    return f"{num_bytes:.1f} PB"

def parse_duration(s):
    """Parse a duration string (hh:mm:ss, mm:ss, or plain seconds) into seconds (float)."""
    s = s.strip()
    if ":" in s:
        parts = s.split(":")
        parts = [float(p) for p in parts]
        if len(parts) == 2:
            return parts[0] * 60 + parts[1]
        elif len(parts) == 3:
            return parts[0] * 3600 + parts[1] * 60 + parts[2]
        else:
            raise ValueError(f"Invalid duration format: {s}")
    return float(s)

def get_keyframe_hashes(filepath):
    """
    Efficiently extracts keyframe images and hashes them.
    -skip_frame nokey: ignores all frames except keyframes (insanely fast).
    -f framemd5: generates hashes of the raw, uncompressed pixel arrays in memory.
    """
    cmd = [
        "ffmpeg", "-loglevel", "error",
        "-skip_frame", "nokey",
        "-i", str(filepath),
        "-map", "0:v:0",
        "-f", "framemd5", "-"
    ]
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, check=True)
        hashes = []
        # Parse the framemd5 output format to extract the MD5 hash of each frame
        for line in result.stdout.splitlines():
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            parts = line.split(",")
            if len(parts) >= 6:
                hashes.append(parts[-1].strip())
        return hashes
    except subprocess.CalledProcessError as e:
        print(f"Error extracting keyframes from {filepath.name}: {e}", file=sys.stderr)
        return None

def remux_normalize_timestamps(filepath, output_dir):
    """Pre-mux a file with -c copy to normalize timestamps (avoid_negative_ts, genpts, consistent timescale).
    Returns the path to the normalized temp file, or None on failure."""
    tmp_file = output_dir / f".norm_{filepath.name}"
    cmd = [
        "ffmpeg", "-loglevel", "error", "-y",
        "-fflags", "+genpts",
        "-i", str(filepath),
        "-c", "copy",
        "-avoid_negative_ts", "make_zero",
        "-video_track_timescale", "90000",
        str(tmp_file)
    ]
    try:
        subprocess.run(cmd, check=True)
        return tmp_file
    except subprocess.CalledProcessError as e:
        print(f"    Error normalizing timestamps for {filepath.name}: {e}", file=sys.stderr)
        if tmp_file.exists():
            tmp_file.unlink()
        return None

def concatenate_group(files, target_file, reencode=False, vcodec="h264_nvenc", crf="20", preset="p4", normalize=True, parallel=4):
    """Concatenates video files using the ffmpeg concat demuxer.
    If reencode=False, uses stream copy (-c copy) with timestamp normalization to prevent frame drops.
    If reencode=True, re-encodes video."""
    concat_txt = Path(target_file).with_suffix('.txt')
    temp_dir = Path(target_file).parent / ".merge_tmp"
    
    files_to_concat = files
    temp_files = []
    
    # For stream copy mode, pre-mux each file to normalize timestamps (prevents frame drops at boundaries)
    if not reencode and normalize:
        print(f"  [*] Normalizing timestamps for {len(files)} files with {parallel} parallel workers...")
        temp_dir.mkdir(exist_ok=True)
        norm_results = {}
        with ThreadPoolExecutor(max_workers=parallel) as executor:
            futures = {executor.submit(remux_normalize_timestamps, f, temp_dir): f for f in files}
            for future in futures:
                f = futures[future]
                norm = future.result()
                if norm is not None:
                    norm_results[f] = norm
                    temp_files.append(norm)
        # Preserve original sort order
        files_to_concat = []
        for f in files:
            if f in norm_results:
                files_to_concat.append(norm_results[f])
            else:
                files_to_concat.append(f)
    
    with open(concat_txt, 'w', encoding='utf-8') as f:
        for filepath in files_to_concat:
            # properly escape single quotes for ffmpeg concat demuxer
            escaped_path = str(filepath.absolute()).replace("'", "'\\''")
            f.write(f"file '{escaped_path}'\n")

    cmd = [
        "ffmpeg", "-loglevel", "error", "-y",
        "-f", "concat",
        "-safe", "0",
        "-i", str(concat_txt),
    ]
    if reencode:
        # NVENC uses -cq instead of -crf for quality-based encoding
        quality_flag = "-cq" if "nvenc" in vcodec else "-crf"
        cmd.extend(["-c:v", vcodec, quality_flag, str(crf), "-preset", str(preset), "-c:a", "copy"])
    else:
        # Stream copy with timestamp fix to eliminate frame drops at segment boundaries
        cmd.extend(["-c", "copy", "-avoid_negative_ts", "make_zero", "-fflags", "+genpts"])
    cmd.append(str(target_file))

    try:
        subprocess.run(cmd, check=True)
        concat_txt.unlink()  # Cleanup the text list
        # Cleanup temp normalized files
        for tf in temp_files:
            tf.unlink()
        if temp_dir.exists():
            try:
                temp_dir.rmdir()
            except OSError:
                pass
        return True
    except subprocess.CalledProcessError as e:
        print(f"Error concatenating to {target_file.name}: {e}", file=sys.stderr)
        # Cleanup temp files on failure
        for tf in temp_files:
            if tf.exists():
                tf.unlink()
        if temp_dir.exists():
            try:
                temp_dir.rmdir()
            except OSError:
                pass
        return False

def verify_concatenation_duration(source_files, target_file, min_tolerance=0.1):
    """Verifies concatenation by comparing total source duration to target duration.
    Tolerance is max(min_tolerance, 0.01% of source total duration)."""
    source_total = 0.0
    for f in source_files:
        d = get_duration(f)
        if d is None:
            print(f"    [X] Could not read duration for {f.name}")
            return False
        source_total += d
    
    target_duration = get_duration(target_file)
    if target_duration is None:
        print(f"    [X] Could not read duration for {target_file.name}")
        return False
    
    # Tolerance: 0.01% of source length, but at least min_tolerance (0.1s)
    tolerance = max(min_tolerance, source_total * 0.0001)
    diff = abs(source_total - target_duration)
    print(f"    -> Source total duration: {source_total:.2f}s")
    print(f"    -> Target duration:       {target_duration:.2f}s")
    print(f"    -> Difference:            {diff:.2f}s (tolerance: {tolerance:.4f}s)")
    
    if diff > tolerance:
        print(f"    [X] VERIFICATION FAILED: Duration mismatch exceeds tolerance.")
        return False
    
    return True

def verify_concatenation(source_files, target_file, parallel=4):
    """Verifies that the sequence of keyframe image hashes matches perfectly."""
    source_hashes = []
    
    print(f"    -> Hashing keyframe images for {len(source_files)} source files with {parallel} parallel workers...")
    with ThreadPoolExecutor(max_workers=parallel) as executor:
        results = list(executor.map(get_keyframe_hashes, source_files))
    for f, hashes in zip(source_files, results):
        if hashes is None:
            print(f"    [X] Error extracting keyframes from {f.name}")
            return False
        source_hashes.extend(hashes)

    print(f"    -> Hashing keyframe images for concatenated {target_file.name}...")
    target_hashes = get_keyframe_hashes(target_file)
    if target_hashes is None:
        return False

    if len(source_hashes) != len(target_hashes):
        print(f"    [X] VERIFICATION FAILED: Mismatch in keyframe count: sources({len(source_hashes)}) != target({len(target_hashes)})")
        return False

    for i, (sh, th) in enumerate(zip(source_hashes, target_hashes)):
        if sh != th:
            print(f"    [X] VERIFICATION FAILED: Visual hash mismatch at keyframe {i}.")
            return False

    return True

def main():
    check_dependencies()
    args = parse_arguments()
    
    target_dir = Path(args.dir)
    if not target_dir.is_dir():
        print(f"Error: Directory '{target_dir}' does not exist.")
        sys.exit(1)

    print(f"Scanning '{target_dir}'...")
    
    # 1. Discover and apply CLI filters
    files_to_process = []
    for filepath in target_dir.iterdir():
        if filepath.is_file():
            if filepath.suffix.lower() not in args.ext:
                continue
            # Skip previously merged output files to avoid re-merging them
            if filepath.name.startswith("merged_"):
                continue
            if args.contains and args.contains not in filepath.name:
                continue
            
            size_mb = filepath.stat().st_size / (1024 * 1024)
            if args.min_size_mb > 0 and size_mb < args.min_size_mb:
                continue
            if args.max_size_mb > 0 and size_mb > args.max_size_mb:
                continue
            
            files_to_process.append(filepath)

    # Sort files alphabetically to ensure continuous concatenation sequence
    files_to_process.sort()

    # 2. Group by exact pixel size and codec (parallel ffprobe)
    groups = defaultdict(list)
    file_durations = {}
    file_sizes = {}
    print(f"Probing {len(files_to_process)} files with {args.parallel} parallel workers...")
    with ThreadPoolExecutor(max_workers=args.parallel) as executor:
        results = executor.map(probe_file, files_to_process)
        for filepath, (w, h, codec), duration, size in results:
            if w is not None and h is not None and codec is not None:
                groups[(w, h, codec)].append(filepath)
                file_durations[filepath] = duration
                file_sizes[filepath] = size

    valid_groups = {k: v for k, v in groups.items() if len(v) >= args.min_files}

    # Apply --max-duration filter: skip files longer than the threshold within each group
    if args.max_duration_s > 0:
        filtered_groups = {}
        for key, files in valid_groups.items():
            kept = [f for f in files if (file_durations.get(f, 0) or 0) <= args.max_duration_s]
            if len(kept) >= args.min_files:
                filtered_groups[key] = kept
            # Report skipped files
            skipped = [f for f in files if (file_durations.get(f, 0) or 0) > args.max_duration_s]
            for f in skipped:
                print(f"  [i] Skipping {f.name} ({format_duration(file_durations.get(f))}) - exceeds max duration {format_duration(args.max_duration_s)}")
        valid_groups = filtered_groups

    # Apply resolution/codec group filter
    if args.res or args.codec:
        wanted = {}
        for (w, h, codec), files in valid_groups.items():
            res_match = (not args.res) or f"{w}x{h}" == args.res
            codec_match = (not args.codec) or codec.lower() == args.codec.lower()
            if res_match and codec_match:
                wanted[(w, h, codec)] = files
        if not wanted:
            print(f"No group matching res={args.res!r} codec={args.codec!r}.")
            print("Available groups:")
            for (w, h, codec), files in valid_groups.items():
                print(f"  {w}x{h} ({codec}) -> {len(files)} files")
            sys.exit(1)
        valid_groups = wanted

    # 3. Sort groups by the selected metric
    sort_key_map = {
        "count": lambda x: len(x[1]),
        "duration": lambda x: sum(file_durations.get(f, 0) or 0 for f in x[1]),
        "size": lambda x: sum(file_sizes.get(f, 0) or 0 for f in x[1]),
        "count-asc": lambda x: len(x[1]),
        "duration-asc": lambda x: sum(file_durations.get(f, 0) or 0 for f in x[1]),
        "size-asc": lambda x: sum(file_sizes.get(f, 0) or 0 for f in x[1]),
    }
    reverse = not args.sort.endswith("-asc")
    sorted_groups = sorted(valid_groups.items(), key=sort_key_map[args.sort], reverse=reverse)

    if not sorted_groups:
        print("No file groups matching your filters were found.")
        sys.exit(0)

    # 4. Dry Run handling (Default Behavior)
    if args.dry_run:
        print(f"\n=== DRY RUN MODE ===")
        print(f"Found {len(sorted_groups)} group(s) capable of being concatenated.\n")
        
        for (w, h, codec), files in sorted_groups:
            group_total_dur = sum(file_durations.get(f, 0) or 0 for f in files)
            group_total_size = sum(file_sizes.get(f, 0) or 0 for f in files)
            print(f"Group: {w}x{h} ({codec}) | Will merge {len(files)} files together | Total duration: {format_duration(group_total_dur)} | Total size: {format_size(group_total_size)}:")
            for f in files:
                d = file_durations.get(f)
                sz = file_sizes.get(f)
                print(f"  - {f.name} ({format_duration(d)}, {format_size(sz)})")
            # Build the command to run this specific group (always stream copy)
            cmd_parts = [sys.executable, sys.argv[0], "--res", f"{w}x{h}", "--codec", codec, "--run"]
            if args.delete:
                cmd_parts.append("-D")
            print(f"  => {' '.join(cmd_parts)}")
            print("")
            
        print("Use --run to actually execute the concatenation.")
        return

    # 5. Execution handling
    for i, ((w, h, codec), files) in enumerate(sorted_groups, 1):
        if args.reencode:
            base_name = f"merged_{w}x{h}_{args.vcodec}_group{i}"
        else:
            base_name = f"merged_{w}x{h}_{codec}_group{i}"
        target_filename = target_dir / f"{base_name}.mp4"
        # Avoid overwriting existing files: append _2, _3, etc.
        if target_filename.exists():
            suffix = 2
            while (target_dir / f"{base_name}_{suffix}.mp4").exists():
                suffix += 1
            target_filename = target_dir / f"{base_name}_{suffix}.mp4"
        print(f"\n[{i}/{len(sorted_groups)}] Processing Group {w}x{h} ({codec}) -> {target_filename.name}")

        success = concatenate_group(files, target_filename,
                                    reencode=args.reencode, vcodec=args.vcodec, crf=args.crf, preset=args.preset,
                                    parallel=args.parallel)
        if not success:
            print(f"  [!] Skipping verification due to concatenation failure.")
            continue

        if args.strict and not args.reencode:
            print(f"  [+] Concatenation done. Running strict keyframe image verification...")
            is_verified = verify_concatenation(files, target_filename, parallel=args.parallel)
        else:
            mode = "Re-encode" if args.reencode else "Concatenation"
            print(f"  [+] {mode} done. Running duration verification (min tolerance: {args.tolerance}s)...")
            is_verified = verify_concatenation_duration(files, target_filename, min_tolerance=args.tolerance)
        
        if is_verified:
            if args.strict and not args.reencode:
                print("  [✓] VERIFICATION SUCCESS: All keyframe pixels match perfectly.")
            else:
                print("  [✓] VERIFICATION SUCCESS: Duration matches within tolerance.")
            if args.delete:
                print("  [!] Deleting verified source files...")
                for f in files:
                    f.unlink()
                    print(f"      Deleted {f.name}")
        else:
            print("  [X] VERIFICATION FAILED: Output corrupted or frames dropped.")
            print(f"      Source files will NOT be deleted. Inspect {target_filename.name}")

if __name__ == "__main__":
    main()

$ python3 merge.py /path/to/videos --res 2160x3840 --codec av1 --run

[1/1] Processing Group 2160x3840 (av1) -> merged_2160x3840_av1_group1.mp4
  [*] Normalizing timestamps for 59 files (prevents frame drops at joins)...
  [+] Concatenation done. Running duration verification (min tolerance: 0.1s)...
    -> Source total duration: 865.55s
    -> Target duration:       865.55s
    -> Difference:            0.00s (tolerance: 0.1000s)
  [✓] VERIFICATION SUCCESS: Duration matches within tolerance.