diff --git a/examples/jsonrpc_client_demo.py b/examples/jsonrpc_client_demo.py deleted file mode 100644 index 4d638e6..0000000 --- a/examples/jsonrpc_client_demo.py +++ /dev/null @@ -1,279 +0,0 @@ -#!/usr/bin/env python3 -""" -JSON-RPC Client Demo -JSON-RPC 客户端演示 - -演示如何使用Python客户端调用场景检测JSON-RPC API -""" - -import json -import requests -import time -from typing import Dict, Any, Optional - - -class SceneDetectionClient: - """场景检测JSON-RPC客户端""" - - def __init__(self, server_url: str = "http://localhost:8080"): - self.server_url = server_url - self.request_id = 0 - - def _call_method(self, method: str, params: Dict[str, Any]) -> Dict[str, Any]: - """调用JSON-RPC方法""" - self.request_id += 1 - - payload = { - "jsonrpc": "2.0", - "method": method, - "params": params, - "id": self.request_id - } - - try: - response = requests.post( - self.server_url, - json=payload, - headers={'Content-Type': 'application/json'}, - timeout=60 - ) - - if response.status_code == 200: - return response.json() - else: - return { - "error": { - "code": response.status_code, - "message": f"HTTP Error: {response.text}" - } - } - - except requests.exceptions.RequestException as e: - return { - "error": { - "code": -1, - "message": f"Request failed: {str(e)}" - } - } - - def detect_scenes(self, video_path: str, detector_type: str = "content", - threshold: float = 30.0, min_scene_length: float = 1.0) -> Dict[str, Any]: - """基础场景检测""" - params = { - "video_path": video_path, - "detector_type": detector_type, - "threshold": threshold, - "min_scene_length": min_scene_length - } - - return self._call_method("scene.detect", params) - - def detect_scenes_workflow(self, video_path: str, detector_type: str = "content", - threshold: float = 30.0, min_scene_length: float = 1.0, - output_path: Optional[str] = None, output_format: str = "json", - enable_ai_analysis: bool = True) -> Dict[str, Any]: - """LangGraph工作流场景检测""" - params = { - "video_path": video_path, - "detector_type": detector_type, - "threshold": threshold, - "min_scene_length": min_scene_length, - "enable_ai_analysis": enable_ai_analysis - } - - if output_path: - params["output_path"] = output_path - params["output_format"] = output_format - - return self._call_method("scene.detect_workflow", params) - - def get_video_info(self, video_path: str) -> Dict[str, Any]: - """获取视频信息""" - params = {"video_path": video_path} - return self._call_method("scene.get_video_info", params) - - def batch_detect(self, directory: str, detector_type: str = "content", - threshold: float = 30.0, min_scene_length: float = 1.0, - output_dir: Optional[str] = None, output_format: str = "json") -> Dict[str, Any]: - """批量场景检测""" - params = { - "directory": directory, - "detector_type": detector_type, - "threshold": threshold, - "min_scene_length": min_scene_length, - "output_format": output_format - } - - if output_dir: - params["output_dir"] = output_dir - - return self._call_method("scene.batch_detect", params) - - -def demo_basic_detection(): - """演示基础场景检测""" - print("🎯 基础场景检测演示") - print("=" * 50) - - client = SceneDetectionClient("http://localhost:8081") - - # 测试视频路径 - video_path = "assets/1/1752032011698.mp4" - - print(f"📹 检测视频: {video_path}") - - # 调用基础检测 - start_time = time.time() - result = client.detect_scenes(video_path, threshold=15.0) - end_time = time.time() - - if "error" in result: - print(f"❌ 检测失败: {result['error']}") - return - - detection_result = result.get("result", {}) - - if detection_result.get("success"): - print(f"✅ 检测成功!") - print(f" 场景数量: {detection_result['total_scenes']}") - print(f" 视频时长: {detection_result['total_duration']:.2f}秒") - print(f" 检测时间: {detection_result['detection_time']:.2f}秒") - print(f" API调用时间: {end_time - start_time:.2f}秒") - - # 显示场景详情 - scenes = detection_result.get("scenes", []) - print(f"\n🎬 场景详情:") - for scene in scenes[:5]: # 只显示前5个 - print(f" 场景 {scene['index']}: {scene['start_time']:.2f}s - {scene['end_time']:.2f}s") - - if len(scenes) > 5: - print(f" ... 还有 {len(scenes) - 5} 个场景") - else: - print(f"❌ 检测失败: {detection_result.get('error', '未知错误')}") - - -def demo_workflow_detection(): - """演示工作流场景检测""" - print("\n🔄 LangGraph工作流检测演示") - print("=" * 50) - - client = SceneDetectionClient("http://localhost:8081") - - # 测试视频路径 - video_path = "assets/1/1752032011698.mp4" - - print(f"📹 检测视频: {video_path}") - - # 调用工作流检测 - start_time = time.time() - result = client.detect_scenes_workflow( - video_path, - threshold=15.0, - enable_ai_analysis=False # 禁用AI分析以避免API密钥问题 - ) - end_time = time.time() - - if "error" in result: - print(f"❌ 工作流检测失败: {result['error']}") - return - - workflow_result = result.get("result", {}) - detection_result = workflow_result.get("detection_result", {}) - video_info = workflow_result.get("video_info", {}) - ai_analysis = workflow_result.get("ai_analysis") - workflow_state = workflow_result.get("workflow_state") - - if detection_result.get("success"): - print(f"✅ 工作流检测成功!") - print(f" 工作流状态: {workflow_state}") - print(f" 场景数量: {detection_result['total_scenes']}") - print(f" 检测时间: {detection_result['detection_time']:.2f}秒") - print(f" API调用时间: {end_time - start_time:.2f}秒") - - # 显示视频信息 - if video_info: - print(f"\n📹 视频信息:") - print(f" 分辨率: {video_info.get('resolution')}") - print(f" 帧率: {video_info.get('fps'):.2f} fps") - print(f" 时长: {video_info.get('duration'):.2f}秒") - - # 显示AI分析结果 - if ai_analysis: - print(f"\n🧠 AI分析: {ai_analysis}") - else: - print(f"❌ 工作流检测失败: {detection_result.get('error', '未知错误')}") - - -def demo_video_info(): - """演示获取视频信息""" - print("\n📊 视频信息获取演示") - print("=" * 50) - - client = SceneDetectionClient("http://localhost:8081") - - # 测试视频路径 - video_path = "assets/1/1752032011698.mp4" - - print(f"📹 获取视频信息: {video_path}") - - # 获取视频信息 - start_time = time.time() - result = client.get_video_info(video_path) - end_time = time.time() - - if "error" in result: - print(f"❌ 获取失败: {result['error']}") - return - - info_result = result.get("result", {}) - - if info_result.get("success"): - info = info_result.get("info", {}) - print(f"✅ 获取成功!") - print(f" 文件名: {info.get('filename')}") - print(f" 分辨率: {info.get('resolution')}") - print(f" 帧率: {info.get('fps'):.2f} fps") - print(f" 总帧数: {info.get('frame_count'):,}") - print(f" 时长: {info.get('duration'):.2f}秒") - print(f" 文件大小: {info.get('file_size'):,} 字节") - print(f" API调用时间: {end_time - start_time:.2f}秒") - else: - print(f"❌ 获取失败: {info_result.get('error', '未知错误')}") - - -def main(): - """主演示函数""" - print("🚀 JSON-RPC 场景检测客户端演示") - print("=" * 60) - - # 检查服务器连接 - client = SceneDetectionClient("http://localhost:8081") - - try: - # 简单的连接测试 - test_result = client.get_video_info("nonexistent.mp4") - if "error" in test_result and "Request failed" in str(test_result["error"]): - print("❌ 无法连接到JSON-RPC服务器") - print("💡 请先启动服务器: python3 -m python_core.cli jsonrpc start --port 8081") - return - except Exception as e: - print(f"❌ 连接测试失败: {e}") - return - - print("✅ 服务器连接正常") - - # 运行演示 - demo_video_info() - demo_basic_detection() - demo_workflow_detection() - - print("\n🎉 演示完成!") - print("\n💡 更多用法:") - print(" • 调整检测阈值以获得不同的场景分割效果") - print(" • 使用不同的检测器类型 (content/threshold/adaptive)") - print(" • 启用AI分析获得智能建议 (需要配置API密钥)") - print(" • 批量处理多个视频文件") - - -if __name__ == "__main__": - main() diff --git a/examples/test_jsonrpc_registration.py b/examples/test_jsonrpc_registration.py deleted file mode 100644 index 9a93029..0000000 --- a/examples/test_jsonrpc_registration.py +++ /dev/null @@ -1,86 +0,0 @@ -#!/usr/bin/env python3 -""" -Test JSON-RPC Registration -测试JSON-RPC方法注册 -""" - -import sys -from pathlib import Path - -# 添加项目根目录到Python路径 -project_root = Path(__file__).parent.parent -sys.path.insert(0, str(project_root)) - -from python_core.utils.jsonrpc_enhanced import method_registry -from python_core.scene_detection import SceneDetector - - -def test_method_registration(): - """测试方法注册""" - print("🧪 测试JSON-RPC方法注册...") - - # 创建检测器并注册方法 - detector = SceneDetector() - detector.register_jsonrpc_methods() - - # 检查注册的方法 - print(f"📋 已注册的方法: {list(method_registry.methods.keys())}") - - # 测试方法调用 - test_request = ''' - { - "jsonrpc": "2.0", - "method": "scene.get_video_info", - "params": { - "video_path": "assets/1/1752032011698.mp4" - }, - "id": 1 - } - ''' - - print("📤 测试方法调用...") - response = method_registry.handle_request(test_request) - - if response: - print("📥 收到响应:") - print(response) - else: - print("📥 收到空响应（异步模式）") - - return True - - -def test_direct_method_call(): - """测试直接方法调用""" - print("\n🧪 测试直接方法调用...") - - detector = SceneDetector() - - # 直接调用方法 - result = detector.jsonrpc_get_video_info("assets/1/1752032011698.mp4") - - print(f"📋 直接调用结果: {result}") - - return True - - -def main(): - """主函数""" - print("🚀 开始测试JSON-RPC方法注册") - print("=" * 50) - - try: - test_method_registration() - test_direct_method_call() - print("\n✅ 所有测试通过") - return True - except Exception as e: - print(f"\n❌ 测试失败: {e}") - import traceback - print(f"详细错误: {traceback.format_exc()}") - return False - - -if __name__ == "__main__": - success = main() - sys.exit(0 if success else 1) diff --git a/examples/test_jsonrpc_workflow.py b/examples/test_jsonrpc_workflow.py deleted file mode 100644 index 46eb9fe..0000000 --- a/examples/test_jsonrpc_workflow.py +++ /dev/null @@ -1,271 +0,0 @@ -#!/usr/bin/env python3 -""" -测试JSON-RPC工作流进度反馈 -Test JSON-RPC Workflow Progress Feedback -""" - -import json -import requests -import time -import threading -import websocket -from typing import Dict, Any - - -class JSONRPCWorkflowTester: - """JSON-RPC工作流测试器""" - - def __init__(self, server_url: str = "http://localhost:8081"): - self.server_url = server_url - self.request_id = 0 - self.progress_updates = [] - - def _call_method(self, method: str, params: Dict[str, Any]) -> Dict[str, Any]: - """调用JSON-RPC方法""" - self.request_id += 1 - - payload = { - "jsonrpc": "2.0", - "method": method, - "params": params, - "id": self.request_id - } - - print(f"📤 发送请求 (ID: {self.request_id}): {method}") - print(f" 参数: {json.dumps(params, indent=2, ensure_ascii=False)}") - - try: - response = requests.post( - self.server_url, - json=payload, - headers={'Content-Type': 'application/json'}, - timeout=120 # 增加超时时间 - ) - - if response.status_code == 200: - result = response.json() - print(f"📥 收到响应 (ID: {self.request_id}): {response.status_code}") - return result - else: - print(f"❌ HTTP错误: {response.status_code}") - return { - "error": { - "code": response.status_code, - "message": f"HTTP Error: {response.text}" - } - } - - except requests.exceptions.RequestException as e: - print(f"❌ 请求失败: {str(e)}") - return { - "error": { - "code": -1, - "message": f"Request failed: {str(e)}" - } - } - - def test_basic_detection(self): - """测试基础场景检测""" - print("\n" + "="*60) - print("🎯 测试基础场景检测") - print("="*60) - - params = { - "video_path": "assets/1/1752032011698.mp4", - "detector_type": "content", - "threshold": 15.0, - "min_scene_length": 1.0 - } - - start_time = time.time() - result = self._call_method("scene.detect", params) - end_time = time.time() - - if "error" in result: - print(f"❌ 检测失败: {result['error']}") - return False - - detection_result = result.get("result", {}) - if detection_result.get("success"): - print(f"✅ 基础检测成功!") - print(f" 场景数量: {detection_result['total_scenes']}") - print(f" 检测时间: {detection_result['detection_time']:.2f}秒") - print(f" API调用时间: {end_time - start_time:.2f}秒") - return True - else: - print(f"❌ 检测失败: {detection_result.get('error', '未知错误')}") - return False - - def test_workflow_detection(self): - """测试工作流场景检测""" - print("\n" + "="*60) - print("🔄 测试LangGraph工作流检测") - print("="*60) - - params = { - "video_path": "assets/1/1752032011698.mp4", - "detector_type": "content", - "threshold": 15.0, - "min_scene_length": 1.0, - "enable_ai_analysis": False # 禁用AI分析避免API密钥问题 - } - - start_time = time.time() - result = self._call_method("scene.detect_workflow", params) - end_time = time.time() - - if "error" in result: - print(f"❌ 工作流检测失败: {result['error']}") - return False - - workflow_result = result.get("result", {}) - detection_result = workflow_result.get("detection_result", {}) - - if detection_result.get("success"): - print(f"✅ 工作流检测成功!") - print(f" 工作流状态: {workflow_result.get('workflow_state')}") - print(f" 场景数量: {detection_result['total_scenes']}") - print(f" 检测时间: {detection_result['detection_time']:.2f}秒") - print(f" API调用时间: {end_time - start_time:.2f}秒") - - # 显示视频信息 - video_info = workflow_result.get("video_info", {}) - if video_info: - print(f" 视频信息: {video_info.get('resolution')}, {video_info.get('fps'):.2f}fps") - - # 显示AI分析结果 - ai_analysis = workflow_result.get("ai_analysis") - if ai_analysis: - print(f" AI分析: {ai_analysis[:100]}...") - - return True - else: - print(f"❌ 工作流检测失败: {detection_result.get('error', '未知错误')}") - return False - - def test_video_info(self): - """测试视频信息获取""" - print("\n" + "="*60) - print("📊 测试视频信息获取") - print("="*60) - - params = { - "video_path": "assets/1/1752032011698.mp4" - } - - start_time = time.time() - result = self._call_method("scene.get_video_info", params) - end_time = time.time() - - if "error" in result: - print(f"❌ 获取失败: {result['error']}") - return False - - info_result = result.get("result", {}) - if info_result.get("success"): - info = info_result.get("info", {}) - print(f"✅ 信息获取成功!") - print(f" 文件名: {info.get('filename')}") - print(f" 分辨率: {info.get('resolution')}") - print(f" 帧率: {info.get('fps'):.2f} fps") - print(f" 时长: {info.get('duration'):.2f}秒") - print(f" 文件大小: {info.get('file_size'):,} 字节") - print(f" API调用时间: {end_time - start_time:.2f}秒") - return True - else: - print(f"❌ 获取失败: {info_result.get('error', '未知错误')}") - return False - - def test_error_handling(self): - """测试错误处理""" - print("\n" + "="*60) - print("🚨 测试错误处理") - print("="*60) - - # 测试不存在的文件 - params = { - "video_path": "nonexistent_video.mp4", - "threshold": 15.0 - } - - result = self._call_method("scene.detect", params) - - if "error" in result: - print(f"✅ 正确处理了请求错误: {result['error']['message']}") - else: - detection_result = result.get("result", {}) - if not detection_result.get("success"): - print(f"✅ 正确处理了应用错误: {detection_result.get('error')}") - else: - print(f"❌ 应该返回错误，但返回了成功结果") - return False - - # 测试无效的方法 - result = self._call_method("scene.invalid_method", {}) - if "error" in result and result["error"]["code"] == -32601: - print(f"✅ 正确处理了方法不存在错误") - else: - print(f"❌ 没有正确处理方法不存在错误") - return False - - return True - - def run_all_tests(self): - """运行所有测试""" - print("🚀 JSON-RPC工作流测试开始") - print("="*60) - - # 检查服务器连接 - try: - test_result = self._call_method("scene.get_video_info", {"video_path": "test.mp4"}) - if "error" in test_result and "Request failed" in str(test_result["error"]): - print("❌ 无法连接到JSON-RPC服务器") - print("💡 请先启动服务器: python3 -m python_core.cli jsonrpc start --port 8081") - return False - except Exception as e: - print(f"❌ 连接测试失败: {e}") - return False - - print("✅ 服务器连接正常") - - # 运行测试 - tests = [ - ("视频信息获取", self.test_video_info), - ("基础场景检测", self.test_basic_detection), - ("工作流场景检测", self.test_workflow_detection), - ("错误处理", self.test_error_handling), - ] - - passed = 0 - total = len(tests) - - for test_name, test_func in tests: - try: - if test_func(): - passed += 1 - print(f"✅ {test_name} - 通过") - else: - print(f"❌ {test_name} - 失败") - except Exception as e: - print(f"❌ {test_name} - 异常: {e}") - - print("\n" + "="*60) - print(f"🎉 测试完成: {passed}/{total} 通过") - print("="*60) - - if passed == total: - print("🎊 所有测试都通过了！JSON-RPC工作流功能正常") - else: - print("⚠️ 部分测试失败，请检查服务器和配置") - - return passed == total - - -def main(): - """主函数""" - tester = JSONRPCWorkflowTester("http://localhost:8081") - tester.run_all_tests() - - -if __name__ == "__main__": - main() diff --git a/examples/test_refactored_modules.py b/examples/test_refactored_modules.py deleted file mode 100644 index 88d6b7e..0000000 --- a/examples/test_refactored_modules.py +++ /dev/null @@ -1,248 +0,0 @@ -#!/usr/bin/env python3 -""" -Test Refactored Modules -测试重构后的模块 - -验证重构后的场景检测模块是否正常工作 -""" - -import sys -from pathlib import Path - -# 添加项目根目录到Python路径 -project_root = Path(__file__).parent.parent -sys.path.insert(0, str(project_root)) - -from python_core.scene_detection import ( - SceneDetector, - DetectorType, - OutputFormat, - SceneInfo, - DetectionResult, - SceneDetectionWorkflowState -) - - -def test_imports(): - """测试模块导入""" - print("🧪 测试模块导入...") - - try: - # 测试类型导入 - assert DetectorType.CONTENT == "content" - assert OutputFormat.JSON == "json" - print("✅ 类型导入成功") - - # 测试数据模型 - scene = SceneInfo(0, 0.0, 5.0, 5.0) - assert scene.index == 0 - assert scene.duration == 5.0 - print("✅ 数据模型导入成功") - - # 测试主接口类 - detector = SceneDetector() - assert detector is not None - assert hasattr(detector, 'detect_scenes') - assert hasattr(detector, 'detect_with_workflow') - print("✅ 主接口类导入成功") - - return True - - except Exception as e: - print(f"❌ 导入测试失败: {e}") - return False - - -def test_basic_functionality(): - """测试基础功能""" - print("\n🧪 测试基础功能...") - - try: - detector = SceneDetector() - - # 测试视频信息获取 - video_path = Path("assets/1/1752032011698.mp4") - if video_path.exists(): - info_result = detector.get_video_info(video_path) - if info_result["success"]: - print("✅ 视频信息获取成功") - print(f" 分辨率: {info_result['info']['resolution']}") - print(f" 时长: {info_result['info']['duration']:.2f}秒") - else: - print(f"⚠️ 视频信息获取失败: {info_result['error']}") - else: - print("⚠️ 测试视频文件不存在，跳过视频信息测试") - - # 测试JSON-RPC方法注册 - detector.register_jsonrpc_methods() - print("✅ JSON-RPC方法注册成功") - - return True - - except Exception as e: - print(f"❌ 基础功能测试失败: {e}") - import traceback - print(f"详细错误: {traceback.format_exc()}") - return False - - -def test_services(): - """测试服务层""" - print("\n🧪 测试服务层...") - - try: - from python_core.scene_detection.services import ( - SceneDetectorService, - VideoInfoService, - AIAnalysisService - ) - - # 测试检测服务 - detector_service = SceneDetectorService() - assert len(detector_service.supported_formats) > 0 - print("✅ 场景检测服务初始化成功") - - # 测试视频信息服务 - video_info_service = VideoInfoService() - assert hasattr(video_info_service, 'extract_video_info') - print("✅ 视频信息服务初始化成功") - - # 测试AI分析服务 - ai_service = AIAnalysisService() - assert hasattr(ai_service, 'analyze_detection_result') - print(f"✅ AI分析服务初始化成功 (AI启用: {ai_service.ai_enabled})") - - return True - - except Exception as e: - print(f"❌ 服务层测试失败: {e}") - return False - - -def test_workflows(): - """测试工作流""" - print("\n🧪 测试工作流...") - - try: - from python_core.scene_detection.workflows import ( - SceneDetectionWorkflowManager, - WorkflowNodes - ) - - # 测试工作流管理器 - workflow_manager = SceneDetectionWorkflowManager() - assert hasattr(workflow_manager, 'create_detection_workflow') - print("✅ 工作流管理器初始化成功") - - # 测试工作流节点 - nodes = WorkflowNodes() - assert hasattr(nodes, 'validate_input') - assert hasattr(nodes, 'detect_scenes') - assert hasattr(nodes, 'finalize_results') - print("✅ 工作流节点初始化成功") - - # 测试工作流创建 - workflow = workflow_manager.create_detection_workflow() - if workflow: - print("✅ LangGraph工作流创建成功") - else: - print("⚠️ LangGraph工作流创建失败（可能是依赖问题）") - - return True - - except Exception as e: - print(f"❌ 工作流测试失败: {e}") - return False - - -def test_utils(): - """测试工具类""" - print("\n🧪 测试工具类...") - - try: - from python_core.scene_detection.utils import ( - ResultSaver, - InputValidator - ) - - # 测试结果保存器 - saver = ResultSaver() - assert hasattr(saver, 'save_results') - print("✅ 结果保存器初始化成功") - - # 测试输入验证器 - validator = InputValidator({'.mp4', '.avi'}) - assert hasattr(validator, 'validate_video_path') - print("✅ 输入验证器初始化成功") - - return True - - except Exception as e: - print(f"❌ 工具类测试失败: {e}") - return False - - -def test_cli_integration(): - """测试CLI集成""" - print("\n🧪 测试CLI集成...") - - try: - # 测试新的CLI模块 - from python_core.cli.scene_detect_new import app - assert app is not None - print("✅ 新CLI模块导入成功") - - return True - - except Exception as e: - print(f"❌ CLI集成测试失败: {e}") - return False - - -def main(): - """主测试函数""" - print("🚀 开始测试重构后的场景检测模块") - print("=" * 60) - - tests = [ - ("模块导入", test_imports), - ("基础功能", test_basic_functionality), - ("服务层", test_services), - ("工作流", test_workflows), - ("工具类", test_utils), - ("CLI集成", test_cli_integration), - ] - - passed = 0 - total = len(tests) - - for test_name, test_func in tests: - try: - if test_func(): - passed += 1 - print(f"✅ {test_name} - 通过") - else: - print(f"❌ {test_name} - 失败") - except Exception as e: - print(f"❌ {test_name} - 异常: {e}") - - print("\n" + "=" * 60) - print(f"🎉 测试完成: {passed}/{total} 通过") - - if passed == total: - print("🎊 所有测试都通过了！重构成功！") - print("\n📋 重构后的模块结构:") - print("├── types/ # 类型定义和数据模型") - print("├── services/ # 核心业务逻辑服务") - print("├── workflows/ # LangGraph工作流") - print("├── utils/ # 工具类和辅助函数") - print("└── scene_detector.py # 主接口类") - else: - print("⚠️ 部分测试失败，请检查重构实现") - - return passed == total - - -if __name__ == "__main__": - success = main() - sys.exit(0 if success else 1) diff --git a/examples/test_scene_detection_fix.py b/examples/test_scene_detection_fix.py deleted file mode 100644 index 5d47e30..0000000 --- a/examples/test_scene_detection_fix.py +++ /dev/null @@ -1,206 +0,0 @@ -#!/usr/bin/env python3 -""" -Test Scene Detection Fix -测试场景检测修复 -""" - -import sys -from pathlib import Path - -# 添加项目根目录到Python路径 -project_root = Path(__file__).parent.parent -sys.path.insert(0, str(project_root)) - -from python_core.scene_detection.services import SceneDetectorService -from python_core.scene_detection.types import DetectorType - - -def test_basic_detection(): - """测试基础场景检测""" - print("🧪 测试基础场景检测...") - - try: - # 创建检测服务 - detector_service = SceneDetectorService() - - # 测试视频路径 - video_path = Path("assets/1/1752032011698.mp4") - - if not video_path.exists(): - print(f"❌ 测试视频不存在: {video_path}") - return False - - print(f"📹 测试视频: {video_path}") - - # 执行检测 - result = detector_service.detect_scenes( - video_path, - DetectorType.CONTENT, - 30.0, - 1.0 - ) - - if result.success: - print(f"✅ 检测成功!") - print(f" 场景数量: {result.total_scenes}") - print(f" 检测时间: {result.detection_time:.2f}秒") - print(f" 视频时长: {result.total_duration:.2f}秒") - - # 显示前几个场景 - for i, scene in enumerate(result.scenes[:3]): - print(f" 场景 {i+1}: {scene.start_time:.2f}s - {scene.end_time:.2f}s (时长: {scene.duration:.2f}s)") - - if len(result.scenes) > 3: - print(f" ... 还有 {len(result.scenes) - 3} 个场景") - - return True - else: - print(f"❌ 检测失败: {result.error}") - return False - - except Exception as e: - print(f"❌ 测试异常: {e}") - import traceback - print(f"详细错误: {traceback.format_exc()}") - return False - - -def test_video_info(): - """测试视频信息提取""" - print("\n🧪 测试视频信息提取...") - - try: - from python_core.scene_detection.services import VideoInfoService - - video_info_service = VideoInfoService() - video_path = Path("assets/1/1752032011698.mp4") - - if not video_path.exists(): - print(f"❌ 测试视频不存在: {video_path}") - return False - - info = video_info_service.extract_video_info(video_path) - - print(f"✅ 视频信息提取成功!") - print(f" 文件名: {info['filename']}") - print(f" 分辨率: {info['resolution']}") - print(f" 帧率: {info['fps']:.2f} fps") - print(f" 时长: {info['duration']:.2f}秒") - print(f" 文件大小: {info['file_size']:,} 字节") - - return True - - except Exception as e: - print(f"❌ 测试异常: {e}") - import traceback - print(f"详细错误: {traceback.format_exc()}") - return False - - -def test_pyscenedetect_version(): - """测试PySceneDetect版本和API""" - print("\n🧪 测试PySceneDetect版本和API...") - - try: - import scenedetect - print(f"📦 PySceneDetect版本: {scenedetect.__version__}") - - # 测试基本API - from scenedetect import open_video, SceneManager - from scenedetect.detectors import ContentDetector - - video_path = Path("assets/1/1752032011698.mp4") - if not video_path.exists(): - print(f"❌ 测试视频不存在: {video_path}") - return False - - # 打开视频 - video = open_video(str(video_path)) - scene_manager = SceneManager() - scene_manager.add_detector(ContentDetector(threshold=30.0)) - - print("📊 执行基础场景检测...") - scene_manager.detect_scenes(video, show_progress=False) - scene_list = scene_manager.get_scene_list() - - print(f"✅ 基础检测成功，找到 {len(scene_list)} 个场景") - - # 测试时间码API - if scene_list: - start_time, end_time = scene_list[0] - print(f"📋 第一个场景时间码类型: {type(start_time)}") - - # 测试不同的时间获取方法 - methods = [] - - if hasattr(start_time, 'total_seconds'): - try: - seconds = start_time.total_seconds() - methods.append(f"total_seconds(): {seconds:.2f}") - except Exception as e: - methods.append(f"total_seconds() 失败: {e}") - - if hasattr(start_time, 'get_seconds'): - try: - seconds = start_time.get_seconds() - methods.append(f"get_seconds(): {seconds:.2f}") - except Exception as e: - methods.append(f"get_seconds() 失败: {e}") - - try: - seconds = float(start_time) - methods.append(f"float(): {seconds:.2f}") - except Exception as e: - methods.append(f"float() 失败: {e}") - - print("🔍 可用的时间获取方法:") - for method in methods: - print(f" - {method}") - - return True - - except Exception as e: - print(f"❌ 测试异常: {e}") - import traceback - print(f"详细错误: {traceback.format_exc()}") - return False - - -def main(): - """主函数""" - print("🚀 开始测试场景检测修复") - print("=" * 60) - - tests = [ - ("PySceneDetect版本和API", test_pyscenedetect_version), - ("视频信息提取", test_video_info), - ("基础场景检测", test_basic_detection), - ] - - passed = 0 - total = len(tests) - - for test_name, test_func in tests: - try: - if test_func(): - passed += 1 - print(f"✅ {test_name} - 通过") - else: - print(f"❌ {test_name} - 失败") - except Exception as e: - print(f"❌ {test_name} - 异常: {e}") - - print("\n" + "=" * 60) - print(f"🎉 测试完成: {passed}/{total} 通过") - - if passed == total: - print("🎊 所有测试都通过了！场景检测修复成功！") - else: - print("⚠️ 部分测试失败，需要进一步调试") - - return passed == total - - -if __name__ == "__main__": - success = main() - sys.exit(0 if success else 1) diff --git a/examples/test_workflow_results.py b/examples/test_workflow_results.py deleted file mode 100644 index d18374a..0000000 --- a/examples/test_workflow_results.py +++ /dev/null @@ -1,303 +0,0 @@ -#!/usr/bin/env python3 -""" -测试工作流结果返回 -Test Workflow Result Return -""" - -import json -import requests -import time -import threading -import queue -from typing import Dict, Any, Optional - - -class WorkflowResultTester: - """工作流结果测试器""" - - def __init__(self, server_url: str = "http://localhost:8081"): - self.server_url = server_url - self.request_id = 0 - self.progress_queue = queue.Queue() - self.result_queue = queue.Queue() - - def _call_method_async(self, method: str, params: Dict[str, Any]) -> str: - """异步调用JSON-RPC方法""" - self.request_id += 1 - request_id = f"test_{self.request_id}_{int(time.time())}" - - payload = { - "jsonrpc": "2.0", - "method": method, - "params": params, - "id": request_id - } - - print(f"📤 发送异步请求 (ID: {request_id}): {method}") - print(f" 参数: {json.dumps(params, indent=2, ensure_ascii=False)}") - - def make_request(): - try: - response = requests.post( - self.server_url, - json=payload, - headers={'Content-Type': 'application/json'}, - timeout=120 - ) - - print(f"📥 收到HTTP响应 (ID: {request_id}): {response.status_code}") - - if response.status_code == 200: - # 检查是否有响应体 - if response.text.strip(): - try: - result = response.json() - print(f"📋 JSON响应内容: {json.dumps(result, indent=2, ensure_ascii=False)}") - self.result_queue.put(("response", request_id, result)) - except json.JSONDecodeError: - print(f"⚠️ 响应不是有效的JSON: {response.text}") - self.result_queue.put(("invalid_json", request_id, response.text)) - else: - print(f"✅ 空响应体 - 表示结果已异步发送") - self.result_queue.put(("empty_response", request_id, None)) - else: - print(f"❌ HTTP错误: {response.status_code} - {response.text}") - self.result_queue.put(("http_error", request_id, { - "status_code": response.status_code, - "text": response.text - })) - - except requests.exceptions.RequestException as e: - print(f"❌ 请求异常: {str(e)}") - self.result_queue.put(("request_error", request_id, str(e))) - - # 在后台线程中发送请求 - thread = threading.Thread(target=make_request) - thread.daemon = True - thread.start() - - return request_id - - def test_workflow_with_progress_monitoring(self): - """测试带进度监控的工作流""" - print("\n" + "="*60) - print("🔄 测试工作流进度监控和结果返回") - print("="*60) - - # 启动进度监控（模拟WebSocket或长轮询） - progress_thread = threading.Thread(target=self._monitor_progress) - progress_thread.daemon = True - progress_thread.start() - - # 发送工作流请求 - params = { - "video_path": "assets/1/1752032011698.mp4", - "detector_type": "content", - "threshold": 15.0, - "min_scene_length": 1.0, - "enable_ai_analysis": False - } - - request_id = self._call_method_async("scene.detect_workflow", params) - - # 等待结果 - print(f"⏳ 等待工作流完成 (Request ID: {request_id})...") - - start_time = time.time() - timeout = 60 # 60秒超时 - - while time.time() - start_time < timeout: - try: - result_type, result_id, result_data = self.result_queue.get(timeout=1) - - if result_id == request_id: - print(f"\n📋 收到结果 (类型: {result_type}):") - - if result_type == "response": - print("✅ 收到JSON-RPC响应:") - print(json.dumps(result_data, indent=2, ensure_ascii=False)) - return True - elif result_type == "empty_response": - print("✅ 收到空响应 - 结果已通过进度通道发送") - return True - elif result_type == "http_error": - print(f"❌ HTTP错误: {result_data}") - return False - elif result_type == "request_error": - print(f"❌ 请求错误: {result_data}") - return False - else: - print(f"⚠️ 未知结果类型: {result_type}") - return False - - except queue.Empty: - continue - - print(f"⏰ 等待超时 ({timeout}秒)") - return False - - def _monitor_progress(self): - """监控进度更新（模拟）""" - # 这里应该是WebSocket连接或长轮询 - # 为了演示，我们只是打印一些模拟的进度信息 - print("📊 进度监控已启动...") - - # 模拟进度更新 - progress_steps = [ - ("validate", "🔍 验证输入参数..."), - ("extract_info", "📊 提取视频信息..."), - ("detect", "🎯 执行场景检测..."), - ("analyze", "🧠 AI分析场景结果..."), - ("finalize", "📋 整理最终结果...") - ] - - for step, message in progress_steps: - time.sleep(2) # 模拟处理时间 - print(f"📈 [进度] {step}: {message}") - - def test_basic_method_comparison(self): - """测试基础方法对比""" - print("\n" + "="*60) - print("🔍 测试基础方法 vs 工作流方法") - print("="*60) - - # 测试基础方法（应该返回结果） - print("1️⃣ 测试基础场景检测方法:") - basic_params = { - "video_path": "assets/1/1752032011698.mp4", - "threshold": 15.0 - } - - basic_request_id = self._call_method_async("scene.detect", basic_params) - - # 等待基础方法结果 - time.sleep(5) - - try: - result_type, result_id, result_data = self.result_queue.get(timeout=1) - if result_id == basic_request_id: - if result_type == "response": - print("✅ 基础方法正确返回了JSON响应") - print(f" 场景数: {result_data.get('result', {}).get('total_scenes', 'N/A')}") - else: - print(f"⚠️ 基础方法返回了意外的结果类型: {result_type}") - except queue.Empty: - print("⏰ 基础方法响应超时") - - print("\n2️⃣ 测试工作流方法:") - # 工作流方法测试在上面的方法中进行 - return self.test_workflow_with_progress_monitoring() - - def test_error_handling(self): - """测试错误处理""" - print("\n" + "="*60) - print("🚨 测试工作流错误处理") - print("="*60) - - # 测试不存在的文件 - params = { - "video_path": "nonexistent_video.mp4", - "threshold": 15.0, - "enable_ai_analysis": False - } - - request_id = self._call_method_async("scene.detect_workflow", params) - - # 等待错误结果 - print(f"⏳ 等待错误处理 (Request ID: {request_id})...") - - start_time = time.time() - timeout = 30 - - while time.time() - start_time < timeout: - try: - result_type, result_id, result_data = self.result_queue.get(timeout=1) - - if result_id == request_id: - if result_type == "response" and "error" in result_data: - print("✅ 正确收到错误响应:") - print(json.dumps(result_data, indent=2, ensure_ascii=False)) - return True - elif result_type == "empty_response": - print("✅ 错误已通过进度通道发送") - return True - else: - print(f"⚠️ 意外的错误处理结果: {result_type}") - return False - - except queue.Empty: - continue - - print(f"⏰ 错误处理等待超时") - return False - - def run_all_tests(self): - """运行所有测试""" - print("🚀 工作流结果返回测试开始") - print("="*60) - - # 检查服务器连接 - try: - response = requests.get(f"{self.server_url.replace('//', '//').replace('http:', 'http:').replace('8081', '8081')}") - except: - try: - # 简单的连接测试 - test_payload = { - "jsonrpc": "2.0", - "method": "scene.get_video_info", - "params": {"video_path": "test.mp4"}, - "id": "test" - } - response = requests.post( - self.server_url, - json=test_payload, - headers={'Content-Type': 'application/json'}, - timeout=5 - ) - except Exception as e: - print("❌ 无法连接到JSON-RPC服务器") - print("💡 请先启动服务器: python3 -m python_core.cli jsonrpc start --port 8081") - return False - - print("✅ 服务器连接正常") - - # 运行测试 - tests = [ - ("基础方法对比", self.test_basic_method_comparison), - ("错误处理", self.test_error_handling), - ] - - passed = 0 - total = len(tests) - - for test_name, test_func in tests: - try: - print(f"\n🧪 运行测试: {test_name}") - if test_func(): - passed += 1 - print(f"✅ {test_name} - 通过") - else: - print(f"❌ {test_name} - 失败") - except Exception as e: - print(f"❌ {test_name} - 异常: {e}") - - print("\n" + "="*60) - print(f"🎉 测试完成: {passed}/{total} 通过") - print("="*60) - - if passed == total: - print("🎊 所有测试都通过了！工作流结果返回功能正常") - else: - print("⚠️ 部分测试失败，请检查实现") - - return passed == total - - -def main(): - """主函数""" - tester = WorkflowResultTester("http://localhost:8081") - tester.run_all_tests() - - -if __name__ == "__main__": - main() diff --git a/examples/textVideoAgentUsage.ts b/examples/textVideoAgentUsage.ts deleted file mode 100644 index ad8f815..0000000 --- a/examples/textVideoAgentUsage.ts +++ /dev/null @@ -1,416 +0,0 @@ -/** - * Text Video Agent API 使用示例 - */ - -import { - TextVideoAgentAPI, - textVideoAgentAPI, - ImageGenerationParams, - VideoGenerationParams, - TaskRequest -} from '../src/services/textVideoAgentAPI' - -import { - TaskType, - AspectRatio, - VideoDuration, - PRESET_CONFIGS -} from '../src/services/textVideoAgentTypes' - -// ==================== 基础使用示例 ==================== - -/** - * 示例1: 基础健康检查 - */ -async function example1_HealthCheck() { - console.log('=== 健康检查示例 ===') - - try { - const health = await textVideoAgentAPI.healthCheck() - console.log('服务状态:', health) - - const mjHealth = await textVideoAgentAPI.mjHealthCheck() - console.log('Midjourney状态:', mjHealth) - - const jmHealth = await textVideoAgentAPI.jmHealthCheck() - console.log('极梦状态:', jmHealth) - } catch (error) { - console.error('健康检查失败:', error) - } -} - -/** - * 示例2: 获取示例提示词 - */ -async function example2_GetSamplePrompts() { - console.log('=== 获取示例提示词 ===') - - try { - const prompts = await textVideoAgentAPI.getSamplePrompt(TaskType.VLOG) - console.log('VLOG类型示例提示词:', prompts) - } catch (error) { - console.error('获取提示词失败:', error) - } -} - -/** - * 示例3: 文件上传 - */ -async function example3_FileUpload() { - console.log('=== 文件上传示例 ===') - - // 注意：这里需要实际的文件对象 - // const fileInput = document.getElementById('fileInput') as HTMLInputElement - // const file = fileInput.files?.[0] - - // 模拟文件对象（实际使用时替换为真实文件） - const mockFile = new File(['mock content'], 'test.jpg', { type: 'image/jpeg' }) - - try { - const uploadResult = await textVideoAgentAPI.uploadFile(mockFile) - console.log('文件上传结果:', uploadResult) - return uploadResult.data // 返回文件URL - } catch (error) { - console.error('文件上传失败:', error) - } -} - -// ==================== 图片生成示例 ==================== - -/** - * 示例4: 基础图片生成 - */ -async function example4_BasicImageGeneration() { - console.log('=== 基础图片生成示例 ===') - - const params: ImageGenerationParams = { - prompt: '一个美丽的女孩在喝茶，温馨的下午时光，自然光线，高质量摄影', - max_wait_time: 120, - poll_interval: 2 - } - - try { - const result = await textVideoAgentAPI.generateImageSync(params) - console.log('图片生成结果:', result) - return result.data?.image_url - } catch (error) { - console.error('图片生成失败:', error) - } -} - -/** - * 示例5: 带参考图片的图片生成 - */ -async function example5_ImageGenerationWithReference() { - console.log('=== 带参考图片的图片生成示例 ===') - - // 模拟参考图片文件 - const referenceImage = new File(['reference'], 'reference.jpg', { type: 'image/jpeg' }) - - const params: ImageGenerationParams = { - prompt: '保持人物特征，改变背景为咖啡厅环境', - img_file: referenceImage, - ...PRESET_CONFIGS.STANDARD - } - - try { - const result = await textVideoAgentAPI.generateImageWithRetry(params, 3) - console.log('带参考图片生成结果:', result) - return result.data?.image_url - } catch (error) { - console.error('带参考图片生成失败:', error) - } -} - -/** - * 示例6: 异步图片生成 - */ -async function example6_AsyncImageGeneration() { - console.log('=== 异步图片生成示例 ===') - - const prompt = '现代简约风格的室内设计，明亮的客厅' - - try { - // 提交异步任务 - const taskResult = await textVideoAgentAPI.generateImageAsync(prompt) - const taskId = taskResult.data?.task_id - - if (!taskId) { - throw new Error('未获取到任务ID') - } - - console.log('任务已提交，ID:', taskId) - - // 轮询任务状态 - const finalResult = await textVideoAgentAPI.pollTaskUntilComplete( - taskId, - 2000, // 2秒轮询间隔 - 120000, // 2分钟超时 - (status) => { - console.log('任务状态更新:', status) - } - ) - - console.log('异步图片生成完成:', finalResult) - return finalResult.data?.image_url - } catch (error) { - console.error('异步图片生成失败:', error) - } -} - -// ==================== 视频生成示例 ==================== - -/** - * 示例7: 基础视频生成 - */ -async function example7_BasicVideoGeneration() { - console.log('=== 基础视频生成示例 ===') - - // 首先生成一张图片作为视频的基础 - const imageUrl = await example4_BasicImageGeneration() - - if (!imageUrl) { - console.error('无法获取基础图片，跳过视频生成') - return - } - - const params: VideoGenerationParams = { - prompt: '女孩优雅地品茶，轻柔的动作，温馨的氛围', - img_url: imageUrl, - duration: VideoDuration.MEDIUM, - max_wait_time: 300, - poll_interval: 5 - } - - try { - const result = await textVideoAgentAPI.generateVideoSync(params) - console.log('视频生成结果:', result) - return result.data?.video_url - } catch (error) { - console.error('视频生成失败:', error) - } -} - -/** - * 示例8: 异步视频生成 - */ -async function example8_AsyncVideoGeneration() { - console.log('=== 异步视频生成示例 ===') - - const params: VideoGenerationParams = { - prompt: '城市夜景延时摄影，车流如水，霓虹闪烁', - duration: VideoDuration.LONG - } - - try { - const taskResult = await textVideoAgentAPI.generateVideoAsync(params) - const taskId = taskResult.data?.task_id - - if (!taskId) { - throw new Error('未获取到任务ID') - } - - console.log('视频生成任务已提交，ID:', taskId) - - // 轮询任务状态 - const finalResult = await textVideoAgentAPI.pollTaskUntilComplete( - taskId, - 5000, // 5秒轮询间隔 - 600000, // 10分钟超时 - (status) => { - console.log('视频生成进度:', status) - } - ) - - console.log('异步视频生成完成:', finalResult) - return finalResult.data?.video_url - } catch (error) { - console.error('异步视频生成失败:', error) - } -} - -// ==================== 高级功能示例 ==================== - -/** - * 示例9: 图片描述功能 - */ -async function example9_ImageDescription() { - console.log('=== 图片描述示例 ===') - - const imageUrl = 'https://example.com/sample-image.jpg' - - try { - const description = await textVideoAgentAPI.describeImageByUrl({ - image_url: imageUrl, - max_wait_time: 60 - }) - - console.log('图片描述结果:', description) - return description.data?.description - } catch (error) { - console.error('图片描述失败:', error) - } -} - -/** - * 示例10: 任务管理 - */ -async function example10_TaskManagement() { - console.log('=== 任务管理示例 ===') - - const taskRequest: TaskRequest = { - task_type: TaskType.VLOG, - prompt: '制作一个关于健康生活方式的短视频', - ar: AspectRatio.PORTRAIT - } - - try { - // 创建任务 - const createResult = await textVideoAgentAPI.createTask(taskRequest) - const taskId = createResult.data?.task_id - - if (!taskId) { - throw new Error('任务创建失败') - } - - console.log('任务创建成功，ID:', taskId) - - // 查询任务状态 - const statusResult = await textVideoAgentAPI.getTaskStatusAsync(taskId) - console.log('任务状态:', statusResult) - - // 等待任务完成（同步方式） - const finalResult = await textVideoAgentAPI.getTaskResultSync(taskId) - console.log('任务最终结果:', finalResult) - - return finalResult - } catch (error) { - console.error('任务管理失败:', error) - } -} - -/** - * 示例11: 端到端内容生成 - */ -async function example11_EndToEndGeneration() { - console.log('=== 端到端内容生成示例 ===') - - try { - const result = await textVideoAgentAPI.generateContentEndToEnd( - '一个关于咖啡文化的精美内容，展现咖啡师的专业技艺', - { - taskType: TaskType.VLOG, - aspectRatio: AspectRatio.PORTRAIT, - videoDuration: VideoDuration.MEDIUM, - generateVideo: true, - onProgress: (step, progress) => { - console.log(`进度更新: ${step} - ${progress}%`) - } - } - ) - - console.log('端到端生成完成:', result) - return result - } catch (error) { - console.error('端到端生成失败:', error) - } -} - -// ==================== 批量处理示例 ==================== - -/** - * 示例12: 批量图片生成 - */ -async function example12_BatchImageGeneration() { - console.log('=== 批量图片生成示例 ===') - - const prompts = [ - '春天的樱花盛开', - '夏日的海滩风光', - '秋天的枫叶满山', - '冬日的雪景如画' - ] - - try { - const results = await Promise.allSettled( - prompts.map(async (prompt, index) => { - const params: ImageGenerationParams = { - prompt, - ...PRESET_CONFIGS.FAST // 使用快速配置 - } - - console.log(`开始生成图片 ${index + 1}: ${prompt}`) - const result = await textVideoAgentAPI.generateImageSync(params) - console.log(`图片 ${index + 1} 生成完成`) - - return { - prompt, - result: result.data?.image_url, - success: result.status - } - }) - ) - - console.log('批量图片生成结果:', results) - return results - } catch (error) { - console.error('批量图片生成失败:', error) - } -} - -// ==================== 主函数 ==================== - -/** - * 运行所有示例 - */ -async function runAllExamples() { - console.log('开始运行 Text Video Agent API 示例...\n') - - // 基础功能示例 - await example1_HealthCheck() - await example2_GetSamplePrompts() - - // 文件操作示例 - // await example3_FileUpload() // 需要实际文件 - - // 图片生成示例 - await example4_BasicImageGeneration() - // await example5_ImageGenerationWithReference() // 需要实际文件 - await example6_AsyncImageGeneration() - - // 视频生成示例 - await example7_BasicVideoGeneration() - await example8_AsyncVideoGeneration() - - // 高级功能示例 - await example9_ImageDescription() - await example10_TaskManagement() - await example11_EndToEndGeneration() - - // 批量处理示例 - await example12_BatchImageGeneration() - - console.log('\n所有示例运行完成！') -} - -// 导出示例函数 -export { - example1_HealthCheck, - example2_GetSamplePrompts, - example3_FileUpload, - example4_BasicImageGeneration, - example5_ImageGenerationWithReference, - example6_AsyncImageGeneration, - example7_BasicVideoGeneration, - example8_AsyncVideoGeneration, - example9_ImageDescription, - example10_TaskManagement, - example11_EndToEndGeneration, - example12_BatchImageGeneration, - runAllExamples -} - -// 如果直接运行此文件，执行所有示例 -if (require.main === module) { - runAllExamples().catch(console.error) -} diff --git a/examples/typer_media_manager.py b/examples/typer_media_manager.py deleted file mode 100644 index c9e9439..0000000 --- a/examples/typer_media_manager.py +++ /dev/null @@ -1,504 +0,0 @@ -#!/usr/bin/env python3 -""" -基于Typer开发规范的媒体管理器实现示例 -""" - -import sys -import time -from pathlib import Path -from typing import Optional, List -from enum import Enum - -# 添加项目根目录到Python路径 -project_root = Path(__file__).parent.parent -sys.path.insert(0, str(project_root)) - -try: - import typer - from rich.console import Console - from rich.table import Table - from rich.panel import Panel - from rich.progress import Progress, SpinnerColumn, TextColumn -except ImportError: - print("❌ 需要安装 typer 和 rich: pip install typer rich") - sys.exit(1) - -from python_core.utils.progress import ProgressJSONRPCCommander -from python_core.services.base import ProgressServiceBase - -console = Console() - -# 1. 类型定义（遵循规范） -class OutputFormat(str, Enum): - """输出格式枚举""" - JSON = "json" - CSV = "csv" - TXT = "txt" - - def __str__(self) -> str: - return self.value - -class ProcessingMode(str, Enum): - """处理模式枚举""" - FAST = "fast" - BALANCED = "balanced" - QUALITY = "quality" - -# 2. 服务类（遵循规范） -class MediaManagerService(ProgressServiceBase): - """媒体管理服务""" - - def get_service_name(self) -> str: - return "media_manager" - - def upload_video(self, video_path: Path, tags: List[str] = None, filename: str = None) -> dict: - """上传单个视频""" - if tags is None: - tags = [] - - # 模拟上传过程 - steps = [ - "计算文件哈希...", - "检查重复文件...", - "复制文件到存储...", - "提取视频信息...", - "检测场景变化..." - ] - - for step in steps: - self.report_progress(step) - time.sleep(0.3) # 模拟处理时间 - - result = { - "video_path": str(video_path), - "filename": filename or video_path.name, - "tags": tags, - "success": True, - "scenes_detected": 3, - "duration": 120.5 - } - - # 保存结果到存储 - self.save_data("uploads", f"upload_{int(time.time())}", result) - - return result - - def batch_upload(self, directory: Path, tags: List[str] = None, recursive: bool = False) -> dict: - """批量上传视频""" - if tags is None: - tags = [] - - # 扫描视频文件 - video_extensions = {'.mp4', '.avi', '.mov', '.mkv', '.wmv'} - video_files = [] - - if recursive: - for ext in video_extensions: - video_files.extend(directory.rglob(f"*{ext}")) - else: - for ext in video_extensions: - video_files.extend(directory.glob(f"*{ext}")) - - results = [] - for i, video_file in enumerate(video_files): - self.report_progress(f"处理文件: {video_file.name} ({i+1}/{len(video_files)})") - - try: - result = self.upload_video(video_file, tags) - results.append(result) - except Exception as e: - results.append({ - "video_path": str(video_file), - "success": False, - "error": str(e) - }) - - batch_result = { - "total_files": len(video_files), - "successful": len([r for r in results if r.get("success")]), - "failed": len([r for r in results if not r.get("success")]), - "results": results - } - - # 保存批量结果 - self.save_data("batch_uploads", f"batch_{int(time.time())}", batch_result) - - return batch_result - - def get_recent_uploads(self, limit: int = 10) -> List[dict]: - """获取最近的上传记录""" - keys = self.list_keys("uploads") - recent_keys = sorted(keys)[-limit:] - return list(self.load_batch_data("uploads", recent_keys).values()) - -# 3. Typer命令行接口（遵循规范） -class MediaManagerCommander(ProgressJSONRPCCommander): - """基于Typer的媒体管理器命令行接口""" - - def __init__(self): - super().__init__("media_manager") - self.app = typer.Typer( - name="media_manager", - help=""" - 🎬 媒体管理器 - - 功能强大的视频处理和管理工具，支持： - • 视频上传和元数据提取 - • 自动场景检测和分割 - • 批量处理和进度跟踪 - • 多种输出格式 - - 使用 --help 查看具体命令帮助。 - """, - rich_markup_mode="rich", - no_args_is_help=True - ) - self.service = MediaManagerService() - self._setup_commands() - - def _register_commands(self): - """注册命令（继承要求）""" - pass # Typer通过装饰器自动注册 - - def _is_progressive_command(self, command: str) -> bool: - """判断是否需要进度报告""" - return command in ["batch_upload", "analyze"] - - def _execute_with_progress(self, command: str, args: dict): - """执行带进度的命令""" - pass # 通过Typer命令直接处理 - - def _execute_simple_command(self, command: str, args: dict): - """执行简单命令""" - pass # 通过Typer命令直接处理 - - def _setup_commands(self): - """设置Typer命令""" - - @self.app.command() - def upload( - video_path: Path = typer.Argument( - ..., - help="📹 视频文件路径", - exists=True - ), - tags: Optional[str] = typer.Option( - None, - "--tags", "-t", - help="🏷️ 标签列表（逗号分隔）" - ), - filename: Optional[str] = typer.Option( - None, - "--filename", "-f", - help="📝 自定义文件名" - ), - verbose: bool = typer.Option( - False, - "--verbose", "-v", - help="📝 详细输出" - ) - ): - """ - 📤 上传视频文件 - - 上传单个视频文件到媒体库，自动进行场景检测和元数据提取。 - - 示例: - media_manager upload video.mp4 --tags "demo,test" - media_manager upload /path/to/video.mp4 --filename "my_video" - - 注意: - - 支持的格式: MP4, AVI, MOV, MKV, WMV - - 自动检测重复文件 - - 自动提取视频元数据 - """ - # 参数验证 - if not video_path.exists(): - console.print("❌ [red]视频文件不存在[/red]") - raise typer.Exit(1) - - # 解析标签 - tag_list = [] - if tags: - tag_list = [tag.strip() for tag in tags.split(",")] - - # 显示开始信息 - console.print(f"🚀 开始上传: [bold blue]{video_path}[/bold blue]") - - try: - # 设置进度回调 - def progress_callback(message: str): - console.print(f"📊 {message}") - - self.service.set_progress_callback(progress_callback) - - # 执行上传 - result = self.service.upload_video(video_path, tag_list, filename) - - # 显示结果 - console.print("✅ [bold green]上传完成[/bold green]") - - if verbose or True: # 总是显示基本信息 - self._show_upload_result(result) - - except Exception as e: - console.print(f"❌ [red]上传失败: {e}[/red]") - if verbose: - console.print_exception() - raise typer.Exit(1) - - @self.app.command() - def batch_upload( - input_directory: Path = typer.Argument( - ..., - help="📁 输入目录路径", - exists=True, - file_okay=False - ), - output_path: Optional[Path] = typer.Option( - None, - "--output", "-o", - help="📄 结果输出文件路径" - ), - tags: Optional[str] = typer.Option( - None, - "--tags", "-t", - help="🏷️ 标签列表（逗号分隔）" - ), - recursive: bool = typer.Option( - False, - "--recursive", "-r", - help="🔄 递归处理子目录" - ), - output_format: OutputFormat = typer.Option( - OutputFormat.JSON, - "--format", "-f", - help="📄 输出格式" - ) - ): - """ - 📦 批量上传视频文件（带进度条） - - 批量处理目录中的所有视频文件，支持递归扫描和进度跟踪。 - - 示例: - media_manager batch-upload /videos --tags "batch,demo" - media_manager batch-upload /videos -r --output results.json - """ - console.print(f"📦 批量上传目录: [bold blue]{input_directory}[/bold blue]") - - # 解析标签 - tag_list = [] - if tags: - tag_list = [tag.strip() for tag in tags.split(",")] - - # 扫描文件数量 - video_extensions = {'.mp4', '.avi', '.mov', '.mkv', '.wmv'} - video_files = [] - - if recursive: - for ext in video_extensions: - video_files.extend(input_directory.rglob(f"*{ext}")) - else: - for ext in video_extensions: - video_files.extend(input_directory.glob(f"*{ext}")) - - if not video_files: - console.print("⚠️ [yellow]未找到可处理的视频文件[/yellow]") - return - - console.print(f"📋 找到 {len(video_files)} 个视频文件") - - # 使用进度任务 - with self.create_task("批量上传", len(video_files)) as task: - def progress_callback(message: str): - # 从消息中提取文件信息更新任务 - if "处理文件:" in message: - task.update(message=message) - - self.service.set_progress_callback(progress_callback) - - # 执行批量上传 - result = self.service.batch_upload(input_directory, tag_list, recursive) - - task.finish(f"批量上传完成: {result['successful']}/{result['total_files']} 成功") - - # 显示结果摘要 - self._show_batch_result(result) - - # 保存结果文件 - if output_path: - self._save_batch_results(result, output_path, output_format) - console.print(f"📄 结果已保存到: {output_path}") - - @self.app.command() - def list_uploads( - limit: int = typer.Option( - 10, - "--limit", "-l", - min=1, - max=100, - help="📊 显示数量限制" - ), - verbose: bool = typer.Option( - False, - "--verbose", "-v", - help="📝 详细信息" - ) - ): - """ - 📋 列出最近的上传记录 - - 显示最近上传的视频文件信息和统计数据。 - """ - console.print("📋 最近的上传记录") - - try: - uploads = self.service.get_recent_uploads(limit) - - if not uploads: - console.print("⚠️ [yellow]暂无上传记录[/yellow]") - return - - self._show_uploads_table(uploads, verbose) - - except Exception as e: - console.print(f"❌ [red]获取记录失败: {e}[/red]") - raise typer.Exit(1) - - @self.app.command() - def status(): - """ - 📊 显示系统状态 - - 显示媒体管理器的当前状态和统计信息。 - """ - console.print("📊 媒体管理器状态") - - try: - # 获取统计信息 - uploads_stats = self.service.get_collection_stats("uploads") - batch_stats = self.service.get_collection_stats("batch_uploads") - - # 创建状态表格 - table = Table(title="系统状态") - table.add_column("组件", style="cyan") - table.add_column("状态", style="green") - table.add_column("详情", style="yellow") - - table.add_row("存储", "✅ 正常", f"JSON文件存储") - table.add_row("上传记录", "✅ 活跃", f"{uploads_stats.get('file_count', 0)} 条记录") - table.add_row("批量任务", "✅ 就绪", f"{batch_stats.get('file_count', 0)} 个任务") - table.add_row("进度系统", "✅ 就绪", "JSON-RPC协议") - - console.print(table) - - # 创建信息面板 - info_panel = Panel( - "[bold blue]媒体管理器运行正常[/bold blue]\n" - "所有组件状态良好，可以开始处理视频文件。", - title="📊 状态摘要", - border_style="green" - ) - console.print(info_panel) - - except Exception as e: - console.print(f"❌ [red]获取状态失败: {e}[/red]") - raise typer.Exit(1) - - def _show_upload_result(self, result: dict): - """显示上传结果""" - table = Table(title="📤 上传结果") - table.add_column("项目", style="cyan") - table.add_column("值", style="green") - - table.add_row("文件名", result["filename"]) - table.add_row("标签", ", ".join(result["tags"]) if result["tags"] else "无") - table.add_row("检测场景", str(result["scenes_detected"])) - table.add_row("视频时长", f"{result['duration']:.1f}秒") - - console.print(table) - - def _show_batch_result(self, result: dict): - """显示批量结果摘要""" - panel = Panel( - f"[bold green]总文件: {result['total_files']}[/bold green]\n" - f"[bold blue]成功: {result['successful']}[/bold blue]\n" - f"[bold red]失败: {result['failed']}[/bold red]", - title="📦 批量上传摘要", - border_style="green" - ) - console.print(panel) - - def _show_uploads_table(self, uploads: List[dict], verbose: bool): - """显示上传记录表格""" - table = Table(title="📋 上传记录") - table.add_column("文件名", style="cyan") - table.add_column("场景数", style="green") - table.add_column("时长", style="yellow") - - if verbose: - table.add_column("标签", style="magenta") - - for upload in uploads: - row = [ - upload["filename"], - str(upload.get("scenes_detected", "N/A")), - f"{upload.get('duration', 0):.1f}s" - ] - - if verbose: - tags = ", ".join(upload.get("tags", [])) or "无" - row.append(tags) - - table.add_row(*row) - - console.print(table) - - def _save_batch_results(self, result: dict, output_path: Path, format: OutputFormat): - """保存批量结果""" - if format == OutputFormat.JSON: - import json - with open(output_path, 'w', encoding='utf-8') as f: - json.dump(result, f, indent=2, ensure_ascii=False) - elif format == OutputFormat.CSV: - import csv - with open(output_path, 'w', newline='', encoding='utf-8') as f: - writer = csv.writer(f) - writer.writerow(['filename', 'success', 'scenes', 'duration', 'error']) - for item in result['results']: - writer.writerow([ - item.get('filename', ''), - item.get('success', False), - item.get('scenes_detected', ''), - item.get('duration', ''), - item.get('error', '') - ]) - else: # TXT - with open(output_path, 'w', encoding='utf-8') as f: - f.write(f"批量上传结果\n") - f.write(f"总文件数: {result['total_files']}\n") - f.write(f"成功: {result['successful']}\n") - f.write(f"失败: {result['failed']}\n\n") - - for item in result['results']: - f.write(f"文件: {item.get('filename', '')}\n") - f.write(f" 状态: {'成功' if item.get('success') else '失败'}\n") - if item.get('success'): - f.write(f" 场景数: {item.get('scenes_detected', 'N/A')}\n") - f.write(f" 时长: {item.get('duration', 0):.1f}秒\n") - else: - f.write(f" 错误: {item.get('error', '')}\n") - f.write("\n") - - def run(self): - """运行CLI""" - self.app() - -def main(): - """主入口函数""" - commander = MediaManagerCommander() - commander.run() - -if __name__ == "__main__": - main() diff --git a/tests/test_media_manager.py b/tests/test_media_manager.py deleted file mode 100644 index d3ce1d9..0000000 --- a/tests/test_media_manager.py +++ /dev/null @@ -1,403 +0,0 @@ -#!/usr/bin/env python3 -""" -MediaManager 测试文件 -测试视频切分、场景检测、信息提取等功能 -""" - -import os -import sys -import unittest -import tempfile -import shutil -from pathlib import Path -from unittest.mock import Mock, patch - -# 添加项目根目录到Python路径 -project_root = Path(__file__).parent.parent -sys.path.insert(0, str(project_root)) - -# 导入要测试的模块 -from python_core.services.media_manager import ( - MediaManager, - DependencyManager, - VideoProcessorFactory, - FFProbeVideoInfoExtractor, - OpenCVVideoInfoExtractor, - PySceneDetectSceneDetector, - OpenCVSceneDetector, - OpenCVThumbnailGenerator, - MD5FileHashCalculator, - OpenCVVideoSegmentCreator -) - -class TestDependencyManager(unittest.TestCase): - """测试依赖管理器""" - - def setUp(self): - self.dependency_manager = DependencyManager() - - def test_dependency_initialization(self): - """测试依赖初始化""" - # 检查依赖管理器是否正确初始化 - self.assertIsInstance(self.dependency_manager._dependencies, dict) - - # 检查是否包含预期的依赖 - self.assertIn('opencv', self.dependency_manager._dependencies) - self.assertIn('scenedetect', self.dependency_manager._dependencies) - - def test_opencv_availability(self): - """测试OpenCV可用性检查""" - opencv_available = self.dependency_manager.is_available('opencv') - print(f"OpenCV available: {opencv_available}") - - if opencv_available: - # 如果OpenCV可用，测试模块获取 - cv2 = self.dependency_manager.get_module('opencv', 'cv2') - self.assertIsNotNone(cv2) - - numpy = self.dependency_manager.get_module('opencv', 'numpy') - self.assertIsNotNone(numpy) - - def test_scenedetect_availability(self): - """测试PySceneDetect可用性检查""" - scenedetect_available = self.dependency_manager.is_available('scenedetect') - print(f"PySceneDetect available: {scenedetect_available}") - - if scenedetect_available: - # 如果PySceneDetect可用，测试模块获取 - VideoManager = self.dependency_manager.get_module('scenedetect', 'VideoManager') - self.assertIsNotNone(VideoManager) - - def test_dependency_info(self): - """测试依赖信息获取""" - opencv_info = self.dependency_manager.get_dependency_info('opencv') - self.assertIsInstance(opencv_info, dict) - self.assertIn('available', opencv_info) - - all_deps = self.dependency_manager.get_all_dependencies() - self.assertIsInstance(all_deps, dict) - self.assertIn('opencv', all_deps) - self.assertIn('scenedetect', all_deps) - -class TestVideoProcessorFactory(unittest.TestCase): - """测试视频处理器工厂""" - - def setUp(self): - self.dependency_manager = DependencyManager() - self.factory = VideoProcessorFactory(self.dependency_manager) - - def test_create_video_info_extractor(self): - """测试创建视频信息提取器""" - try: - extractor = self.factory.create_video_info_extractor() - self.assertIsNotNone(extractor) - print(f"Created video info extractor: {type(extractor).__name__}") - except Exception as e: - self.fail(f"Failed to create video info extractor: {e}") - - def test_create_scene_detector(self): - """测试创建场景检测器""" - try: - detector = self.factory.create_scene_detector() - self.assertIsNotNone(detector) - print(f"Created scene detector: {type(detector).__name__}") - except Exception as e: - self.fail(f"Failed to create scene detector: {e}") - - def test_create_thumbnail_generator(self): - """测试创建缩略图生成器""" - if self.dependency_manager.is_available('opencv'): - try: - generator = self.factory.create_thumbnail_generator() - self.assertIsNotNone(generator) - print(f"Created thumbnail generator: {type(generator).__name__}") - except Exception as e: - self.fail(f"Failed to create thumbnail generator: {e}") - else: - print("OpenCV not available, skipping thumbnail generator test") - - def test_create_hash_calculator(self): - """测试创建哈希计算器""" - calculator = self.factory.create_hash_calculator() - self.assertIsNotNone(calculator) - self.assertIsInstance(calculator, MD5FileHashCalculator) - -class TestVideoInfoExtractors(unittest.TestCase): - """测试视频信息提取器""" - - def setUp(self): - # 使用assets文件夹中的测试视频 - self.test_video = project_root / "assets" / "1" / "1752031789460.mp4" - if not self.test_video.exists(): - self.skipTest(f"Test video not found: {self.test_video}") - - def test_ffprobe_extractor(self): - """测试FFProbe视频信息提取器""" - try: - extractor = FFProbeVideoInfoExtractor() - info = extractor.extract_video_info(str(self.test_video)) - - # 验证返回的信息结构 - self.assertIsInstance(info, dict) - self.assertIn('duration', info) - self.assertIn('width', info) - self.assertIn('height', info) - self.assertIn('fps', info) - self.assertIn('file_size', info) - - print(f"FFProbe video info: {info}") - - # 验证数值合理性 - self.assertGreater(info['duration'], 0) - self.assertGreater(info['width'], 0) - self.assertGreater(info['height'], 0) - self.assertGreater(info['fps'], 0) - self.assertGreater(info['file_size'], 0) - - except Exception as e: - print(f"FFProbe extractor failed (expected if ffprobe not available): {e}") - - def test_opencv_extractor(self): - """测试OpenCV视频信息提取器""" - dependency_manager = DependencyManager() - if not dependency_manager.is_available('opencv'): - self.skipTest("OpenCV not available") - - try: - extractor = OpenCVVideoInfoExtractor(dependency_manager) - info = extractor.extract_video_info(str(self.test_video)) - - # 验证返回的信息结构 - self.assertIsInstance(info, dict) - self.assertIn('duration', info) - self.assertIn('width', info) - self.assertIn('height', info) - self.assertIn('fps', info) - self.assertIn('file_size', info) - - print(f"OpenCV video info: {info}") - - # 验证数值合理性 - self.assertGreater(info['duration'], 0) - self.assertGreater(info['width'], 0) - self.assertGreater(info['height'], 0) - self.assertGreater(info['fps'], 0) - self.assertGreater(info['file_size'], 0) - - except Exception as e: - self.fail(f"OpenCV extractor failed: {e}") - -class TestSceneDetectors(unittest.TestCase): - """测试场景检测器""" - - def setUp(self): - # 使用assets文件夹中的测试视频 - self.test_video = project_root / "assets" / "1" / "1752031789460.mp4" - if not self.test_video.exists(): - self.skipTest(f"Test video not found: {self.test_video}") - - self.dependency_manager = DependencyManager() - - def test_pyscenedetect_detector(self): - """测试PySceneDetect场景检测器""" - if not self.dependency_manager.is_available('scenedetect'): - self.skipTest("PySceneDetect not available") - - try: - detector = PySceneDetectSceneDetector(self.dependency_manager) - scene_changes = detector.detect_scenes(str(self.test_video), threshold=30.0) - - # 验证返回的场景变化点 - self.assertIsInstance(scene_changes, list) - self.assertGreater(len(scene_changes), 0) - - # 第一个应该是0.0 - self.assertEqual(scene_changes[0], 0.0) - - # 场景变化点应该是递增的 - for i in range(1, len(scene_changes)): - self.assertGreater(scene_changes[i], scene_changes[i-1]) - - print(f"PySceneDetect found {len(scene_changes)} scene changes: {scene_changes}") - - except Exception as e: - print(f"PySceneDetect detector failed: {e}") - - def test_opencv_detector(self): - """测试OpenCV场景检测器""" - if not self.dependency_manager.is_available('opencv'): - self.skipTest("OpenCV not available") - - try: - detector = OpenCVSceneDetector(self.dependency_manager) - scene_changes = detector.detect_scenes(str(self.test_video), threshold=30.0) - - # 验证返回的场景变化点 - self.assertIsInstance(scene_changes, list) - self.assertGreater(len(scene_changes), 0) - - # 第一个应该是0.0 - self.assertEqual(scene_changes[0], 0.0) - - # 场景变化点应该是递增的 - for i in range(1, len(scene_changes)): - self.assertGreater(scene_changes[i], scene_changes[i-1]) - - print(f"OpenCV found {len(scene_changes)} scene changes: {scene_changes}") - - except Exception as e: - self.fail(f"OpenCV detector failed: {e}") - -class TestMediaManager(unittest.TestCase): - """测试媒体管理器""" - - def setUp(self): - # 创建临时目录用于测试 - self.temp_dir = tempfile.mkdtemp() - self.test_video = project_root / "assets" / "1" / "1752031789460.mp4" - - if not self.test_video.exists(): - self.skipTest(f"Test video not found: {self.test_video}") - - # 使用临时目录创建MediaManager - with patch('python_core.config.settings') as mock_settings: - mock_settings.temp_dir = Path(self.temp_dir) - self.media_manager = MediaManager() - - def tearDown(self): - # 清理临时目录 - shutil.rmtree(self.temp_dir, ignore_errors=True) - - def test_media_manager_initialization(self): - """测试媒体管理器初始化""" - self.assertIsNotNone(self.media_manager) - self.assertIsNotNone(self.media_manager.dependency_manager) - self.assertIsNotNone(self.media_manager.factory) - - def test_video_info_extraction(self): - """测试视频信息提取""" - try: - video_info = self.media_manager._get_video_info(str(self.test_video)) - - self.assertIsInstance(video_info, dict) - self.assertIn('duration', video_info) - self.assertIn('width', video_info) - self.assertIn('height', video_info) - self.assertIn('fps', video_info) - - print(f"Video info extracted: {video_info}") - - except Exception as e: - self.fail(f"Video info extraction failed: {e}") - - def test_scene_detection(self): - """测试场景检测""" - try: - scene_changes = self.media_manager._detect_scene_changes(str(self.test_video)) - - self.assertIsInstance(scene_changes, list) - self.assertGreater(len(scene_changes), 0) - - print(f"Scene changes detected: {scene_changes}") - - except Exception as e: - self.fail(f"Scene detection failed: {e}") - - def test_video_splitting(self): - """测试视频切分功能""" - try: - # 首先检测场景变化 - scene_changes = self.media_manager._detect_scene_changes(str(self.test_video)) - - # 如果场景变化太少，手动添加一些切分点 - if len(scene_changes) < 3: - video_info = self.media_manager._get_video_info(str(self.test_video)) - duration = video_info.get('duration', 10.0) - scene_changes = [0.0, duration / 2, duration] - - print(f"Using scene changes for splitting: {scene_changes}") - - # 执行视频切分 - segments = self.media_manager._split_video_by_scenes( - str(self.test_video), - scene_changes, - "test_original_id", - ["测试", "分镜"] - ) - - print(f"Created {len(segments)} video segments") - - # 验证切分结果 - self.assertIsInstance(segments, list) - - for i, segment in enumerate(segments): - print(f"Segment {i}: {segment.filename}, duration: {segment.duration}s") - - # 验证片段文件是否存在（如果创建成功的话） - if hasattr(segment, 'file_path') and segment.file_path: - segment_path = Path(segment.file_path) - if segment_path.exists(): - print(f" File exists: {segment_path}") - self.assertGreater(segment_path.stat().st_size, 0) - - except Exception as e: - print(f"Video splitting failed (may be expected if dependencies missing): {e}") - -def run_comprehensive_test(): - """运行全面的测试""" - print("=" * 60) - print("MediaManager 综合测试") - print("=" * 60) - - # 检查测试视频 - test_videos = list((project_root / "assets").rglob("*.mp4")) - print(f"Found {len(test_videos)} test videos:") - for video in test_videos[:5]: # 只显示前5个 - print(f" - {video}") - - if not test_videos: - print("❌ No test videos found in assets folder") - return - - print("\n" + "=" * 60) - print("开始测试...") - print("=" * 60) - - # 运行测试套件 - loader = unittest.TestLoader() - suite = unittest.TestSuite() - - # 添加测试类 - suite.addTests(loader.loadTestsFromTestCase(TestDependencyManager)) - suite.addTests(loader.loadTestsFromTestCase(TestVideoProcessorFactory)) - suite.addTests(loader.loadTestsFromTestCase(TestVideoInfoExtractors)) - suite.addTests(loader.loadTestsFromTestCase(TestSceneDetectors)) - suite.addTests(loader.loadTestsFromTestCase(TestMediaManager)) - - # 运行测试 - runner = unittest.TextTestRunner(verbosity=2) - result = runner.run(suite) - - print("\n" + "=" * 60) - print("测试总结") - print("=" * 60) - print(f"运行测试: {result.testsRun}") - print(f"失败: {len(result.failures)}") - print(f"错误: {len(result.errors)}") - print(f"跳过: {len(result.skipped) if hasattr(result, 'skipped') else 0}") - - if result.failures: - print("\n失败的测试:") - for test, traceback in result.failures: - print(f" - {test}: {traceback}") - - if result.errors: - print("\n错误的测试:") - for test, traceback in result.errors: - print(f" - {test}: {traceback}") - - return result.wasSuccessful() - -if __name__ == "__main__": - success = run_comprehensive_test() - sys.exit(0 if success else 1) diff --git a/tests/textVideoAgentAPI.test.ts b/tests/textVideoAgentAPI.test.ts deleted file mode 100644 index 1e98ccd..0000000 --- a/tests/textVideoAgentAPI.test.ts +++ /dev/null @@ -1,442 +0,0 @@ -/** - * Text Video Agent API 测试文件 - */ - -import { textVideoAgentAPI, TextVideoAgentAPI } from '../src/services/textVideoAgentAPI' -import { TaskType, AspectRatio, VideoDuration } from '../src/services/textVideoAgentTypes' - -// Mock fetch for testing -global.fetch = jest.fn() - -describe('TextVideoAgentAPI', () => { - beforeEach(() => { - jest.clearAllMocks() - }) - - describe('基础功能测试', () => { - test('健康检查', async () => { - const mockResponse = { - status: true, - msg: 'Service is healthy' - } - - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - json: async () => mockResponse - }) - - const result = await textVideoAgentAPI.healthCheck() - - expect(fetch).toHaveBeenCalledWith( - expect.stringContaining('/health'), - expect.any(Object) - ) - expect(result).toEqual(mockResponse) - }) - - test('获取示例提示词', async () => { - const mockResponse = { - status: true, - msg: 'Success', - data: { - prompts: ['示例提示词1', '示例提示词2'] - } - } - - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - json: async () => mockResponse - }) - - const result = await textVideoAgentAPI.getSamplePrompt(TaskType.VLOG) - - expect(fetch).toHaveBeenCalledWith( - expect.stringContaining('/api/prompt/default?task_type=vlog'), - expect.any(Object) - ) - expect(result).toEqual(mockResponse) - }) - }) - - describe('图片生成测试', () => { - test('同步生成图片', async () => { - const mockResponse = { - status: true, - msg: 'Image generated successfully', - data: { - image_url: 'https://example.com/generated-image.jpg' - } - } - - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - json: async () => mockResponse - }) - - const result = await textVideoAgentAPI.generateImageSync({ - prompt: '测试图片生成', - max_wait_time: 60 - }) - - expect(result).toEqual(mockResponse) - expect(result.data?.image_url).toBe('https://example.com/generated-image.jpg') - }) - - test('异步生成图片', async () => { - const mockResponse = { - status: true, - msg: 'Task submitted', - data: { - task_id: 'test-task-123' - } - } - - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - json: async () => mockResponse - }) - - const result = await textVideoAgentAPI.generateImageAsync('测试异步图片生成') - - expect(result).toEqual(mockResponse) - expect(result.data?.task_id).toBe('test-task-123') - }) - - test('带重试的图片生成', async () => { - // 第一次失败 - ;(fetch as jest.Mock).mockRejectedValueOnce(new Error('Network error')) - - // 第二次成功 - const mockResponse = { - status: true, - msg: 'Image generated successfully', - data: { - image_url: 'https://example.com/generated-image.jpg' - } - } - - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - json: async () => mockResponse - }) - - const result = await textVideoAgentAPI.generateImageWithRetry({ - prompt: '测试重试机制', - max_wait_time: 60 - }, 2) - - expect(fetch).toHaveBeenCalledTimes(2) - expect(result).toEqual(mockResponse) - }) - }) - - describe('视频生成测试', () => { - test('同步生成视频', async () => { - const mockResponse = { - status: true, - msg: 'Video generated successfully', - data: { - video_url: 'https://example.com/generated-video.mp4' - } - } - - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - json: async () => mockResponse - }) - - const result = await textVideoAgentAPI.generateVideoSync({ - prompt: '测试视频生成', - img_url: 'https://example.com/base-image.jpg', - duration: '5' - }) - - expect(result).toEqual(mockResponse) - expect(result.data?.video_url).toBe('https://example.com/generated-video.mp4') - }) - - test('异步生成视频', async () => { - const mockResponse = { - status: true, - msg: 'Video task submitted', - data: { - task_id: 'video-task-456' - } - } - - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - json: async () => mockResponse - }) - - const result = await textVideoAgentAPI.generateVideoAsync({ - prompt: '测试异步视频生成', - duration: '5' - }) - - expect(result).toEqual(mockResponse) - expect(result.data?.task_id).toBe('video-task-456') - }) - }) - - describe('任务管理测试', () => { - test('创建任务', async () => { - const mockResponse = { - status: true, - msg: 'Task created successfully', - data: { - task_id: 'new-task-789' - } - } - - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - json: async () => mockResponse - }) - - const result = await textVideoAgentAPI.createTask({ - task_type: TaskType.VLOG, - prompt: '测试任务创建', - ar: AspectRatio.PORTRAIT - }) - - expect(result).toEqual(mockResponse) - expect(result.data?.task_id).toBe('new-task-789') - }) - - test('查询任务状态', async () => { - const mockResponse = { - status: true, - msg: 'Task status retrieved', - data: { - task_id: 'test-task-123', - status: 'completed', - progress: 100, - result: { - image_url: 'https://example.com/result.jpg' - } - } - } - - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - json: async () => mockResponse - }) - - const result = await textVideoAgentAPI.getTaskStatusAsync('test-task-123') - - expect(result).toEqual(mockResponse) - expect(result.data?.status).toBe('completed') - }) - }) - - describe('文件操作测试', () => { - test('文件上传', async () => { - const mockResponse = { - status: true, - msg: 'File uploaded successfully', - data: 'https://example.com/uploaded-file.jpg' - } - - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - json: async () => mockResponse - }) - - const mockFile = new File(['test content'], 'test.jpg', { type: 'image/jpeg' }) - const result = await textVideoAgentAPI.uploadFile(mockFile) - - expect(result).toEqual(mockResponse) - expect(result.data).toBe('https://example.com/uploaded-file.jpg') - }) - }) - - describe('图片描述测试', () => { - test('通过URL描述图片', async () => { - const mockResponse = { - status: true, - msg: 'Image described successfully', - data: { - description: '这是一张美丽的风景照片' - } - } - - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - json: async () => mockResponse - }) - - const result = await textVideoAgentAPI.describeImageByUrl({ - image_url: 'https://example.com/image.jpg' - }) - - expect(result).toEqual(mockResponse) - expect(result.data?.description).toBe('这是一张美丽的风景照片') - }) - - test('通过文件描述图片', async () => { - const mockResponse = { - status: true, - msg: 'Image described successfully', - data: { - description: '这是一张人物照片' - } - } - - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - json: async () => mockResponse - }) - - const mockFile = new File(['image content'], 'portrait.jpg', { type: 'image/jpeg' }) - const result = await textVideoAgentAPI.describeImageByFile({ - img_file: mockFile - }) - - expect(result).toEqual(mockResponse) - expect(result.data?.description).toBe('这是一张人物照片') - }) - }) - - describe('高级功能测试', () => { - test('端到端内容生成', async () => { - // Mock 创建任务 - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - json: async () => ({ - status: true, - data: { task_id: 'end-to-end-task' } - }) - }) - - // Mock 图片生成 - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - json: async () => ({ - status: true, - data: { image_url: 'https://example.com/generated.jpg' } - }) - }) - - // Mock 视频生成 - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - json: async () => ({ - status: true, - data: { video_url: 'https://example.com/generated.mp4' } - }) - }) - - const result = await textVideoAgentAPI.generateContentEndToEnd( - '测试端到端生成', - { - taskType: TaskType.VLOG, - aspectRatio: AspectRatio.PORTRAIT, - videoDuration: VideoDuration.MEDIUM, - generateVideo: true - } - ) - - expect(result).toEqual({ - imageUrl: 'https://example.com/generated.jpg', - videoUrl: 'https://example.com/generated.mp4', - taskId: 'end-to-end-task' - }) - }) - - test('轮询任务直到完成', async () => { - // 第一次查询 - 进行中 - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - json: async () => ({ - status: true, - data: { status: 'running', progress: 50 } - }) - }) - - // 第二次查询 - 完成 - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - json: async () => ({ - status: true, - data: { - status: 'completed', - progress: 100, - result: { image_url: 'https://example.com/final.jpg' } - } - }) - }) - - const progressCallback = jest.fn() - - const result = await textVideoAgentAPI.pollTaskUntilComplete( - 'test-task-id', - 100, // 快速轮询用于测试 - 5000, // 5秒超时 - progressCallback - ) - - expect(fetch).toHaveBeenCalledTimes(2) - expect(progressCallback).toHaveBeenCalledTimes(2) - expect(result.data?.status).toBe('completed') - }) - }) - - describe('错误处理测试', () => { - test('网络错误处理', async () => { - ;(fetch as jest.Mock).mockRejectedValueOnce(new Error('Network error')) - - await expect(textVideoAgentAPI.healthCheck()).rejects.toThrow('Network error') - }) - - test('HTTP错误处理', async () => { - ;(fetch as jest.Mock).mockResolvedValueOnce({ - ok: false, - status: 500, - json: async () => ({ error: 'Internal server error' }) - }) - - await expect(textVideoAgentAPI.healthCheck()).rejects.toThrow('HTTP error! status: 500') - }) - - test('参数验证', async () => { - await expect( - textVideoAgentAPI.describeImageByUrl({} as any) - ).rejects.toThrow('image_url is required') - - await expect( - textVideoAgentAPI.describeImageByFile({} as any) - ).rejects.toThrow('img_file is required') - }) - }) - - describe('自定义配置测试', () => { - test('自定义API端点', () => { - const customAPI = new TextVideoAgentAPI('https://custom-endpoint.com') - expect(customAPI).toBeInstanceOf(TextVideoAgentAPI) - }) - }) -}) - -// 集成测试（需要真实API） -describe('集成测试 (需要真实API)', () => { - // 这些测试需要真实的API端点，通常在CI/CD环境中跳过 - const shouldRunIntegrationTests = process.env.RUN_INTEGRATION_TESTS === 'true' - - test.skipIf(!shouldRunIntegrationTests)('真实健康检查', async () => { - const result = await textVideoAgentAPI.healthCheck() - expect(result.status).toBe(true) - }) - - test.skipIf(!shouldRunIntegrationTests)('真实图片生成', async () => { - const result = await textVideoAgentAPI.generateImageSync({ - prompt: '一朵美丽的花', - max_wait_time: 60 - }) - - expect(result.status).toBe(true) - expect(result.data?.image_url).toBeDefined() - }, 120000) // 2分钟超时 -}) - -export {}