diff --git a/app/utils/video_processor.py b/app/utils/video_processor.py index 920fe53..ebafabf 100644 --- a/app/utils/video_processor.py +++ b/app/utils/video_processor.py @@ -212,12 +212,18 @@ class VideoProcessor: return True logger.debug(f"硬件加速方案失败，回退到软件方案") - # 策略3: 软件方案（最后的备用方案） - return self._try_extract_with_software(timestamp, output_path) + # 策略3: 软件方案 + if self._try_extract_with_software(timestamp, output_path): + return True + logger.debug(f"软件方案失败，尝试超级兼容性方案") + + # 策略4: 超级兼容性方案（Windows 特殊处理） + return self._try_extract_with_ultra_compatibility(timestamp, output_path) def _try_extract_with_software_decode(self, timestamp: float, output_path: str) -> bool: """ 使用纯软件解码提取帧（推荐用于 Windows N 卡） + 参考 clip_video.py 中的成功实现 Args: timestamp: 时间戳 @@ -226,13 +232,19 @@ class VideoProcessor: Returns: bool: 是否成功 """ - # 使用 Windows NVIDIA 优化配置 - cmd = FFmpegConfigManager.get_extraction_command( - input_path=self.video_path, - output_path=output_path, - timestamp=timestamp, - profile_name="windows_nvidia" - ) + # 参考 clip_video.py 中的兼容性方案，专门针对图片输出优化 + cmd = [ + "ffmpeg", + "-hide_banner", + "-loglevel", "error", + "-ss", str(timestamp), # 先定位时间戳 + "-i", self.video_path, + "-vframes", "1", # 只提取一帧 + "-q:v", "2", # 高质量 + "-pix_fmt", "yuv420p", # 明确指定像素格式 + "-y", + output_path + ] return self._execute_ffmpeg_command(cmd, f"软件解码提取帧 {timestamp:.1f}s") @@ -272,6 +284,7 @@ class VideoProcessor: def _try_extract_with_software(self, timestamp: float, output_path: str) -> bool: """ 使用纯软件方案提取帧（最后的备用方案） + 参考 clip_video.py 中的基本编码方案 Args: timestamp: 时间戳 @@ -280,22 +293,125 @@ class VideoProcessor: Returns: bool: 是否成功 """ - # 使用最高兼容性配置 - cmd = FFmpegConfigManager.get_extraction_command( - input_path=self.video_path, - output_path=output_path, - timestamp=timestamp, - profile_name="compatibility" - ) - - # 软件方案使用更详细的日志 - cmd[cmd.index("-loglevel") + 1] = "warning" + # 最基本的兼容性方案，参考 clip_video.py 的 try_basic_fallback + cmd = [ + "ffmpeg", + "-hide_banner", + "-loglevel", "warning", # 更详细的日志用于调试 + "-ss", str(timestamp), + "-i", self.video_path, + "-vframes", "1", + "-q:v", "3", # 稍微降低质量以提高兼容性 + "-pix_fmt", "yuv420p", + "-avoid_negative_ts", "make_zero", # 避免时间戳问题 + "-y", + output_path + ] return self._execute_ffmpeg_command(cmd, f"软件方案提取帧 {timestamp:.1f}s") + def _try_extract_with_ultra_compatibility(self, timestamp: float, output_path: str) -> bool: + """ + 超级兼容性方案，专门解决 Windows 系统的 MJPEG 编码问题 + + Args: + timestamp: 时间戳 + output_path: 输出路径 + + Returns: + bool: 是否成功 + """ + # 方案1: 使用 PNG 格式避免 MJPEG 问题 + png_output = output_path.replace('.jpg', '.png') + cmd1 = [ + "ffmpeg", + "-hide_banner", + "-loglevel", "error", + "-ss", str(timestamp), + "-i", self.video_path, + "-vframes", "1", + "-f", "image2", # 明确指定图片格式 + "-y", + png_output + ] + + if self._execute_ffmpeg_command(cmd1, f"PNG格式提取帧 {timestamp:.1f}s"): + # 如果 PNG 成功，转换为 JPG + try: + from PIL import Image + with Image.open(png_output) as img: + # 转换为 RGB 模式（去除 alpha 通道） + if img.mode in ('RGBA', 'LA'): + background = Image.new('RGB', img.size, (255, 255, 255)) + background.paste(img, mask=img.split()[-1] if img.mode == 'RGBA' else None) + img = background + img.save(output_path, 'JPEG', quality=90) + + # 删除临时 PNG 文件 + os.remove(png_output) + return True + except Exception as e: + logger.debug(f"PNG 转 JPG 失败: {e}") + # 如果转换失败，直接重命名 PNG 为 JPG + try: + os.rename(png_output, output_path) + return True + except Exception: + pass + + # 方案2: 使用最简单的参数 + cmd2 = [ + "ffmpeg", + "-hide_banner", + "-loglevel", "error", + "-i", self.video_path, + "-ss", str(timestamp), # 把 -ss 放在 -i 后面 + "-vframes", "1", + "-f", "mjpeg", # 明确指定 MJPEG 格式 + "-q:v", "5", # 降低质量要求 + "-y", + output_path + ] + + if self._execute_ffmpeg_command(cmd2, f"MJPEG格式提取帧 {timestamp:.1f}s"): + return True + + # 方案3: 最后的尝试 - 使用 BMP 格式 + bmp_output = output_path.replace('.jpg', '.bmp') + cmd3 = [ + "ffmpeg", + "-hide_banner", + "-loglevel", "error", + "-i", self.video_path, + "-ss", str(timestamp), + "-vframes", "1", + "-f", "bmp", + "-y", + bmp_output + ] + + if self._execute_ffmpeg_command(cmd3, f"BMP格式提取帧 {timestamp:.1f}s"): + # 尝试转换 BMP 为 JPG + try: + from PIL import Image + with Image.open(bmp_output) as img: + img.save(output_path, 'JPEG', quality=90) + os.remove(bmp_output) + return True + except Exception: + # 如果转换失败，直接重命名 + try: + os.rename(bmp_output, output_path) + return True + except Exception: + pass + + return False + def _execute_ffmpeg_command(self, cmd: List[str], description: str) -> bool: """ 执行 FFmpeg 命令并处理结果 + 参考 clip_video.py 中的错误处理机制 Args: cmd: FFmpeg 命令列表 @@ -305,31 +421,44 @@ class VideoProcessor: bool: 是否成功 """ try: - # 在 Windows 上使用 UTF-8 编码 + # 参考 clip_video.py 中的 Windows 处理方式 is_windows = os.name == 'nt' process_kwargs = { + "stdout": subprocess.PIPE, + "stderr": subprocess.PIPE, + "text": True, "check": True, - "capture_output": True, "timeout": 30 # 30秒超时 } if is_windows: process_kwargs["encoding"] = 'utf-8' - process_kwargs["text"] = True + logger.debug(f"执行命令: {' '.join(cmd)}") result = subprocess.run(cmd, **process_kwargs) # 验证输出文件 output_path = cmd[-1] if os.path.exists(output_path) and os.path.getsize(output_path) > 0: + logger.debug(f"{description} - 成功") return True else: - logger.debug(f"{description} - 输出文件无效") + logger.debug(f"{description} - 输出文件无效: {output_path}") return False except subprocess.CalledProcessError as e: error_msg = e.stderr if hasattr(e, 'stderr') and e.stderr else str(e) - logger.debug(f"{description} - 命令执行失败: {error_msg}") + + # 分析错误类型，提供更好的调试信息 + if "mjpeg" in error_msg.lower() and "non full-range yuv" in error_msg.lower(): + logger.debug(f"{description} - MJPEG YUV 格式问题: {error_msg[:200]}") + elif "codec avOption" in error_msg.lower(): + logger.debug(f"{description} - 编码器参数问题: {error_msg[:200]}") + elif "filter" in error_msg.lower(): + logger.debug(f"{description} - 滤镜链问题: {error_msg[:200]}") + else: + logger.debug(f"{description} - 命令执行失败: {error_msg[:200]}") + return False except subprocess.TimeoutExpired: logger.debug(f"{description} - 命令执行超时") diff --git a/test_video_extraction.py b/test_video_extraction.py deleted file mode 100644 index 2e539dd..0000000 --- a/test_video_extraction.py +++ /dev/null @@ -1,189 +0,0 @@ -#!/usr/bin/env python3 -""" -视频关键帧提取测试脚本 -用于验证 Windows 系统 FFmpeg 兼容性修复效果 -""" - -import os -import sys -import tempfile -import traceback -from pathlib import Path - -# 添加项目根目录到 Python 路径 -project_root = Path(__file__).parent -sys.path.insert(0, str(project_root)) - -from loguru import logger -from app.utils import video_processor, ffmpeg_utils - - -def test_ffmpeg_compatibility(): - """测试 FFmpeg 兼容性""" - print("=" * 60) - print("🔧 FFmpeg 兼容性测试") - print("=" * 60) - - # 检查 FFmpeg 安装 - if not ffmpeg_utils.check_ffmpeg_installation(): - print("❌ FFmpeg 未安装或不在系统 PATH 中") - return False - - print("✅ FFmpeg 已安装") - - # 获取硬件加速信息 - hwaccel_info = ffmpeg_utils.get_ffmpeg_hwaccel_info() - print(f"🎮 硬件加速状态: {hwaccel_info.get('message', '未知')}") - print(f"🔧 加速类型: {hwaccel_info.get('type', 'software')}") - print(f"🎯 编码器: {hwaccel_info.get('encoder', 'libx264')}") - - return True - - -def test_video_extraction(video_path: str, output_dir: str = None): - """测试视频关键帧提取""" - print("\n" + "=" * 60) - print("🎬 视频关键帧提取测试") - print("=" * 60) - - if not os.path.exists(video_path): - print(f"❌ 视频文件不存在: {video_path}") - return False - - # 创建临时输出目录 - if output_dir is None: - output_dir = tempfile.mkdtemp(prefix="keyframes_test_") - - try: - # 初始化视频处理器 - print(f"📁 输入视频: {video_path}") - print(f"📁 输出目录: {output_dir}") - - processor = video_processor.VideoProcessor(video_path) - - # 显示视频信息 - print(f"📊 视频信息:") - print(f" - 分辨率: {processor.width}x{processor.height}") - print(f" - 帧率: {processor.fps:.1f} fps") - print(f" - 时长: {processor.duration:.1f} 秒") - print(f" - 总帧数: {processor.total_frames}") - - # 测试关键帧提取 - print("\n🚀 开始提取关键帧...") - - # 先测试硬件加速方案 - print("\n1️⃣ 测试硬件加速方案:") - try: - processor.process_video_pipeline( - output_dir=output_dir, - interval_seconds=10.0, # 10秒间隔，减少测试时间 - use_hw_accel=True - ) - - # 检查结果 - extracted_files = [f for f in os.listdir(output_dir) if f.endswith('.jpg')] - print(f"✅ 硬件加速成功，提取了 {len(extracted_files)} 个关键帧") - - if len(extracted_files) > 0: - return True - - except Exception as e: - print(f"⚠️ 硬件加速失败: {str(e)}") - - # 清理失败的文件 - for f in os.listdir(output_dir): - if f.endswith('.jpg'): - os.remove(os.path.join(output_dir, f)) - - # 测试软件方案 - print("\n2️⃣ 测试软件方案:") - try: - # 强制使用软件编码 - ffmpeg_utils.force_software_encoding() - - processor.process_video_pipeline( - output_dir=output_dir, - interval_seconds=10.0, - use_hw_accel=False - ) - - # 检查结果 - extracted_files = [f for f in os.listdir(output_dir) if f.endswith('.jpg')] - print(f"✅ 软件方案成功，提取了 {len(extracted_files)} 个关键帧") - - if len(extracted_files) > 0: - return True - else: - print("❌ 软件方案也未能提取到关键帧") - return False - - except Exception as e: - print(f"❌ 软件方案也失败: {str(e)}") - return False - - except Exception as e: - print(f"❌ 测试过程中发生错误: {str(e)}") - print(f"详细错误信息:\n{traceback.format_exc()}") - return False - - finally: - # 清理临时文件 - try: - import shutil - if output_dir and os.path.exists(output_dir): - shutil.rmtree(output_dir) - print(f"🧹 已清理临时目录: {output_dir}") - except Exception as e: - print(f"⚠️ 清理临时目录失败: {e}") - - -def main(): - """主函数""" - print("🎯 视频关键帧提取兼容性测试工具") - print("专门用于测试 Windows 系统 FFmpeg 兼容性修复效果") - - # 测试 FFmpeg 兼容性 - if not test_ffmpeg_compatibility(): - return - - # 获取测试视频路径 - if len(sys.argv) > 1: - video_path = sys.argv[1] - else: - # 尝试找到项目中的测试视频 - possible_paths = [ - "./resource/videos/test.mp4", - "./storage/videos/test.mp4", - "./test_video.mp4" - ] - - video_path = None - for path in possible_paths: - if os.path.exists(path): - video_path = path - break - - if not video_path: - print("\n❌ 未找到测试视频文件") - print("请提供视频文件路径作为参数:") - print(f"python {sys.argv[0]} ") - return - - # 执行测试 - success = test_video_extraction(video_path) - - print("\n" + "=" * 60) - if success: - print("🎉 测试成功！关键帧提取功能正常工作") - print("💡 建议：如果之前遇到问题，现在应该已经修复") - else: - print("❌ 测试失败！可能需要进一步调试") - print("💡 建议：") - print(" 1. 检查视频文件是否损坏") - print(" 2. 尝试更新显卡驱动") - print(" 3. 检查 FFmpeg 版本是否过旧") - print("=" * 60) - - -if __name__ == "__main__": - main()