github/NarratoAI
1
0
Fork 0
mirror of https://github.com/linyqh/NarratoAI.git synced 2025-12-11 10:32:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(video): 优化视频裁剪和合并功能，增强硬件加速兼容性

Browse Source 
更新编码器配置，优先使用纯NVENC编码器以避免滤镜链错误，确保视频裁剪和合并过程中的兼容性和性能。改进错误处理机制，智能分析FFmpeg错误类型并选择合适的回退方案，提升整体稳定性和用户体验。
This commit is contained in:
linyqh 2025-07-07 02:21:02 +08:00
parent 13ed28626f
commit c252b2a782
5 changed files with 597 additions and 250 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

386
app/services/clip_video.py
View File

@ -87,7 +87,7 @@ def check_hardware_acceleration() -> Optional[str]:
def get_safe_encoder_config(hwaccel_type: Optional[str] = None) -> Dict[str, str]:
"""
获取安全的编码器配置，针对Windows平台优化
获取安全的编码器配置，基于ffmpeg_demo.py成功方案优化
Args:
hwaccel_type: 硬件加速类型
@ -95,28 +95,48 @@ def get_safe_encoder_config(hwaccel_type: Optional[str] = None) -> Dict[str, str
Returns:
Dict[str, str]: 编码器配置字典
"""
# 基础配置 - 参考ffmpeg_demo.py的成功方案
config = {
"video_codec": "libx264",
"audio_codec": "aac",
"pixel_format": "yuv420p",
"preset": "fast",
"crf": "23"
"preset": "medium",
"quality_param": "crf", # 质量参数类型
"quality_value": "23" # 质量值
}
# 根据硬件加速类型调整配置
if hwaccel_type == "cuda":
# 根据硬件加速类型调整配置（简化版本）
if hwaccel_type in ["nvenc_pure", "nvenc_software", "cuda_careful", "nvenc", "cuda", "cuda_decode"]:
# NVIDIA硬件加速 - 使用ffmpeg_demo.py中验证有效的参数
config["video_codec"] = "h264_nvenc"
config["preset"] = "fast"
config["preset"] = "medium"
config["quality_param"] = "cq" # CQ质量控制，而不是CRF
config["quality_value"] = "23"
config["pixel_format"] = "yuv420p"
elif hwaccel_type == "amf":
# AMD AMF编码器
config["video_codec"] = "h264_amf"
config["preset"] = "balanced"
config["quality_param"] = "qp_i"
config["quality_value"] = "23"
elif hwaccel_type == "qsv":
# Intel QSV编码器
config["video_codec"] = "h264_qsv"
config["preset"] = "fast"
elif hwaccel_type == "d3d11va" or hwaccel_type == "dxva2":
# Windows平台的硬件解码，但使用软件编码
config["video_codec"] = "libx264"
config["preset"] = "fast"
config["preset"] = "medium"
config["quality_param"] = "global_quality"
config["quality_value"] = "23"
elif hwaccel_type == "videotoolbox":
# macOS VideoToolbox编码器
config["video_codec"] = "h264_videotoolbox"
config["preset"] = "medium"
config["quality_param"] = "b:v"
config["quality_value"] = "5M"
else:
# 软件编码（默认）
config["video_codec"] = "libx264"
config["preset"] = "medium"
config["quality_param"] = "crf"
config["quality_value"] = "23"
return config
@ -130,7 +150,10 @@ def build_ffmpeg_command(
hwaccel_args: List[str] = None
) -> List[str]:
"""
构建优化的ffmpeg命令
构建优化的ffmpeg命令，基于测试结果使用正确的硬件加速方案
重要发现：对于视频裁剪场景，CUDA硬件解码会导致滤镜链错误，
应该使用纯NVENC编码器（无硬件解码）来获得最佳兼容性
Args:
input_path: 输入视频路径
@ -145,8 +168,14 @@ def build_ffmpeg_command(
"""
cmd = ["ffmpeg", "-y"]
# 添加硬件加速参数（如果有）
if hwaccel_args:
# 关键修正：对于视频裁剪，不使用CUDA硬件解码，只使用NVENC编码器
# 这样能避免滤镜链格式转换错误，同时保持编码性能优势
if encoder_config["video_codec"] == "h264_nvenc":
# 不添加硬件解码参数，让FFmpeg自动处理
# 这避免了 "Impossible to convert between the formats" 错误
pass
elif hwaccel_args:
# 对于其他编码器，可以使用硬件解码参数
cmd.extend(hwaccel_args)
# 输入文件
@ -159,25 +188,39 @@ def build_ffmpeg_command(
cmd.extend(["-c:v", encoder_config["video_codec"]])
cmd.extend(["-c:a", encoder_config["audio_codec"]])
# 像素格式（关键：避免滤镜链问题）
# 像素格式
cmd.extend(["-pix_fmt", encoder_config["pixel_format"]])
# 编码质量设置
if encoder_config["video_codec"] == "libx264":
# 质量和预设参数 - 针对NVENC优化
if encoder_config["video_codec"] == "h264_nvenc":
# 纯NVENC编码器配置（无硬件解码，兼容性最佳）
cmd.extend(["-preset", encoder_config["preset"]])
cmd.extend(["-crf", encoder_config["crf"]])
elif encoder_config["video_codec"] == "h264_nvenc":
cmd.extend(["-preset", encoder_config["preset"]])
cmd.extend(["-rc", "vbr", "-cq", encoder_config["crf"]])
cmd.extend(["-cq", encoder_config["quality_value"]])
cmd.extend(["-profile:v", "main"]) # 提高兼容性
logger.debug("使用纯NVENC编码器（无硬件解码，避免滤镜链问题）")
elif encoder_config["video_codec"] == "h264_amf":
# AMD AMF编码器
cmd.extend(["-quality", encoder_config["preset"]])
cmd.extend(["-qp_i", encoder_config["quality_value"]])
elif encoder_config["video_codec"] == "h264_qsv":
# Intel QSV编码器
cmd.extend(["-preset", encoder_config["preset"]])
cmd.extend(["-global_quality", encoder_config["crf"]])
cmd.extend(["-global_quality", encoder_config["quality_value"]])
elif encoder_config["video_codec"] == "h264_videotoolbox":
# macOS VideoToolbox编码器
cmd.extend(["-profile:v", "high"])
cmd.extend(["-b:v", encoder_config["quality_value"]])
else:
# 软件编码器（libx264）
cmd.extend(["-preset", encoder_config["preset"]])
cmd.extend(["-crf", encoder_config["quality_value"]])
# 音频设置
cmd.extend(["-ar", "44100", "-ac", "2"])
# 避免滤镜链问题的关键参数
# 优化参数
cmd.extend(["-avoid_negative_ts", "make_zero"])
cmd.extend(["-movflags", "+faststart"])
# 输出文件
cmd.append(output_path)
@ -194,7 +237,7 @@ def execute_ffmpeg_with_fallback(
end_time: str
) -> bool:
"""
执行ffmpeg命令，带有fallback机制
执行ffmpeg命令，带有智能fallback机制
Args:
cmd: 主要的ffmpeg命令
@ -222,11 +265,11 @@ def execute_ffmpeg_with_fallback(
if is_windows:
process_kwargs["encoding"] = 'utf-8'
subprocess.run(cmd, **process_kwargs)
result = subprocess.run(cmd, **process_kwargs)
# 验证输出文件
if os.path.exists(output_path) and os.path.getsize(output_path) > 0:
logger.info(f"视频裁剪成功: {timestamp}")
logger.info(f"✓ 视频裁剪成功: {timestamp}")
return True
else:
logger.warning(f"输出文件无效: {output_path}")
@ -236,14 +279,234 @@ def execute_ffmpeg_with_fallback(
error_msg = e.stderr if e.stderr else str(e)
logger.warning(f"主要命令失败: {error_msg}")
# 尝试fallback命令（纯软件编码）
logger.info(f"尝试fallback方案: {timestamp}")
return try_fallback_encoding(input_path, output_path, start_time, end_time, timestamp)
# 智能错误分析
error_type = analyze_ffmpeg_error(error_msg)
logger.debug(f"错误类型分析: {error_type}")
# 根据错误类型选择fallback策略
if error_type == "filter_chain_error":
logger.info(f"检测到滤镜链错误，尝试兼容性模式: {timestamp}")
return try_compatibility_fallback(input_path, output_path, start_time, end_time, timestamp)
elif error_type == "hardware_error":
logger.info(f"检测到硬件加速错误，尝试软件编码: {timestamp}")
return try_software_fallback(input_path, output_path, start_time, end_time, timestamp)
elif error_type == "encoder_error":
logger.info(f"检测到编码器错误，尝试基本编码: {timestamp}")
return try_basic_fallback(input_path, output_path, start_time, end_time, timestamp)
else:
logger.info(f"尝试通用fallback方案: {timestamp}")
return try_fallback_encoding(input_path, output_path, start_time, end_time, timestamp)
except Exception as e:
logger.error(f"执行ffmpeg命令时发生异常: {str(e)}")
return False
def analyze_ffmpeg_error(error_msg: str) -> str:
"""
分析ffmpeg错误信息，返回错误类型
Args:
error_msg: 错误信息
Returns:
str: 错误类型
"""
error_msg_lower = error_msg.lower()
# 滤镜链错误
if any(keyword in error_msg_lower for keyword in [
"impossible to convert", "filter", "format", "scale", "auto_scale",
"null", "parsed_null", "reinitializing filters"
]):
return "filter_chain_error"
# 硬件加速错误
if any(keyword in error_msg_lower for keyword in [
"cuda", "nvenc", "amf", "qsv", "d3d11va", "dxva2", "videotoolbox",
"hardware", "hwaccel", "gpu", "device"
]):
return "hardware_error"
# 编码器错误
if any(keyword in error_msg_lower for keyword in [
"encoder", "codec", "h264", "libx264", "bitrate", "preset"
]):
return "encoder_error"
# 文件访问错误
if any(keyword in error_msg_lower for keyword in [
"no such file", "permission denied", "access denied", "file not found"
]):
return "file_error"
return "unknown_error"
def try_compatibility_fallback(
input_path: str,
output_path: str,
start_time: str,
end_time: str,
timestamp: str
) -> bool:
"""
尝试兼容性fallback方案（解决滤镜链问题）
Args:
input_path: 输入路径
output_path: 输出路径
start_time: 开始时间
end_time: 结束时间
timestamp: 时间戳
Returns:
bool: 是否成功
"""
# 兼容性模式：避免所有可能的滤镜链问题
fallback_cmd = [
"ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
"-i", input_path,
"-ss", start_time,
"-to", end_time,
"-c:v", "libx264",
"-c:a", "aac",
"-pix_fmt", "yuv420p", # 明确指定像素格式
"-preset", "fast",
"-crf", "23",
"-ar", "44100", "-ac", "2", # 标准化音频
"-avoid_negative_ts", "make_zero",
"-movflags", "+faststart",
"-max_muxing_queue_size", "1024", # 增加缓冲区大小
output_path
]
return execute_simple_command(fallback_cmd, timestamp, "兼容性模式")
def try_software_fallback(
input_path: str,
output_path: str,
start_time: str,
end_time: str,
timestamp: str
) -> bool:
"""
尝试软件编码fallback方案
Args:
input_path: 输入路径
output_path: 输出路径
start_time: 开始时间
end_time: 结束时间
timestamp: 时间戳
Returns:
bool: 是否成功
"""
# 纯软件编码
fallback_cmd = [
"ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
"-i", input_path,
"-ss", start_time,
"-to", end_time,
"-c:v", "libx264",
"-c:a", "aac",
"-pix_fmt", "yuv420p",
"-preset", "fast",
"-crf", "23",
"-ar", "44100", "-ac", "2",
"-avoid_negative_ts", "make_zero",
"-movflags", "+faststart",
output_path
]
return execute_simple_command(fallback_cmd, timestamp, "软件编码")
def try_basic_fallback(
input_path: str,
output_path: str,
start_time: str,
end_time: str,
timestamp: str
) -> bool:
"""
尝试基本编码fallback方案
Args:
input_path: 输入路径
output_path: 输出路径
start_time: 开始时间
end_time: 结束时间
timestamp: 时间戳
Returns:
bool: 是否成功
"""
# 最基本的编码参数
fallback_cmd = [
"ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
"-i", input_path,
"-ss", start_time,
"-to", end_time,
"-c:v", "libx264",
"-c:a", "aac",
"-pix_fmt", "yuv420p",
"-preset", "ultrafast", # 最快速度
"-crf", "28", # 稍微降低质量
"-avoid_negative_ts", "make_zero",
output_path
]
return execute_simple_command(fallback_cmd, timestamp, "基本编码")
def execute_simple_command(cmd: List[str], timestamp: str, method_name: str) -> bool:
"""
执行简单的ffmpeg命令
Args:
cmd: 命令列表
timestamp: 时间戳
method_name: 方法名称
Returns:
bool: 是否成功
"""
try:
logger.debug(f"执行{method_name}命令: {' '.join(cmd)}")
is_windows = os.name == 'nt'
process_kwargs = {
"stdout": subprocess.PIPE,
"stderr": subprocess.PIPE,
"text": True,
"check": True
}
if is_windows:
process_kwargs["encoding"] = 'utf-8'
subprocess.run(cmd, **process_kwargs)
output_path = cmd[-1] # 输出路径总是最后一个参数
if os.path.exists(output_path) and os.path.getsize(output_path) > 0:
logger.info(f"✓ {method_name}成功: {timestamp}")
return True
else:
logger.error(f"{method_name}失败，输出文件无效: {output_path}")
return False
except subprocess.CalledProcessError as e:
error_msg = e.stderr if e.stderr else str(e)
logger.error(f"{method_name}失败: {error_msg}")
return False
except Exception as e:
logger.error(f"{method_name}异常: {str(e)}")
return False
def try_fallback_encoding(
input_path: str,
output_path: str,
@ -252,7 +515,7 @@ def try_fallback_encoding(
timestamp: str
) -> bool:
"""
尝试fallback编码方案（纯软件编码）
尝试fallback编码方案（通用方案）
Args:
input_path: 输入路径
@ -280,36 +543,7 @@ def try_fallback_encoding(
output_path
]
try:
logger.debug(f"执行fallback命令: {' '.join(fallback_cmd)}")
is_windows = os.name == 'nt'
process_kwargs = {
"stdout": subprocess.PIPE,
"stderr": subprocess.PIPE,
"text": True,
"check": True
}
if is_windows:
process_kwargs["encoding"] = 'utf-8'
subprocess.run(fallback_cmd, **process_kwargs)
if os.path.exists(output_path) and os.path.getsize(output_path) > 0:
logger.info(f"Fallback编码成功: {timestamp}")
return True
else:
logger.error(f"Fallback编码失败，输出文件无效: {output_path}")
return False
except subprocess.CalledProcessError as e:
error_msg = e.stderr if e.stderr else str(e)
logger.error(f"Fallback编码也失败: {error_msg}")
return False
except Exception as e:
logger.error(f"Fallback编码异常: {str(e)}")
return False
return execute_simple_command(fallback_cmd, timestamp, "通用Fallback")
def clip_video(
@ -355,19 +589,24 @@ def clip_video(
if hwaccel_type:
hwaccel_args = ffmpeg_utils.get_ffmpeg_hwaccel_args()
logger.info(f"使用硬件加速: {hwaccel_type}")
hwaccel_info = ffmpeg_utils.get_ffmpeg_hwaccel_info()
logger.info(f"🚀 使用硬件加速: {hwaccel_type} ({hwaccel_info.get('message', '')})")
else:
logger.info("使用软件编码")
logger.info("🔧 使用软件编码")
# 获取编码器配置
encoder_config = get_safe_encoder_config(hwaccel_type)
logger.debug(f"编码器配置: {encoder_config}")
# 存储裁剪结果
# 统计信息
total_clips = len(tts_result)
result = {}
failed_clips = []
success_count = 0
for item in tts_result:
logger.info(f"📹 开始裁剪视频，总共{total_clips}个片段")
for i, item in enumerate(tts_result, 1):
_id = item.get("_id", item.get("timestamp", "unknown"))
timestamp = item["timestamp"]
start_time, _ = parse_timestamp(timestamp)
@ -397,7 +636,7 @@ def clip_video(
)
# 执行FFmpeg命令
logger.info(f"裁剪视频片段: {timestamp} -> {ffmpeg_start_time}到{ffmpeg_end_time}")
logger.info(f"📹 [{i}/{total_clips}] 裁剪视频片段: {timestamp} -> {ffmpeg_start_time}到{ffmpeg_end_time}")
success = execute_ffmpeg_with_fallback(
ffmpeg_cmd,
@ -410,17 +649,26 @@ def clip_video(
if success:
result[_id] = output_path
success_count += 1
logger.info(f"✅ [{i}/{total_clips}] 片段裁剪成功: {timestamp}")
else:
failed_clips.append(timestamp)
logger.error(f"裁剪视频片段失败: {timestamp}")
logger.error(f"❌ [{i}/{total_clips}] 片段裁剪失败: {timestamp}")
# 最终统计
logger.info(f"📊 视频裁剪完成: 成功 {success_count}/{total_clips}, 失败 {len(failed_clips)}")
# 检查是否有失败的片段
if failed_clips:
logger.warning(f"以下片段裁剪失败: {failed_clips}")
if len(failed_clips) == len(tts_result):
logger.warning(f"⚠️ 以下片段裁剪失败: {failed_clips}")
if len(failed_clips) == total_clips:
raise RuntimeError("所有视频片段裁剪都失败了，请检查视频文件和ffmpeg配置")
elif len(failed_clips) > total_clips / 2:
logger.warning(f"⚠️ 超过一半的片段裁剪失败 ({len(failed_clips)}/{total_clips})，请检查硬件加速配置")
logger.info(f"视频裁剪完成，成功: {len(result)}, 失败: {len(failed_clips)}")
if success_count > 0:
logger.info(f"🎉 视频裁剪任务完成! 输出目录: {output_dir}")
return result

77
app/services/merger_video.py
View File

@ -137,6 +137,9 @@ def process_single_video(
) -> str:
"""
处理单个视频：调整分辨率、帧率等
重要修复：避免在视频滤镜处理时使用CUDA硬件解码，
因为这会导致滤镜链格式转换错误。使用纯NVENC编码器获得最佳兼容性。
Args:
input_path: 输入视频路径
@ -178,24 +181,11 @@ def process_single_video(
logger.warning(f"视频探测出错，禁用硬件加速: {str(e)}")
hwaccel = None
# 添加硬件加速参数（使用新的智能检测机制）
if hwaccel:
try:
# 使用新的硬件加速检测API
hwaccel_args = ffmpeg_utils.get_ffmpeg_hwaccel_args()
if hwaccel_args:
command.extend(hwaccel_args)
logger.debug(f"应用硬件加速参数: {hwaccel_args}")
else:
logger.info("硬件加速不可用，将使用软件编码")
hwaccel = False # 标记为不使用硬件加速
except Exception as e:
logger.warning(f"应用硬件加速参数时出错: {str(e)}，将使用软件编码")
hwaccel = False # 标记为不使用硬件加速
# 重置命令，移除可能添加了一半的硬件加速参数
command = ['ffmpeg', '-y']
# 输入文件
# 关键修复：对于涉及滤镜处理的场景，不使用CUDA硬件解码
# 这避免了 "Impossible to convert between the formats" 错误
# 我们将只使用纯NVENC编码器来获得硬件加速优势
# 输入文件（不添加硬件解码参数）
command.extend(['-i', input_path])
# 处理音频
@ -218,27 +208,36 @@ def process_single_video(
'-r', '30', # 设置帧率为30fps
])
# 选择编码器 - 使用新的智能编码器选择
encoder = ffmpeg_utils.get_optimal_ffmpeg_encoder()
if hwaccel and encoder != "libx264":
logger.info(f"使用硬件编码器: {encoder}")
command.extend(['-c:v', encoder])
# 根据编码器类型添加特定参数
if "nvenc" in encoder:
command.extend(['-preset', 'p4', '-profile:v', 'high'])
elif "videotoolbox" in encoder:
command.extend(['-profile:v', 'high'])
elif "qsv" in encoder:
command.extend(['-preset', 'medium'])
elif "vaapi" in encoder:
command.extend(['-profile', '100'])
elif "amf" in encoder:
command.extend(['-quality', 'balanced'])
else:
command.extend(['-preset', 'medium', '-profile:v', 'high'])
else:
# 关键修复：选择编码器时优先使用纯NVENC（无硬件解码）
if hwaccel:
try:
# 检查是否为NVIDIA硬件加速
hwaccel_info = ffmpeg_utils.detect_hardware_acceleration()
if hwaccel_info.get("type") in ["cuda", "nvenc"] and hwaccel_info.get("encoder") == "h264_nvenc":
# 使用纯NVENC编码器（最佳兼容性）
logger.info("使用纯NVENC编码器（避免滤镜链问题）")
command.extend(['-c:v', 'h264_nvenc'])
command.extend(['-preset', 'medium', '-cq', '23', '-profile:v', 'main'])
else:
# 其他硬件编码器
encoder = ffmpeg_utils.get_optimal_ffmpeg_encoder()
logger.info(f"使用硬件编码器: {encoder}")
command.extend(['-c:v', encoder])
# 根据编码器类型添加特定参数
if "amf" in encoder:
command.extend(['-quality', 'balanced'])
elif "qsv" in encoder:
command.extend(['-preset', 'medium'])
elif "videotoolbox" in encoder:
command.extend(['-profile:v', 'high'])
else:
command.extend(['-preset', 'medium', '-profile:v', 'high'])
except Exception as e:
logger.warning(f"硬件编码器检测失败: {str(e)}，将使用软件编码")
hwaccel = None
if not hwaccel:
logger.info("使用软件编码器(libx264)")
command.extend(['-c:v', 'libx264', '-preset', 'medium', '-profile:v', 'high'])

380
app/utils/ffmpeg_utils.py
View File

@ -469,167 +469,171 @@ def _detect_macos_acceleration(supported_hwaccels: str) -> None:
def _detect_windows_acceleration(supported_hwaccels: str) -> None:
"""
检测Windows系统的硬件加速
检测Windows系统的硬件加速 - 基于实际测试结果优化
重要发现：CUDA硬件解码在视频裁剪场景下会导致滤镜链错误，
因此优先使用纯NVENC编码器方案，既保证性能又确保兼容性。
Args:
supported_hwaccels: FFmpeg支持的硬件加速器列表
"""
global _FFMPEG_HW_ACCEL_INFO
# 在Windows上，首先检查显卡信息
gpu_info = _get_windows_gpu_info()
# 检查是否为AMD显卡
if 'amd' in gpu_info.lower() or 'radeon' in gpu_info.lower():
logger.info("检测到AMD显卡，为避免兼容性问题，将使用软件编码")
_FFMPEG_HW_ACCEL_INFO["message"] = "检测到AMD显卡，为避免兼容性问题，将使用软件编码"
return
logger.debug(f"Windows GPU信息: {gpu_info}")
# 检查是否为Intel集成显卡
is_intel_integrated = False
if 'intel' in gpu_info.lower() and ('hd graphics' in gpu_info.lower() or 'uhd graphics' in gpu_info.lower()):
logger.info("检测到Intel集成显卡")
is_intel_integrated = True
# 检测NVIDIA CUDA支持
if 'cuda' in supported_hwaccels and 'nvidia' in gpu_info.lower():
# 添加调试日志
logger.debug(f"Windows检测到NVIDIA显卡，尝试CUDA加速")
# 1. 优先检测NVIDIA硬件加速 - 基于实际测试的最佳方案
if 'nvidia' in gpu_info.lower() or 'geforce' in gpu_info.lower() or 'quadro' in gpu_info.lower():
logger.info("检测到NVIDIA显卡，开始测试硬件加速")
# 检查NVENC编码器是否可用
try:
# 先检查NVENC编码器是否可用，使用UTF-8编码
encoders_cmd = subprocess.run(
["ffmpeg", "-hide_banner", "-encoders"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, encoding='utf-8', text=True, check=False
stderr=subprocess.PIPE, stdout=subprocess.PIPE,
encoding='utf-8', text=True, check=False
)
has_nvenc = "h264_nvenc" in encoders_cmd.stdout.lower()
logger.debug(f"NVENC编码器检测结果: {'可用' if has_nvenc else '不可用'}")
# 测试CUDA硬件加速，使用UTF-8编码
test_cmd = subprocess.run(
["ffmpeg", "-hwaccel", "cuda", "-i", "NUL", "-f", "null", "-t", "0.1", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, encoding='utf-8', text=True, check=False
)
# 记录详细的返回信息以便调试
logger.debug(f"CUDA测试返回码: {test_cmd.returncode}")
logger.debug(f"CUDA测试错误输出: {test_cmd.stderr[:200]}..." if len(test_cmd.stderr) > 200 else f"CUDA测试错误输出: {test_cmd.stderr}")
if test_cmd.returncode == 0 or has_nvenc:
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "cuda"
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264_nvenc"
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "cuda"]
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = True
return
# 如果上面的测试失败，尝试另一种方式，使用UTF-8编码
test_cmd2 = subprocess.run(
["ffmpeg", "-hide_banner", "-loglevel", "error", "-hwaccel", "cuda", "-hwaccel_output_format", "cuda", "-i", "NUL", "-f", "null", "-t", "0.1", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, encoding='utf-8', text=True, check=False
)
if test_cmd2.returncode == 0:
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "cuda"
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264_nvenc"
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "cuda", "-hwaccel_output_format", "cuda"]
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = True
return
except Exception as e:
logger.debug(f"测试CUDA失败: {str(e)}")
# 检测Intel QSV支持（如果是Intel显卡）
if 'qsv' in supported_hwaccels and 'intel' in gpu_info.lower():
try:
test_cmd = subprocess.run(
["ffmpeg", "-hwaccel", "qsv", "-i", "/dev/null", "-f", "null", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, check=False
)
if test_cmd.returncode == 0:
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "qsv"
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264_qsv"
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "qsv"]
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = not is_intel_integrated
return
except Exception as e:
logger.debug(f"测试QSV失败: {str(e)}")
# 检测D3D11VA支持
if 'd3d11va' in supported_hwaccels:
logger.debug("Windows尝试D3D11VA加速")
try:
test_cmd = subprocess.run(
["ffmpeg", "-hwaccel", "d3d11va", "-i", "NUL", "-f", "null", "-t", "0.1", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, encoding='utf-8', text=True, check=False
)
# 记录详细的返回信息以便调试
logger.debug(f"D3D11VA测试返回码: {test_cmd.returncode}")
if test_cmd.returncode == 0:
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "d3d11va"
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264" # D3D11VA只用于解码，编码仍使用软件编码器
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "d3d11va"]
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = not is_intel_integrated
return
except Exception as e:
logger.debug(f"测试D3D11VA失败: {str(e)}")
# 检测DXVA2支持
if 'dxva2' in supported_hwaccels:
logger.debug("Windows尝试DXVA2加速")
try:
test_cmd = subprocess.run(
["ffmpeg", "-hwaccel", "dxva2", "-i", "NUL", "-f", "null", "-t", "0.1", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, encoding='utf-8', text=True, check=False
)
# 记录详细的返回信息以便调试
logger.debug(f"DXVA2测试返回码: {test_cmd.returncode}")
if test_cmd.returncode == 0:
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "dxva2"
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264" # DXVA2只用于解码，编码仍使用软件编码器
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "dxva2"]
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = not is_intel_integrated
return
except Exception as e:
logger.debug(f"测试DXVA2失败: {str(e)}")
# 如果检测到NVIDIA显卡但前面的测试都失败，尝试直接使用NVENC编码器
if 'nvidia' in gpu_info.lower():
logger.debug("Windows检测到NVIDIA显卡，尝试直接使用NVENC编码器")
try:
# 检查NVENC编码器是否可用，使用UTF-8编码
encoders_cmd = subprocess.run(
["ffmpeg", "-hide_banner", "-encoders"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, encoding='utf-8', text=True, check=False
)
if "h264_nvenc" in encoders_cmd.stdout.lower():
logger.debug("NVENC编码器可用，尝试直接使用")
# 测试NVENC编码器，使用UTF-8编码
test_cmd = subprocess.run(
["ffmpeg", "-f", "lavfi", "-i", "color=c=black:s=640x360:r=30", "-c:v", "h264_nvenc", "-t", "0.1", "-f", "null", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, encoding='utf-8', text=True, check=False
)
logger.debug(f"NVENC编码器测试返回码: {test_cmd.returncode}")
if has_nvenc:
# 优先方案：纯NVENC编码器（测试证明最兼容）
logger.debug("测试纯NVENC编码器（推荐方案，避免滤镜链问题）")
test_cmd = subprocess.run([
"ffmpeg", "-hide_banner", "-loglevel", "error",
"-f", "lavfi", "-i", "testsrc=duration=0.1:size=640x480:rate=30",
"-c:v", "h264_nvenc", "-preset", "medium", "-cq", "23",
"-pix_fmt", "yuv420p", "-f", "null", "-"
], stderr=subprocess.PIPE, stdout=subprocess.PIPE,
encoding='utf-8', text=True, check=False)
if test_cmd.returncode == 0:
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "nvenc"
_FFMPEG_HW_ACCEL_INFO["type"] = "nvenc" # 使用nvenc类型标识纯编码器
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264_nvenc"
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = [] # 不使用hwaccel参数，直接使用编码器
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = [] # 不使用硬件解码参数
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = True
_FFMPEG_HW_ACCEL_INFO["message"] = "纯NVENC编码器（最佳兼容性）"
logger.info("✓ 纯NVENC编码器测试成功")
return
# 备用方案：如果需要的话，可以测试CUDA硬件解码（但不推荐用于视频裁剪）
if 'cuda' in supported_hwaccels:
logger.debug("测试CUDA硬件解码（仅用于非裁剪场景）")
test_cmd = subprocess.run([
"ffmpeg", "-hide_banner", "-loglevel", "error",
"-hwaccel", "cuda", "-hwaccel_output_format", "cuda",
"-f", "lavfi", "-i", "testsrc=duration=0.1:size=640x480:rate=30",
"-c:v", "h264_nvenc", "-preset", "medium", "-cq", "23",
"-pix_fmt", "yuv420p", "-f", "null", "-"
], stderr=subprocess.PIPE, stdout=subprocess.PIPE,
encoding='utf-8', text=True, check=False)
if test_cmd.returncode == 0:
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "cuda" # 保留cuda类型用于特殊场景
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264_nvenc"
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "cuda", "-hwaccel_output_format", "cuda"]
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = True
_FFMPEG_HW_ACCEL_INFO["message"] = "CUDA+NVENC（限特殊场景使用）"
_FFMPEG_HW_ACCEL_INFO["fallback_available"] = True
_FFMPEG_HW_ACCEL_INFO["fallback_encoder"] = "h264_nvenc"
logger.info("✓ CUDA+NVENC硬件加速测试成功（备用方案）")
return
except Exception as e:
logger.debug(f"测试NVENC编码器失败: {str(e)}")
_FFMPEG_HW_ACCEL_INFO["message"] = f"Windows系统未检测到可用的硬件加速，显卡信息: {gpu_info}"
logger.debug(f"NVIDIA硬件加速测试失败: {str(e)}")
# 2. 检测AMD硬件加速
if 'amd' in gpu_info.lower() or 'radeon' in gpu_info.lower():
logger.info("检测到AMD显卡，开始测试硬件加速")
# 检查AMF编码器是否可用
try:
encoders_cmd = subprocess.run(
["ffmpeg", "-hide_banner", "-encoders"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE,
encoding='utf-8', text=True, check=False
)
has_amf = "h264_amf" in encoders_cmd.stdout.lower()
logger.debug(f"AMF编码器检测结果: {'可用' if has_amf else '不可用'}")
if has_amf:
# 测试AMF编码器
logger.debug("测试AMF编码器")
test_cmd = subprocess.run([
"ffmpeg", "-hide_banner", "-loglevel", "error",
"-f", "lavfi", "-i", "testsrc=duration=0.1:size=640x480:rate=30",
"-c:v", "h264_amf", "-quality", "balanced", "-qp_i", "23",
"-pix_fmt", "yuv420p", "-f", "null", "-"
], stderr=subprocess.PIPE, stdout=subprocess.PIPE,
encoding='utf-8', text=True, check=False)
if test_cmd.returncode == 0:
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "amf"
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264_amf"
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = []
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = True
_FFMPEG_HW_ACCEL_INFO["message"] = "AMD AMF编码器"
logger.info("✓ AMD AMF编码器测试成功")
return
except Exception as e:
logger.debug(f"AMD硬件加速测试失败: {str(e)}")
# 3. 检测Intel硬件加速
if 'intel' in gpu_info.lower() and 'qsv' in supported_hwaccels:
logger.info("检测到Intel显卡，开始测试硬件加速")
try:
encoders_cmd = subprocess.run(
["ffmpeg", "-hide_banner", "-encoders"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE,
encoding='utf-8', text=True, check=False
)
has_qsv = "h264_qsv" in encoders_cmd.stdout.lower()
logger.debug(f"QSV编码器检测结果: {'可用' if has_qsv else '不可用'}")
if has_qsv:
# 测试QSV编码器
logger.debug("测试QSV编码器")
test_cmd = subprocess.run([
"ffmpeg", "-hide_banner", "-loglevel", "error",
"-f", "lavfi", "-i", "testsrc=duration=0.1:size=640x480:rate=30",
"-c:v", "h264_qsv", "-preset", "medium", "-global_quality", "23",
"-pix_fmt", "yuv420p", "-f", "null", "-"
], stderr=subprocess.PIPE, stdout=subprocess.PIPE,
encoding='utf-8', text=True, check=False)
if test_cmd.returncode == 0:
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "qsv"
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264_qsv"
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = []
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = not is_intel_integrated
_FFMPEG_HW_ACCEL_INFO["message"] = "Intel QSV编码器"
logger.info("✓ Intel QSV编码器测试成功")
return
except Exception as e:
logger.debug(f"Intel硬件加速测试失败: {str(e)}")
# 4. 如果没有硬件编码器，使用软件编码
logger.info("未检测到可用的硬件编码器，使用软件编码")
_FFMPEG_HW_ACCEL_INFO["available"] = False
_FFMPEG_HW_ACCEL_INFO["type"] = "software"
_FFMPEG_HW_ACCEL_INFO["encoder"] = "libx264"
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = []
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = False
_FFMPEG_HW_ACCEL_INFO["message"] = "使用软件编码"
def _detect_linux_acceleration(supported_hwaccels: str) -> None:
@ -997,9 +1001,15 @@ def force_software_encoding() -> None:
def reset_hwaccel_detection() -> None:
"""
重置硬件加速检测结果，强制重新检测
这在以下情况下很有用：
1. 驱动程序更新后
2. 系统配置改变后
3. 需要重新测试硬件加速时
"""
global _FFMPEG_HW_ACCEL_INFO
logger.info("🔄 重置硬件加速检测，将重新检测...")
_FFMPEG_HW_ACCEL_INFO = {
"available": False,
"type": None,
@ -1014,4 +1024,94 @@ def reset_hwaccel_detection() -> None:
"tested_methods": []
}
logger.info("已重置硬件加速检测结果")
def test_nvenc_directly() -> bool:
"""
直接测试NVENC编码器是否可用（无硬件解码）
Returns:
bool: NVENC是否可用
"""
try:
logger.info("🧪 直接测试NVENC编码器...")
# 测试纯NVENC编码器
test_cmd = subprocess.run([
"ffmpeg", "-hide_banner", "-loglevel", "error",
"-f", "lavfi", "-i", "testsrc=duration=1:size=640x480:rate=30",
"-c:v", "h264_nvenc", "-preset", "fast", "-profile:v", "main",
"-pix_fmt", "yuv420p", "-t", "1", "-f", "null", "-"
], stderr=subprocess.PIPE, stdout=subprocess.PIPE,
encoding='utf-8', text=True, check=False)
if test_cmd.returncode == 0:
logger.info("✅ NVENC编码器测试成功！")
return True
else:
logger.warning(f"❌ NVENC编码器测试失败: {test_cmd.stderr}")
return False
except Exception as e:
logger.error(f"NVENC测试异常: {str(e)}")
return False
def force_use_nvenc_pure() -> None:
"""
强制使用纯NVENC编码器模式
当自动检测失败但你确定NVENC可用时使用
"""
global _FFMPEG_HW_ACCEL_INFO
logger.info("🎯 强制启用纯NVENC编码器模式...")
# 先测试NVENC是否真的可用
if test_nvenc_directly():
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "nvenc_pure"
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264_nvenc"
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = []
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = True
_FFMPEG_HW_ACCEL_INFO["message"] = "强制启用纯NVENC编码器"
logger.info("✅ 已强制启用纯NVENC编码器模式")
else:
logger.error("❌ NVENC编码器不可用，无法强制启用")
def get_hwaccel_status() -> Dict[str, any]:
"""
获取当前硬件加速状态的详细信息
Returns:
Dict: 硬件加速状态信息
"""
hwaccel_info = get_ffmpeg_hwaccel_info()
status = {
"available": hwaccel_info.get("available", False),
"type": hwaccel_info.get("type", "software"),
"encoder": hwaccel_info.get("encoder", "libx264"),
"message": hwaccel_info.get("message", ""),
"is_dedicated_gpu": hwaccel_info.get("is_dedicated_gpu", False),
"platform": platform.system(),
"gpu_vendor": detect_gpu_vendor(),
"ffmpeg_available": check_ffmpeg_installation()
}
return status
# 自动重置检测（在模块导入时执行）
def _auto_reset_on_import():
"""模块导入时自动重置硬件加速检测"""
try:
# 检查是否需要重置（比如检测到配置变化）
current_platform = platform.system()
if _FFMPEG_HW_ACCEL_INFO.get("platform") != current_platform:
reset_hwaccel_detection()
except Exception as e:
logger.debug(f"自动重置检测失败: {str(e)}")
# 执行自动重置
_auto_reset_on_import()

2
config.example.toml
View File

@ -1,5 +1,5 @@
[app]
project_version="0.6.5"
project_version="0.6.6"
# 支持视频理解的大模型提供商
# gemini (谷歌, 需要 VPN)
# siliconflow (硅基流动)

2
project_version
View File

@ -1 +1 @@
0.6.5
0.6.6