Python使用ffmpeg concat实现录屏片段无损合并

ffprobe -v error -select_streams v:0 \
  -show_entries stream=codec_name,profile,width,height,pix_fmt,time_base,avg_frame_rate \
  -of default=noprint_wrappers=1 "录屏2026-09-11 22.34.21.mov"

h264 / Main@L5.2 / 2920x1820 / yuv420p / tb=1/12800 / AAC 48kHz stereo

file '/Users/liuxiaowei/Desktop/录屏2026-09-11 22.34.21.mov'
file '/Users/liuxiaowei/Desktop/录屏2026-09-11 23.03.50.mov'

ffmpeg -f concat -safe 0 -i concat_list.txt \
       -c copy -movflags +faststart \
       "录屏合并2026-09-11-勤云漏洞复现完整视频.mov"

ffprobe 探测 → 参数一致性判定 → 生成 concat 清单 → -c copy 拼接 → 帧数/时长守恒复核

def probe(path: str) -> dict:
    cmd = [FFPROBE, "-v", "error", "-print_format", "json",
           "-show_format", "-show_streams", path]
    data = json.loads(subprocess.run(cmd, capture_output=True, text=True).stdout)
    fmt = data.get("format", {})
    info = {"path": path,
            "size": int(fmt.get("size", 0)),
            "duration": float(fmt.get("duration", 0.0)),
            "video": None, "audio": None}
    for s in data.get("streams", []):
        if s.get("codec_type") == "video" and info["video"] is None:
            info["video"] = {
                "codec": s.get("codec_name", ""),
                "profile": s.get("profile", ""),
                "width": s.get("width", 0), "height": s.get("height", 0),
                "pix_fmt": s.get("pix_fmt", ""),
                "time_base": s.get("time_base", ""),
                "frames": int(s.get("nb_frames") or 0),
                "fps": s.get("avg_frame_rate", "0/0"),
            }
        elif s.get("codec_type") == "audio" and info["audio"] is None:
            info["audio"] = {
                "codec": s.get("codec_name", ""),
                "sample_rate": s.get("sample_rate", ""),
                "channels": s.get("channels", 0),
                "time_base": s.get("time_base", ""),
                "frames": int(s.get("nb_frames") or 0),
            }
    return info

# level（规格等级）不参与判定：-c copy 不重编码，等级差异不影响拷贝拼接
VIDEO_KEYS = ("codec", "profile", "width", "height", "pix_fmt", "time_base")
AUDIO_KEYS = ("codec", "sample_rate", "channels", "time_base")

def compatibility(infos: list) -> list:
    """返回不一致项列表，空列表表示可无损拼接。"""
    problems, ref = [], infos[0]
    for info in infos[1:]:
        for key in VIDEO_KEYS:
            if ref["video"][key] != info["video"][key]:
                problems.append(f"视频 {key}: {ref['path']}={ref['video'][key]} "
                                f"vs {info['path']}={info['video'][key]}")
        if (ref["audio"] is None) != (info["audio"] is None):
            problems.append(f"音轨缺失不一致：{ref['path']} / {info['path']}")
        elif ref["audio"] and info["audio"]:
            for key in AUDIO_KEYS:
                if ref["audio"][key] != info["audio"][key]:
                    problems.append(f"音频 {key}: ...")
    return problems

def write_concat_list(paths: list, list_file: str) -> str:
    with open(list_file, "w", encoding="utf-8") as f:
        for p in paths:
            safe = os.path.abspath(p).replace("'", "'\\''")   # ' -> '\''
            f.write(f"file '{safe}'\n")
    return list_file

def concat_copy(paths: list, output: str) -> None:
    workdir = tempfile.mkdtemp(prefix="merge_recordings_")
    list_file = write_concat_list(paths, os.path.join(workdir, "concat_list.txt"))
    cmd = [FFMPEG, "-hide_banner", "-loglevel", "warning", "-nostdin", "-y",
           "-f", "concat", "-safe", "0", "-i", list_file, "-c", "copy"]
    if output.lower().endswith((".mp4", ".mov", ".m4v")):
        cmd += ["-movflags", "+faststart"]
    cmd += [output]
    subprocess.run(cmd, check=True)

ffprobe -v error -select_streams v -count_frames \
  -show_entries stream=nb_read_frames -print_format json out.mov

$ python3 merge_recordings.py -o merged_demo.mov "录屏2026-09-11 22.34.21.mov" "录屏2026-09-11 23.03.50.mov"
== 片段参数 ==
  录屏2026-09-11 22.34.21.mov: h264 Main L5.2 1280x800 yuv420p tb=1/12800 | aac 48000Hz stereo
  录屏2026-09-11 23.03.50.mov: h264 Main L5.2 1280x800 yuv420p tb=1/12800 | aac 48000Hz stereo
[+] 参数完全一致，走无损拼接（-c copy）
[*] ffmpeg -hide_banner -loglevel warning -nostdin -y -f concat -safe 0 \
    -i /tmp/merge_recordings_fcp46ksk/concat_list.txt -c copy -movflags +faststart merged_demo.mov
序号  文件                                      时长(s)      视频帧          大小
1   录屏2026-09-11 22.34.21.mov                    3.00          75     94.62 KiB
2   录屏2026-09-11 23.03.50.mov                    2.00          50     66.13 KiB
[+] 输出：merged_demo.mov
    时长 5.02s（源合计 5.00s，差 0.02s）
    视频帧 125（源合计 125，一致 ✔）
    音频帧 237（源合计 235，偏差 2 帧（AAC priming/padding，属正常））
    大小 159.43 KiB

[!] 检测到参数不一致，无法无损拼接：
    - 视频 profile: ...22.34.21.mov=Main vs tmp_other.mov=High
    - 视频 width: ...22.34.21.mov=1280 vs tmp_other.mov=640
    - 视频 height: ...22.34.21.mov=800 vs tmp_other.mov=480
    - 视频 time_base: ...22.34.21.mov=1/12800 vs tmp_other.mov=1/15360
    - 音轨缺失不一致：...22.34.21.mov / tmp_other.mov
    可加 --force 触发重编码拼接（画质有一次编码损失）

python3 merge_recordings.py -o out.mov a.mov b.mov                  # 按命令行顺序
python3 merge_recordings.py -o out.mov --sort *.mov                 # 按文件名（时间）排序
python3 merge_recordings.py -o out.mov --check a.mov b.mov          # 只体检，不合并
python3 merge_recordings.py -o out.mp4 --force a.mov b.mov          # 参数不一致时重编码

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
merge_recordings.py —— 录屏片段无损合并工具（macOS / Linux / Windows 通用）

原理：调用 ffmpeg 的 concat demuxer + `-c copy`，在容器层把多段录屏首尾串联，
      视频与音频流均不重新编码，画质、音轨、时间戳零损失。

用法：
    # 1) 按命令行给定顺序合并（默认无损）
    python3 merge_recordings.py -o 合并结果.mov 片段A.mov 片段B.mov

    # 2) 让脚本按文件名（含时间）自动排序，等价于“按时间顺序拼接”
    python3 merge_recordings.py -o 合并结果.mov --sort 片段1.mov 片段2.mov 片段3.mov

    # 3) 各片段编码参数不一致时（无法无损拼接），强制重编码输出
    python3 merge_recordings.py -o 合并结果.mp4 --force 片段A.mov 片段B.mov

    # 4) 只做参数体检，不合并
    python3 merge_recordings.py -o /dev/null --check 片段A.mov 片段B.mov

依赖：ffmpeg / ffprobe（macOS: brew install ffmpeg）
作者：White_hat@9808    许可：MIT
"""

import argparse
import json
import os
import shutil
import subprocess
import sys
import tempfile

# ---------- 环境定位：兼容 GUI 应用启动的非交互 shell（PATH 里没有 /opt/homebrew/bin） ----------
EXTRA_BIN_DIRS = ["/opt/homebrew/bin", "/usr/local/bin", "/usr/bin"]


def find_binary(name: str) -> str:
    """定位 ffmpeg / ffprobe，找不到直接退出并给出安装提示。"""
    path = shutil.which(name)
    if path:
        return path
    for d in EXTRA_BIN_DIRS:
        cand = os.path.join(d, name)
        if os.path.isfile(cand) and os.access(cand, os.X_OK):
            return cand
    sys.exit(f"[!] 未找到 {name}，请先安装：macOS 执行 `brew install ffmpeg`")


FFMPEG = find_binary("ffmpeg")
FFPROBE = find_binary("ffprobe")


# ---------- 探测：把每个片段的容器信息、视频流、音频流结构化成 dict ----------
def probe(path: str) -> dict:
    cmd = [FFPROBE, "-v", "error", "-print_format", "json",
           "-show_format", "-show_streams", path]
    out = subprocess.run(cmd, capture_output=True, text=True)
    if out.returncode != 0:
        sys.exit(f"[!] ffprobe 读取失败：{path}\n{out.stderr}")

    data = json.loads(out.stdout)
    fmt = data.get("format", {})
    info = {
        "path": path,
        "size": int(fmt.get("size", 0)),
        "duration": float(fmt.get("duration", 0.0)),
        "format_name": fmt.get("format_name", ""),
        "video": None,
        "audio": None,
    }
    for s in data.get("streams", []):
        if s.get("codec_type") == "video" and info["video"] is None:
            info["video"] = {
                "codec": s.get("codec_name", ""),
                "profile": s.get("profile", ""),
                "level": s.get("level", ""),
                "width": s.get("width", 0),
                "height": s.get("height", 0),
                "pix_fmt": s.get("pix_fmt", ""),
                "time_base": s.get("time_base", ""),
                "tag": s.get("codec_tag_string", ""),
                "frames": int(s.get("nb_frames") or 0),   # 部分容器不写 nb_frames，需靠解码复核
                "fps": s.get("avg_frame_rate", "0/0"),
            }
        elif s.get("codec_type") == "audio" and info["audio"] is None:
            info["audio"] = {
                "codec": s.get("codec_name", ""),
                "profile": s.get("profile", ""),
                "sample_rate": s.get("sample_rate", ""),
                "channels": s.get("channels", 0),
                "channel_layout": s.get("channel_layout", ""),
                "time_base": s.get("time_base", ""),
                "tag": s.get("codec_tag_string", ""),
                "frames": int(s.get("nb_frames") or 0),
            }
    return info


# ---------- 兼容性判定：这些字段全部一致，才允许 -c copy 直接拼接 ----------
# 注意 level（规格等级）不参与判定：-c copy 不重编码，等级差异不影响拷贝拼接
VIDEO_KEYS = ("codec", "profile", "width", "height", "pix_fmt", "time_base")
AUDIO_KEYS = ("codec", "sample_rate", "channels", "time_base")


def compatibility(infos: list) -> list:
    """返回不一致项列表，空列表表示可无损拼接。"""
    problems = []
    ref = infos[0]
    for info in infos[1:]:
        for key in VIDEO_KEYS:
            a, b = ref["video"][key], info["video"][key]
            if a != b:
                problems.append(f"视频 {key}: {ref['path']}={a} vs {info['path']}={b}")
        if (ref["audio"] is None) != (info["audio"] is None):
            problems.append(f"音轨缺失不一致：{ref['path']} / {info['path']}")
        elif ref["audio"] and info["audio"]:
            for key in AUDIO_KEYS:
                a, b = ref["audio"][key], info["audio"][key]
                if a != b:
                    problems.append(f"音频 {key}: {ref['path']}={a} vs {info['path']}={b}")
    return problems


# ---------- concat 清单：路径含空格/中文/单引号都要能安全转义 ----------
def write_concat_list(paths: list, list_file: str) -> str:
    with open(list_file, "w", encoding="utf-8") as f:
        for p in paths:
            safe = os.path.abspath(p).replace("'", "'\\''")  # 单引号 -> '\'' 标准转义
            f.write(f"file '{safe}'\n")
    return list_file


# ---------- 无损拼接 ----------
def concat_copy(paths: list, output: str) -> None:
    workdir = tempfile.mkdtemp(prefix="merge_recordings_")
    list_file = write_concat_list(paths, os.path.join(workdir, "concat_list.txt"))
    cmd = [FFMPEG, "-hide_banner", "-loglevel", "warning", "-nostdin", "-y",
           "-f", "concat", "-safe", "0", "-i", list_file,
           "-c", "copy"]
    if output.lower().endswith((".mp4", ".mov", ".m4v")):
        cmd += ["-movflags", "+faststart"]  # moov 前置，网盘/网页播放可边下边播
    cmd += [output]
    print("[*] " + " ".join(cmd))
    subprocess.run(cmd, check=True)


# ---------- 参数不一致时的兜底：重编码拼接 ----------
def concat_reencode(paths: list, output: str) -> None:
    inputs = []
    for p in paths:
        inputs += ["-i", p]
    has_audio = all(bool(probe(p)["audio"]) for p in paths)
    n = len(paths)
    if has_audio:
        streams = "".join(f"[{i}:v][{i}:a]" for i in range(n))
        fc = f"{streams}concat=n={n}:v=1:a=1[v][a]"
        maps = ["-map", "[v]", "-map", "[a]", "-c:a", "aac", "-b:a", "192k"]
    else:
        streams = "".join(f"[{i}:v]" for i in range(n))
        fc = f"{streams}concat=n={n}:v=1:a=0[v]"
        maps = ["-map", "[v]", "-an"]
    cmd = [FFMPEG, "-hide_banner", "-loglevel", "warning", "-nostdin", "-y", *inputs,
           "-filter_complex", fc, *maps,
           "-c:v", "libx264", "-crf", "20", "-preset", "medium", "-pix_fmt", "yuv420p"]
    if output.lower().endswith((".mp4", ".mov", ".m4v")):
        cmd += ["-movflags", "+faststart"]
    cmd += [output]
    print("[*] " + " ".join(cmd))
    subprocess.run(cmd, check=True)


# ---------- 复核：帧数与时长是否守恒 ----------
def count_frames(path: str, kind: str) -> int:
    """用解码器真实计数（nb_frames 在部分容器里为空或不准）。"""
    cmd = [FFPROBE, "-v", "error", "-select_streams", kind[0],
           "-count_frames", "-show_entries", f"stream=nb_read_frames",
           "-print_format", "json", path]
    out = subprocess.run(cmd, capture_output=True, text=True)
    try:
        streams = json.loads(out.stdout).get("streams", [])
        return int(streams[0].get("nb_read_frames") or 0) if streams else 0
    except Exception:
        return 0


def human(n: int) -> str:
    for unit in ("B", "KiB", "MiB", "GiB"):
        if n < 1024 or unit == "GiB":
            return f"{n:.2f} {unit}" if unit != "B" else f"{n} B"
        n /= 1024.0


def level_str(v: dict) -> str:
    """level=52 -> L5.2，便于和 QuickTime/FFmpeg 显示对上。"""
    lv = str(v.get("level") or "")
    if len(lv) == 2 and lv.isdigit():
        return f"L{lv[0]}.{lv[1]}"
    return f"L{lv}" if lv else ""


def report(infos: list, output: str) -> None:
    print("\n{:<4}{:<44}{:>12}{:>12}{:>12}".format("序号", "文件", "时长(s)", "视频帧", "大小"))
    for i, info in enumerate(infos, 1):
        name = os.path.basename(info["path"])
        print("{:<4}{:<44}{:>12.2f}{:>12}{:>12}".format(
            i, name[:42], info["duration"], info["video"]["frames"] or "-", human(info["size"])))

    out = probe(output)
    out_v = count_frames(output, "video")
    out_a = count_frames(output, "audio") if out["audio"] else 0
    src_v = sum(count_frames(i["path"], "video") for i in infos)
    src_a = sum(count_frames(i["path"], "audio") for i in infos) if out["audio"] else 0
    src_d = sum(i["duration"] for i in infos)

    print(f"\n[+] 输出：{output}")
    print(f"    时长 {out['duration']:.2f}s（源合计 {src_d:.2f}s，差 {abs(out['duration']-src_d):.2f}s）")
    print(f"    视频帧 {out_v}（源合计 {src_v}，{'一致 ✔' if out_v == src_v else '不一致 ✘'}）")
    if out["audio"]:
        diff = abs(out_a - src_a)
        if diff == 0:
            note = "一致 ✔"
        elif src_a and diff / src_a <= 0.02:
            # AAC 每段自带 priming/padding 帧，-c copy 拼接后允许 ≤2% 偏差，听感无影响
            note = f"偏差 {diff} 帧（AAC priming/padding，属正常）"
        else:
            note = f"不一致 ✘（差 {diff} 帧）"
        print(f"    音频帧 {out_a}（源合计 {src_a}，{note}）")
    print(f"    大小 {human(out['size'])}")


def main() -> None:
    ap = argparse.ArgumentParser(description="录屏片段无损合并（ffmpeg concat + stream copy）")
    ap.add_argument("inputs", nargs="+", help="待合并的视频文件（按命令行顺序拼接）")
    ap.add_argument("-o", "--output", required=True, help="输出文件路径")
    ap.add_argument("--sort", action="store_true", help="按文件名排序后再拼接（录制时间命名时等价于时间顺序）")
    ap.add_argument("--force", action="store_true", help="参数不一致时强制重编码拼接")
    ap.add_argument("--check", action="store_true", help="只做参数体检，不执行合并")
    args = ap.parse_args()

    inputs = sorted(args.inputs) if args.sort else args.inputs
    # 防止通配符把输出文件自身也卷进输入（重复合并的常见误操作）
    out_abs = os.path.abspath(args.output)
    inputs = [p for p in inputs if os.path.abspath(p) != out_abs]
    for p in inputs:
        if not os.path.isfile(p):
            sys.exit(f"[!] 文件不存在：{p}")
    if len(inputs) < 2:
        sys.exit("[!] 至少需要两个输入文件")

    infos = [probe(p) for p in inputs]

    print("== 片段参数 ==")
    for info in infos:
        v, a = info["video"], info["audio"]
        print(f"  {os.path.basename(info['path'])}: "
              f"{v['codec']} {v['profile']} {level_str(v)} {v['width']}x{v['height']} {v['pix_fmt']} "
              f"tb={v['time_base']} | "
              + (f"{a['codec']} {a['sample_rate']}Hz {a['channel_layout']}" if a else "无音轨"))

    problems = compatibility(infos)
    if problems:
        print("\n[!] 检测到参数不一致，无法无损拼接：")
        for p in problems:
            print("    - " + p)
        if not args.force:
            sys.exit("\n    可加 --force 触发重编码拼接（画质有一次编码损失）")
        print("\n[*] --force 已开启，改用 filter_complex concat 重编码拼接")
    else:
        print("\n[+] 参数完全一致，走无损拼接（-c copy）")

    if args.check:
        return

    if problems:
        concat_reencode(inputs, args.output)
    else:
        concat_copy(inputs, args.output)

    report(infos, args.output)


if __name__ == "__main__":
    main()