group_videos.py: Concatenate many small videos into larger files by total length

ffmpeg -y -f concat -safe 0 -i group_001.mp4.concatlist \
  -c copy -movflags +faststart group_001.mp4

clip_01.mp4  03:00
clip_02.mp4  03:00
clip_03.mp4  03:00
clip_04.mp4  03:00
clip_05.mp4  03:00
clip_06.mp4  03:00
clip_07.mp4  03:00
clip_08.mp4  03:00

Group 1: clip_01 + clip_02 + clip_03 + clip_04  (12:00)
Group 2: clip_05 + clip_06 + clip_07 + clip_08  (12:00)

Group 1: clip_01 + clip_02          (06:00)
Group 2: clip_03                     (03:00)
Group 3: clip_04 + clip_05 + clip_06 (09:00)
Group 4: clip_07 + clip_08           (06:00)

python3 group_videos.py

python3 group_videos.py --dry-run

python3 group_videos.py --sort size --parallel 4

python3 group_videos.py -s /path/to/videos -o /path/to/output
python3 group_videos.py -t 1800          # target 30 minutes
python3 group_videos.py --sort duration --reverse
python3 group_videos.py -p merged        # output prefix

#!/usr/bin/env python3
"""Group compatible video files by duration and concatenate them with ffmpeg (copy streams)."""

import argparse
import concurrent.futures
import json
import re
import subprocess
import sys
from pathlib import Path


def concat_escape(path_str):
    """Escape a path for ffmpeg's concat demuxer file list."""
    return "'" + path_str.replace("'", "'\\''") + "'"


def get_media_info(path):
    """Return (duration, compatibility_key) for a media file using ffprobe."""
    cmd = [
        "ffprobe",
        "-v", "error",
        "-show_format", "-show_streams",
        "-show_entries", "format=duration,stream=codec_type,codec_name,width,height,pix_fmt,r_frame_rate,sample_rate,channels",
        "-of", "json",
        str(path),
    ]
    result = subprocess.run(cmd, capture_output=True, text=True)
    if result.returncode != 0:
        raise RuntimeError(f"ffprobe failed for {path}: {result.stderr}")
    data = json.loads(result.stdout)
    duration_str = data.get("format", {}).get("duration")
    if duration_str is None:
        raise RuntimeError(f"Could not get duration for {path}")

    video = {}
    audio = {}
    for stream in data.get("streams", []):
        if stream.get("codec_type") == "video" and not video:
            video = stream
        elif stream.get("codec_type") == "audio" and not audio:
            audio = stream

    key = (
        video.get("codec_name"),
        video.get("width"),
        video.get("height"),
        video.get("pix_fmt"),
        video.get("r_frame_rate"),
        audio.get("codec_name"),
        audio.get("sample_rate"),
        audio.get("channels"),
    )
    return float(duration_str), key


def fmt_duration(seconds):
    """Format seconds as HH:MM:SS."""
    hours, rem = divmod(int(seconds), 3600)
    minutes, secs = divmod(rem, 60)
    return f"{hours:02d}:{minutes:02d}:{secs:02d}"


def find_video_files(source_dir, extensions, prefix):
    """Find video files in source_dir matching the given extensions."""
    files = []
    for ext in extensions:
        files.extend(source_dir.glob(f"*{ext}"))

    # Exclude output files we may have generated in a previous run.
    exclude_re = re.compile(re.escape(prefix) + r"_\d+\.mp4$", re.IGNORECASE)
    files = [f for f in files if not exclude_re.match(f.name)]

    return files


def collect_file_infos(paths, sort_mode, reverse):
    """Probe files and return them sorted by the requested mode."""
    infos = []
    for path in paths:
        duration, key = get_media_info(path)
        infos.append((path, duration, key))

    if sort_mode == "name":
        infos.sort(key=lambda x: x[0].name.lower(), reverse=reverse)
    elif sort_mode == "date":
        infos.sort(key=lambda x: x[0].stat().st_mtime, reverse=reverse)
    elif sort_mode == "size":
        infos.sort(key=lambda x: x[0].stat().st_size, reverse=reverse)
    elif sort_mode == "duration":
        infos.sort(key=lambda x: x[1], reverse=reverse)

    return infos


def group_files(file_infos, target_seconds):
    """Group files by codec compatibility and target duration."""
    groups = []
    current = []
    current_duration = 0.0
    current_key = None

    for path, duration, key in file_infos:
        # Start a new group if this file is incompatible with the current one.
        if current and key != current_key:
            groups.append((current, current_duration))
            current = []
            current_duration = 0.0

        current.append((path, duration))
        current_duration += duration
        current_key = key

        # Close the group once it reaches or exceeds the target.
        if current_duration >= target_seconds:
            groups.append((current, current_duration))
            current = []
            current_duration = 0.0
            current_key = None

    if current:
        groups.append((current, current_duration))

    return groups


def concat_group(group_files, output_path):
    """Use ffmpeg concat demuxer to concatenate a group without re-encoding."""
    list_path = output_path.with_suffix(output_path.suffix + ".concatlist")
    lines = [f"file {concat_escape(str(path))}" for path, _ in group_files]
    list_text = "\n".join(lines) + "\n"

    list_path.write_text(list_text, encoding="utf-8")
    try:
        cmd = [
            "ffmpeg",
            "-y",
            "-f", "concat",
            "-safe", "0",
            "-i", str(list_path),
            "-c", "copy",
            "-movflags", "+faststart",
            str(output_path),
        ]
        subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
    finally:
        list_path.unlink(missing_ok=True)
    return output_path


def print_group(i, group, duration, output_path=None, dry_run=False):
    print(f"\nGroup {i}: {len(group)} file(s), total {fmt_duration(duration)} ({duration:.2f}s)")
    for path, d in group:
        print(f"  {path.name} ({fmt_duration(d)})")
    if dry_run:
        print(f"  [dry-run] would create {output_path}")
    elif output_path:
        print(f"  -> {output_path}")


def main():
    parser = argparse.ArgumentParser(
        description="Group compatible video files into larger concatenated clips using ffmpeg -c copy."
    )
    parser.add_argument(
        "-s", "--source",
        default=".",
        help="Directory containing video files (default: current directory)",
    )
    parser.add_argument(
        "-o", "--output-dir",
        default=".",
        help="Directory for output group files (default: current directory)",
    )
    parser.add_argument(
        "-p", "--prefix",
        default="group",
        help="Output filename prefix (default: group)",
    )
    parser.add_argument(
        "-t", "--target",
        type=float,
        default=3600.0,
        help="Target group duration in seconds (default: 3600 = 1 hour)",
    )
    parser.add_argument(
        "-e", "--ext",
        default=".mp4",
        help="Input file extension to look for (default: .mp4)",
    )
    parser.add_argument(
        "--sort",
        choices=["name", "date", "size", "duration"],
        default="name",
        help="How to sort input files before grouping (default: name)",
    )
    parser.add_argument(
        "--reverse",
        action="store_true",
        help="Reverse the sort order",
    )
    parser.add_argument(
        "-j", "--parallel",
        type=int,
        default=1,
        help="Number of ffmpeg concat jobs to run in parallel (default: 1)",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Show groups and commands without running ffmpeg",
    )
    args = parser.parse_args()

    source_dir = Path(args.source).resolve()
    output_dir = Path(args.output_dir).resolve()
    output_dir.mkdir(parents=True, exist_ok=True)

    extensions = [args.ext]
    if args.ext.lower() != args.ext:
        extensions.append(args.ext.lower())

    paths = find_video_files(source_dir, extensions, args.prefix)
    if not paths:
        print(f"No {args.ext} files found in {source_dir}")
        sys.exit(1)

    print(f"Found {len(paths)} input file(s) in {source_dir}; probing & sorting by '{args.sort}'")
    file_infos = collect_file_infos(paths, args.sort, args.reverse)
    groups = group_files(file_infos, args.target)

    group_jobs = []
    for i, (group, duration) in enumerate(groups, start=1):
        output_name = f"{args.prefix}_{i:03d}.mp4"
        output_path = output_dir / output_name
        group_jobs.append((i, group, duration, output_path))

    if args.dry_run:
        for i, group, duration, output_path in group_jobs:
            print_group(i, group, duration, output_path, dry_run=True)
        return

    if args.parallel > 1:
        with concurrent.futures.ThreadPoolExecutor(max_workers=args.parallel) as executor:
            future_to_job = {
                executor.submit(concat_group, group, output_path): (i, group, duration, output_path)
                for i, group, duration, output_path in group_jobs
            }
            for future in concurrent.futures.as_completed(future_to_job):
                i, group, duration, output_path = future_to_job[future]
                try:
                    future.result()
                    print_group(i, group, duration, output_path)
                except subprocess.CalledProcessError as e:
                    print_group(i, group, duration)
                    print(f"  ERROR: ffmpeg failed for {output_path}")
                    if e.stderr:
                        print(e.stderr.decode(errors="replace")[-1000:])
    else:
        for i, group, duration, output_path in group_jobs:
            try:
                concat_group(group, output_path)
                print_group(i, group, duration, output_path)
            except subprocess.CalledProcessError as e:
                print_group(i, group, duration)
                print(f"  ERROR: ffmpeg failed for {output_path}")
                if e.stderr:
                    print(e.stderr.decode(errors="replace")[-1000:])


if __name__ == "__main__":
    main()