345 lines
14 KiB
Python
345 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
场景检测命令模块
|
||
"""
|
||
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
import typer
|
||
|
||
from python_core.cli.const import progress_reporter, console
|
||
from json import dumps
|
||
# 创建场景检测命令组
|
||
scene_app = typer.Typer(help="🎯 场景检测工具")
|
||
|
||
@scene_app.command()
|
||
def detect(
|
||
video_path: Path = typer.Argument(..., help="📹 视频文件路径", exists=True),
|
||
detector: str = typer.Option("content", help="🔧 检测器类型 (content/threshold/adaptive)"),
|
||
threshold: float = typer.Option(30.0, help="🎚️ 检测阈值 (0-100)"),
|
||
min_scene_length: float = typer.Option(1.0, help="⏱️ 最小场景长度（秒）"),
|
||
output: Optional[Path] = typer.Option(None, "--output", "-o", help="📄 输出文件路径"),
|
||
format: str = typer.Option("json", help="📋 输出格式 (json/csv/txt)")
|
||
):
|
||
"""🎯 检测单个视频的场景"""
|
||
try:
|
||
from python_core.cli.scene_detect import detector as scene_detector, DetectorType, OutputFormat
|
||
|
||
# 验证参数
|
||
try:
|
||
detector_type = DetectorType(detector)
|
||
except ValueError:
|
||
progress_reporter.error(f"❌ 无效的检测器类型: {detector}")
|
||
progress_reporter.info("💡 可用类型: content, threshold, adaptive")
|
||
raise typer.Exit(1)
|
||
|
||
try:
|
||
output_format = OutputFormat(format)
|
||
except ValueError:
|
||
progress_reporter.error(f"❌ 无效的输出格式: {format}")
|
||
progress_reporter.info("💡 可用格式: json, csv, txt")
|
||
raise typer.Exit(1)
|
||
|
||
# 执行检测
|
||
result = scene_detector.detect_scenes(
|
||
video_path, detector_type, threshold, min_scene_length
|
||
)
|
||
|
||
if not result.success:
|
||
progress_reporter.error(f"❌ 检测失败: {result.error}")
|
||
raise typer.Exit(1)
|
||
|
||
# 显示结果摘要
|
||
console.print(f"📊 检测结果摘要:")
|
||
console.print(f" 文件: {result.filename}")
|
||
console.print(f" 检测器: {result.detector_type}")
|
||
console.print(f" 阈值: {result.threshold}")
|
||
console.print(f" 场景数: {result.total_scenes}")
|
||
console.print(f" 总时长: {result.total_duration:.2f}秒")
|
||
console.print(f" 检测时间: {result.detection_time:.2f}秒")
|
||
|
||
# 显示场景详情
|
||
if result.scenes:
|
||
console.print(f"\n🎬 场景列表:")
|
||
for scene in result.scenes[:10]: # 只显示前10个场景
|
||
console.print(f" 场景 {scene.index}: {scene.start_time:.2f}s - {scene.end_time:.2f}s ({scene.duration:.2f}s)")
|
||
|
||
if len(result.scenes) > 10:
|
||
console.print(f" ... 还有 {len(result.scenes) - 10} 个场景")
|
||
|
||
# 保存结果
|
||
if output:
|
||
scene_detector.save_results(result, output, output_format)
|
||
progress_reporter.success(f"📄 结果已保存到: {output}")
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
progress_reporter.error(f"❌ 命令执行失败: {e}")
|
||
raise typer.Exit(1)
|
||
|
||
@scene_app.command()
|
||
def batch_detect(
|
||
input_directory: Path = typer.Argument(..., help="📁 输入目录路径", exists=True),
|
||
detector: str = typer.Option("content", help="🔧 检测器类型 (content/threshold/adaptive)"),
|
||
threshold: float = typer.Option(30.0, help="🎚️ 检测阈值 (0-100)"),
|
||
recursive: bool = typer.Option(False, "--recursive", "-r", help="🔄 递归扫描子目录"),
|
||
output: Optional[Path] = typer.Option(None, "--output", "-o", help="📄 输出文件路径"),
|
||
format: str = typer.Option("json", help="📋 输出格式 (json/csv/txt)")
|
||
):
|
||
"""📦 批量检测目录中的所有视频"""
|
||
try:
|
||
from python_core.cli.scene_detect import detector as scene_detector, DetectorType, OutputFormat
|
||
|
||
# 验证参数
|
||
try:
|
||
detector_type = DetectorType(detector)
|
||
output_format = OutputFormat(format)
|
||
except ValueError as e:
|
||
progress_reporter.error(f"❌ 参数错误: {e}")
|
||
raise typer.Exit(1)
|
||
|
||
# 执行批量检测
|
||
results = scene_detector.batch_detect(
|
||
input_directory, detector_type, threshold, recursive
|
||
)
|
||
|
||
if not results:
|
||
progress_reporter.warning("⚠️ 没有检测到任何视频文件")
|
||
return
|
||
|
||
# 统计结果
|
||
successful = len([r for r in results if r.success])
|
||
failed = len(results) - successful
|
||
total_scenes = sum(r.total_scenes for r in results if r.success)
|
||
total_duration = sum(r.total_duration for r in results if r.success)
|
||
|
||
console.print(f"📊 批量检测结果:")
|
||
console.print(f" 总文件数: {len(results)}")
|
||
console.print(f" 成功: {successful}")
|
||
console.print(f" 失败: {failed}")
|
||
console.print(f" 总场景数: {total_scenes}")
|
||
console.print(f" 总时长: {total_duration:.2f}秒")
|
||
|
||
# 显示详细的场景信息
|
||
console.print(f"\n🎬 详细场景信息:")
|
||
for result in results:
|
||
if result.success and result.scenes:
|
||
console.print(f"\n📹 {result.filename} ({result.total_scenes} 个场景):")
|
||
for scene in result.scenes:
|
||
console.print(f" 场景 {scene.index}: {scene.start_time:.2f}s - {scene.end_time:.2f}s (时长: {scene.duration:.2f}s)")
|
||
elif result.success:
|
||
console.print(f"\n📹 {result.filename}: 无场景数据")
|
||
|
||
# 显示失败的文件
|
||
failed_files = [r for r in results if not r.success]
|
||
if failed_files:
|
||
console.print(f"\n❌ 失败的文件:")
|
||
for result in failed_files[:5]: # 只显示前5个失败文件
|
||
console.print(f" {result.filename}: {result.error}")
|
||
|
||
if len(failed_files) > 5:
|
||
console.print(f" ... 还有 {len(failed_files) - 5} 个失败文件")
|
||
|
||
# 保存结果
|
||
if output:
|
||
scene_detector.save_results(results, output, output_format)
|
||
progress_reporter.success(f"📄 结果已保存到: {output}")
|
||
|
||
return results
|
||
|
||
except Exception as e:
|
||
progress_reporter.error(f"❌ 批量检测失败: {e}")
|
||
raise typer.Exit(1)
|
||
|
||
@scene_app.command()
|
||
def compare(
|
||
video_path: Path = typer.Argument(..., help="📹 视频文件路径", exists=True),
|
||
thresholds: str = typer.Option("20,30,40", help="🎚️ 测试阈值列表（逗号分隔）"),
|
||
output: Optional[Path] = typer.Option(None, "--output", "-o", help="📄 输出文件路径")
|
||
):
|
||
"""🔬 比较不同检测器的效果"""
|
||
try:
|
||
from python_core.cli.scene_detect import detector as scene_detector
|
||
|
||
# 解析阈值列表
|
||
try:
|
||
threshold_list = [float(t.strip()) for t in thresholds.split(",")]
|
||
except ValueError:
|
||
progress_reporter.error("❌ 无效的阈值格式，请使用逗号分隔的数字")
|
||
raise typer.Exit(1)
|
||
|
||
# 执行比较
|
||
result = scene_detector.compare_detectors(video_path, threshold_list)
|
||
|
||
# 显示分析结果
|
||
analysis = result["analysis"]
|
||
console.print(f"🔬 检测器比较结果:")
|
||
console.print(f" 视频: {Path(result['video_path']).name}")
|
||
console.print(f" 总测试数: {result['total_tests']}")
|
||
console.print(f" 成功测试数: {analysis['total_successful']}")
|
||
console.print(f" 推荐检测器: {analysis['best_detector']}")
|
||
console.print(f" 建议: {analysis['recommendation']}")
|
||
|
||
# 显示详细分析
|
||
console.print(f"\n📊 各检测器表现:")
|
||
for detector_name, stats in analysis["detector_analysis"].items():
|
||
console.print(f" 🔧 {detector_name}:")
|
||
console.print(f" 平均场景数: {stats['average_scenes']:.1f}")
|
||
console.print(f" 平均检测时间: {stats['average_detection_time']:.2f}秒")
|
||
console.print(f" 测试次数: {stats['test_count']}")
|
||
|
||
# 显示详细测试结果
|
||
console.print(f"\n🧪 详细测试结果:")
|
||
for test_result in result["results"]:
|
||
if test_result["success"]:
|
||
console.print(f" {test_result['detector']} (阈值: {test_result['threshold']}): "
|
||
f"{test_result['scenes']} 场景, {test_result['detection_time']:.2f}s")
|
||
else:
|
||
console.print(f" {test_result['detector']} (阈值: {test_result['threshold']}): "
|
||
f"❌ {test_result['error']}")
|
||
|
||
# 保存结果
|
||
if output:
|
||
scene_detector.save_results(result, output)
|
||
progress_reporter.success(f"📄 结果已保存到: {output}")
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
progress_reporter.error(f"❌ 比较测试失败: {e}")
|
||
raise typer.Exit(1)
|
||
|
||
@scene_app.command()
|
||
def split(
|
||
video_path: Path = typer.Argument(..., help="📹 视频文件路径", exists=True),
|
||
detector: str = typer.Option("content", help="🔧 检测器类型 (content/threshold/adaptive)"),
|
||
threshold: float = typer.Option(30.0, help="🎚️ 检测阈值 (0-100)"),
|
||
output_dir: Optional[Path] = typer.Option(None, "--output-dir", "-d", help="📁 输出目录"),
|
||
filename_template: str = typer.Option("scene_{:03d}.mp4", help="📝 文件名模板")
|
||
):
|
||
"""✂️ 根据场景检测结果分割视频"""
|
||
try:
|
||
from python_core.cli.scene_detect import detector as scene_detector, DetectorType
|
||
from scenedetect.video_splitter import split_video_ffmpeg
|
||
|
||
# 验证参数
|
||
try:
|
||
detector_type = DetectorType(detector)
|
||
except ValueError:
|
||
progress_reporter.error(f"❌ 无效的检测器类型: {detector}")
|
||
raise typer.Exit(1)
|
||
|
||
# 设置输出目录
|
||
if output_dir is None:
|
||
output_dir = video_path.parent / f"{video_path.stem}_scenes"
|
||
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 先检测场景
|
||
progress_reporter.info("🎯 正在检测场景...")
|
||
result = scene_detector.detect_scenes(video_path, detector_type, threshold)
|
||
|
||
if not result.success:
|
||
progress_reporter.error(f"❌ 场景检测失败: {result.error}")
|
||
raise typer.Exit(1)
|
||
|
||
if not result.scenes:
|
||
progress_reporter.warning("⚠️ 未检测到任何场景")
|
||
return
|
||
|
||
# 构建场景列表（PySceneDetect格式）
|
||
from scenedetect import FrameTimecode
|
||
scene_list = []
|
||
|
||
# 假设视频帧率（实际应该从视频中获取）
|
||
fps = 25.0 # 默认帧率，实际使用时应该从视频文件中获取
|
||
|
||
for scene in result.scenes:
|
||
start_tc = FrameTimecode(timecode=scene.start_time, fps=fps)
|
||
end_tc = FrameTimecode(timecode=scene.end_time, fps=fps)
|
||
scene_list.append((start_tc, end_tc))
|
||
|
||
# 分割视频
|
||
progress_reporter.info(f"✂️ 正在分割视频到 {len(scene_list)} 个场景...")
|
||
|
||
try:
|
||
split_video_ffmpeg(
|
||
input_video_path=str(video_path),
|
||
scene_list=scene_list,
|
||
output_file_template=str(output_dir / filename_template),
|
||
video_name=video_path.stem,
|
||
arg_override=None,
|
||
hide_progress=False
|
||
)
|
||
|
||
progress_reporter.success(f"✅ 视频分割完成，输出到: {output_dir}")
|
||
console.print(f"📁 输出目录: {output_dir}")
|
||
console.print(f"🎬 场景数量: {len(scene_list)}")
|
||
|
||
# 列出生成的文件
|
||
output_files = list(output_dir.glob("*.mp4"))
|
||
if output_files:
|
||
console.print(f"\n📄 生成的文件:")
|
||
for file_path in sorted(output_files)[:10]:
|
||
console.print(f" {file_path.name}")
|
||
|
||
if len(output_files) > 10:
|
||
console.print(f" ... 还有 {len(output_files) - 10} 个文件")
|
||
|
||
except Exception as e:
|
||
progress_reporter.error(f"❌ 视频分割失败: {e}")
|
||
raise typer.Exit(1)
|
||
|
||
except Exception as e:
|
||
progress_reporter.error(f"❌ 分割命令执行失败: {e}")
|
||
raise typer.Exit(1)
|
||
|
||
@scene_app.command()
|
||
def info(
|
||
video_path: Path = typer.Argument(..., help="📹 视频文件路径", exists=True)
|
||
):
|
||
"""📋 显示视频基本信息"""
|
||
try:
|
||
import cv2
|
||
|
||
progress_reporter.info(f"📋 获取视频信息: {video_path.name}")
|
||
|
||
# 使用OpenCV获取视频信息
|
||
cap = cv2.VideoCapture(str(video_path))
|
||
|
||
if not cap.isOpened():
|
||
raise Exception("无法打开视频文件")
|
||
|
||
# 获取视频信息
|
||
fps = cap.get(cv2.CAP_PROP_FPS)
|
||
frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
|
||
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
|
||
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
|
||
duration = frame_count / fps if fps > 0 else 0
|
||
resolution = (width, height)
|
||
|
||
cap.release()
|
||
|
||
# 显示信息
|
||
console.print(f"📹 视频信息:")
|
||
console.print(f" 文件名: {video_path.name}")
|
||
console.print(f" 文件大小: {video_path.stat().st_size / (1024*1024):.2f} MB")
|
||
console.print(f" 分辨率: {resolution[0]}x{resolution[1]}")
|
||
console.print(f" 帧率: {fps:.2f} fps")
|
||
console.print(f" 总帧数: {frame_count}")
|
||
console.print(f" 时长: {duration:.2f}秒 ({duration//60:.0f}分{duration%60:.0f}秒)")
|
||
|
||
progress_reporter.success(dumps({
|
||
"filename": video_path.name,
|
||
"file_size_mb": video_path.stat().st_size / (1024*1024),
|
||
"resolution": resolution,
|
||
"fps": fps,
|
||
"frame_count": frame_count,
|
||
"duration": duration
|
||
}))
|
||
|
||
except Exception as e:
|
||
progress_reporter.error(f"❌ 获取视频信息失败: {e}")
|
||
raise typer.Exit(1)
|