From 31de1e5a4d145133377330a26f0082edf51783f6 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 12 Jul 2025 13:44:57 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E5=91=BD=E4=BB=A4?= =?UTF-8?q?=E8=A1=8C=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- examples/test_jsonrpc_registration.py | 86 ++ examples/test_jsonrpc_workflow.py | 271 ++++++ examples/test_refactored_modules.py | 248 +++++ examples/test_scene_detection_fix.py | 206 +++++ examples/test_workflow_results.py | 303 ++++++ python_core/cli/cli.py | 25 +- python_core/cli/commands/__init__.py | 4 +- python_core/cli/commands/jsonrpc_server.py | 303 ------ python_core/cli/commands/scene.py | 549 ----------- python_core/cli/commands/scene_detect.py | 99 ++ python_core/cli/const.py | 6 - python_core/cli/scene_detect.py | 871 ------------------ python_core/scene_detection/__init__.py | 67 ++ python_core/scene_detection/scene_detector.py | 228 +++++ .../scene_detection/services/__init__.py | 17 + .../services/ai_analysis_service.py | 100 ++ .../services/detector_service.py | 158 ++++ .../services/video_info_service.py | 65 ++ python_core/scene_detection/types/__init__.py | 19 + python_core/scene_detection/types/enums.py | 21 + python_core/scene_detection/types/models.py | 34 + .../scene_detection/types/workflow_state.py | 82 ++ python_core/scene_detection/utils/__init__.py | 15 + .../scene_detection/utils/result_saver.py | 110 +++ .../scene_detection/utils/validators.py | 84 ++ .../scene_detection/workflows/__init__.py | 15 + .../workflows/workflow_manager.py | 176 ++++ .../workflows/workflow_nodes.py | 242 +++++ python_core/utils/jsonrpc_enhanced.py | 65 +- python_core/utils/jsonrpc_server.py | 14 +- 30 files changed, 2728 insertions(+), 1755 deletions(-) create mode 100644 examples/test_jsonrpc_registration.py create mode 100644 examples/test_jsonrpc_workflow.py create mode 100644 examples/test_refactored_modules.py create mode 100644 examples/test_scene_detection_fix.py create mode 100644 examples/test_workflow_results.py delete mode 100644 python_core/cli/commands/jsonrpc_server.py delete mode 100644 python_core/cli/commands/scene.py create mode 100644 python_core/cli/commands/scene_detect.py delete mode 100644 python_core/cli/const.py delete mode 100644 python_core/cli/scene_detect.py create mode 100644 python_core/scene_detection/__init__.py create mode 100644 python_core/scene_detection/scene_detector.py create mode 100644 python_core/scene_detection/services/__init__.py create mode 100644 python_core/scene_detection/services/ai_analysis_service.py create mode 100644 python_core/scene_detection/services/detector_service.py create mode 100644 python_core/scene_detection/services/video_info_service.py create mode 100644 python_core/scene_detection/types/__init__.py create mode 100644 python_core/scene_detection/types/enums.py create mode 100644 python_core/scene_detection/types/models.py create mode 100644 python_core/scene_detection/types/workflow_state.py create mode 100644 python_core/scene_detection/utils/__init__.py create mode 100644 python_core/scene_detection/utils/result_saver.py create mode 100644 python_core/scene_detection/utils/validators.py create mode 100644 python_core/scene_detection/workflows/__init__.py create mode 100644 python_core/scene_detection/workflows/workflow_manager.py create mode 100644 python_core/scene_detection/workflows/workflow_nodes.py diff --git a/examples/test_jsonrpc_registration.py b/examples/test_jsonrpc_registration.py new file mode 100644 index 0000000..9a93029 --- /dev/null +++ b/examples/test_jsonrpc_registration.py @@ -0,0 +1,86 @@ +#!/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 new file mode 100644 index 0000000..46eb9fe --- /dev/null +++ b/examples/test_jsonrpc_workflow.py @@ -0,0 +1,271 @@ +#!/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 new file mode 100644 index 0000000..88d6b7e --- /dev/null +++ b/examples/test_refactored_modules.py @@ -0,0 +1,248 @@ +#!/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 new file mode 100644 index 0000000..5d47e30 --- /dev/null +++ b/examples/test_scene_detection_fix.py @@ -0,0 +1,206 @@ +#!/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 new file mode 100644 index 0000000..d18374a --- /dev/null +++ b/examples/test_workflow_results.py @@ -0,0 +1,303 @@ +#!/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/python_core/cli/cli.py b/python_core/cli/cli.py index bb379fb..3a4b5f6 100644 --- a/python_core/cli/cli.py +++ b/python_core/cli/cli.py @@ -4,14 +4,11 @@ MixVideo 主命令行接口 """ import sys -from pathlib import Path -from typing import Optional -from python_core.cli.const import progress_reporter, console, project_root +from python_core.utils.logger import logger import typer # 导入命令模块 -from python_core.cli.commands import scene_app -from python_core.cli.commands.jsonrpc_server import jsonrpc_app +from python_core.cli.commands import scene_detect app = typer.Typer( name="mixvideo", @@ -23,38 +20,30 @@ app = typer.Typer( • 📤 媒体管理 - 上传、处理、组织视频文件 • 📋 模板管理 - 视频模板导入导出 • ⚙️ 系统管理 - 配置、状态、存储管理 - - 快速开始: - mixvideo scene detect video.mp4 # 检测场景 - mixvideo scene batch-detect /videos # 批量检测 - mixvideo scene split video.mp4 # 分割视频 - mixvideo scene info video.mp4 # 视频信息 - mixvideo jsonrpc start # 启动JSON-RPC服务器 """, rich_markup_mode="rich", no_args_is_help=True ) # 添加命令组到主应用 -app.add_typer(scene_app, name="scene") -app.add_typer(jsonrpc_app, name="jsonrpc") +app.add_typer(scene_detect, name="scene") @app.command() def init(): """🚀 初始化MixVideo工作环境""" - progress_reporter.info("🚀 初始化MixVideo环境...") + logger.info("🚀 初始化MixVideo环境...") # TODO: 实现初始化逻辑 - progress_reporter.success("✅ 初始化完成") + logger.success("✅ 初始化完成") def main(): """主入口函数""" try: app() except KeyboardInterrupt: - progress_reporter.error("\n👋 用户取消操作") + logger.error("\n👋 用户取消操作") sys.exit(0) except Exception as e: - progress_reporter.error(f"\n❌ [red]程序异常: {e}[/red]") + logger.error(f"\n❌ [red]程序异常: {e}[/red]") sys.exit(1) if __name__ == "__main__": diff --git a/python_core/cli/commands/__init__.py b/python_core/cli/commands/__init__.py index 57cf584..88de5dc 100644 --- a/python_core/cli/commands/__init__.py +++ b/python_core/cli/commands/__init__.py @@ -3,6 +3,6 @@ CLI 命令模块 """ -from .scene import scene_app +from .scene_detect import scene_detect -__all__ = ["scene_app"] +__all__ = ["scene_detect"] diff --git a/python_core/cli/commands/jsonrpc_server.py b/python_core/cli/commands/jsonrpc_server.py deleted file mode 100644 index 7d0b9f0..0000000 --- a/python_core/cli/commands/jsonrpc_server.py +++ /dev/null @@ -1,303 +0,0 @@ -#!/usr/bin/env python3 -""" -JSON-RPC Server Command -JSON-RPC 服务器命令 - -Provides HTTP and WebSocket JSON-RPC server for scene detection services. -""" - -from pathlib import Path -from typing import Optional -import signal -import sys - -import typer -from python_core.cli.const import console -from python_core.utils.jsonrpc_server import JSONRPCServer, JSONRPCWebSocketServer, ServerConfig - -# 创建子应用 -jsonrpc_app = typer.Typer(help="🌐 JSON-RPC 服务器") - - -@jsonrpc_app.command() -def start( - host: str = typer.Option("localhost", help="🌐 服务器主机地址"), - port: int = typer.Option(8080, help="🔌 服务器端口"), - debug: bool = typer.Option(False, help="🐛 启用调试模式"), - cors: bool = typer.Option(True, help="🔗 启用CORS支持"), - websocket: bool = typer.Option(False, help="🔌 启用WebSocket服务器"), - max_request_size: int = typer.Option(1024*1024, help="📦 最大请求大小（字节）") -): - """🚀 启动JSON-RPC服务器""" - - config = ServerConfig( - host=host, - port=port, - debug=debug, - cors_enabled=cors, - max_request_size=max_request_size - ) - - console.print(f"🚀 [bold blue]启动JSON-RPC服务器[/bold blue]") - console.print(f"📍 地址: {host}:{port}") - console.print(f"🔧 模式: {'WebSocket' if websocket else 'HTTP'}") - console.print(f"🐛 调试: {'启用' if debug else '禁用'}") - console.print(f"🔗 CORS: {'启用' if cors else '禁用'}") - - # 导入并注册所有JSON-RPC方法 - console.print("📋 注册JSON-RPC方法...") - try: - # 导入场景检测模块以注册方法 - from python_core.cli.scene_detect import detector - console.print("✅ 场景检测方法已注册") - except Exception as e: - console.print(f"⚠️ 注册方法时出错: {e}") - - try: - if websocket: - # WebSocket服务器 - import asyncio - - server = JSONRPCWebSocketServer(config) - - # 注册信号处理 - def signal_handler(sig, frame): - console.print("\n🛑 [yellow]收到停止信号，正在关闭服务器...[/yellow]") - sys.exit(0) - - signal.signal(signal.SIGINT, signal_handler) - signal.signal(signal.SIGTERM, signal_handler) - - # 启动异步服务器 - asyncio.run(server.start()) - - else: - # HTTP服务器 - server = JSONRPCServer(config) - - # 注册信号处理 - def signal_handler(sig, frame): - console.print("\n🛑 [yellow]收到停止信号，正在关闭服务器...[/yellow]") - server.stop() - sys.exit(0) - - signal.signal(signal.SIGINT, signal_handler) - signal.signal(signal.SIGTERM, signal_handler) - - # 启动服务器 - server.start(blocking=True) - - except Exception as e: - console.print(f"❌ [red]服务器启动失败: {e}[/red]") - raise typer.Exit(1) - - -@jsonrpc_app.command() -def test( - host: str = typer.Option("localhost", help="🌐 服务器主机地址"), - port: int = typer.Option(8080, help="🔌 服务器端口"), - method: str = typer.Option("scene.detect", help="🎯 测试方法名"), - video_path: Optional[str] = typer.Option(None, help="📹 测试视频路径") -): - """🧪 测试JSON-RPC服务器""" - - import requests - import json - - if not video_path: - video_path = "assets/1/1752032011698.mp4" # 默认测试视频 - - console.print(f"🧪 [bold blue]测试JSON-RPC服务器[/bold blue]") - console.print(f"📍 服务器: http://{host}:{port}") - console.print(f"🎯 方法: {method}") - console.print(f"📹 视频: {video_path}") - - # 准备测试请求 - test_requests = { - "scene.detect": { - "video_path": video_path, - "detector_type": "content", - "threshold": 30.0, - "min_scene_length": 1.0 - }, - "scene.get_video_info": { - "video_path": video_path - }, - "scene.detect_workflow": { - "video_path": video_path, - "detector_type": "content", - "threshold": 15.0, - "enable_ai_analysis": False - } - } - - params = test_requests.get(method, {"video_path": video_path}) - - payload = { - "jsonrpc": "2.0", - "method": method, - "params": params, - "id": 1 - } - - try: - console.print("📤 发送请求...") - response = requests.post( - f"http://{host}:{port}", - json=payload, - headers={'Content-Type': 'application/json'}, - timeout=30 - ) - - console.print(f"📥 响应状态: {response.status_code}") - - if response.status_code == 200: - result = response.json() - console.print("✅ [green]请求成功[/green]") - console.print("📋 响应内容:") - console.print(json.dumps(result, indent=2, ensure_ascii=False)) - else: - console.print(f"❌ [red]请求失败: {response.text}[/red]") - - except requests.exceptions.ConnectionError: - console.print(f"❌ [red]无法连接到服务器 http://{host}:{port}[/red]") - console.print("💡 请确保服务器已启动") - raise typer.Exit(1) - except Exception as e: - console.print(f"❌ [red]测试失败: {e}[/red]") - raise typer.Exit(1) - - -@jsonrpc_app.command() -def methods(): - """📋 列出可用的JSON-RPC方法""" - - console.print("📋 [bold blue]可用的JSON-RPC方法[/bold blue]") - console.print("=" * 60) - - methods_info = [ - { - "method": "scene.detect", - "description": "基础场景检测", - "params": ["video_path", "detector_type?", "threshold?", "min_scene_length?"] - }, - { - "method": "scene.detect_workflow", - "description": "LangGraph工作流场景检测", - "params": ["video_path", "detector_type?", "threshold?", "min_scene_length?", - "output_path?", "output_format?", "enable_ai_analysis?"] - }, - { - "method": "scene.get_video_info", - "description": "获取视频信息", - "params": ["video_path"] - }, - { - "method": "scene.batch_detect", - "description": "批量场景检测", - "params": ["directory", "detector_type?", "threshold?", "min_scene_length?", - "output_dir?", "output_format?"] - } - ] - - for info in methods_info: - console.print(f"\n🎯 [bold]{info['method']}[/bold]") - console.print(f" 📝 {info['description']}") - console.print(f" 📋 参数: {', '.join(info['params'])}") - - console.print("\n💡 [yellow]参数说明:[/yellow]") - console.print(" • ? 表示可选参数") - console.print(" • detector_type: content/threshold/adaptive") - console.print(" • output_format: json/csv/txt") - console.print(" • threshold: 0-100") - - -@jsonrpc_app.command() -def client_example(): - """📖 显示客户端调用示例""" - - console.print("📖 [bold blue]JSON-RPC 客户端调用示例[/bold blue]") - console.print("=" * 60) - - examples = [ - { - "title": "Python requests 示例", - "code": '''import requests -import json - -def call_scene_detect(video_path, threshold=30.0): - payload = { - "jsonrpc": "2.0", - "method": "scene.detect", - "params": { - "video_path": video_path, - "threshold": threshold - }, - "id": 1 - } - - response = requests.post( - "http://localhost:8080", - json=payload, - headers={'Content-Type': 'application/json'} - ) - - return response.json() - -# 调用示例 -result = call_scene_detect("video.mp4", 15.0) -print(result)''' - }, - { - "title": "curl 命令示例", - "code": '''curl -X POST http://localhost:8080 \\ - -H "Content-Type: application/json" \\ - -d '{ - "jsonrpc": "2.0", - "method": "scene.detect", - "params": { - "video_path": "video.mp4", - "threshold": 15.0 - }, - "id": 1 - }' -''' - }, - { - "title": "JavaScript fetch 示例", - "code": '''async function detectScenes(videoPath, threshold = 30.0) { - const response = await fetch('http://localhost:8080', { - method: 'POST', - headers: { - 'Content-Type': 'application/json', - }, - body: JSON.stringify({ - jsonrpc: '2.0', - method: 'scene.detect', - params: { - video_path: videoPath, - threshold: threshold - }, - id: 1 - }) - }); - - return await response.json(); -} - -// 调用示例 -detectScenes('video.mp4', 15.0).then(result => { - console.log(result); -});''' - } - ] - - for example in examples: - console.print(f"\n📝 [bold]{example['title']}[/bold]") - console.print("```") - console.print(example['code']) - console.print("```") - - -if __name__ == "__main__": - jsonrpc_app() diff --git a/python_core/cli/commands/scene.py b/python_core/cli/commands/scene.py deleted file mode 100644 index d0234d8..0000000 --- a/python_core/cli/commands/scene.py +++ /dev/null @@ -1,549 +0,0 @@ -#!/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)"), - use_workflow: bool = typer.Option(False, "--workflow", help="🔄 使用LangGraph工作流"), - enable_ai: bool = typer.Option(True, "--ai/--no-ai", help="🧠 启用AI分析") -): - """🎯 检测单个视频的场景""" - 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) - - # 选择执行方式 - if use_workflow: - # 使用LangGraph工作流 - progress_reporter.info("🔄 使用LangGraph工作流进行检测...") - - workflow_result = scene_detector.detect_with_workflow( - video_path, detector_type, threshold, min_scene_length, - output, output_format, enable_ai - ) - - result = workflow_result.get("detection_result") - ai_analysis = workflow_result.get("ai_analysis") - video_info = workflow_result.get("video_info") - errors = workflow_result.get("errors", []) - - if errors: - for error in errors: - progress_reporter.error(f"❌ {error}") - raise typer.Exit(1) - - if not result or not result.success: - progress_reporter.error(f"❌ 工作流检测失败: {result.error if result else '未知错误'}") - raise typer.Exit(1) - - # 显示工作流结果 - console.print(f"🔄 LangGraph工作流检测完成") - 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 video_info: - console.print(f"\n📹 视频信息:") - console.print(f" 分辨率: {video_info.get('resolution', 'Unknown')}") - console.print(f" 帧率: {video_info.get('fps', 0):.2f} fps") - console.print(f" 总帧数: {video_info.get('frame_count', 0)}") - - # 显示AI分析结果 - if ai_analysis and enable_ai: - console.print(f"\n🧠 AI分析结果:") - console.print(f"{ai_analysis}") - - # 显示场景详情 - if result.scenes: - console.print(f"\n🎬 场景列表:") - for scene in result.scenes[: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} 个场景") - - return workflow_result - - else: - # 使用传统方法 - 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) - -@scene_app.command() -def workflow( - 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)"), - enable_ai: bool = typer.Option(True, "--ai/--no-ai", help="🧠 启用AI分析"), - interactive: bool = typer.Option(False, "--interactive", "-i", help="🔄 交互式工作流") -): - """🔄 使用LangGraph工作流进行智能场景检测""" - 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) - - console.print("🔄 [bold blue]LangGraph智能场景检测工作流[/bold blue]") - console.print("=" * 60) - - if interactive: - # 交互式模式 - console.print("🎯 交互式模式启动...") - - # 确认参数 - console.print(f"\n📋 检测参数:") - console.print(f" 视频文件: {video_path}") - console.print(f" 检测器: {detector}") - console.print(f" 阈值: {threshold}") - console.print(f" 最小场景长度: {min_scene_length}秒") - console.print(f" AI分析: {'启用' if enable_ai else '禁用'}") - - if not typer.confirm("\n是否继续执行？"): - console.print("❌ 用户取消操作") - return - - # 执行工作流 - progress_reporter.info("🚀 启动LangGraph工作流...") - - workflow_result = scene_detector.detect_with_workflow( - video_path, detector_type, threshold, min_scene_length, - output, output_format, enable_ai - ) - - result = workflow_result.get("detection_result") - ai_analysis = workflow_result.get("ai_analysis") - video_info = workflow_result.get("video_info") - workflow_state = workflow_result.get("workflow_state") - errors = workflow_result.get("errors", []) - - # 检查错误 - if errors: - console.print("\n❌ [red]工作流执行中发现错误:[/red]") - for error in errors: - console.print(f" • {error}") - raise typer.Exit(1) - - if not result or not result.success: - progress_reporter.error(f"❌ 工作流检测失败: {result.error if result else '未知错误'}") - raise typer.Exit(1) - - # 显示完整结果 - console.print("\n" + "=" * 60) - console.print("🎉 [bold green]LangGraph工作流执行完成[/bold green]") - console.print("=" * 60) - - # 工作流状态 - console.print(f"\n🔄 工作流状态: [bold]{workflow_state}[/bold]") - - # 视频信息 - if video_info: - console.print(f"\n📹 [bold]视频信息[/bold]:") - console.print(f" 文件名: {result.filename}") - console.print(f" 分辨率: {video_info.get('resolution', 'Unknown')}") - console.print(f" 帧率: {video_info.get('fps', 0):.2f} fps") - console.print(f" 总帧数: {video_info.get('frame_count', 0):,}") - console.print(f" 时长: {result.total_duration:.2f}秒") - - # 检测结果 - console.print(f"\n🎯 [bold]检测结果[/bold]:") - console.print(f" 检测器类型: {result.detector_type}") - console.print(f" 检测阈值: {result.threshold}") - console.print(f" 场景数量: [bold green]{result.total_scenes}[/bold green]") - console.print(f" 检测耗时: {result.detection_time:.2f}秒") - - # 场景详情 - if result.scenes: - console.print(f"\n🎬 [bold]场景详情[/bold]:") - for scene in result.scenes: - duration_color = "green" if scene.duration >= 2.0 else "yellow" if scene.duration >= 1.0 else "red" - console.print( - f" 场景 {scene.index:2d}: " - f"{scene.start_time:6.2f}s - {scene.end_time:6.2f}s " - f"([{duration_color}]{scene.duration:5.2f}s[/{duration_color}])" - ) - - # AI分析结果 - if ai_analysis and enable_ai: - console.print(f"\n🧠 [bold]AI智能分析[/bold]:") - console.print("-" * 50) - console.print(ai_analysis) - console.print("-" * 50) - elif enable_ai: - console.print(f"\n⚠️ AI分析不可用") - - # 保存信息 - if output: - console.print(f"\n💾 结果已保存到: [bold]{output}[/bold]") - - # 交互式后续操作 - if interactive: - console.print(f"\n🎯 [bold]后续操作选项[/bold]:") - console.print("1. 保存结果到文件") - console.print("2. 调整参数重新检测") - console.print("3. 分割视频") - console.print("4. 退出") - - choice = typer.prompt("请选择操作 (1-4)", type=int, default=4) - - if choice == 1 and not output: - output_path = typer.prompt("请输入输出文件路径", type=str) - scene_detector.save_results(result, Path(output_path), output_format) - console.print(f"✅ 结果已保存到: {output_path}") - - elif choice == 2: - console.print("🔄 参数调整功能开发中...") - - elif choice == 3: - console.print("✂️ 视频分割功能开发中...") - - else: - console.print("👋 感谢使用LangGraph工作流！") - - return workflow_result - - except Exception as e: - progress_reporter.error(f"❌ 工作流命令执行失败: {e}") - raise typer.Exit(1) diff --git a/python_core/cli/commands/scene_detect.py b/python_core/cli/commands/scene_detect.py new file mode 100644 index 0000000..f497958 --- /dev/null +++ b/python_core/cli/commands/scene_detect.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +""" +Scene Detection CLI - Refactored +场景检测命令行工具 - 重构版 + +使用重构后的场景检测模块，代码更简洁、模块化更好。 +""" + +import typer +from pathlib import Path +from typing import Optional, List +from rich.console import Console +from rich.table import Table + +from python_core.scene_detection import ( + SceneDetector, + DetectorType, + OutputFormat +) +from python_core.utils.logger import logger + +scene_detect = typer.Typer(help="场景检测工具 - 重构版") +console = Console() + +@scene_detect.command("split") +def split( + video_path: Path = typer.Argument(..., help="视频文件路径"), + detector_type: DetectorType = typer.Option(DetectorType.CONTENT, "--detector", "-d", help="检测器类型"), + threshold: float = typer.Option(30.0, "--threshold", "-t", help="检测阈值"), + min_scene_length: float = typer.Option(1.0, "--min-length", "-m", help="最小场景长度(秒)"), + output: Optional[Path] = typer.Option(None, "--output", "-o", help="输出文件路径"), + output_format: OutputFormat = typer.Option(OutputFormat.JSON, "--format", "-f", help="输出格式"), + ai_analysis: bool = typer.Option(True, "--ai/--no-ai", help="启用/禁用AI分析"), + verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出") +): + """使用LangGraph工作流进行场景检测""" + console.print(f"🔄 使用工作流检测视频: [bold blue]{video_path}[/bold blue]") + + try: + # 创建检测器 + detector = SceneDetector() + + # 执行工作流检测 + result = detector.detect_with_workflow( + video_path, detector_type, threshold, min_scene_length, + output, output_format, ai_analysis + ) + + # 显示结果 + if result.get("workflow_state") == "completed": + detection_result = result.get("detection_result") + if detection_result and detection_result.success: + console.print(f"✅ 工作流完成: [bold green]{detection_result.total_scenes}[/bold green] 个场景") + console.print(f"📊 检测时间: {detection_result.detection_time:.2f}秒") + + # 显示AI分析结果 + ai_analysis_result = result.get("ai_analysis") + if ai_analysis_result and ai_analysis_result != "AI分析已禁用": + console.print("\n🧠 AI分析结果:") + console.print(ai_analysis_result[:500] + "..." if len(ai_analysis_result) > 500 else ai_analysis_result) + + # 显示场景列表 + if verbose: + _display_scenes_table(detection_result.scenes) + else: + console.print(f"❌ 检测失败: [bold red]{detection_result.error if detection_result else '未知错误'}[/bold red]") + raise typer.Exit(1) + else: + errors = result.get("errors", []) + error_msg = "; ".join(errors) if errors else "工作流执行失败" + console.print(f"❌ 工作流失败: [bold red]{error_msg}[/bold red]") + raise typer.Exit(1) + + except Exception as e: + console.print(f"❌ 执行失败: [bold red]{str(e)}[/bold red]") + raise typer.Exit(1) + + +def _display_scenes_table(scenes): + """显示场景表格""" + table = Table(title="检测到的场景") + table.add_column("场景", style="cyan") + table.add_column("开始时间", style="green") + table.add_column("结束时间", style="green") + table.add_column("时长", style="yellow") + + for scene in scenes: + table.add_row( + str(scene.index + 1), + f"{scene.start_time:.2f}s", + f"{scene.end_time:.2f}s", + f"{scene.duration:.2f}s" + ) + + console.print(table) + + +if __name__ == "__main__": + scene_detect() \ No newline at end of file diff --git a/python_core/cli/const.py b/python_core/cli/const.py deleted file mode 100644 index 7209740..0000000 --- a/python_core/cli/const.py +++ /dev/null @@ -1,6 +0,0 @@ -from python_core.utils.jsonrpc_enhanced import create_progress_reporter -from rich.console import Console -from python_core.config import settings -console = Console() -progress_reporter = create_progress_reporter() -project_root = settings.project_root \ No newline at end of file diff --git a/python_core/cli/scene_detect.py b/python_core/cli/scene_detect.py deleted file mode 100644 index 0937cbf..0000000 --- a/python_core/cli/scene_detect.py +++ /dev/null @@ -1,871 +0,0 @@ -#!/usr/bin/env python3 -""" -PySceneDetect 场景检测命令行工具 - LangGraph增强版 -""" - -import json -import time -from pathlib import Path -from typing import Optional, List, Literal, Dict, Any -from enum import Enum -from dataclasses import dataclass, asdict - -from python_core.cli.const import progress_reporter - -# PySceneDetect 依赖 -from scenedetect import open_video, SceneManager -from scenedetect.detectors import ContentDetector, ThresholdDetector - -# LangGraph 依赖 -from langgraph.graph import StateGraph, START, END -from langgraph.graph.state import CompiledStateGraph -from langgraph.checkpoint.memory import MemorySaver -from langchain_anthropic import ChatAnthropic - -class DetectorType(str, Enum): - """检测器类型""" - CONTENT = "content" - THRESHOLD = "threshold" - ADAPTIVE = "adaptive" - -class OutputFormat(str, Enum): - """输出格式""" - JSON = "json" - CSV = "csv" - TXT = "txt" - -@dataclass -class SceneInfo: - """场景信息""" - index: int - start_time: float - end_time: float - duration: float - start_frame: int - end_frame: int - -@dataclass -class DetectionResult: - """检测结果""" - video_path: str - filename: str - detector_type: str - threshold: float - total_scenes: int - total_duration: float - detection_time: float - scenes: List[SceneInfo] - success: bool - error: Optional[str] = None - -# LangGraph 工作流状态 -@dataclass -class SceneDetectionWorkflowState: - """场景检测工作流状态""" - # 输入参数 - 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 - - # 工作流状态 - current_stage: str = "init" - progress: int = 0 - total_steps: int = 5 - - # 中间结果 - video_info: Dict[str, Any] = None - raw_scenes: List[Any] = None - processed_scenes: List[SceneInfo] = None - - # 最终结果 - detection_result: Optional[DetectionResult] = None - ai_analysis: Optional[str] = None - - # 错误处理 - errors: List[str] = None - - def __post_init__(self): - if self.video_info is None: - self.video_info = {} - if self.raw_scenes is None: - self.raw_scenes = [] - if self.processed_scenes is None: - self.processed_scenes = [] - if self.errors is None: - self.errors = [] - -class SceneDetector: - """场景检测器""" - - def __init__(self): - self.supported_formats = {'.mp4', '.avi', '.mov', '.mkv', '.wmv', '.flv', '.webm', '.m4v'} - - # 初始化AI分析器（如果可用） - try: - self.llm = ChatAnthropic(model="claude-3-5-sonnet-20241022") - self.ai_enabled = True - except Exception as e: - progress_reporter.warning(f"⚠️ AI分析器初始化失败: {e}") - self.ai_enabled = False - - def detect_scenes(self, video_path: Path, detector_type: DetectorType = DetectorType.CONTENT, - threshold: float = 30.0, min_scene_length: float = 1.0) -> DetectionResult: - """检测单个视频的场景""" - start_time = time.time() - filename = video_path.name - - try: - progress_reporter.info(f"🎬 开始检测: {filename}") - - # 打开视频文件 - video = open_video(str(video_path)) - scene_manager = SceneManager() - - # 根据类型添加检测器 - if detector_type == DetectorType.CONTENT: - scene_manager.add_detector(ContentDetector(threshold=threshold)) - progress_reporter.info(f"📊 使用内容检测器，阈值: {threshold}") - elif detector_type == DetectorType.THRESHOLD: - scene_manager.add_detector(ThresholdDetector(threshold=threshold)) - progress_reporter.info(f"📊 使用阈值检测器，阈值: {threshold}") - elif detector_type == DetectorType.ADAPTIVE: - # 自适应：同时使用两种检测器 - scene_manager.add_detector(ContentDetector(threshold=threshold)) - scene_manager.add_detector(ThresholdDetector(threshold=threshold * 0.8)) - progress_reporter.info(f"📊 使用自适应检测器，阈值: {threshold}") - - # 开始检测 - progress_reporter.info("🔍 正在分析视频帧...") - scene_manager.detect_scenes(video) - scene_list = scene_manager.get_scene_list() - progress_reporter.info(f"✅ 检测到 {len(scene_list)} 个场景") - - # 获取视频信息 - fps = video.frame_rate - total_duration = video.duration.get_seconds() - progress_reporter.info(f"📊 视频信息: {fps:.2f}fps, {total_duration:.2f}秒") - - # 构建场景信息 - scenes = [] - - if not scene_list: - # 如果没有检测到场景变化，整个视频就是一个场景 - scene_info = SceneInfo( - index=0, - start_time=0.0, - end_time=total_duration, - duration=total_duration, - start_frame=0, - end_frame=int(total_duration * fps) if fps > 0 else 0 - ) - scenes.append(scene_info) - progress_reporter.info(f"📝 无场景变化，整个视频作为单一场景: {total_duration:.2f}秒") - else: - # 处理检测到的场景 - for start_time_scene, end_time_scene in scene_list: - start_seconds = start_time_scene.get_seconds() - end_seconds = end_time_scene.get_seconds() - duration = end_seconds - start_seconds - - # 跳过太短的场景 - if duration < min_scene_length: - continue - - scene_info = SceneInfo( - index=len(scenes), - start_time=start_seconds, - end_time=end_seconds, - duration=duration, - start_frame=start_time_scene.get_frames(), - end_frame=end_time_scene.get_frames() - ) - scenes.append(scene_info) - detection_time = time.time() - start_time - - result = DetectionResult( - video_path=str(video_path), - filename=filename, - detector_type=detector_type.value, - threshold=threshold, - total_scenes=len(scenes), - total_duration=total_duration, - detection_time=detection_time, - scenes=scenes, - success=True - ) - - progress_reporter.success(f"🎯 检测完成: {len(scenes)} 个场景，耗时 {detection_time:.2f}秒") - return result - - except Exception as e: - detection_time = time.time() - start_time - error_msg = str(e) - progress_reporter.error(f"❌ 检测失败: {error_msg}") - - return DetectionResult( - video_path=str(video_path), - filename=filename, - detector_type=detector_type.value, - threshold=threshold, - total_scenes=0, - total_duration=0.0, - detection_time=detection_time, - scenes=[], - success=False, - error=error_msg - ) - - def batch_detect(self, input_directory: Path, detector_type: DetectorType = DetectorType.CONTENT, - threshold: float = 30.0, recursive: bool = False) -> List[DetectionResult]: - """批量检测场景""" - progress_reporter.info(f"📦 开始批量检测: {input_directory}") - - # 扫描视频文件 - video_files = self._scan_video_files(input_directory, recursive) - - if not video_files: - progress_reporter.warning("⚠️ 未找到视频文件") - return [] - - progress_reporter.info(f"📋 找到 {len(video_files)} 个视频文件") - - results = [] - for i, video_file in enumerate(video_files): - progress_reporter.info(f"📊 处理进度: {i+1}/{len(video_files)} - {video_file.name}") - - result = self.detect_scenes(video_file, detector_type, threshold) - results.append(result) - - successful = len([r for r in results if r.success]) - progress_reporter.success(f"🎉 批量检测完成: {successful}/{len(results)} 成功") - - return results - - def compare_detectors(self, video_path: Path, thresholds: List[float] = None) -> dict: - """比较不同检测器效果""" - if thresholds is None: - thresholds = [20.0, 30.0, 40.0] - - progress_reporter.info(f"🔬 开始检测器比较: {video_path.name}") - - detectors = [DetectorType.CONTENT, DetectorType.THRESHOLD, DetectorType.ADAPTIVE] - results = [] - - total_tests = len(detectors) * len(thresholds) - current_test = 0 - - for detector in detectors: - for threshold in thresholds: - current_test += 1 - progress_reporter.info(f"🧪 测试 {current_test}/{total_tests}: {detector.value} (阈值: {threshold})") - - result = self.detect_scenes(video_path, detector, threshold) - results.append({ - "detector": detector.value, - "threshold": threshold, - "scenes": result.total_scenes, - "duration": result.total_duration, - "detection_time": result.detection_time, - "success": result.success, - "error": result.error - }) - - # 分析结果 - analysis = self._analyze_comparison(results) - - comparison_result = { - "video_path": str(video_path), - "total_tests": total_tests, - "results": results, - "analysis": analysis - } - - progress_reporter.success("🔬 检测器比较完成") - return comparison_result - - def _scan_video_files(self, directory: Path, recursive: bool = False) -> List[Path]: - """扫描视频文件""" - video_files = [] - - if recursive: - for ext in self.supported_formats: - video_files.extend(directory.rglob(f"*{ext}")) - else: - for ext in self.supported_formats: - video_files.extend(directory.glob(f"*{ext}")) - - return sorted(video_files) - - def _analyze_comparison(self, results: List[dict]) -> dict: - """分析比较结果""" - successful_results = [r for r in results if r["success"]] - - if not successful_results: - return {"message": "所有测试都失败了"} - - # 按检测器分组 - by_detector = {} - for result in successful_results: - detector = result["detector"] - if detector not in by_detector: - by_detector[detector] = [] - by_detector[detector].append(result) - - # 分析每个检测器 - detector_analysis = {} - for detector, detector_results in by_detector.items(): - avg_scenes = sum(r["scenes"] for r in detector_results) / len(detector_results) - avg_time = sum(r["detection_time"] for r in detector_results) / len(detector_results) - - detector_analysis[detector] = { - "average_scenes": avg_scenes, - "average_detection_time": avg_time, - "test_count": len(detector_results) - } - - # 推荐最佳检测器 - best_detector = max(detector_analysis.keys(), - key=lambda d: detector_analysis[d]["average_scenes"]) - - return { - "total_successful": len(successful_results), - "detector_analysis": detector_analysis, - "best_detector": best_detector, - "recommendation": f"推荐使用 {best_detector} 检测器" - } - - def save_results(self, results, output_path: Path, format: OutputFormat = OutputFormat.JSON): - """保存检测结果""" - output_path.parent.mkdir(parents=True, exist_ok=True) - - if format == OutputFormat.JSON: - self._save_json(results, output_path) - elif format == OutputFormat.CSV: - self._save_csv(results, output_path) - elif format == OutputFormat.TXT: - self._save_txt(results, output_path) - - progress_reporter.success(f"📄 结果已保存: {output_path}") - - def _save_json(self, results, output_path: Path): - """保存JSON格式""" - if isinstance(results, list): - # 批量结果 - data = [asdict(result) for result in results] - else: - # 单个结果或比较结果 - if hasattr(results, '__dict__'): - data = asdict(results) - else: - data = results - - with open(output_path, 'w', encoding='utf-8') as f: - json.dump(data, f, indent=2, ensure_ascii=False) - - def _save_csv(self, results, output_path: Path): - """保存CSV格式""" - import csv - - with open(output_path, 'w', newline='', encoding='utf-8') as f: - writer = csv.writer(f) - - if isinstance(results, list) and results: - # 批量结果 - writer.writerow(['filename', 'detector', 'threshold', 'scenes', 'duration', 'detection_time', 'success']) - - for result in results: - writer.writerow([ - result.filename, - result.detector_type, - result.threshold, - result.total_scenes, - result.total_duration, - result.detection_time, - result.success - ]) - - def _save_txt(self, results, output_path: Path): - """保存文本格式""" - with open(output_path, 'w', encoding='utf-8') as f: - f.write("PySceneDetect 场景检测结果\n") - f.write("=" * 50 + "\n\n") - - if isinstance(results, list): - # 批量结果 - for result in results: - 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.scenes: - f.write(" 场景列表:\n") - for scene in result.scenes: - f.write(f" 场景 {scene.index}: {scene.start_time:.2f}s - {scene.end_time:.2f}s ({scene.duration:.2f}s)\n") - - f.write("\n") - - # ==================== LangGraph 工作流方法 ==================== - - def create_detection_workflow(self) -> Optional[CompiledStateGraph]: - """创建场景检测工作流""" - - # 定义工作流节点 - def validate_input(state: SceneDetectionWorkflowState) -> Dict[str, Any]: - """验证输入参数""" - progress_reporter.info("🔍 验证输入参数...") - - video_path = Path(state.video_path) - 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}") - - # 验证参数范围 - if not (0 <= state.threshold <= 100): - errors.append(f"阈值超出范围 (0-100): {state.threshold}") - - if state.min_scene_length < 0: - errors.append(f"最小场景长度不能为负数: {state.min_scene_length}") - - return { - "current_stage": "validated" if not errors else "error", - "progress": 1, - "errors": errors - } - - def extract_video_info(state: SceneDetectionWorkflowState) -> Dict[str, Any]: - """提取视频信息""" - progress_reporter.info("📊 提取视频信息...") - - try: - import cv2 - cap = cv2.VideoCapture(state.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 - - cap.release() - - video_info = { - "fps": fps, - "frame_count": frame_count, - "width": width, - "height": height, - "duration": duration, - "resolution": f"{width}x{height}" - } - - progress_reporter.info(f"📹 视频信息: {video_info['resolution']}, {fps:.2f}fps, {duration:.2f}s") - - return { - "current_stage": "info_extracted", - "progress": 2, - "video_info": video_info - } - - except Exception as e: - return { - "current_stage": "error", - "errors": state.errors + [f"提取视频信息失败: {e}"] - } - - def detect_scenes(state: SceneDetectionWorkflowState) -> Dict[str, Any]: - """执行场景检测""" - progress_reporter.info("🎯 执行场景检测...") - - try: - # 使用现有的检测逻辑 - result = self.detect_scenes( - Path(state.video_path), - DetectorType(state.detector_type), - state.threshold, - state.min_scene_length - ) - - return { - "current_stage": "scenes_detected", - "progress": 3, - "detection_result": result, - "processed_scenes": result.scenes - } - - except Exception as e: - return { - "current_stage": "error", - "errors": state.errors + [f"场景检测失败: {e}"] - } - - def analyze_with_ai(state: SceneDetectionWorkflowState) -> Dict[str, Any]: - """AI分析场景结果""" - if not self.ai_enabled or not state.enable_ai_analysis: - progress_reporter.info("⚠️ AI分析已禁用，跳过此步骤") - return { - "current_stage": "analysis_skipped", - "progress": 4, - "ai_analysis": "AI分析已禁用" - } - - progress_reporter.info("🧠 AI分析场景结果...") - - try: - result = state.detection_result - video_info = state.video_info - - analysis_prompt = f""" - 请分析以下视频场景检测结果： - - 视频信息: - - 文件: {result.filename} - - 分辨率: {video_info.get('resolution', 'Unknown')} - - 时长: {result.total_duration:.2f}秒 - - 帧率: {video_info.get('fps', 0):.2f}fps - - 检测结果: - - 检测器: {result.detector_type} - - 阈值: {result.threshold} - - 场景数: {result.total_scenes} - - 检测时间: {result.detection_time:.2f}秒 - - 场景详情: - {self._format_scenes_for_ai(result.scenes)} - - 请提供: - 1. 场景分布分析 - 2. 检测质量评估 - 3. 参数优化建议 - 4. 潜在问题识别 - """ - - response = self.llm.invoke([{"role": "user", "content": analysis_prompt}]) - - return { - "current_stage": "ai_analyzed", - "progress": 4, - "ai_analysis": response.content - } - - except Exception as e: - progress_reporter.warning(f"⚠️ AI分析失败: {e}") - return { - "current_stage": "analysis_failed", - "progress": 4, - "ai_analysis": f"AI分析失败: {e}" - } - - def finalize_results(state: SceneDetectionWorkflowState) -> Dict[str, Any]: - """整理最终结果""" - progress_reporter.info("📋 整理最终结果...") - - # 保存结果（如果指定了输出路径） - if state.output_path and state.detection_result: - try: - output_path = Path(state.output_path) - output_format = OutputFormat(state.output_format) - self.save_results(state.detection_result, output_path, output_format) - except Exception as e: - progress_reporter.warning(f"⚠️ 保存结果失败: {e}") - - return { - "current_stage": "completed", - "progress": 5 - } - - # 路由函数 - def route_next_step(state: SceneDetectionWorkflowState) -> Literal["extract_info", "detect", "analyze", "finalize", "error"]: - if state.errors: - return "error" - elif state.current_stage == "validated": - return "extract_info" - elif state.current_stage == "info_extracted": - return "detect" - elif state.current_stage == "scenes_detected": - return "analyze" - elif state.current_stage in ["ai_analyzed", "analysis_skipped", "analysis_failed"]: - return "finalize" - else: - return "error" - - def handle_error(state: SceneDetectionWorkflowState) -> Dict[str, Any]: - """处理错误""" - error_msg = "; ".join(state.errors) - progress_reporter.error(f"❌ 工作流错误: {error_msg}") - return {"current_stage": "failed"} - - # 构建工作流图 - workflow = StateGraph(SceneDetectionWorkflowState) - - # 添加节点 - workflow.add_node("validate", validate_input) - workflow.add_node("extract_info", extract_video_info) - workflow.add_node("detect", detect_scenes) - workflow.add_node("analyze", analyze_with_ai) - workflow.add_node("finalize", finalize_results) - workflow.add_node("error", handle_error) - - # 添加边 - workflow.add_edge(START, "validate") - workflow.add_conditional_edges("validate", route_next_step) - workflow.add_conditional_edges("extract_info", route_next_step) - workflow.add_conditional_edges("detect", route_next_step) - workflow.add_conditional_edges("analyze", route_next_step) - workflow.add_edge("finalize", END) - workflow.add_edge("error", END) - - # 编译工作流 - memory = MemorySaver() - return workflow.compile(checkpointer=memory) - - def _format_scenes_for_ai(self, scenes: List[SceneInfo]) -> str: - """格式化场景信息供AI分析""" - if not scenes: - return "无场景数据" - - formatted = [] - for scene in scenes[:10]: # 只显示前10个场景 - formatted.append( - f"场景 {scene.index}: {scene.start_time:.2f}s - {scene.end_time:.2f}s " - f"(时长: {scene.duration:.2f}s)" - ) - - if len(scenes) > 10: - formatted.append(f"... 还有 {len(scenes) - 10} 个场景") - - return "\n".join(formatted) - - def detect_with_workflow(self, video_path: Path, detector_type: DetectorType = DetectorType.CONTENT, - threshold: float = 30.0, min_scene_length: float = 1.0, - output_path: Optional[Path] = None, output_format: OutputFormat = OutputFormat.JSON, - enable_ai_analysis: bool = True) -> Dict[str, Any]: - """使用LangGraph工作流进行场景检测""" - - # 创建工作流 - workflow = self.create_detection_workflow() - if not workflow: - raise Exception("无法创建工作流") - - # 初始化状态 - initial_state = SceneDetectionWorkflowState( - video_path=str(video_path), - detector_type=detector_type.value, - threshold=threshold, - min_scene_length=min_scene_length, - output_path=str(output_path) if output_path else None, - output_format=output_format.value, - enable_ai_analysis=enable_ai_analysis # 使用参数 - ) - - # 执行工作流 - config = {"configurable": {"thread_id": f"detection_{int(time.time())}"}} - - try: - final_state = workflow.invoke(initial_state, config) - - return { - "detection_result": final_state.get("detection_result"), - "ai_analysis": final_state.get("ai_analysis"), - "video_info": final_state.get("video_info"), - "workflow_state": final_state.get("current_stage"), - "errors": final_state.get("errors", []) - } - - except Exception as e: - progress_reporter.error(f"❌ 工作流执行失败: {e}") - raise - - # ==================== JSON-RPC 方法注册 ==================== - - def register_jsonrpc_methods(self): - """注册JSON-RPC方法到全局注册器""" - from python_core.utils.jsonrpc_enhanced import method_registry - - # 注册方法到全局注册器 - method_registry.register_function(self.jsonrpc_detect_scenes, "scene.detect") - method_registry.register_function(self.jsonrpc_detect_with_workflow, "scene.detect_workflow") - method_registry.register_function(self.jsonrpc_get_video_info, "scene.get_video_info") - method_registry.register_function(self.jsonrpc_batch_detect, "scene.batch_detect") - - def jsonrpc_detect_scenes(self, video_path: str, detector_type: str = "content", - threshold: float = 30.0, min_scene_length: float = 1.0) -> Dict[str, Any]: - """JSON-RPC方法：场景检测""" - try: - result = self.detect_scenes( - Path(video_path), - DetectorType(detector_type), - threshold, - min_scene_length - ) - - return { - "success": result.success, - "filename": result.filename, - "detector_type": result.detector_type, - "threshold": result.threshold, - "total_scenes": result.total_scenes, - "total_duration": result.total_duration, - "detection_time": result.detection_time, - "scenes": [asdict(scene) for scene in result.scenes], - "error": result.error - } - except Exception as e: - return { - "success": False, - "error": str(e) - } - - def jsonrpc_detect_with_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]: - """JSON-RPC方法：工作流场景检测""" - try: - output_path_obj = Path(output_path) if output_path else None - - result = self.detect_with_workflow( - Path(video_path), - DetectorType(detector_type), - threshold, - min_scene_length, - output_path_obj, - OutputFormat(output_format), - enable_ai_analysis - ) - - # 序列化结果 - serialized_result = {} - for key, value in result.items(): - if key == "detection_result" and value: - serialized_result[key] = { - "success": value.success, - "filename": value.filename, - "detector_type": value.detector_type, - "threshold": value.threshold, - "total_scenes": value.total_scenes, - "total_duration": value.total_duration, - "detection_time": value.detection_time, - "scenes": [asdict(scene) for scene in value.scenes], - "error": value.error - } - else: - serialized_result[key] = value - - return serialized_result - - except Exception as e: - return { - "success": False, - "error": str(e) - } - - def get_video_info(self, video_path: Path) -> Dict[str, Any]: - """获取视频信息""" - try: - import cv2 - 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 - - cap.release() - - return { - "filename": video_path.name, - "fps": fps, - "frame_count": frame_count, - "width": width, - "height": height, - "duration": duration, - "resolution": f"{width}x{height}", - "file_size": video_path.stat().st_size if video_path.exists() else 0 - } - - except Exception as e: - raise Exception(f"获取视频信息失败: {e}") - - def jsonrpc_get_video_info(self, video_path: str) -> Dict[str, Any]: - """JSON-RPC方法：获取视频信息""" - try: - info = self.get_video_info(Path(video_path)) - return { - "success": True, - "info": info - } - except Exception as e: - return { - "success": False, - "error": str(e) - } - - def jsonrpc_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]: - """JSON-RPC方法：批量场景检测""" - try: - output_dir_obj = Path(output_dir) if output_dir else None - - results = self.batch_detect( - Path(directory), - DetectorType(detector_type), - threshold, - min_scene_length, - output_dir_obj, - OutputFormat(output_format) - ) - - # 序列化结果 - serialized_results = [] - for result in results: - serialized_results.append({ - "success": result.success, - "filename": result.filename, - "detector_type": result.detector_type, - "threshold": result.threshold, - "total_scenes": result.total_scenes, - "total_duration": result.total_duration, - "detection_time": result.detection_time, - "scenes": [asdict(scene) for scene in result.scenes], - "error": result.error - }) - - return { - "success": True, - "results": serialized_results, - "total_processed": len(results) - } - - except Exception as e: - return { - "success": False, - "error": str(e) - } - - -# 创建全局检测器实例 -detector = SceneDetector() - -# 注册JSON-RPC方法 -detector.register_jsonrpc_methods() \ No newline at end of file diff --git a/python_core/scene_detection/__init__.py b/python_core/scene_detection/__init__.py new file mode 100644 index 0000000..2f06049 --- /dev/null +++ b/python_core/scene_detection/__init__.py @@ -0,0 +1,67 @@ +#!/usr/bin/env python3 +""" +Scene Detection Module +场景检测模块 + +重构后的场景检测功能，按照功能模块化组织： +- types: 类型定义和数据模型 +- services: 核心业务逻辑服务 +- workflows: LangGraph工作流 +- utils: 工具类和辅助函数 +""" + +# 导出主要类型 +from .types import ( + DetectorType, + OutputFormat, + SceneInfo, + DetectionResult, + SceneDetectionWorkflowState +) + +# 导出主要服务 +from .services import ( + SceneDetectorService, + AIAnalysisService, + VideoInfoService +) + +# 导出工作流 +from .workflows import ( + SceneDetectionWorkflowManager, + WorkflowNodes +) + +# 导出工具类 +from .utils import ( + ResultSaver, + InputValidator +) + +# 导出主要接口类 +from .scene_detector import SceneDetector + +__all__ = [ + # Types + "DetectorType", + "OutputFormat", + "SceneInfo", + "DetectionResult", + "SceneDetectionWorkflowState", + + # Services + "SceneDetectorService", + "AIAnalysisService", + "VideoInfoService", + + # Workflows + "SceneDetectionWorkflowManager", + "WorkflowNodes", + + # Utils + "ResultSaver", + "InputValidator", + + # Main Interface + "SceneDetector" +] diff --git a/python_core/scene_detection/scene_detector.py b/python_core/scene_detection/scene_detector.py new file mode 100644 index 0000000..17cf1c1 --- /dev/null +++ b/python_core/scene_detection/scene_detector.py @@ -0,0 +1,228 @@ +#!/usr/bin/env python3 +""" +Scene Detector Main Interface +场景检测主接口类 + +整合所有功能的主要接口类，提供简洁的API +""" + +from pathlib import Path +from typing import Dict, Any, Optional, List +from dataclasses import asdict + +from python_core.utils.logger import logger +from .types import DetectorType, OutputFormat, DetectionResult +from .services import SceneDetectorService, VideoInfoService, AIAnalysisService +from .workflows import SceneDetectionWorkflowManager +from .utils import ResultSaver, InputValidator + + +class SceneDetector: + """场景检测器主类""" + + def __init__(self): + self.supported_formats = {'.mp4', '.avi', '.mov', '.mkv', '.wmv', '.flv', '.webm', '.m4v'} + + # 初始化服务 + self.detector_service = SceneDetectorService() + self.video_info_service = VideoInfoService() + self.ai_analysis_service = AIAnalysisService() + + # 初始化工作流管理器 + self.workflow_manager = SceneDetectionWorkflowManager() + + # 初始化工具类 + self.result_saver = ResultSaver() + self.input_validator = InputValidator(self.supported_formats) + + def detect_scenes(self, video_path: Path, detector_type: DetectorType = DetectorType.CONTENT, + threshold: float = 30.0, min_scene_length: float = 1.0) -> DetectionResult: + """基础场景检测方法""" + return self.detector_service.detect_scenes(video_path, detector_type, threshold, min_scene_length) + + def get_video_info(self, video_path: Path) -> Dict[str, Any]: + """获取视频信息""" + try: + # 验证文件 + self.video_info_service.validate_video_file(video_path, self.supported_formats) + + # 提取信息 + video_info = self.video_info_service.extract_video_info(video_path) + + return { + "success": True, + "info": video_info + } + + except Exception as e: + return { + "success": False, + "error": str(e) + } + + def detect_with_workflow(self, video_path: Path, detector_type: DetectorType = DetectorType.CONTENT, + threshold: float = 30.0, min_scene_length: float = 1.0, + output_path: Optional[Path] = None, output_format: OutputFormat = OutputFormat.JSON, + enable_ai_analysis: bool = True, request_id: Optional[str] = None) -> Dict[str, Any]: + """使用LangGraph工作流进行场景检测""" + return self.workflow_manager.detect_with_workflow( + video_path, detector_type, threshold, min_scene_length, + output_path, output_format, enable_ai_analysis, request_id + ) + + def save_results(self, result: DetectionResult, output_path: Path, + output_format: OutputFormat) -> None: + """保存检测结果""" + self.result_saver.save_results(result, output_path, output_format) + + def batch_detect(self, video_paths: List[Path], detector_type: DetectorType = DetectorType.CONTENT, + threshold: float = 30.0, min_scene_length: float = 1.0, + output_dir: Optional[Path] = None, output_format: OutputFormat = OutputFormat.JSON) -> List[Dict[str, Any]]: + """批量检测场景""" + results = [] + + for video_path in video_paths: + try: + logger.info(f"🎬 处理视频: {video_path.name}") + + # 检测场景 + result = self.detect_scenes(video_path, detector_type, threshold, min_scene_length) + + # 保存结果（如果指定了输出目录） + if output_dir and result.success: + output_file = output_dir / f"{video_path.stem}_scenes.{output_format.value}" + self.save_results(result, output_file, output_format) + + # 序列化结果 + serialized_result = { + "video_path": str(video_path), + "success": result.success, + "result": asdict(result) if result.success else None, + "error": result.error + } + + results.append(serialized_result) + + except Exception as e: + logger.error(f"❌ 处理视频失败 {video_path.name}: {e}") + results.append({ + "video_path": str(video_path), + "success": False, + "result": None, + "error": str(e) + }) + + return results + + # JSON-RPC方法 + def jsonrpc_detect_scenes(self, video_path: str, detector_type: str = "content", + threshold: float = 30.0, min_scene_length: float = 1.0) -> Dict[str, Any]: + """JSON-RPC方法：基础场景检测""" + try: + result = self.detect_scenes( + Path(video_path), + DetectorType(detector_type), + threshold, + min_scene_length + ) + + return asdict(result) + + except Exception as e: + return { + "success": False, + "error": str(e) + } + + def jsonrpc_get_video_info(self, video_path: str) -> Dict[str, Any]: + """JSON-RPC方法：获取视频信息""" + try: + return self.get_video_info(Path(video_path)) + except Exception as e: + return { + "success": False, + "error": str(e) + } + + def jsonrpc_detect_with_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, request_id: Optional[str] = None) -> Optional[Dict[str, Any]]: + """JSON-RPC方法：工作流场景检测""" + try: + output_path_obj = Path(output_path) if output_path else None + + result = self.detect_with_workflow( + Path(video_path), + DetectorType(detector_type), + threshold, + min_scene_length, + output_path_obj, + OutputFormat(output_format), + enable_ai_analysis, + request_id + ) + + # 检查是否是JSON-RPC模式 + if result.get("jsonrpc_mode"): + # JSON-RPC模式：结果已经通过工作流发送，返回None + return None + else: + # 非JSON-RPC模式：序列化并返回结果 + serialized_result = {} + for key, value in result.items(): + if key == "detection_result" and value: + serialized_result[key] = asdict(value) + else: + serialized_result[key] = value + return serialized_result + + except Exception as e: + # 如果有request_id，错误已经在detect_with_workflow中发送 + if request_id: + return None + else: + return { + "success": False, + "error": str(e) + } + + def jsonrpc_batch_detect(self, video_paths: List[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]: + """JSON-RPC方法：批量检测""" + try: + paths = [Path(p) for p in video_paths] + output_dir_obj = Path(output_dir) if output_dir else None + + results = self.batch_detect( + paths, + DetectorType(detector_type), + threshold, + min_scene_length, + output_dir_obj, + OutputFormat(output_format) + ) + + return { + "success": True, + "results": results, + "total_videos": len(video_paths), + "successful_detections": sum(1 for r in results if r["success"]) + } + + except Exception as e: + return { + "success": False, + "error": str(e) + } + + def register_jsonrpc_methods(self): + """注册JSON-RPC方法到全局注册器""" + from python_core.utils.jsonrpc_enhanced import method_registry + + # 直接注册方法 + method_registry.register_function(self.jsonrpc_detect_scenes, "scene.detect") + method_registry.register_function(self.jsonrpc_detect_with_workflow, "scene.detect_workflow") + method_registry.register_function(self.jsonrpc_get_video_info, "scene.get_video_info") + method_registry.register_function(self.jsonrpc_batch_detect, "scene.batch_detect") diff --git a/python_core/scene_detection/services/__init__.py b/python_core/scene_detection/services/__init__.py new file mode 100644 index 0000000..01e4456 --- /dev/null +++ b/python_core/scene_detection/services/__init__.py @@ -0,0 +1,17 @@ +#!/usr/bin/env python3 +""" +Scene Detection Services +场景检测服务层 + +导出所有服务类 +""" + +from .detector_service import SceneDetectorService +from .ai_analysis_service import AIAnalysisService +from .video_info_service import VideoInfoService + +__all__ = [ + "SceneDetectorService", + "AIAnalysisService", + "VideoInfoService" +] diff --git a/python_core/scene_detection/services/ai_analysis_service.py b/python_core/scene_detection/services/ai_analysis_service.py new file mode 100644 index 0000000..7589ed4 --- /dev/null +++ b/python_core/scene_detection/services/ai_analysis_service.py @@ -0,0 +1,100 @@ +#!/usr/bin/env python3 +""" +AI Analysis Service +AI分析服务 +""" + +from typing import List, Dict, Any, Optional + +from python_core.utils.logger import logger +from ..types import SceneInfo, DetectionResult + + +class AIAnalysisService: + """AI分析服务""" + + def __init__(self): + self.ai_enabled = False + self.llm = None + + # 尝试初始化AI分析器 + try: + import os + from langchain_anthropic import ChatAnthropic + + # 检查是否有API密钥 + api_key = os.getenv('ANTHROPIC_API_KEY') + if not api_key: + logger.info("ℹ️ 未设置ANTHROPIC_API_KEY环境变量，AI分析功能已禁用") + self.ai_enabled = False + return + + self.llm = ChatAnthropic(model="claude-3-5-sonnet-20241022", api_key=api_key) + self.ai_enabled = True + logger.info("✅ AI分析器初始化成功") + except ImportError: + logger.info("ℹ️ langchain_anthropic未安装，AI分析功能已禁用") + self.ai_enabled = False + except Exception as e: + logger.warning(f"⚠️ AI分析器初始化失败: {e}") + self.ai_enabled = False + + def analyze_detection_result(self, detection_result: DetectionResult, + video_info: Dict[str, Any]) -> str: + """分析检测结果""" + if not self.ai_enabled: + return "AI分析器未启用" + + try: + analysis_prompt = self._create_analysis_prompt(detection_result, video_info) + response = self.llm.invoke([{"role": "user", "content": analysis_prompt}]) + return response.content + + except Exception as e: + error_msg = f"AI分析失败: {e}" + logger.warning(error_msg) + return error_msg + + def _create_analysis_prompt(self, result: DetectionResult, video_info: Dict[str, Any]) -> str: + """创建分析提示词""" + return f""" +请分析以下视频场景检测结果： + +视频信息: +- 文件: {result.filename} +- 分辨率: {video_info.get('resolution', 'Unknown')} +- 时长: {result.total_duration:.2f}秒 +- 帧率: {video_info.get('fps', 0):.2f}fps + +检测结果: +- 检测器: {result.detector_type} +- 阈值: {result.threshold} +- 场景数: {result.total_scenes} +- 检测时间: {result.detection_time:.2f}秒 + +场景详情: +{self._format_scenes_for_ai(result.scenes)} + +请提供: +1. 场景分布分析 +2. 检测质量评估 +3. 参数优化建议 +4. 潜在问题识别 +""" + + def _format_scenes_for_ai(self, scenes: List[SceneInfo]) -> str: + """格式化场景信息供AI分析""" + if not scenes: + return "无场景数据" + + formatted = [] + for i, scene in enumerate(scenes[:10]): # 只显示前10个场景 + formatted.append( + f"场景 {i+1}: {scene.start_time:.2f}s - {scene.end_time:.2f}s " + f"(时长: {scene.duration:.2f}s)" + ) + + if len(scenes) > 10: + formatted.append(f"... 还有 {len(scenes) - 10} 个场景") + + return "\n".join(formatted) diff --git a/python_core/scene_detection/services/detector_service.py b/python_core/scene_detection/services/detector_service.py new file mode 100644 index 0000000..1ddcfab --- /dev/null +++ b/python_core/scene_detection/services/detector_service.py @@ -0,0 +1,158 @@ +#!/usr/bin/env python3 +""" +Scene Detector Service +场景检测服务 +""" + +import time +from pathlib import Path +from typing import List, Set + +from scenedetect import open_video, SceneManager +from scenedetect.detectors import ContentDetector, ThresholdDetector, AdaptiveDetector + +from python_core.utils.logger import logger +from ..types import DetectorType, SceneInfo, DetectionResult + + +def _timecode_to_seconds(timecode) -> float: + """将FrameTimecode对象转换为秒数""" + try: + # 新版本PySceneDetect + if hasattr(timecode, 'total_seconds'): + return timecode.total_seconds() + # 旧版本PySceneDetect + elif hasattr(timecode, 'get_seconds'): + return timecode.get_seconds() + # 如果是数字，直接返回 + elif isinstance(timecode, (int, float)): + return float(timecode) + # 尝试直接转换 + else: + return float(timecode) + except Exception as e: + logger.warning(f"时间码转换失败: {e}, 使用默认值0.0") + return 0.0 + + +class SceneDetectorService: + """场景检测服务""" + + def __init__(self): + self.supported_formats: Set[str] = { + '.mp4', '.avi', '.mov', '.mkv', '.wmv', '.flv', '.webm', '.m4v' + } + + def detect_scenes(self, video_path: Path, detector_type: DetectorType = DetectorType.CONTENT, + threshold: float = 30.0, min_scene_length: float = 1.0) -> DetectionResult: + """检测视频场景""" + detection_start_time = time.time() + + try: + logger.info(f"🎬 开始检测: {video_path.name}") + + # 打开视频 + video = open_video(str(video_path)) + scene_manager = SceneManager() + + # 选择检测器 + detector = self._create_detector(detector_type, threshold) + scene_manager.add_detector(detector) + + logger.info(f"📊 使用{detector_type.value}检测器，阈值: {threshold}") + + # 执行检测 + logger.info("🔍 正在分析视频帧...") + scene_manager.detect_scenes(video, show_progress=False) + + # 获取场景列表 + scene_list = scene_manager.get_scene_list() + + # 过滤短场景 + if min_scene_length > 0: + filtered_scenes = [] + for scene in scene_list: + start_time, end_time = scene + start_seconds = _timecode_to_seconds(start_time) + end_seconds = _timecode_to_seconds(end_time) + duration = end_seconds - start_seconds + + if duration >= min_scene_length: + filtered_scenes.append(scene) + scene_list = filtered_scenes + + # 转换为SceneInfo对象 + scenes = self._convert_scenes(scene_list, video.frame_rate) + + detection_time = time.time() - detection_start_time + total_duration = _timecode_to_seconds(video.duration) + + logger.info(f"✅ 检测到 {len(scenes)} 个场景") + logger.info(f"📊 视频信息: {video.frame_rate:.2f}fps, {total_duration:.2f}秒") + logger.info(f"🎯 检测完成: {len(scenes)} 个场景，耗时 {detection_time:.2f}秒") + + return DetectionResult( + success=True, + filename=video_path.name, + detector_type=detector_type, + threshold=threshold, + total_scenes=len(scenes), + total_duration=total_duration, + detection_time=detection_time, + scenes=scenes + ) + + except Exception as e: + detection_time = time.time() - detection_start_time + error_msg = f"场景检测失败: {str(e)}" + logger.error(error_msg) + logger.error(f"详细错误信息: {type(e).__name__}: {e}") + import traceback + logger.error(f"错误堆栈: {traceback.format_exc()}") + + return DetectionResult( + success=False, + filename=video_path.name, + detector_type=detector_type, + threshold=threshold, + total_scenes=0, + total_duration=0, + detection_time=detection_time, + scenes=[], + error=error_msg + ) + + def _create_detector(self, detector_type: DetectorType, threshold: float): + """创建检测器""" + if detector_type == DetectorType.CONTENT: + return ContentDetector(threshold=threshold) + elif detector_type == DetectorType.THRESHOLD: + return ThresholdDetector(threshold=threshold) + elif detector_type == DetectorType.ADAPTIVE: + return AdaptiveDetector() + else: + raise ValueError(f"不支持的检测器类型: {detector_type}") + + def _convert_scenes(self, scene_list, frame_rate: float) -> List[SceneInfo]: + """转换场景列表为SceneInfo对象""" + scenes = [] + + for i, (start_time, end_time) in enumerate(scene_list): + start_seconds = _timecode_to_seconds(start_time) + end_seconds = _timecode_to_seconds(end_time) + duration = end_seconds - start_seconds + + start_frame = int(start_seconds * frame_rate) + end_frame = int(end_seconds * frame_rate) + + scene_info = SceneInfo( + index=i, + start_time=start_seconds, + end_time=end_seconds, + duration=duration, + start_frame=start_frame, + end_frame=end_frame + ) + scenes.append(scene_info) + + return scenes diff --git a/python_core/scene_detection/services/video_info_service.py b/python_core/scene_detection/services/video_info_service.py new file mode 100644 index 0000000..4a45bd4 --- /dev/null +++ b/python_core/scene_detection/services/video_info_service.py @@ -0,0 +1,65 @@ +#!/usr/bin/env python3 +""" +Video Info Service +视频信息服务 +""" + +import cv2 +from pathlib import Path +from typing import Dict, Any + +from python_core.utils.logger import logger + + +class VideoInfoService: + """视频信息服务""" + + def extract_video_info(self, video_path: Path) -> Dict[str, Any]: + """提取视频信息""" + try: + 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 + + cap.release() + + # 获取文件大小 + file_size = video_path.stat().st_size + + video_info = { + "filename": video_path.name, + "file_path": str(video_path), + "file_size": file_size, + "fps": fps, + "frame_count": frame_count, + "width": width, + "height": height, + "duration": duration, + "resolution": f"{width}x{height}", + "format": video_path.suffix.lower() + } + + logger.info(f"📹 视频信息: {video_info['resolution']}, {fps:.2f}fps, {duration:.2f}s") + + return video_info + + except Exception as e: + logger.error(f"提取视频信息失败: {e}") + raise + + def validate_video_file(self, video_path: Path, supported_formats: set) -> bool: + """验证视频文件""" + if not video_path.exists(): + raise FileNotFoundError(f"视频文件不存在: {video_path}") + + if video_path.suffix.lower() not in supported_formats: + raise ValueError(f"不支持的文件格式: {video_path.suffix}") + + return True diff --git a/python_core/scene_detection/types/__init__.py b/python_core/scene_detection/types/__init__.py new file mode 100644 index 0000000..5eb3263 --- /dev/null +++ b/python_core/scene_detection/types/__init__.py @@ -0,0 +1,19 @@ +#!/usr/bin/env python3 +""" +Scene Detection Types +场景检测类型定义 + +导出所有类型定义 +""" + +from .enums import DetectorType, OutputFormat +from .models import SceneInfo, DetectionResult +from .workflow_state import SceneDetectionWorkflowState + +__all__ = [ + "DetectorType", + "OutputFormat", + "SceneInfo", + "DetectionResult", + "SceneDetectionWorkflowState" +] diff --git a/python_core/scene_detection/types/enums.py b/python_core/scene_detection/types/enums.py new file mode 100644 index 0000000..cb15355 --- /dev/null +++ b/python_core/scene_detection/types/enums.py @@ -0,0 +1,21 @@ +#!/usr/bin/env python3 +""" +Scene Detection Enums +场景检测枚举类型 +""" + +from enum import Enum + + +class DetectorType(str, Enum): + """检测器类型""" + CONTENT = "content" + THRESHOLD = "threshold" + ADAPTIVE = "adaptive" + + +class OutputFormat(str, Enum): + """输出格式""" + JSON = "json" + CSV = "csv" + TXT = "txt" diff --git a/python_core/scene_detection/types/models.py b/python_core/scene_detection/types/models.py new file mode 100644 index 0000000..de15027 --- /dev/null +++ b/python_core/scene_detection/types/models.py @@ -0,0 +1,34 @@ +#!/usr/bin/env python3 +""" +Scene Detection Models +场景检测数据模型 +""" + +from dataclasses import dataclass +from typing import List, Optional +from .enums import DetectorType + + +@dataclass +class SceneInfo: + """场景信息""" + index: int + start_time: float + end_time: float + duration: float + start_frame: int = 0 + end_frame: int = 0 + + +@dataclass +class DetectionResult: + """检测结果""" + success: bool + filename: str + detector_type: DetectorType + threshold: float + total_scenes: int + total_duration: float + detection_time: float + scenes: List[SceneInfo] + error: Optional[str] = None diff --git a/python_core/scene_detection/types/workflow_state.py b/python_core/scene_detection/types/workflow_state.py new file mode 100644 index 0000000..bc0037e --- /dev/null +++ b/python_core/scene_detection/types/workflow_state.py @@ -0,0 +1,82 @@ +#!/usr/bin/env python3 +""" +Workflow State +工作流状态定义 +""" + +from dataclasses import dataclass +from typing import Dict, Any, List, Optional +from .models import SceneInfo, DetectionResult +from python_core.utils.jsonrpc_enhanced import EnhancedJSONRPCResponse, ProgressLevel + + +@dataclass +class SceneDetectionWorkflowState: + """场景检测工作流状态""" + # 输入参数 + 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 + + # 工作流状态 + current_stage: str = "init" + progress: int = 0 + total_steps: int = 5 + + # JSON-RPC支持 + request_id: Optional[str] = None + enable_jsonrpc: bool = False + + # 中间结果 + video_info: Dict[str, Any] = None + raw_scenes: List[Any] = None + processed_scenes: List[SceneInfo] = None + + # 最终结果 + detection_result: Optional[DetectionResult] = None + ai_analysis: Optional[str] = None + + # 错误处理 + errors: List[str] = None + + def __post_init__(self): + if self.video_info is None: + self.video_info = {} + if self.raw_scenes is None: + self.raw_scenes = [] + if self.processed_scenes is None: + self.processed_scenes = [] + if self.errors is None: + self.errors = [] + + def get_jsonrpc_handler(self) -> Optional[EnhancedJSONRPCResponse]: + """获取JSON-RPC响应处理器""" + if self.enable_jsonrpc and self.request_id: + return EnhancedJSONRPCResponse(self.request_id) + return None + + def send_progress(self, step: str, message: str, level: ProgressLevel = ProgressLevel.INFO, + data: Optional[Dict[str, Any]] = None) -> None: + """发送进度更新""" + handler = self.get_jsonrpc_handler() + if handler: + progress_percent = int((self.progress / self.total_steps * 100)) if self.total_steps > 0 else -1 + handler.progress(step, progress_percent, message, level, data) + + def send_final_result(self, result: Dict[str, Any]) -> None: + """发送最终结果""" + handler = self.get_jsonrpc_handler() + if handler: + # 发送最终结果作为成功响应 + handler.success(result) + + def send_error_result(self, error_code: int, error_message: str, error_data: Any = None) -> None: + """发送错误结果""" + handler = self.get_jsonrpc_handler() + if handler: + # 发送错误响应 + handler.error(error_code, error_message, error_data) diff --git a/python_core/scene_detection/utils/__init__.py b/python_core/scene_detection/utils/__init__.py new file mode 100644 index 0000000..d2bd592 --- /dev/null +++ b/python_core/scene_detection/utils/__init__.py @@ -0,0 +1,15 @@ +#!/usr/bin/env python3 +""" +Scene Detection Utils +场景检测工具类 + +导出所有工具类 +""" + +from .result_saver import ResultSaver +from .validators import InputValidator + +__all__ = [ + "ResultSaver", + "InputValidator" +] diff --git a/python_core/scene_detection/utils/result_saver.py b/python_core/scene_detection/utils/result_saver.py new file mode 100644 index 0000000..b0baa3a --- /dev/null +++ b/python_core/scene_detection/utils/result_saver.py @@ -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") diff --git a/python_core/scene_detection/utils/validators.py b/python_core/scene_detection/utils/validators.py new file mode 100644 index 0000000..d9766c6 --- /dev/null +++ b/python_core/scene_detection/utils/validators.py @@ -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 diff --git a/python_core/scene_detection/workflows/__init__.py b/python_core/scene_detection/workflows/__init__.py new file mode 100644 index 0000000..95631b5 --- /dev/null +++ b/python_core/scene_detection/workflows/__init__.py @@ -0,0 +1,15 @@ +#!/usr/bin/env python3 +""" +Scene Detection Workflows +场景检测工作流 + +导出所有工作流类 +""" + +from .workflow_manager import SceneDetectionWorkflowManager +from .workflow_nodes import WorkflowNodes + +__all__ = [ + "SceneDetectionWorkflowManager", + "WorkflowNodes" +] diff --git a/python_core/scene_detection/workflows/workflow_manager.py b/python_core/scene_detection/workflows/workflow_manager.py new file mode 100644 index 0000000..397a6a0 --- /dev/null +++ b/python_core/scene_detection/workflows/workflow_manager.py @@ -0,0 +1,176 @@ +#!/usr/bin/env python3 +""" +Workflow Manager +工作流管理器 +""" + +import time +from pathlib import Path +from typing import Dict, Any, Optional, Literal + +from python_core.utils.logger import logger +from python_core.utils.jsonrpc_enhanced import EnhancedJSONRPCResponse +from ..types import SceneDetectionWorkflowState, DetectorType, OutputFormat +from .workflow_nodes import WorkflowNodes + + +class SceneDetectionWorkflowManager: + """场景检测工作流管理器""" + + def __init__(self): + self.nodes = WorkflowNodes() + self.workflow = None + + def create_detection_workflow(self): + """创建检测工作流""" + try: + from langgraph.graph import StateGraph, END + + # 创建状态图 + workflow = StateGraph(SceneDetectionWorkflowState) + + # 添加节点 + workflow.add_node("validate", self.nodes.validate_input) + workflow.add_node("extract_info", self.nodes.extract_video_info) + workflow.add_node("detect", self.nodes.detect_scenes) + workflow.add_node("analyze", self.nodes.analyze_with_ai) + workflow.add_node("finalize", self.nodes.finalize_results) + workflow.add_node("error", self.nodes.handle_error) + + # 设置入口点 + workflow.set_entry_point("validate") + + # 添加条件边 + workflow.add_conditional_edges( + "validate", + self._route_next_step, + { + "extract_info": "extract_info", + "error": "error" + } + ) + + workflow.add_conditional_edges( + "extract_info", + self._route_next_step, + { + "detect": "detect", + "error": "error" + } + ) + + workflow.add_conditional_edges( + "detect", + self._route_next_step, + { + "analyze": "analyze", + "error": "error" + } + ) + + workflow.add_conditional_edges( + "analyze", + self._route_next_step, + { + "finalize": "finalize", + "error": "error" + } + ) + + # 结束节点 + workflow.add_edge("finalize", END) + workflow.add_edge("error", END) + + # 编译工作流 + self.workflow = workflow.compile() + return self.workflow + + except ImportError: + logger.error("❌ LangGraph未安装，无法创建工作流") + return None + except Exception as e: + logger.error(f"❌ 创建工作流失败: {e}") + return None + + def _route_next_step(self, state: SceneDetectionWorkflowState) -> Literal["extract_info", "detect", "analyze", "finalize", "error"]: + """路由下一步""" + if state.errors: + return "error" + elif state.current_stage == "validated": + return "extract_info" + elif state.current_stage == "info_extracted": + return "detect" + elif state.current_stage == "scenes_detected": + return "analyze" + elif state.current_stage in ["ai_analyzed", "analysis_skipped", "analysis_failed"]: + return "finalize" + else: + return "error" + + def detect_with_workflow(self, video_path: Path, detector_type: DetectorType = DetectorType.CONTENT, + threshold: float = 30.0, min_scene_length: float = 1.0, + output_path: Optional[Path] = None, output_format: OutputFormat = OutputFormat.JSON, + enable_ai_analysis: bool = True, request_id: Optional[str] = None) -> Dict[str, Any]: + """使用LangGraph工作流进行场景检测""" + + # 创建工作流 + workflow = self.create_detection_workflow() + if not workflow: + raise Exception("无法创建工作流") + + # 初始化状态 + initial_state = SceneDetectionWorkflowState( + video_path=str(video_path), + detector_type=detector_type.value, + threshold=threshold, + min_scene_length=min_scene_length, + output_path=str(output_path) if output_path else None, + output_format=output_format.value, + enable_ai_analysis=enable_ai_analysis, + request_id=request_id, + enable_jsonrpc=request_id is not None # 如果有request_id就启用JSON-RPC + ) + + # 执行工作流 + config = {"configurable": {"thread_id": f"detection_{int(time.time())}"}} + + try: + final_state = workflow.invoke(initial_state, config) + + # 如果是JSON-RPC模式，结果已经通过JSON-RPC发送了 + if request_id is not None: + # JSON-RPC模式：结果已经在工作流节点中发送 + # 这里返回简化的状态信息供内部使用 + return { + "jsonrpc_mode": True, + "request_id": request_id, + "workflow_state": final_state.get("current_stage"), + "final_result_sent": True + } + else: + # 非JSON-RPC模式：返回完整结果 + return { + "detection_result": final_state.get("detection_result"), + "ai_analysis": final_state.get("ai_analysis"), + "video_info": final_state.get("video_info"), + "workflow_state": final_state.get("current_stage"), + "errors": final_state.get("errors", []) + } + + except Exception as e: + error_msg = f"工作流执行失败: {e}" + + # 如果是JSON-RPC模式，发送错误响应 + if request_id is not None: + handler = EnhancedJSONRPCResponse(request_id) + handler.error(-32603, error_msg, {"exception": str(e)}) + return { + "jsonrpc_mode": True, + "request_id": request_id, + "error_sent": True, + "error": error_msg + } + else: + # 非JSON-RPC模式：抛出异常 + logger.error(f"❌ {error_msg}") + raise diff --git a/python_core/scene_detection/workflows/workflow_nodes.py b/python_core/scene_detection/workflows/workflow_nodes.py new file mode 100644 index 0000000..f227eab --- /dev/null +++ b/python_core/scene_detection/workflows/workflow_nodes.py @@ -0,0 +1,242 @@ +#!/usr/bin/env python3 +""" +Workflow Nodes +工作流节点定义 +""" + +from pathlib import Path +from typing import Dict, Any +from dataclasses import asdict + +from python_core.utils.logger import logger +from python_core.utils.jsonrpc_enhanced import ProgressLevel +from ..types import SceneDetectionWorkflowState, DetectorType +from ..services import SceneDetectorService, VideoInfoService, AIAnalysisService + + +class WorkflowNodes: + """工作流节点集合""" + + def __init__(self): + self.detector_service = SceneDetectorService() + self.video_info_service = VideoInfoService() + self.ai_analysis_service = AIAnalysisService() + + def validate_input(self, state: SceneDetectionWorkflowState) -> Dict[str, Any]: + """验证输入参数""" + state.send_progress("validate", "🔍 验证输入参数...", ProgressLevel.INFO) + + video_path = Path(state.video_path) + errors = [] + + # 验证文件存在 + if not video_path.exists(): + errors.append(f"视频文件不存在: {video_path}") + state.send_progress("validate", f"❌ 文件不存在: {video_path}", ProgressLevel.ERROR) + + # 验证文件格式 + if video_path.suffix.lower() not in self.detector_service.supported_formats: + errors.append(f"不支持的文件格式: {video_path.suffix}") + state.send_progress("validate", f"❌ 不支持的格式: {video_path.suffix}", ProgressLevel.ERROR) + + # 验证参数范围 + if not (0 <= state.threshold <= 100): + errors.append(f"阈值超出范围 (0-100): {state.threshold}") + state.send_progress("validate", f"❌ 阈值超出范围: {state.threshold}", ProgressLevel.ERROR) + + if state.min_scene_length < 0: + errors.append(f"最小场景长度不能为负数: {state.min_scene_length}") + state.send_progress("validate", f"❌ 最小场景长度无效: {state.min_scene_length}", ProgressLevel.ERROR) + + if not errors: + state.send_progress("validate", "✅ 输入参数验证通过", ProgressLevel.SUCCESS) + + return { + "current_stage": "validated" if not errors else "error", + "progress": 1, + "errors": errors + } + + def extract_video_info(self, state: SceneDetectionWorkflowState) -> Dict[str, Any]: + """提取视频信息""" + state.send_progress("extract_info", "📊 提取视频信息...", ProgressLevel.INFO) + + try: + video_info = self.video_info_service.extract_video_info(Path(state.video_path)) + + state.send_progress("extract_info", + f"📹 视频信息: {video_info['resolution']}, {video_info['fps']:.2f}fps, {video_info['duration']:.2f}s", + ProgressLevel.SUCCESS, + {"video_info": video_info} + ) + + return { + "current_stage": "info_extracted", + "progress": 2, + "video_info": video_info + } + + except Exception as e: + error_msg = f"提取视频信息失败: {e}" + state.send_progress("extract_info", f"❌ {error_msg}", ProgressLevel.ERROR) + return { + "current_stage": "error", + "errors": state.errors + [error_msg] + } + + def detect_scenes(self, state: SceneDetectionWorkflowState) -> Dict[str, Any]: + """执行场景检测""" + state.send_progress("detect", "🎯 执行场景检测...", ProgressLevel.INFO) + + try: + result = self.detector_service.detect_scenes( + Path(state.video_path), + DetectorType(state.detector_type), + state.threshold, + state.min_scene_length + ) + + state.send_progress("detect", + f"✅ 检测完成: {result.total_scenes} 个场景，耗时 {result.detection_time:.2f}秒", + ProgressLevel.SUCCESS, + { + "total_scenes": result.total_scenes, + "detection_time": result.detection_time, + "total_duration": result.total_duration + } + ) + + return { + "current_stage": "scenes_detected", + "progress": 3, + "detection_result": result, + "processed_scenes": result.scenes + } + + except Exception as e: + error_msg = f"场景检测失败: {e}" + state.send_progress("detect", f"❌ {error_msg}", ProgressLevel.ERROR) + return { + "current_stage": "error", + "errors": state.errors + [error_msg] + } + + def analyze_with_ai(self, state: SceneDetectionWorkflowState) -> Dict[str, Any]: + """AI分析场景结果""" + if not self.ai_analysis_service.ai_enabled or not state.enable_ai_analysis: + state.send_progress("analyze", "⚠️ AI分析已禁用，跳过此步骤", ProgressLevel.WARNING) + return { + "current_stage": "analysis_skipped", + "progress": 4, + "ai_analysis": "AI分析已禁用" + } + + state.send_progress("analyze", "🧠 AI分析场景结果...", ProgressLevel.INFO) + + try: + state.send_progress("analyze", "🤖 正在调用AI分析服务...", ProgressLevel.INFO) + analysis = self.ai_analysis_service.analyze_detection_result( + state.detection_result, + state.video_info + ) + + state.send_progress("analyze", "✅ AI分析完成", ProgressLevel.SUCCESS, + {"analysis_length": len(analysis)}) + + return { + "current_stage": "ai_analyzed", + "progress": 4, + "ai_analysis": analysis + } + + except Exception as e: + error_msg = f"AI分析失败: {e}" + state.send_progress("analyze", f"⚠️ {error_msg}", ProgressLevel.WARNING) + return { + "current_stage": "analysis_failed", + "progress": 4, + "ai_analysis": error_msg + } + + def finalize_results(self, state: SceneDetectionWorkflowState) -> Dict[str, Any]: + """整理最终结果""" + state.send_progress("finalize", "📋 整理最终结果...", ProgressLevel.INFO) + + # 保存结果（如果指定了输出路径） + if state.output_path and state.detection_result: + try: + from ..utils import ResultSaver + from ..types import OutputFormat + + output_path = Path(state.output_path) + output_format = OutputFormat(state.output_format) + + saver = ResultSaver() + saver.save_results(state.detection_result, output_path, output_format) + state.send_progress("finalize", f"💾 结果已保存到: {output_path}", ProgressLevel.SUCCESS) + except Exception as e: + state.send_progress("finalize", f"⚠️ 保存结果失败: {e}", ProgressLevel.WARNING) + + # 准备最终结果 + final_result = { + "success": True, + "workflow_state": "completed", + "detection_result": None, + "ai_analysis": state.ai_analysis, + "video_info": state.video_info, + "errors": state.errors or [] + } + + # 序列化检测结果 + if state.detection_result: + final_result["detection_result"] = { + "success": state.detection_result.success, + "filename": state.detection_result.filename, + "detector_type": state.detection_result.detector_type, + "threshold": state.detection_result.threshold, + "total_scenes": state.detection_result.total_scenes, + "total_duration": state.detection_result.total_duration, + "detection_time": state.detection_result.detection_time, + "scenes": [asdict(scene) for scene in state.detection_result.scenes], + "error": state.detection_result.error + } + + state.send_progress("finalize", "🎉 工作流执行完成", ProgressLevel.SUCCESS, { + "total_scenes": state.detection_result.total_scenes if state.detection_result else 0, + "workflow_stage": "completed" + }) + + # 发送最终结果到JSON-RPC客户端 + state.send_final_result(final_result) + + return { + "current_stage": "completed", + "progress": 5, + "final_result": final_result + } + + def handle_error(self, state: SceneDetectionWorkflowState) -> Dict[str, Any]: + """处理错误""" + error_msg = "; ".join(state.errors) if state.errors else "未知错误" + + # 发送错误进度 + state.send_progress("error", f"❌ 工作流错误: {error_msg}", ProgressLevel.ERROR) + + # 准备错误结果 + error_result = { + "success": False, + "workflow_state": "failed", + "error": error_msg, + "errors": state.errors or [], + "detection_result": None, + "ai_analysis": None, + "video_info": state.video_info + } + + # 发送错误结果到JSON-RPC客户端 + state.send_error_result(-32603, f"工作流执行失败: {error_msg}", error_result) + + return { + "current_stage": "failed", + "error_result": error_result + } diff --git a/python_core/utils/jsonrpc_enhanced.py b/python_core/utils/jsonrpc_enhanced.py index c792143..85689ca 100644 --- a/python_core/utils/jsonrpc_enhanced.py +++ b/python_core/utils/jsonrpc_enhanced.py @@ -150,10 +150,69 @@ class JSONRPCMethodRegistry: self.dispatcher.add_method(func, method_name) - def handle_request(self, request_data: str) -> str: + def handle_request(self, request_data: str) -> Optional[str]: """处理JSON-RPC请求""" - response = JSONRPCResponseManager.handle(request_data, self.dispatcher) - return response.json + try: + # 解析请求 + request_json = json.loads(request_data) + method_name = request_json.get("method") + params = request_json.get("params", {}) + request_id = request_json.get("id") + + # 检查方法是否存在 + if method_name not in self.methods: + error_response = { + "jsonrpc": "2.0", + "id": request_id, + "error": { + "code": -32601, + "message": f"Method '{method_name}' not found" + } + } + return json.dumps(error_response, ensure_ascii=False, separators=(',', ':')) + + try: + # 调用方法 + if isinstance(params, dict): + result = self.methods[method_name](**params) + elif isinstance(params, list): + result = self.methods[method_name](*params) + else: + result = self.methods[method_name](params) + + # 如果方法返回None，表示响应已经异步发送，不需要额外响应 + if result is None: + return None + + # 正常响应 + success_response = { + "jsonrpc": "2.0", + "id": request_id, + "result": result + } + return json.dumps(success_response, ensure_ascii=False, separators=(',', ':')) + + except Exception as e: + error_response = { + "jsonrpc": "2.0", + "id": request_id, + "error": { + "code": -32603, + "message": f"Internal error: {str(e)}" + } + } + return json.dumps(error_response, ensure_ascii=False, separators=(',', ':')) + + except Exception as e: + error_response = { + "jsonrpc": "2.0", + "id": None, + "error": { + "code": -32700, + "message": f"Parse error: {str(e)}" + } + } + return json.dumps(error_response, ensure_ascii=False, separators=(',', ':')) # 全局实例 enhanced_progress_reporter = EnhancedProgressReporter() diff --git a/python_core/utils/jsonrpc_server.py b/python_core/utils/jsonrpc_server.py index db8e2f5..ec58a19 100644 --- a/python_core/utils/jsonrpc_server.py +++ b/python_core/utils/jsonrpc_server.py @@ -62,9 +62,17 @@ class JSONRPCHTTPHandler(BaseHTTPRequestHandler): # 处理JSON-RPC请求 response_data = self.method_registry.handle_request(request_data) - - # 发送响应 - self._send_json_response(response_data) + + # 如果响应为None，表示响应已经异步发送，不需要额外响应 + if response_data is not None: + # 发送响应 + self._send_json_response(response_data) + else: + # 发送空的200响应，表示请求已处理但无需响应体 + self.send_response(200) + if self.config.cors_enabled: + self._send_cors_headers() + self.end_headers() except Exception as e: if self.config.debug: