362 lines
13 KiB
Python
362 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
测试重构后的媒体管理器
|
|
"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# 添加项目根目录到Python路径
|
|
project_root = Path(__file__).parent.parent
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
def test_imports():
|
|
"""测试模块导入"""
|
|
print("🔍 测试模块导入")
|
|
print("=" * 50)
|
|
|
|
try:
|
|
# 测试数据类型导入
|
|
from python_core.services.media_manager.types import (
|
|
VideoSegment, OriginalVideo, VideoInfo, UploadResult, BatchUploadResult
|
|
)
|
|
print("✅ 数据类型导入成功")
|
|
|
|
# 测试组件导入
|
|
from python_core.services.media_manager.video_info import (
|
|
VideoInfoExtractor, create_video_info_extractor
|
|
)
|
|
from python_core.services.media_manager.scene_detector import (
|
|
SceneDetector, create_scene_detector
|
|
)
|
|
from python_core.services.media_manager.video_processor import (
|
|
VideoProcessor, create_video_processor
|
|
)
|
|
from python_core.services.media_manager.storage import MediaStorage
|
|
print("✅ 组件导入成功")
|
|
|
|
# 测试主要类导入
|
|
from python_core.services.media_manager.manager import MediaManager, get_media_manager
|
|
from python_core.services.media_manager.cli import MediaManagerCommander
|
|
print("✅ 主要类导入成功")
|
|
|
|
# 测试统一导入
|
|
from python_core.services.media_manager import (
|
|
VideoSegment, MediaManager, MediaManagerCommander
|
|
)
|
|
print("✅ 统一导入成功")
|
|
|
|
return True
|
|
|
|
except ImportError as e:
|
|
print(f"❌ 导入失败: {e}")
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ 测试失败: {e}")
|
|
return False
|
|
|
|
def test_video_info_extractor():
|
|
"""测试视频信息提取器"""
|
|
print("\n📹 测试视频信息提取器")
|
|
print("=" * 50)
|
|
|
|
try:
|
|
from python_core.services.media_manager.video_info import create_video_info_extractor
|
|
|
|
# 创建提取器
|
|
extractor = create_video_info_extractor()
|
|
print(f"✅ 视频信息提取器创建成功: {type(extractor).__name__}")
|
|
|
|
# 查找测试视频
|
|
assets_dir = project_root / "assets"
|
|
video_files = list(assets_dir.rglob("*.mp4"))
|
|
|
|
if video_files:
|
|
test_video = str(video_files[0])
|
|
print(f"📹 测试视频: {test_video}")
|
|
|
|
try:
|
|
video_info = extractor.extract_video_info(test_video)
|
|
print(f"✅ 视频信息提取成功:")
|
|
print(f" 时长: {video_info.duration:.2f}秒")
|
|
print(f" 分辨率: {video_info.width}x{video_info.height}")
|
|
print(f" 帧率: {video_info.fps:.2f}")
|
|
print(f" 文件大小: {video_info.file_size} bytes")
|
|
except Exception as e:
|
|
print(f"⚠️ 视频信息提取失败: {e}")
|
|
else:
|
|
print("⚠️ 没有找到测试视频，跳过功能测试")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ 视频信息提取器测试失败: {e}")
|
|
return False
|
|
|
|
def test_scene_detector():
|
|
"""测试场景检测器"""
|
|
print("\n🎬 测试场景检测器")
|
|
print("=" * 50)
|
|
|
|
try:
|
|
from python_core.services.media_manager.scene_detector import create_scene_detector
|
|
|
|
# 创建检测器
|
|
detector = create_scene_detector()
|
|
print(f"✅ 场景检测器创建成功: {type(detector).__name__}")
|
|
|
|
# 查找测试视频
|
|
assets_dir = project_root / "assets"
|
|
video_files = list(assets_dir.rglob("*.mp4"))
|
|
|
|
if video_files:
|
|
test_video = str(video_files[0])
|
|
print(f"📹 测试视频: {test_video}")
|
|
|
|
try:
|
|
scene_changes = detector.detect_scenes(test_video, threshold=30.0)
|
|
print(f"✅ 场景检测成功:")
|
|
print(f" 检测到 {len(scene_changes)-1} 个场景变化")
|
|
print(f" 场景时间点: {[f'{t:.2f}s' for t in scene_changes[:5]]}")
|
|
except Exception as e:
|
|
print(f"⚠️ 场景检测失败: {e}")
|
|
else:
|
|
print("⚠️ 没有找到测试视频，跳过功能测试")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ 场景检测器测试失败: {e}")
|
|
return False
|
|
|
|
def test_media_storage():
|
|
"""测试媒体存储"""
|
|
print("\n💾 测试媒体存储")
|
|
print("=" * 50)
|
|
|
|
try:
|
|
from python_core.services.media_manager.storage import MediaStorage
|
|
from python_core.services.media_manager.types import VideoSegment, OriginalVideo
|
|
|
|
# 创建存储管理器
|
|
storage = MediaStorage()
|
|
print("✅ 媒体存储创建成功")
|
|
|
|
# 测试加载数据
|
|
segments = storage.load_video_segments()
|
|
videos = storage.load_original_videos()
|
|
print(f"✅ 数据加载成功: {len(segments)} 个片段, {len(videos)} 个视频")
|
|
|
|
# 测试搜索功能
|
|
if segments:
|
|
# 测试标签搜索
|
|
test_segments = storage.get_segments_by_tags(segments, ["分镜"], match_all=False)
|
|
print(f"✅ 标签搜索测试: 找到 {len(test_segments)} 个分镜片段")
|
|
|
|
# 测试关键词搜索
|
|
keyword_segments = storage.search_segments(segments, "mp4")
|
|
print(f"✅ 关键词搜索测试: 找到 {len(keyword_segments)} 个mp4片段")
|
|
|
|
# 测试热门片段
|
|
popular_segments = storage.get_popular_segments(segments, limit=5)
|
|
print(f"✅ 热门片段测试: 找到 {len(popular_segments)} 个热门片段")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ 媒体存储测试失败: {e}")
|
|
return False
|
|
|
|
def test_media_manager():
|
|
"""测试媒体管理器"""
|
|
print("\n🎛️ 测试媒体管理器")
|
|
print("=" * 50)
|
|
|
|
try:
|
|
from python_core.services.media_manager.manager import get_media_manager
|
|
|
|
# 获取全局实例
|
|
manager = get_media_manager()
|
|
print("✅ 媒体管理器创建成功")
|
|
|
|
# 测试查询方法
|
|
all_segments = manager.get_all_segments()
|
|
all_videos = manager.get_all_original_videos()
|
|
print(f"✅ 查询测试: {len(all_segments)} 个片段, {len(all_videos)} 个视频")
|
|
|
|
# 测试搜索方法
|
|
if all_segments:
|
|
search_results = manager.search_segments("mp4")
|
|
print(f"✅ 搜索测试: 找到 {len(search_results)} 个结果")
|
|
|
|
tag_results = manager.get_segments_by_tags(["分镜"])
|
|
print(f"✅ 标签搜索测试: 找到 {len(tag_results)} 个分镜片段")
|
|
|
|
popular_results = manager.get_popular_segments(limit=3)
|
|
print(f"✅ 热门片段测试: 找到 {len(popular_results)} 个热门片段")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ 媒体管理器测试失败: {e}")
|
|
return False
|
|
|
|
def test_commander():
|
|
"""测试命令行接口"""
|
|
print("\n⌨️ 测试命令行接口")
|
|
print("=" * 50)
|
|
|
|
try:
|
|
from python_core.services.media_manager.cli import MediaManagerCommander
|
|
|
|
# 创建Commander
|
|
commander = MediaManagerCommander()
|
|
print("✅ 媒体管理器Commander创建成功")
|
|
|
|
# 检查注册的命令
|
|
commands = list(commander.commands.keys())
|
|
expected_commands = [
|
|
"upload", "batch_upload", "get_all_segments", "get_all_videos",
|
|
"search_segments", "get_segments_by_tags", "add_tags", "delete_segment"
|
|
]
|
|
|
|
for cmd in expected_commands:
|
|
if cmd in commands:
|
|
print(f"✅ 命令 '{cmd}' 已注册")
|
|
else:
|
|
print(f"❌ 命令 '{cmd}' 未注册")
|
|
return False
|
|
|
|
# 测试参数解析
|
|
test_args = ["get_all_segments"]
|
|
try:
|
|
command, parsed_args = commander.parse_arguments(test_args)
|
|
print(f"✅ 参数解析成功: {command}")
|
|
except Exception as e:
|
|
print(f"❌ 参数解析失败: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ Commander测试失败: {e}")
|
|
return False
|
|
|
|
def test_file_structure():
|
|
"""测试文件结构"""
|
|
print("\n📁 测试文件结构")
|
|
print("=" * 50)
|
|
|
|
try:
|
|
# 检查新的模块文件
|
|
module_files = [
|
|
"python_core/services/media_manager/__init__.py",
|
|
"python_core/services/media_manager/types.py",
|
|
"python_core/services/media_manager/video_info.py",
|
|
"python_core/services/media_manager/scene_detector.py",
|
|
"python_core/services/media_manager/video_processor.py",
|
|
"python_core/services/media_manager/storage.py",
|
|
"python_core/services/media_manager/manager.py",
|
|
"python_core/services/media_manager/cli.py"
|
|
]
|
|
|
|
for file_path in module_files:
|
|
full_path = project_root / file_path
|
|
if full_path.exists():
|
|
lines = len(full_path.read_text().splitlines())
|
|
print(f"✅ {file_path} 存在 ({lines} 行)")
|
|
else:
|
|
print(f"❌ {file_path} 不存在")
|
|
return False
|
|
|
|
# 检查原文件是否已删除
|
|
old_file = project_root / "python_core/services/media_manager.py"
|
|
if not old_file.exists():
|
|
print("✅ 原始大文件已删除")
|
|
else:
|
|
print("⚠️ 原始大文件仍然存在")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ 文件结构测试失败: {e}")
|
|
return False
|
|
|
|
def main():
|
|
"""主函数"""
|
|
print("🚀 测试重构后的媒体管理器")
|
|
|
|
try:
|
|
# 运行所有测试
|
|
tests = [
|
|
test_imports,
|
|
test_video_info_extractor,
|
|
test_scene_detector,
|
|
test_media_storage,
|
|
test_media_manager,
|
|
test_commander,
|
|
test_file_structure
|
|
]
|
|
|
|
results = []
|
|
for test in tests:
|
|
try:
|
|
result = test()
|
|
results.append(result)
|
|
except Exception as e:
|
|
print(f"❌ 测试 {test.__name__} 异常: {e}")
|
|
results.append(False)
|
|
|
|
# 总结
|
|
print("\n" + "=" * 60)
|
|
print("📊 媒体管理器重构测试总结")
|
|
print("=" * 60)
|
|
|
|
passed = sum(results)
|
|
total = len(results)
|
|
|
|
print(f"通过测试: {passed}/{total}")
|
|
|
|
if passed == total:
|
|
print("🎉 所有重构测试通过！")
|
|
print("\n✅ 重构成果:")
|
|
print(" 1. 模块化设计 - 8个专门的模块文件")
|
|
print(" 2. 单一职责 - 每个模块功能明确")
|
|
print(" 3. 代码简化 - 从1506行减少到~200行/文件")
|
|
print(" 4. 易于维护 - 修改影响范围小")
|
|
print(" 5. 可测试性 - 每个组件可独立测试")
|
|
|
|
print("\n📁 新的模块结构:")
|
|
print(" media_manager/")
|
|
print(" ├── types.py - 数据类型定义")
|
|
print(" ├── video_info.py - 视频信息提取")
|
|
print(" ├── scene_detector.py - 场景检测")
|
|
print(" ├── video_processor.py - 视频处理")
|
|
print(" ├── storage.py - 存储管理")
|
|
print(" ├── manager.py - 主要管理器")
|
|
print(" ├── cli.py - 命令行接口")
|
|
print(" └── __init__.py - 统一导入")
|
|
|
|
print("\n🎯 使用方式:")
|
|
print(" # 简单使用")
|
|
print(" from python_core.services.media_manager import MediaManager")
|
|
print(" manager = MediaManager()")
|
|
print(" # 命令行使用")
|
|
print(" from python_core.services.media_manager import MediaManagerCommander")
|
|
print(" commander = MediaManagerCommander()")
|
|
|
|
return 0
|
|
else:
|
|
print("⚠️ 部分重构测试失败")
|
|
return 1
|
|
|
|
except Exception as e:
|
|
print(f"❌ 测试过程中出错: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return 1
|
|
|
|
if __name__ == "__main__":
|
|
exit_code = main()
|
|
sys.exit(exit_code)
|