mirror of
https://github.com/linyqh/NarratoAI.git
synced 2025-12-12 11:22:51 +00:00
fa2541f95c
- 在 Windows 系统上增加显卡信息检测，避免使用不兼容的硬件加速 - 添加强制使用软件编码的选项，提升兼容性 - 增强错误处理机制，确保在硬件加速失败时能够回退到软件编码 - 更新日志信息，提供更清晰的处理反馈
708 lines
29 KiB
Python
708 lines
29 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: UTF-8 -*-
|
||
|
||
'''
|
||
@Project: NarratoAI
|
||
@File : merger_video
|
||
@Author : 小林同学
|
||
@Date : 2025/5/6 下午7:38
|
||
'''
|
||
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
from enum import Enum
|
||
from typing import List, Optional, Tuple
|
||
from loguru import logger
|
||
|
||
|
||
class VideoAspect(Enum):
|
||
"""视频宽高比枚举"""
|
||
landscape = "16:9" # 横屏 16:9
|
||
landscape_2 = "4:3"
|
||
portrait = "9:16" # 竖屏 9:16
|
||
portrait_2 = "3:4"
|
||
square = "1:1" # 方形 1:1
|
||
|
||
def to_resolution(self) -> Tuple[int, int]:
|
||
"""根据宽高比返回标准分辨率"""
|
||
if self == VideoAspect.portrait:
|
||
return 1080, 1920 # 竖屏 9:16
|
||
elif self == VideoAspect.portrait_2:
|
||
return 720, 1280 # 竖屏 4:3
|
||
elif self == VideoAspect.landscape:
|
||
return 1920, 1080 # 横屏 16:9
|
||
elif self == VideoAspect.landscape_2:
|
||
return 1280, 720 # 横屏 4:3
|
||
elif self == VideoAspect.square:
|
||
return 1080, 1080 # 方形 1:1
|
||
else:
|
||
return 1080, 1920 # 默认竖屏
|
||
|
||
|
||
def check_ffmpeg_installation() -> bool:
|
||
"""
|
||
检查ffmpeg是否已安装
|
||
|
||
Returns:
|
||
bool: 如果安装则返回True，否则返回False
|
||
"""
|
||
try:
|
||
subprocess.run(['ffmpeg', '-version'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
|
||
return True
|
||
except (subprocess.SubprocessError, FileNotFoundError):
|
||
logger.error("ffmpeg未安装或不在系统PATH中，请安装ffmpeg")
|
||
return False
|
||
|
||
|
||
def get_hardware_acceleration_option() -> Optional[str]:
|
||
"""
|
||
根据系统环境选择合适的硬件加速选项
|
||
|
||
Returns:
|
||
Optional[str]: 硬件加速参数，如果不支持则返回None
|
||
"""
|
||
try:
|
||
# 检测操作系统
|
||
is_windows = os.name == 'nt'
|
||
|
||
# 检查NVIDIA GPU支持
|
||
nvidia_check = subprocess.run(
|
||
['ffmpeg', '-hide_banner', '-hwaccels'],
|
||
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
|
||
)
|
||
output = nvidia_check.stdout.lower()
|
||
|
||
# 首先尝试获取系统信息，Windows系统使用更安全的检测方法
|
||
if is_windows:
|
||
try:
|
||
# 尝试检测显卡信息
|
||
gpu_info = subprocess.run(
|
||
['wmic', 'path', 'win32_VideoController', 'get', 'name'],
|
||
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False
|
||
)
|
||
gpu_info_output = gpu_info.stdout.lower()
|
||
|
||
# 检测是否为AMD显卡
|
||
if 'amd' in gpu_info_output or 'radeon' in gpu_info_output:
|
||
logger.info("检测到AMD显卡，为避免兼容性问题，将使用软件编码")
|
||
return None
|
||
|
||
# 检测是否为集成显卡
|
||
if 'intel' in gpu_info_output and ('hd graphics' in gpu_info_output or 'uhd graphics' in gpu_info_output):
|
||
# 在Windows上，Intel集成显卡可能不稳定，建议使用软件编码
|
||
logger.info("检测到Intel集成显卡，为避免兼容性问题，将使用软件编码")
|
||
return None
|
||
except Exception as e:
|
||
logger.warning(f"获取显卡信息失败: {str(e)}，将谨慎处理硬件加速")
|
||
|
||
# 根据ffmpeg支持的硬件加速器决定使用哪种
|
||
if 'cuda' in output and not is_windows:
|
||
# 在非Windows系统上使用CUDA
|
||
return 'cuda'
|
||
elif 'nvenc' in output and not is_windows:
|
||
# 在非Windows系统上使用NVENC
|
||
return 'nvenc'
|
||
elif 'qsv' in output and not (is_windows and ('amd' in gpu_info_output if 'gpu_info_output' in locals() else False)):
|
||
# 只有在非AMD系统上使用QSV
|
||
return 'qsv'
|
||
elif 'videotoolbox' in output: # macOS
|
||
return 'videotoolbox'
|
||
elif 'vaapi' in output and not is_windows: # Linux VA-API
|
||
return 'vaapi'
|
||
else:
|
||
logger.info("没有找到支持的硬件加速器或系统不兼容，将使用软件编码")
|
||
return None
|
||
except Exception as e:
|
||
logger.warning(f"检测硬件加速器时出错: {str(e)}，将使用软件编码")
|
||
return None
|
||
|
||
|
||
def check_video_has_audio(video_path: str) -> bool:
|
||
"""
|
||
检查视频是否包含音频流
|
||
|
||
Args:
|
||
video_path: 视频文件路径
|
||
|
||
Returns:
|
||
bool: 如果视频包含音频流则返回True，否则返回False
|
||
"""
|
||
if not os.path.exists(video_path):
|
||
logger.warning(f"视频文件不存在: {video_path}")
|
||
return False
|
||
|
||
probe_cmd = [
|
||
'ffprobe', '-v', 'error',
|
||
'-select_streams', 'a:0',
|
||
'-show_entries', 'stream=codec_type',
|
||
'-of', 'csv=p=0',
|
||
video_path
|
||
]
|
||
|
||
try:
|
||
result = subprocess.run(probe_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False)
|
||
return result.stdout.strip() == 'audio'
|
||
except Exception as e:
|
||
logger.warning(f"检测视频音频流时出错: {str(e)}")
|
||
return False
|
||
|
||
|
||
def create_ffmpeg_concat_file(video_paths: List[str], concat_file_path: str) -> str:
|
||
"""
|
||
创建ffmpeg合并所需的concat文件
|
||
|
||
Args:
|
||
video_paths: 需要合并的视频文件路径列表
|
||
concat_file_path: concat文件的输出路径
|
||
|
||
Returns:
|
||
str: concat文件的路径
|
||
"""
|
||
with open(concat_file_path, 'w', encoding='utf-8') as f:
|
||
for video_path in video_paths:
|
||
# 获取绝对路径
|
||
abs_path = os.path.abspath(video_path)
|
||
# 在Windows上将反斜杠替换为正斜杠
|
||
if os.name == 'nt': # Windows系统
|
||
abs_path = abs_path.replace('\\', '/')
|
||
else: # Unix/Mac系统
|
||
# 转义特殊字符
|
||
abs_path = abs_path.replace('\\', '\\\\').replace(':', '\\:')
|
||
|
||
# 处理路径中的单引号 (如果有)
|
||
abs_path = abs_path.replace("'", "\\'")
|
||
|
||
f.write(f"file '{abs_path}'\n")
|
||
return concat_file_path
|
||
|
||
|
||
def process_single_video(
|
||
input_path: str,
|
||
output_path: str,
|
||
target_width: int,
|
||
target_height: int,
|
||
keep_audio: bool = True,
|
||
hwaccel: Optional[str] = None
|
||
) -> str:
|
||
"""
|
||
处理单个视频：调整分辨率、帧率等
|
||
|
||
Args:
|
||
input_path: 输入视频路径
|
||
output_path: 输出视频路径
|
||
target_width: 目标宽度
|
||
target_height: 目标高度
|
||
keep_audio: 是否保留音频
|
||
hwaccel: 硬件加速选项
|
||
|
||
Returns:
|
||
str: 处理后的视频路径
|
||
"""
|
||
if not os.path.exists(input_path):
|
||
raise FileNotFoundError(f"找不到视频文件: {input_path}")
|
||
|
||
# 构建基本命令
|
||
command = ['ffmpeg', '-y']
|
||
|
||
# 安全检查：如果在Windows上，则慎用硬件加速
|
||
is_windows = os.name == 'nt'
|
||
if is_windows and hwaccel:
|
||
logger.info("在Windows系统上检测到硬件加速请求，将进行额外的兼容性检查")
|
||
try:
|
||
# 对视频进行快速探测，检测其基本信息
|
||
probe_cmd = [
|
||
'ffprobe', '-v', 'error',
|
||
'-select_streams', 'v:0',
|
||
'-show_entries', 'stream=codec_name,width,height',
|
||
'-of', 'csv=p=0',
|
||
input_path
|
||
]
|
||
result = subprocess.run(probe_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False)
|
||
|
||
# 如果探测成功，使用硬件加速；否则降级到软件编码
|
||
if result.returncode != 0:
|
||
logger.warning(f"视频探测失败，为安全起见，禁用硬件加速: {result.stderr}")
|
||
hwaccel = None
|
||
except Exception as e:
|
||
logger.warning(f"视频探测出错，禁用硬件加速: {str(e)}")
|
||
hwaccel = None
|
||
|
||
# 添加硬件加速参数（根据前面的安全检查可能已经被禁用）
|
||
if hwaccel:
|
||
try:
|
||
if hwaccel == 'cuda' or hwaccel == 'nvenc':
|
||
command.extend(['-hwaccel', 'cuda'])
|
||
elif hwaccel == 'qsv':
|
||
command.extend(['-hwaccel', 'qsv'])
|
||
elif hwaccel == 'videotoolbox':
|
||
command.extend(['-hwaccel', 'videotoolbox'])
|
||
elif hwaccel == 'vaapi':
|
||
command.extend(['-hwaccel', 'vaapi', '-vaapi_device', '/dev/dri/renderD128'])
|
||
logger.info(f"应用硬件加速: {hwaccel}")
|
||
except Exception as e:
|
||
logger.warning(f"应用硬件加速参数时出错: {str(e)}，将使用软件编码")
|
||
# 重置命令，移除可能添加了一半的硬件加速参数
|
||
command = ['ffmpeg', '-y']
|
||
|
||
# 输入文件
|
||
command.extend(['-i', input_path])
|
||
|
||
# 处理音频
|
||
if not keep_audio:
|
||
command.extend(['-an']) # 移除音频
|
||
else:
|
||
# 检查输入视频是否有音频流
|
||
has_audio = check_video_has_audio(input_path)
|
||
if has_audio:
|
||
command.extend(['-c:a', 'aac', '-b:a', '128k']) # 音频编码为AAC
|
||
else:
|
||
logger.warning(f"视频 {input_path} 没有音频流，将会忽略音频设置")
|
||
command.extend(['-an']) # 没有音频流时移除音频设置
|
||
|
||
# 视频处理参数：缩放并添加填充以保持比例
|
||
scale_filter = f"scale={target_width}:{target_height}:force_original_aspect_ratio=decrease"
|
||
pad_filter = f"pad={target_width}:{target_height}:(ow-iw)/2:(oh-ih)/2"
|
||
command.extend([
|
||
'-vf', f"{scale_filter},{pad_filter}",
|
||
'-r', '30', # 设置帧率为30fps
|
||
])
|
||
|
||
# 选择编码器 - 考虑到Windows和特定硬件的兼容性
|
||
use_software_encoder = True
|
||
|
||
if hwaccel:
|
||
if hwaccel == 'cuda' or hwaccel == 'nvenc':
|
||
try:
|
||
# 在使用NVIDIA编码器前先检查其可用性
|
||
subprocess.run(['ffmpeg', '-hide_banner', '-encoders'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
||
command.extend(['-c:v', 'h264_nvenc', '-preset', 'p4', '-profile:v', 'high'])
|
||
use_software_encoder = False
|
||
except Exception:
|
||
logger.warning("NVENC编码器不可用，将使用软件编码")
|
||
elif hwaccel == 'qsv' and not is_windows: # 在Windows上避免使用QSV
|
||
command.extend(['-c:v', 'h264_qsv', '-preset', 'medium'])
|
||
use_software_encoder = False
|
||
elif hwaccel == 'videotoolbox': # macOS
|
||
command.extend(['-c:v', 'h264_videotoolbox', '-profile:v', 'high'])
|
||
use_software_encoder = False
|
||
elif hwaccel == 'vaapi' and not is_windows: # Linux VA-API
|
||
command.extend(['-c:v', 'h264_vaapi', '-profile', '100'])
|
||
use_software_encoder = False
|
||
|
||
# 如果前面的条件未能应用硬件编码器，使用软件编码
|
||
if use_software_encoder:
|
||
logger.info("使用软件编码器(libx264)")
|
||
command.extend(['-c:v', 'libx264', '-preset', 'medium', '-profile:v', 'high'])
|
||
|
||
# 设置视频比特率和其他参数
|
||
command.extend([
|
||
'-b:v', '5M',
|
||
'-maxrate', '8M',
|
||
'-bufsize', '10M',
|
||
'-pix_fmt', 'yuv420p', # 兼容性更好的颜色格式
|
||
])
|
||
|
||
# 输出文件
|
||
command.append(output_path)
|
||
|
||
# 执行命令
|
||
try:
|
||
# logger.info(f"执行FFmpeg命令: {' '.join(command)}")
|
||
process = subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||
logger.info(f"视频处理成功: {output_path}")
|
||
return output_path
|
||
except subprocess.CalledProcessError as e:
|
||
error_msg = e.stderr.decode() if e.stderr else str(e)
|
||
logger.error(f"处理视频失败: {error_msg}")
|
||
|
||
# 如果使用硬件加速失败，尝试使用软件编码
|
||
if hwaccel:
|
||
logger.info("尝试使用软件编码作为备选方案")
|
||
try:
|
||
# 构建新的命令，使用软件编码
|
||
fallback_cmd = ['ffmpeg', '-y', '-i', input_path]
|
||
|
||
# 保持原有的音频设置
|
||
if not keep_audio:
|
||
fallback_cmd.extend(['-an'])
|
||
else:
|
||
has_audio = check_video_has_audio(input_path)
|
||
if has_audio:
|
||
fallback_cmd.extend(['-c:a', 'aac', '-b:a', '128k'])
|
||
else:
|
||
fallback_cmd.extend(['-an'])
|
||
|
||
# 保持原有的视频过滤器
|
||
fallback_cmd.extend([
|
||
'-vf', f"{scale_filter},{pad_filter}",
|
||
'-r', '30',
|
||
'-c:v', 'libx264',
|
||
'-preset', 'medium',
|
||
'-profile:v', 'high',
|
||
'-b:v', '5M',
|
||
'-maxrate', '8M',
|
||
'-bufsize', '10M',
|
||
'-pix_fmt', 'yuv420p',
|
||
output_path
|
||
])
|
||
|
||
logger.info(f"执行备选FFmpeg命令: {' '.join(fallback_cmd)}")
|
||
subprocess.run(fallback_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||
logger.info(f"使用软件编码成功处理视频: {output_path}")
|
||
return output_path
|
||
except subprocess.CalledProcessError as fallback_error:
|
||
fallback_error_msg = fallback_error.stderr.decode() if fallback_error.stderr else str(fallback_error)
|
||
logger.error(f"备选软件编码也失败: {fallback_error_msg}")
|
||
raise RuntimeError(f"无法处理视频 {input_path}: 硬件加速和软件编码都失败")
|
||
|
||
# 如果不是硬件加速导致的问题，或者备选方案也失败了，抛出原始错误
|
||
raise RuntimeError(f"处理视频失败: {error_msg}")
|
||
|
||
|
||
def combine_clip_videos(
|
||
output_video_path: str,
|
||
video_paths: List[str],
|
||
video_ost_list: List[int],
|
||
video_aspect: VideoAspect = VideoAspect.portrait,
|
||
threads: int = 4,
|
||
force_software_encoding: bool = False, # 新参数，强制使用软件编码
|
||
) -> str:
|
||
"""
|
||
合并子视频
|
||
Args:
|
||
output_video_path: 合并后的存储路径
|
||
video_paths: 子视频路径列表
|
||
video_ost_list: 原声播放列表 (0: 不保留原声, 1: 只保留原声, 2: 保留原声并保留解说)
|
||
video_aspect: 屏幕比例
|
||
threads: 线程数
|
||
force_software_encoding: 是否强制使用软件编码（忽略硬件加速检测）
|
||
|
||
Returns:
|
||
str: 合并后的视频路径
|
||
"""
|
||
# 检查ffmpeg是否安装
|
||
if not check_ffmpeg_installation():
|
||
raise RuntimeError("未找到ffmpeg，请先安装")
|
||
|
||
# 准备输出目录
|
||
output_dir = os.path.dirname(output_video_path)
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
|
||
# 获取目标分辨率
|
||
aspect = VideoAspect(video_aspect)
|
||
video_width, video_height = aspect.to_resolution()
|
||
|
||
# 检测可用的硬件加速选项
|
||
hwaccel = None if force_software_encoding else get_hardware_acceleration_option()
|
||
if hwaccel:
|
||
logger.info(f"将使用 {hwaccel} 硬件加速")
|
||
elif force_software_encoding:
|
||
logger.info("已强制使用软件编码，跳过硬件加速检测")
|
||
else:
|
||
logger.info("未检测到兼容的硬件加速，将使用软件编码")
|
||
|
||
# Windows系统上，默认使用软件编码以提高兼容性
|
||
if os.name == 'nt' and hwaccel:
|
||
logger.warning("在Windows系统上检测到硬件加速，但为了提高兼容性，建议使用软件编码")
|
||
# 不强制禁用hwaccel，而是在process_single_video中进行额外安全检查
|
||
|
||
# 重组视频路径和原声设置为一个字典列表结构
|
||
video_segments = []
|
||
|
||
# 检查视频路径和原声设置列表长度是否匹配
|
||
if len(video_paths) != len(video_ost_list):
|
||
logger.warning(f"视频路径列表({len(video_paths)})和原声设置列表({len(video_ost_list)})长度不匹配")
|
||
# 调整长度以匹配较短的列表
|
||
min_length = min(len(video_paths), len(video_ost_list))
|
||
video_paths = video_paths[:min_length]
|
||
video_ost_list = video_ost_list[:min_length]
|
||
|
||
# 创建视频处理配置字典列表
|
||
for i, (video_path, video_ost) in enumerate(zip(video_paths, video_ost_list)):
|
||
if not os.path.exists(video_path):
|
||
logger.warning(f"视频不存在，跳过: {video_path}")
|
||
continue
|
||
|
||
# 检查是否有音频流
|
||
has_audio = check_video_has_audio(video_path)
|
||
|
||
# 构建视频片段配置
|
||
segment = {
|
||
"index": i,
|
||
"path": video_path,
|
||
"ost": video_ost,
|
||
"has_audio": has_audio,
|
||
"keep_audio": video_ost > 0 and has_audio # 只有当ost>0且实际有音频时才保留
|
||
}
|
||
|
||
# 记录日志
|
||
if video_ost > 0 and not has_audio:
|
||
logger.warning(f"视频 {video_path} 设置为保留原声(ost={video_ost})，但该视频没有音频流")
|
||
|
||
video_segments.append(segment)
|
||
|
||
# 处理每个视频片段
|
||
processed_videos = []
|
||
temp_dir = os.path.join(output_dir, "temp_videos")
|
||
os.makedirs(temp_dir, exist_ok=True)
|
||
|
||
try:
|
||
# 第一阶段：处理所有视频片段到中间文件
|
||
for segment in video_segments:
|
||
# 处理单个视频，去除或保留音频
|
||
temp_output = os.path.join(temp_dir, f"processed_{segment['index']}.mp4")
|
||
try:
|
||
process_single_video(
|
||
input_path=segment['path'],
|
||
output_path=temp_output,
|
||
target_width=video_width,
|
||
target_height=video_height,
|
||
keep_audio=segment['keep_audio'],
|
||
hwaccel=hwaccel
|
||
)
|
||
processed_videos.append({
|
||
"index": segment["index"],
|
||
"path": temp_output,
|
||
"keep_audio": segment["keep_audio"]
|
||
})
|
||
logger.info(f"视频 {segment['index'] + 1}/{len(video_segments)} 处理完成")
|
||
except Exception as e:
|
||
logger.error(f"处理视频 {segment['path']} 时出错: {str(e)}")
|
||
# 如果使用硬件加速失败，尝试使用软件编码
|
||
if hwaccel and not force_software_encoding:
|
||
logger.info(f"尝试使用软件编码处理视频 {segment['path']}")
|
||
try:
|
||
process_single_video(
|
||
input_path=segment['path'],
|
||
output_path=temp_output,
|
||
target_width=video_width,
|
||
target_height=video_height,
|
||
keep_audio=segment['keep_audio'],
|
||
hwaccel=None # 使用软件编码
|
||
)
|
||
processed_videos.append({
|
||
"index": segment["index"],
|
||
"path": temp_output,
|
||
"keep_audio": segment["keep_audio"]
|
||
})
|
||
logger.info(f"使用软件编码成功处理视频 {segment['index'] + 1}/{len(video_segments)}")
|
||
except Exception as fallback_error:
|
||
logger.error(f"使用软件编码处理视频 {segment['path']} 也失败: {str(fallback_error)}")
|
||
continue
|
||
else:
|
||
continue
|
||
|
||
if not processed_videos:
|
||
raise ValueError("没有有效的视频片段可以合并")
|
||
|
||
# 按原始索引排序处理后的视频
|
||
processed_videos.sort(key=lambda x: x["index"])
|
||
|
||
# 第二阶段：分步骤合并视频 - 避免复杂的filter_complex滤镜
|
||
try:
|
||
# 1. 首先，将所有没有音频的视频或音频被禁用的视频合并到一个临时文件中
|
||
video_paths_only = [video["path"] for video in processed_videos]
|
||
video_concat_path = os.path.join(temp_dir, "video_concat.mp4")
|
||
|
||
# 创建concat文件，用于合并视频流
|
||
concat_file = os.path.join(temp_dir, "concat_list.txt")
|
||
create_ffmpeg_concat_file(video_paths_only, concat_file)
|
||
|
||
# 合并所有视频流，但不包含音频
|
||
concat_cmd = [
|
||
'ffmpeg', '-y',
|
||
'-f', 'concat',
|
||
'-safe', '0',
|
||
'-i', concat_file,
|
||
'-c:v', 'libx264',
|
||
'-preset', 'medium',
|
||
'-profile:v', 'high',
|
||
'-an', # 不包含音频
|
||
'-threads', str(threads),
|
||
video_concat_path
|
||
]
|
||
|
||
subprocess.run(concat_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||
logger.info("视频流合并完成")
|
||
|
||
# 2. 提取并合并有音频的片段
|
||
audio_segments = [video for video in processed_videos if video["keep_audio"]]
|
||
|
||
if not audio_segments:
|
||
# 如果没有音频片段，直接使用无音频的合并视频作为最终结果
|
||
shutil.copy(video_concat_path, output_video_path)
|
||
logger.info("无音频视频合并完成")
|
||
return output_video_path
|
||
|
||
# 创建音频中间文件
|
||
audio_files = []
|
||
for i, segment in enumerate(audio_segments):
|
||
# 提取音频
|
||
audio_file = os.path.join(temp_dir, f"audio_{i}.aac")
|
||
extract_audio_cmd = [
|
||
'ffmpeg', '-y',
|
||
'-i', segment["path"],
|
||
'-vn', # 不包含视频
|
||
'-c:a', 'aac',
|
||
'-b:a', '128k',
|
||
audio_file
|
||
]
|
||
subprocess.run(extract_audio_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||
audio_files.append({
|
||
"index": segment["index"],
|
||
"path": audio_file
|
||
})
|
||
logger.info(f"提取音频 {i+1}/{len(audio_segments)} 完成")
|
||
|
||
# 3. 计算每个音频片段的时间位置
|
||
audio_timings = []
|
||
current_time = 0.0
|
||
|
||
# 获取每个视频片段的时长
|
||
for i, video in enumerate(processed_videos):
|
||
duration_cmd = [
|
||
'ffprobe', '-v', 'error',
|
||
'-show_entries', 'format=duration',
|
||
'-of', 'csv=p=0',
|
||
video["path"]
|
||
]
|
||
result = subprocess.run(duration_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
||
duration = float(result.stdout.strip())
|
||
|
||
# 如果当前片段需要保留音频，记录时间位置
|
||
if video["keep_audio"]:
|
||
for audio in audio_files:
|
||
if audio["index"] == video["index"]:
|
||
audio_timings.append({
|
||
"file": audio["path"],
|
||
"start": current_time,
|
||
"index": video["index"]
|
||
})
|
||
break
|
||
|
||
current_time += duration
|
||
|
||
# 4. 创建静音音频轨道作为基础
|
||
silence_audio = os.path.join(temp_dir, "silence.aac")
|
||
create_silence_cmd = [
|
||
'ffmpeg', '-y',
|
||
'-f', 'lavfi',
|
||
'-i', f'anullsrc=r=44100:cl=stereo',
|
||
'-t', str(current_time), # 总时长
|
||
'-c:a', 'aac',
|
||
'-b:a', '128k',
|
||
silence_audio
|
||
]
|
||
subprocess.run(create_silence_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||
|
||
# 5. 创建复杂滤镜命令以混合音频
|
||
filter_script = os.path.join(temp_dir, "filter_script.txt")
|
||
with open(filter_script, 'w') as f:
|
||
f.write(f"[0:a]volume=0.0[silence];\n") # 首先静音背景轨道
|
||
|
||
# 添加每个音频文件
|
||
for i, timing in enumerate(audio_timings):
|
||
f.write(f"[{i+1}:a]adelay={int(timing['start']*1000)}|{int(timing['start']*1000)}[a{i}];\n")
|
||
|
||
# 混合所有音频
|
||
mix_str = "[silence]"
|
||
for i in range(len(audio_timings)):
|
||
mix_str += f"[a{i}]"
|
||
mix_str += f"amix=inputs={len(audio_timings)+1}:duration=longest[aout]"
|
||
f.write(mix_str)
|
||
|
||
# 6. 构建音频合并命令
|
||
audio_inputs = ['-i', silence_audio]
|
||
for timing in audio_timings:
|
||
audio_inputs.extend(['-i', timing["file"]])
|
||
|
||
mixed_audio = os.path.join(temp_dir, "mixed_audio.aac")
|
||
audio_mix_cmd = [
|
||
'ffmpeg', '-y'
|
||
] + audio_inputs + [
|
||
'-filter_complex_script', filter_script,
|
||
'-map', '[aout]',
|
||
'-c:a', 'aac',
|
||
'-b:a', '128k',
|
||
mixed_audio
|
||
]
|
||
|
||
subprocess.run(audio_mix_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||
logger.info("音频混合完成")
|
||
|
||
# 7. 将合并的视频和混合的音频组合在一起
|
||
final_cmd = [
|
||
'ffmpeg', '-y',
|
||
'-i', video_concat_path,
|
||
'-i', mixed_audio,
|
||
'-c:v', 'copy',
|
||
'-c:a', 'aac',
|
||
'-map', '0:v:0',
|
||
'-map', '1:a:0',
|
||
'-shortest',
|
||
output_video_path
|
||
]
|
||
|
||
subprocess.run(final_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||
logger.info("视频最终合并完成")
|
||
|
||
return output_video_path
|
||
|
||
except subprocess.CalledProcessError as e:
|
||
logger.error(f"合并视频过程中出错: {e.stderr.decode() if e.stderr else str(e)}")
|
||
|
||
# 尝试备用合并方法 - 最简单的无音频合并
|
||
logger.info("尝试备用合并方法 - 无音频合并")
|
||
try:
|
||
concat_file = os.path.join(temp_dir, "concat_list.txt")
|
||
video_paths_only = [video["path"] for video in processed_videos]
|
||
create_ffmpeg_concat_file(video_paths_only, concat_file)
|
||
|
||
backup_cmd = [
|
||
'ffmpeg', '-y',
|
||
'-f', 'concat',
|
||
'-safe', '0',
|
||
'-i', concat_file,
|
||
'-c:v', 'copy',
|
||
'-an', # 无音频
|
||
output_video_path
|
||
]
|
||
|
||
subprocess.run(backup_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||
logger.warning("使用备用方法（无音频）成功合并视频")
|
||
return output_video_path
|
||
except Exception as backup_error:
|
||
logger.error(f"备用合并方法也失败: {str(backup_error)}")
|
||
raise RuntimeError(f"无法合并视频: {str(backup_error)}")
|
||
|
||
except Exception as e:
|
||
logger.error(f"合并视频时出错: {str(e)}")
|
||
raise
|
||
finally:
|
||
# 清理临时文件
|
||
try:
|
||
if os.path.exists(temp_dir):
|
||
shutil.rmtree(temp_dir)
|
||
logger.info("已清理临时文件")
|
||
except Exception as e:
|
||
logger.warning(f"清理临时文件时出错: {str(e)}")
|
||
|
||
|
||
if __name__ == '__main__':
|
||
video_paths = [
|
||
'/Users/apple/Desktop/home/NarratoAI/storage/temp/clip_video/S01E02_00_14_09_440.mp4',
|
||
'/Users/apple/Desktop/home/NarratoAI/storage/temp/clip_video/S01E08_00_27_11_110.mp4',
|
||
'/Users/apple/Desktop/home/NarratoAI/storage/temp/clip_video/S01E08_00_34_44_480.mp4',
|
||
'/Users/apple/Desktop/home/NarratoAI/storage/temp/clip_video/S01E08_00_42_47_630.mp4',
|
||
'/Users/apple/Desktop/home/NarratoAI/storage/temp/clip_video/S01E09_00_29_48_160.mp4'
|
||
]
|
||
|
||
combine_clip_videos(
|
||
output_video_path="/Users/apple/Desktop/home/NarratoAI/storage/temp/merge/merged_123.mp4",
|
||
video_paths=video_paths,
|
||
video_ost_list=[1, 1, 1,1,1],
|
||
video_aspect=VideoAspect.portrait,
|
||
force_software_encoding=False # 默认不强制使用软件编码，让系统自动决定
|
||
)
|