fix: 修复命令行工具

This commit is contained in:
root
2025-07-12 13:44:57 +08:00
parent 81035caf0e
commit 31de1e5a4d
30 changed files with 2728 additions and 1755 deletions

View File

@@ -0,0 +1,15 @@
#!/usr/bin/env python3
"""
Scene Detection Utils
场景检测工具类
导出所有工具类
"""
from .result_saver import ResultSaver
from .validators import InputValidator
__all__ = [
"ResultSaver",
"InputValidator"
]

View File

@@ -0,0 +1,110 @@
#!/usr/bin/env python3
"""
Result Saver
结果保存工具
"""
import json
import csv
from pathlib import Path
from dataclasses import asdict
from python_core.utils.logger import logger
from ..types import DetectionResult, OutputFormat
class ResultSaver:
"""结果保存器"""
def save_results(self, result: DetectionResult, output_path: Path,
output_format: OutputFormat) -> None:
"""保存检测结果"""
try:
if output_format == OutputFormat.JSON:
self._save_json(result, output_path)
elif output_format == OutputFormat.CSV:
self._save_csv(result, output_path)
elif output_format == OutputFormat.TXT:
self._save_txt(result, output_path)
else:
raise ValueError(f"不支持的输出格式: {output_format}")
logger.info(f"💾 结果已保存到: {output_path}")
except Exception as e:
logger.error(f"保存结果失败: {e}")
raise
def _save_json(self, result: DetectionResult, output_path: Path) -> None:
"""保存为JSON格式"""
# 确保输出目录存在
output_path.parent.mkdir(parents=True, exist_ok=True)
# 转换为字典
result_dict = asdict(result)
# 保存JSON文件
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(result_dict, f, indent=2, ensure_ascii=False)
def _save_csv(self, result: DetectionResult, output_path: Path) -> None:
"""保存为CSV格式"""
# 确保输出目录存在
output_path.parent.mkdir(parents=True, exist_ok=True)
with open(output_path, 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
# 写入头部信息
writer.writerow(['# 场景检测结果'])
writer.writerow(['# 文件名', result.filename])
writer.writerow(['# 检测器', result.detector_type])
writer.writerow(['# 阈值', result.threshold])
writer.writerow(['# 总场景数', result.total_scenes])
writer.writerow(['# 总时长', f"{result.total_duration:.2f}"])
writer.writerow(['# 检测时间', f"{result.detection_time:.2f}"])
writer.writerow([])
# 写入场景数据
writer.writerow(['场景索引', '开始时间(秒)', '结束时间(秒)', '时长(秒)', '开始帧', '结束帧'])
for scene in result.scenes:
writer.writerow([
scene.index,
f"{scene.start_time:.2f}",
f"{scene.end_time:.2f}",
f"{scene.duration:.2f}",
scene.start_frame,
scene.end_frame
])
def _save_txt(self, result: DetectionResult, output_path: Path) -> None:
"""保存为TXT格式"""
# 确保输出目录存在
output_path.parent.mkdir(parents=True, exist_ok=True)
with open(output_path, 'w', encoding='utf-8') as f:
f.write("场景检测结果\n")
f.write("=" * 50 + "\n\n")
# 基本信息
f.write(f"文件名: {result.filename}\n")
f.write(f"检测器: {result.detector_type}\n")
f.write(f"阈值: {result.threshold}\n")
f.write(f"总场景数: {result.total_scenes}\n")
f.write(f"总时长: {result.total_duration:.2f}\n")
f.write(f"检测时间: {result.detection_time:.2f}\n")
f.write(f"检测状态: {'成功' if result.success else '失败'}\n")
if result.error:
f.write(f"错误信息: {result.error}\n")
f.write("\n场景详情:\n")
f.write("-" * 50 + "\n")
# 场景列表
for scene in result.scenes:
f.write(f"场景 {scene.index + 1:2d}: ")
f.write(f"{scene.start_time:7.2f}s - {scene.end_time:7.2f}s ")
f.write(f"(时长: {scene.duration:6.2f}s, ")
f.write(f"帧: {scene.start_frame:5d} - {scene.end_frame:5d})\n")

View File

@@ -0,0 +1,84 @@
#!/usr/bin/env python3
"""
Input Validators
输入验证器
"""
from pathlib import Path
from typing import List, Set
from ..types import DetectorType, OutputFormat
class InputValidator:
"""输入验证器"""
def __init__(self, supported_formats: Set[str]):
self.supported_formats = supported_formats
def validate_video_path(self, video_path: Path) -> List[str]:
"""验证视频路径"""
errors = []
if not video_path.exists():
errors.append(f"视频文件不存在: {video_path}")
if video_path.suffix.lower() not in self.supported_formats:
errors.append(f"不支持的文件格式: {video_path.suffix}")
return errors
def validate_detector_type(self, detector_type: str) -> List[str]:
"""验证检测器类型"""
errors = []
try:
DetectorType(detector_type)
except ValueError:
valid_types = [dt.value for dt in DetectorType]
errors.append(f"无效的检测器类型: {detector_type},支持的类型: {valid_types}")
return errors
def validate_threshold(self, threshold: float) -> List[str]:
"""验证阈值"""
errors = []
if not (0 <= threshold <= 100):
errors.append(f"阈值超出范围 (0-100): {threshold}")
return errors
def validate_min_scene_length(self, min_scene_length: float) -> List[str]:
"""验证最小场景长度"""
errors = []
if min_scene_length < 0:
errors.append(f"最小场景长度不能为负数: {min_scene_length}")
return errors
def validate_output_format(self, output_format: str) -> List[str]:
"""验证输出格式"""
errors = []
try:
OutputFormat(output_format)
except ValueError:
valid_formats = [of.value for of in OutputFormat]
errors.append(f"无效的输出格式: {output_format},支持的格式: {valid_formats}")
return errors
def validate_all(self, video_path: Path, detector_type: str, threshold: float,
min_scene_length: float, output_format: str) -> List[str]:
"""验证所有输入参数"""
errors = []
errors.extend(self.validate_video_path(video_path))
errors.extend(self.validate_detector_type(detector_type))
errors.extend(self.validate_threshold(threshold))
errors.extend(self.validate_min_scene_length(min_scene_length))
errors.extend(self.validate_output_format(output_format))
return errors