mirror of
https://github.com/linyqh/NarratoAI.git
synced 2025-12-11 10:32:49 +00:00
删除视频关键帧提取测试脚本,优化视频处理器中的提取逻辑,增加超级兼容性方案以解决Windows系统的MJPEG编码问题。更新了软件方案的提取命令,增强了错误处理和调试信息,提升了整体兼容性和用户体验。
This commit is contained in:
parent
6715c29057
commit
6270224d45
@ -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} - 命令执行超时")
|
||||
|
||||
@ -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]} <video_path>")
|
||||
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()
|
||||
Loading…
x
Reference in New Issue
Block a user