#!/usr/bin/env python3 """ JSON-RPC Client Demo JSON-RPC 客户端演示 演示如何使用Python客户端调用场景检测JSON-RPC API """ import json import requests import time from typing import Dict, Any, Optional class SceneDetectionClient: """场景检测JSON-RPC客户端""" def __init__(self, server_url: str = "http://localhost:8080"): self.server_url = server_url self.request_id = 0 def _call_method(self, method: str, params: Dict[str, Any]) -> Dict[str, Any]: """调用JSON-RPC方法""" self.request_id += 1 payload = { "jsonrpc": "2.0", "method": method, "params": params, "id": self.request_id } try: response = requests.post( self.server_url, json=payload, headers={'Content-Type': 'application/json'}, timeout=60 ) if response.status_code == 200: return response.json() else: return { "error": { "code": response.status_code, "message": f"HTTP Error: {response.text}" } } except requests.exceptions.RequestException as e: return { "error": { "code": -1, "message": f"Request failed: {str(e)}" } } def detect_scenes(self, video_path: str, detector_type: str = "content", threshold: float = 30.0, min_scene_length: float = 1.0) -> Dict[str, Any]: """基础场景检测""" params = { "video_path": video_path, "detector_type": detector_type, "threshold": threshold, "min_scene_length": min_scene_length } return self._call_method("scene.detect", params) def detect_scenes_workflow(self, video_path: str, detector_type: str = "content", threshold: float = 30.0, min_scene_length: float = 1.0, output_path: Optional[str] = None, output_format: str = "json", enable_ai_analysis: bool = True) -> Dict[str, Any]: """LangGraph工作流场景检测""" params = { "video_path": video_path, "detector_type": detector_type, "threshold": threshold, "min_scene_length": min_scene_length, "enable_ai_analysis": enable_ai_analysis } if output_path: params["output_path"] = output_path params["output_format"] = output_format return self._call_method("scene.detect_workflow", params) def get_video_info(self, video_path: str) -> Dict[str, Any]: """获取视频信息""" params = {"video_path": video_path} return self._call_method("scene.get_video_info", params) def batch_detect(self, directory: str, detector_type: str = "content", threshold: float = 30.0, min_scene_length: float = 1.0, output_dir: Optional[str] = None, output_format: str = "json") -> Dict[str, Any]: """批量场景检测""" params = { "directory": directory, "detector_type": detector_type, "threshold": threshold, "min_scene_length": min_scene_length, "output_format": output_format } if output_dir: params["output_dir"] = output_dir return self._call_method("scene.batch_detect", params) def demo_basic_detection(): """演示基础场景检测""" print("🎯 基础场景检测演示") print("=" * 50) client = SceneDetectionClient("http://localhost:8081") # 测试视频路径 video_path = "assets/1/1752032011698.mp4" print(f"📹 检测视频: {video_path}") # 调用基础检测 start_time = time.time() result = client.detect_scenes(video_path, threshold=15.0) end_time = time.time() if "error" in result: print(f"❌ 检测失败: {result['error']}") return detection_result = result.get("result", {}) if detection_result.get("success"): print(f"✅ 检测成功!") print(f" 场景数量: {detection_result['total_scenes']}") print(f" 视频时长: {detection_result['total_duration']:.2f}秒") print(f" 检测时间: {detection_result['detection_time']:.2f}秒") print(f" API调用时间: {end_time - start_time:.2f}秒") # 显示场景详情 scenes = detection_result.get("scenes", []) print(f"\n🎬 场景详情:") for scene in scenes[:5]: # 只显示前5个 print(f" 场景 {scene['index']}: {scene['start_time']:.2f}s - {scene['end_time']:.2f}s") if len(scenes) > 5: print(f" ... 还有 {len(scenes) - 5} 个场景") else: print(f"❌ 检测失败: {detection_result.get('error', '未知错误')}") def demo_workflow_detection(): """演示工作流场景检测""" print("\n🔄 LangGraph工作流检测演示") print("=" * 50) client = SceneDetectionClient("http://localhost:8081") # 测试视频路径 video_path = "assets/1/1752032011698.mp4" print(f"📹 检测视频: {video_path}") # 调用工作流检测 start_time = time.time() result = client.detect_scenes_workflow( video_path, threshold=15.0, enable_ai_analysis=False # 禁用AI分析以避免API密钥问题 ) end_time = time.time() if "error" in result: print(f"❌ 工作流检测失败: {result['error']}") return workflow_result = result.get("result", {}) detection_result = workflow_result.get("detection_result", {}) video_info = workflow_result.get("video_info", {}) ai_analysis = workflow_result.get("ai_analysis") workflow_state = workflow_result.get("workflow_state") if detection_result.get("success"): print(f"✅ 工作流检测成功!") print(f" 工作流状态: {workflow_state}") print(f" 场景数量: {detection_result['total_scenes']}") print(f" 检测时间: {detection_result['detection_time']:.2f}秒") print(f" API调用时间: {end_time - start_time:.2f}秒") # 显示视频信息 if video_info: print(f"\n📹 视频信息:") print(f" 分辨率: {video_info.get('resolution')}") print(f" 帧率: {video_info.get('fps'):.2f} fps") print(f" 时长: {video_info.get('duration'):.2f}秒") # 显示AI分析结果 if ai_analysis: print(f"\n🧠 AI分析: {ai_analysis}") else: print(f"❌ 工作流检测失败: {detection_result.get('error', '未知错误')}") def demo_video_info(): """演示获取视频信息""" print("\n📊 视频信息获取演示") print("=" * 50) client = SceneDetectionClient("http://localhost:8081") # 测试视频路径 video_path = "assets/1/1752032011698.mp4" print(f"📹 获取视频信息: {video_path}") # 获取视频信息 start_time = time.time() result = client.get_video_info(video_path) end_time = time.time() if "error" in result: print(f"❌ 获取失败: {result['error']}") return info_result = result.get("result", {}) if info_result.get("success"): info = info_result.get("info", {}) print(f"✅ 获取成功!") print(f" 文件名: {info.get('filename')}") print(f" 分辨率: {info.get('resolution')}") print(f" 帧率: {info.get('fps'):.2f} fps") print(f" 总帧数: {info.get('frame_count'):,}") print(f" 时长: {info.get('duration'):.2f}秒") print(f" 文件大小: {info.get('file_size'):,} 字节") print(f" API调用时间: {end_time - start_time:.2f}秒") else: print(f"❌ 获取失败: {info_result.get('error', '未知错误')}") def main(): """主演示函数""" print("🚀 JSON-RPC 场景检测客户端演示") print("=" * 60) # 检查服务器连接 client = SceneDetectionClient("http://localhost:8081") try: # 简单的连接测试 test_result = client.get_video_info("nonexistent.mp4") if "error" in test_result and "Request failed" in str(test_result["error"]): print("❌ 无法连接到JSON-RPC服务器") print("💡 请先启动服务器: python3 -m python_core.cli jsonrpc start --port 8081") return except Exception as e: print(f"❌ 连接测试失败: {e}") return print("✅ 服务器连接正常") # 运行演示 demo_video_info() demo_basic_detection() demo_workflow_detection() print("\n🎉 演示完成!") print("\n💡 更多用法:") print(" • 调整检测阈值以获得不同的场景分割效果") print(" • 使用不同的检测器类型 (content/threshold/adaptive)") print(" • 启用AI分析获得智能建议 (需要配置API密钥)") print(" • 批量处理多个视频文件") if __name__ == "__main__": main()