179 lines
7.7 KiB
Python
179 lines
7.7 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
'''
|
||
# @Time : 2025/9/25 15:30
|
||
# @Author : reenrr
|
||
# @File : merge_video.py
|
||
'''
|
||
# !/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
'''
|
||
# @Time : 2025/09/25
|
||
# @Author : reenrr
|
||
# @File : video_merger.py
|
||
# 功能描述：基于 FFmpeg 合并多个视频文件（支持跨格式、进度显示）
|
||
# 依赖：需先安装 FFmpeg 并配置环境变量
|
||
'''
|
||
import os
|
||
import subprocess
|
||
import platform
|
||
from typing import List
|
||
|
||
|
||
class VideoMerger:
|
||
def __init__(self):
|
||
# 检查 FFmpeg 是否安装（通过调用 ffmpeg -version 验证）
|
||
self._check_ffmpeg_installed()
|
||
|
||
def _check_ffmpeg_installed(self) -> None:
|
||
"""检查 FFmpeg 是否已安装并配置环境变量"""
|
||
try:
|
||
# 调用 FFmpeg 版本命令，无异常则说明安装成功
|
||
subprocess.run(
|
||
["ffmpeg", "-version"],
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
check=True
|
||
)
|
||
except FileNotFoundError:
|
||
raise EnvironmentError(
|
||
"未找到 FFmpeg！请先安装 FFmpeg 并配置系统环境变量。\n"
|
||
"下载地址：https://ffmpeg.org/download.html\n"
|
||
"Windows 配置：将 FFmpeg 的 bin 目录添加到 PATH 环境变量；\n"
|
||
"macOS：使用 brew install ffmpeg；\n"
|
||
"Linux：使用 sudo apt install ffmpeg（Ubuntu）。"
|
||
)
|
||
except subprocess.CalledProcessError:
|
||
raise RuntimeError("FFmpeg 安装异常，请重新安装。")
|
||
|
||
def _create_video_list_file(self, video_paths: List[str], list_file_path: str = "video_list.txt") -> str:
|
||
"""
|
||
创建 FFmpeg 所需的视频列表文件（用于合并多个视频）
|
||
FFmpeg 合并需通过文本文件指定视频路径，格式为：file '视频路径'
|
||
"""
|
||
# 检查所有视频文件是否存在
|
||
for idx, path in enumerate(video_paths):
|
||
if not os.path.exists(path):
|
||
raise FileNotFoundError(f"第 {idx + 1} 个视频文件不存在：{path}")
|
||
# 检查是否为文件（而非目录）
|
||
if not os.path.isfile(path):
|
||
raise IsADirectoryError(f"第 {idx + 1} 个路径不是文件：{path}")
|
||
|
||
# 写入视频列表（处理路径中的特殊字符，兼容 Windows 反斜杠）
|
||
with open(list_file_path, "w", encoding="utf-8") as f:
|
||
for path in video_paths:
|
||
# 统一路径分隔符为正斜杠（FFmpeg 兼容正斜杠）
|
||
normalized_path = path.replace("\\", "/")
|
||
# 写入格式：file '路径'（单引号避免路径含空格/特殊字符）
|
||
f.write(f"file '{normalized_path}'\n")
|
||
|
||
return list_file_path
|
||
|
||
def merge_videos(self,
|
||
video_paths: List[str],
|
||
output_path: str = "merged_output.mp4",
|
||
overwrite: bool = False) -> None:
|
||
"""
|
||
合并多个视频文件
|
||
:param video_paths: 待合并的视频路径列表（顺序即合并顺序）
|
||
:param output_path: 输出视频路径（默认当前目录 merged_output.mp4）
|
||
:param overwrite: 是否覆盖已存在的输出文件（默认不覆盖）
|
||
"""
|
||
# 1. 基础参数校验
|
||
if len(video_paths) < 2:
|
||
raise ValueError("至少需要 2 个视频文件才能合并！")
|
||
|
||
# 2. 处理输出文件（若已存在且不允许覆盖，直接退出）
|
||
if os.path.exists(output_path) and not overwrite:
|
||
raise FileExistsError(f"输出文件已存在：{output_path}，若需覆盖请设置 overwrite=True。")
|
||
|
||
# 3. 创建 FFmpeg 视频列表文件
|
||
list_file = self._create_video_list_file(video_paths)
|
||
print(f"已创建视频列表文件：{list_file}")
|
||
|
||
# 4. 构造 FFmpeg 合并命令
|
||
# -f concat：使用 concat 协议（合并多个文件）
|
||
# -safe 0：允许处理任意路径的文件（避免 FFmpeg 限制相对路径）
|
||
# -i {list_file}：输入视频列表文件
|
||
# -c copy：直接复制音视频流（不重新编码，速度快；若格式不兼容会自动 fallback 编码）
|
||
# -y：强制覆盖输出文件（仅 overwrite=True 时启用）
|
||
ffmpeg_cmd = [
|
||
"ffmpeg",
|
||
"-f", "concat",
|
||
"-safe", "0",
|
||
"-i", list_file,
|
||
"-c", "copy", # 核心参数：复制流（快速合并），若格式不兼容可删除此句（自动编码）
|
||
"-hide_banner", # 隐藏 FFmpeg 版本横幅（简化输出）
|
||
"-loglevel", "info" # 显示进度和关键日志（可改为 warning 减少输出）
|
||
]
|
||
# 若允许覆盖，添加 -y 参数
|
||
if overwrite:
|
||
ffmpeg_cmd.append("-y")
|
||
# 添加输出路径
|
||
ffmpeg_cmd.append(output_path)
|
||
|
||
try:
|
||
print(f"\n开始合并视频（共 {len(video_paths)} 个）：")
|
||
for idx, path in enumerate(video_paths, 1):
|
||
print(f" {idx}. {os.path.basename(path)}")
|
||
print(f"\n输出路径：{os.path.abspath(output_path)}")
|
||
print("=" * 50)
|
||
|
||
# 5. 执行 FFmpeg 命令（实时打印进度）
|
||
process = subprocess.Popen(
|
||
ffmpeg_cmd,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.STDOUT, # 合并 stdout 和 stderr，统一捕获进度
|
||
text=True,
|
||
bufsize=1 # 行缓冲，实时输出日志
|
||
)
|
||
|
||
# 实时打印 FFmpeg 输出（显示合并进度）
|
||
for line in process.stdout:
|
||
print(line.strip(), end="\r" if "time=" in line else "\n") # 进度行覆盖显示，其他行换行
|
||
|
||
# 等待命令执行完成，获取返回码
|
||
process.wait()
|
||
if process.returncode == 0:
|
||
print("\n" + "=" * 50)
|
||
print(f"视频合并成功！输出文件：{os.path.abspath(output_path)}")
|
||
else:
|
||
raise RuntimeError(f"FFmpeg 执行失败（返回码：{process.returncode}），请检查日志排查问题。")
|
||
|
||
finally:
|
||
# 6. 清理临时文件（视频列表文件）
|
||
if os.path.exists(list_file):
|
||
os.remove(list_file)
|
||
print(f"\n已清理临时列表文件：{list_file}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# -------------------------- 配置参数（根据需求修改） --------------------------
|
||
# 1. 待合并的视频路径列表（顺序即合并顺序，支持绝对路径/相对路径）
|
||
# 示例1：相对路径（视频文件与脚本在同一目录）
|
||
# VIDEO_PATHS = [
|
||
# "video1.mp4",
|
||
# "video2.mkv",
|
||
# "video3.avi"
|
||
# ]
|
||
# 示例2：绝对路径（Windows 需用双反斜杠或正斜杠）
|
||
VIDEO_PATHS = [
|
||
r"C:\Project\zjsh\data_collection\data_collection\camera01_videos\video_20250925_062802_part1.mp4", # Windows 绝对路径（r 表示原始字符串，避免转义）
|
||
r"C:\Project\zjsh\data_collection\data_collection\camera01_videos\video_20250925_062905_part2.mp4"
|
||
]
|
||
# 2. 输出视频路径（默认当前目录，可自定义）
|
||
OUTPUT_PATH = "merged_video.mp4"
|
||
# 3. 是否覆盖已存在的输出文件（True=覆盖，False=不覆盖）
|
||
OVERWRITE = True
|
||
|
||
# -------------------------- 执行合并 --------------------------
|
||
try:
|
||
merger = VideoMerger()
|
||
merger.merge_videos(
|
||
video_paths=VIDEO_PATHS,
|
||
output_path=OUTPUT_PATH,
|
||
overwrite=OVERWRITE
|
||
)
|
||
except Exception as e:
|
||
print(f"\n合并失败：{str(e)}") |