一个 4 小时的视频单机转码太慢：切片并行编码与无缝拼接

原方案：
  [============ 4 小时视频 ============] → 1 个进程（8 线程）
                                          耗时 6 小时

新方案：
  [段1][段2][段3][段4][段5][段6][段7][段8]
   8 个进程，每个 1 线程，同时跑
                                          耗时 ~1 小时

ffprobe -v error -select_streams v:0 -skip_frame nokey \
  -show_entries frame=pts_time -of csv=p=0 input.mp4 > keyframes.txt

head -5 keyframes.txt
# 0.000000
# 4.000000
# 8.000000
# 12.000000

import bisect


def snap_to_keyframe(target: float, keyframes: list) -> float:
    """把目标时间点吸附到最近的关键帧（不早于 target 的那个）。"""
    i = bisect.bisect_left(keyframes, target)
    if i >= len(keyframes):
        return keyframes[-1]
    return keyframes[i]

#!/bin/bash
# chunk_encode.sh input.mp4 chunks output.mp4
set -euo pipefail

IN="$1"
N="${2:-8}"
OUT="$3"
TMP=$(mktemp -d)
trap 'rm -rf "$TMP"' EXIT

# 1. 拿总时长
DUR=$(ffprobe -v error -show_entries format=duration -of csv=p=0 "$IN")

# 2. 拿关键帧列表
ffprobe -v error -select_streams v:0 -skip_frame nokey \
  -show_entries frame=pts_time -of csv=p=0 "$IN" > "$TMP/keys.txt"

# 3. 生成切分点（这里用 python 做吸附）
python3 - "$DUR" "$N" "$TMP/keys.txt" > "$TMP/cuts.txt" <<'PY'
import sys, bisect
dur, n = float(sys.argv[1]), int(sys.argv[2])
keys = [float(x) for x in open(sys.argv[3]) if x.strip()]
keys.sort()
step = dur / n
cuts = [0.0]
for i in range(1, n):
    t = step * i
    idx = bisect.bisect_left(keys, t)
    if idx < len(keys):
        cuts.append(keys[idx])
    else:
        cuts.append(dur)
cuts.append(dur)
print('\n'.join(f'{a:.3f} {b:.3f}' for a, b in zip(cuts, cuts[1:])))
PY

# 4. 并行编码每一段（每段 1 线程，关键帧对齐）
i=0
while read -r start end; do
    dur=$(awk "BEGIN{print $end-$start}")
    [ "$(awk "BEGIN{print ($dur <= 0.1) ? 1 : 0}")" -eq 1 ] && { i=$((i+1)); continue; }
    printf '%03d %s %s\n' "$i" "$start" "$dur"
    i=$((i+1))
done < "$TMP/cuts.txt" > "$TMP/jobs.txt"

cat "$TMP/jobs.txt" | xargs -P "$N" -L 1 bash -c '
    set -e
    idx=$0; start=$1; dur=$2
    ffmpeg -v error -y \
      -ss "$start" -i "'"$IN"'" \
      -t "$dur" \
      -vf "scale=1920:-2" \
      -c:v libx265 -crf 26 -preset slow -x265-params "log-level=error" \
      -an -threads 1 \
      "'"$TMP"'/seg_${idx}.mp4"
'

# 5. 拼接（按序号）
ls "$TMP"/seg_*.mp4 | sort | sed "s|^|file '|; s|$|'|" > "$TMP/list.txt"
ffmpeg -v error -y -f concat -safe 0 -i "$TMP/list.txt" -c copy "$TMP/video.mp4"

# 6. 音频单独处理并合并
ffmpeg -v error -y -i "$IN" -vn -c:a aac -b:a 128k "$TMP/audio.m4a"
ffmpeg -v error -y -i "$TMP/video.mp4" -i "$TMP/audio.m4a" \
  -c copy -shortest -movflags +faststart "$OUT"

echo "done: $OUT"

视频：切 8 段并行编码 → 拼接 → video.mp4（无音轨）
音频：整段单独编码 → audio.m4a
合并：video.mp4 + audio.m4a → 最终文件

ffmpeg -i video.mp4 -i audio.m4a -c copy -shortest -movflags +faststart out.mp4

ffmpeg -i video.mp4 -i a1.m4a -i a2.m4a \
  -map 0:v -map 1:a -map 2:a \
  -c copy -metadata:s:a:0 language=chi -metadata:s:a:1 language=eng out.mp4

Non-monotonous DTS in output stream

原来：1 个文件，N 个关键帧
切 8 段：8 个文件，N + 7 个关键帧（多了 7 个 IDR）

import os
import subprocess
import tempfile
import shutil
from concurrent.futures import ProcessPoolExecutor


def encode_segment(args):
    idx, src, start, dur, out_dir, vfilter, vcodec_opts = args
    out = os.path.join(out_dir, f'seg_{idx:04d}.mp4')
    cmd = ['ffmpeg', '-v', 'error', '-y',
           '-ss', f'{start:.3f}', '-i', src, '-t', f'{dur:.3f}']
    if vfilter:
        cmd += ['-vf', vfilter]
    cmd += ['-c:v', 'libx265'] + vcodec_opts
    cmd += ['-an', '-threads', '1', out]
    subprocess.run(cmd, check=True)
    return out


def chunked_encode(src, out, segments=8, vfilter='scale=1920:-2',
                   vcodec_opts=('-crf', '26', '-preset', 'slow')):
    tmp = tempfile.mkdtemp(prefix='chunk_')
    try:
        dur = float(subprocess.run(
            ['ffprobe', '-v', 'error', '-show_entries', 'format=duration',
             '-of', 'csv=p=0', src], capture_output=True, text=True, check=True).stdout)
        keys = get_keyframes(src)
        cuts = build_cuts(dur, segments, keys)

        jobs = [(i, src, a, b - a, tmp, vfilter, list(vcodec_opts))
                for i, (a, b) in enumerate(cuts)]

        with ProcessPoolExecutor(max_workers=segments) as ex:
            segs = list(ex.map(encode_segment, jobs))

        # 拼接
        list_file = os.path.join(tmp, 'list.txt')
        with open(list_file, 'w') as f:
            for s in sorted(segs):
                f.write(f"file '{s}'\n")
        video_only = os.path.join(tmp, 'video.mp4')
        subprocess.run(['ffmpeg', '-v', 'error', '-y', '-f', 'concat', '-safe', '0',
                        '-i', list_file, '-c', 'copy', video_only], check=True)

        # 音频
        audio = os.path.join(tmp, 'audio.m4a')
        subprocess.run(['ffmpeg', '-v', 'error', '-y', '-i', src, '-vn',
                        '-c:a', 'aac', '-b:a', '128k', audio], check=True)

        # 合并
        subprocess.run(['ffmpeg', '-v', 'error', '-y', '-i', video_only, '-i', audio,
                        '-c', 'copy', '-shortest', '-movflags', '+faststart', out],
                       check=True)
    finally:
        shutil.rmtree(tmp, ignore_errors=True)
    return out

8 台机器，每台领一个大文件
  → 每台内部把文件切成 8 段并行编码
  → 段内 1 线程
  → 总并行度 = 8 机 × 8 段 = 64