diff --git a/examples/jsonrpc_client_demo.py b/examples/jsonrpc_client_demo.py new file mode 100644 index 0000000..4d638e6 --- /dev/null +++ b/examples/jsonrpc_client_demo.py @@ -0,0 +1,279 @@ +#!/usr/bin/env python3 +""" +JSON-RPC Client Demo +JSON-RPC 客户端演示 + +演示如何使用Python客户端调用场景检测JSON-RPC API +""" + +import json +import requests +import time +from typing import Dict, Any, Optional + + +class SceneDetectionClient: + """场景检测JSON-RPC客户端""" + + def __init__(self, server_url: str = "http://localhost:8080"): + self.server_url = server_url + self.request_id = 0 + + def _call_method(self, method: str, params: Dict[str, Any]) -> Dict[str, Any]: + """调用JSON-RPC方法""" + self.request_id += 1 + + payload = { + "jsonrpc": "2.0", + "method": method, + "params": params, + "id": self.request_id + } + + try: + response = requests.post( + self.server_url, + json=payload, + headers={'Content-Type': 'application/json'}, + timeout=60 + ) + + if response.status_code == 200: + return response.json() + else: + return { + "error": { + "code": response.status_code, + "message": f"HTTP Error: {response.text}" + } + } + + except requests.exceptions.RequestException as e: + return { + "error": { + "code": -1, + "message": f"Request failed: {str(e)}" + } + } + + def detect_scenes(self, video_path: str, detector_type: str = "content", + threshold: float = 30.0, min_scene_length: float = 1.0) -> Dict[str, Any]: + """基础场景检测""" + params = { + "video_path": video_path, + "detector_type": detector_type, + "threshold": threshold, + "min_scene_length": min_scene_length + } + + return self._call_method("scene.detect", params) + + def detect_scenes_workflow(self, video_path: str, detector_type: str = "content", + threshold: float = 30.0, min_scene_length: float = 1.0, + output_path: Optional[str] = None, output_format: str = "json", + enable_ai_analysis: bool = True) -> Dict[str, Any]: + """LangGraph工作流场景检测""" + params = { + "video_path": video_path, + "detector_type": detector_type, + "threshold": threshold, + "min_scene_length": min_scene_length, + "enable_ai_analysis": enable_ai_analysis + } + + if output_path: + params["output_path"] = output_path + params["output_format"] = output_format + + return self._call_method("scene.detect_workflow", params) + + def get_video_info(self, video_path: str) -> Dict[str, Any]: + """获取视频信息""" + params = {"video_path": video_path} + return self._call_method("scene.get_video_info", params) + + def batch_detect(self, directory: str, detector_type: str = "content", + threshold: float = 30.0, min_scene_length: float = 1.0, + output_dir: Optional[str] = None, output_format: str = "json") -> Dict[str, Any]: + """批量场景检测""" + params = { + "directory": directory, + "detector_type": detector_type, + "threshold": threshold, + "min_scene_length": min_scene_length, + "output_format": output_format + } + + if output_dir: + params["output_dir"] = output_dir + + return self._call_method("scene.batch_detect", params) + + +def demo_basic_detection(): + """演示基础场景检测""" + print("🎯 基础场景检测演示") + print("=" * 50) + + client = SceneDetectionClient("http://localhost:8081") + + # 测试视频路径 + video_path = "assets/1/1752032011698.mp4" + + print(f"📹 检测视频: {video_path}") + + # 调用基础检测 + start_time = time.time() + result = client.detect_scenes(video_path, threshold=15.0) + end_time = time.time() + + if "error" in result: + print(f"❌ 检测失败: {result['error']}") + return + + detection_result = result.get("result", {}) + + if detection_result.get("success"): + print(f"✅ 检测成功!") + print(f" 场景数量: {detection_result['total_scenes']}") + print(f" 视频时长: {detection_result['total_duration']:.2f}秒") + print(f" 检测时间: {detection_result['detection_time']:.2f}秒") + print(f" API调用时间: {end_time - start_time:.2f}秒") + + # 显示场景详情 + scenes = detection_result.get("scenes", []) + print(f"\n🎬 场景详情:") + for scene in scenes[:5]: # 只显示前5个 + print(f" 场景 {scene['index']}: {scene['start_time']:.2f}s - {scene['end_time']:.2f}s") + + if len(scenes) > 5: + print(f" ... 还有 {len(scenes) - 5} 个场景") + else: + print(f"❌ 检测失败: {detection_result.get('error', '未知错误')}") + + +def demo_workflow_detection(): + """演示工作流场景检测""" + print("\n🔄 LangGraph工作流检测演示") + print("=" * 50) + + client = SceneDetectionClient("http://localhost:8081") + + # 测试视频路径 + video_path = "assets/1/1752032011698.mp4" + + print(f"📹 检测视频: {video_path}") + + # 调用工作流检测 + start_time = time.time() + result = client.detect_scenes_workflow( + video_path, + threshold=15.0, + enable_ai_analysis=False # 禁用AI分析以避免API密钥问题 + ) + end_time = time.time() + + if "error" in result: + print(f"❌ 工作流检测失败: {result['error']}") + return + + workflow_result = result.get("result", {}) + detection_result = workflow_result.get("detection_result", {}) + video_info = workflow_result.get("video_info", {}) + ai_analysis = workflow_result.get("ai_analysis") + workflow_state = workflow_result.get("workflow_state") + + if detection_result.get("success"): + print(f"✅ 工作流检测成功!") + print(f" 工作流状态: {workflow_state}") + print(f" 场景数量: {detection_result['total_scenes']}") + print(f" 检测时间: {detection_result['detection_time']:.2f}秒") + print(f" API调用时间: {end_time - start_time:.2f}秒") + + # 显示视频信息 + if video_info: + print(f"\n📹 视频信息:") + print(f" 分辨率: {video_info.get('resolution')}") + print(f" 帧率: {video_info.get('fps'):.2f} fps") + print(f" 时长: {video_info.get('duration'):.2f}秒") + + # 显示AI分析结果 + if ai_analysis: + print(f"\n🧠 AI分析: {ai_analysis}") + else: + print(f"❌ 工作流检测失败: {detection_result.get('error', '未知错误')}") + + +def demo_video_info(): + """演示获取视频信息""" + print("\n📊 视频信息获取演示") + print("=" * 50) + + client = SceneDetectionClient("http://localhost:8081") + + # 测试视频路径 + video_path = "assets/1/1752032011698.mp4" + + print(f"📹 获取视频信息: {video_path}") + + # 获取视频信息 + start_time = time.time() + result = client.get_video_info(video_path) + end_time = time.time() + + if "error" in result: + print(f"❌ 获取失败: {result['error']}") + return + + info_result = result.get("result", {}) + + if info_result.get("success"): + info = info_result.get("info", {}) + print(f"✅ 获取成功!") + print(f" 文件名: {info.get('filename')}") + print(f" 分辨率: {info.get('resolution')}") + print(f" 帧率: {info.get('fps'):.2f} fps") + print(f" 总帧数: {info.get('frame_count'):,}") + print(f" 时长: {info.get('duration'):.2f}秒") + print(f" 文件大小: {info.get('file_size'):,} 字节") + print(f" API调用时间: {end_time - start_time:.2f}秒") + else: + print(f"❌ 获取失败: {info_result.get('error', '未知错误')}") + + +def main(): + """主演示函数""" + print("🚀 JSON-RPC 场景检测客户端演示") + print("=" * 60) + + # 检查服务器连接 + client = SceneDetectionClient("http://localhost:8081") + + try: + # 简单的连接测试 + test_result = client.get_video_info("nonexistent.mp4") + if "error" in test_result and "Request failed" in str(test_result["error"]): + print("❌ 无法连接到JSON-RPC服务器") + print("💡 请先启动服务器: python3 -m python_core.cli jsonrpc start --port 8081") + return + except Exception as e: + print(f"❌ 连接测试失败: {e}") + return + + print("✅ 服务器连接正常") + + # 运行演示 + demo_video_info() + demo_basic_detection() + demo_workflow_detection() + + print("\n🎉 演示完成!") + print("\n💡 更多用法:") + print(" • 调整检测阈值以获得不同的场景分割效果") + print(" • 使用不同的检测器类型 (content/threshold/adaptive)") + print(" • 启用AI分析获得智能建议 (需要配置API密钥)") + print(" • 批量处理多个视频文件") + + +if __name__ == "__main__": + main() diff --git a/python_core/cli/cli.py b/python_core/cli/cli.py index 7755885..bb379fb 100644 --- a/python_core/cli/cli.py +++ b/python_core/cli/cli.py @@ -11,6 +11,7 @@ import typer # 导入命令模块 from python_core.cli.commands import scene_app +from python_core.cli.commands.jsonrpc_server import jsonrpc_app app = typer.Typer( name="mixvideo", @@ -28,13 +29,15 @@ app = typer.Typer( 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.command() def init(): diff --git a/python_core/cli/commands/jsonrpc_server.py b/python_core/cli/commands/jsonrpc_server.py new file mode 100644 index 0000000..7d0b9f0 --- /dev/null +++ b/python_core/cli/commands/jsonrpc_server.py @@ -0,0 +1,303 @@ +#!/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 index 98116e4..d0234d8 100644 --- a/python_core/cli/commands/scene.py +++ b/python_core/cli/commands/scene.py @@ -19,7 +19,9 @@ def detect( 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)") + 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: @@ -40,39 +42,97 @@ def detect( progress_reporter.info("💡 可用格式: json, csv, txt") raise typer.Exit(1) - # 执行检测 - 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 + # 选择执行方式 + 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}") @@ -342,3 +402,148 @@ def info( 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/const.py b/python_core/cli/const.py index 44b81c3..7209740 100644 --- a/python_core/cli/const.py +++ b/python_core/cli/const.py @@ -1,4 +1,4 @@ -from python_core.utils.jsonrpc import create_progress_reporter +from python_core.utils.jsonrpc_enhanced import create_progress_reporter from rich.console import Console from python_core.config import settings console = Console() diff --git a/python_core/cli/scene_detect.py b/python_core/cli/scene_detect.py index 0b36b78..0937cbf 100644 --- a/python_core/cli/scene_detect.py +++ b/python_core/cli/scene_detect.py @@ -1,23 +1,26 @@ #!/usr/bin/env python3 """ -PySceneDetect 场景检测命令行工具 +PySceneDetect 场景检测命令行工具 - LangGraph增强版 """ -import os import json import time from pathlib import Path -from typing import Optional, List +from typing import Optional, List, Literal, Dict, Any from enum import Enum from dataclasses import dataclass, asdict -import typer -from python_core.cli.const import progress_reporter, console, project_root +from python_core.cli.const import progress_reporter -# 检查 PySceneDetect 依赖 +# PySceneDetect 依赖 from scenedetect import open_video, SceneManager from scenedetect.detectors import ContentDetector, ThresholdDetector -from scenedetect.video_splitter import split_video_ffmpeg + +# 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): """检测器类型""" @@ -55,11 +58,59 @@ class DetectionResult: 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: @@ -361,5 +412,460 @@ class SceneDetector: 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() \ No newline at end of file +detector = SceneDetector() + +# 注册JSON-RPC方法 +detector.register_jsonrpc_methods() \ No newline at end of file diff --git a/python_core/requirements.txt b/python_core/requirements.txt index f37092a..0762924 100644 --- a/python_core/requirements.txt +++ b/python_core/requirements.txt @@ -26,4 +26,8 @@ pyyaml pydantic_settings scenedetect[opencv] typer -rich \ No newline at end of file +rich +langgraph +json-rpc +langchain_core +langchain_anthropic \ No newline at end of file diff --git a/python_core/server/__init__.py b/python_core/server/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/python_core/server/__main__.py b/python_core/server/__main__.py new file mode 100644 index 0000000..e69de29 diff --git a/python_core/server/server.py b/python_core/server/server.py new file mode 100644 index 0000000..e69de29 diff --git a/python_core/utils/jsonrpc.py b/python_core/utils/jsonrpc.py index 2e1e8a1..6815492 100644 --- a/python_core/utils/jsonrpc.py +++ b/python_core/utils/jsonrpc.py @@ -112,8 +112,9 @@ class JSONRPCResponse: class ProgressReporter: """Progress reporting using JSON-RPC notifications""" - - def __init__(self): + step: int = 0 + total: int = 0 + def __init__(self, total: int = 0): self.rpc = JSONRPCResponse() def report(self, step: str, progress: float, message: str, details: Dict[str, Any] = None) -> None: diff --git a/python_core/utils/jsonrpc_enhanced.py b/python_core/utils/jsonrpc_enhanced.py new file mode 100644 index 0000000..c792143 --- /dev/null +++ b/python_core/utils/jsonrpc_enhanced.py @@ -0,0 +1,186 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Enhanced JSON-RPC Communication Module +增强版 JSON-RPC 通信模块 + +Uses json-rpc library for robust JSON-RPC 2.0 communication. +""" + +import json +import sys +import time +import uuid +import asyncio +from typing import Any, Dict, Optional, Union, Callable, List +from dataclasses import dataclass, asdict +from enum import Enum + +# json-rpc library imports +from jsonrpc import JSONRPCResponseManager, dispatcher +from jsonrpc.exceptions import JSONRPCError, JSONRPCInvalidRequest, JSONRPCMethodNotFound +from jsonrpc.jsonrpc2 import JSONRPC20Request, JSONRPC20Response + +class ProgressLevel(str, Enum): + """进度级别""" + INFO = "info" + SUCCESS = "success" + WARNING = "warning" + ERROR = "error" + DEBUG = "debug" + + +@dataclass +class ProgressUpdate: + """进度更新数据结构""" + step: str + type: str + progress: int + message: str + timestamp: float + level: ProgressLevel = ProgressLevel.INFO + data: Optional[Dict[str, Any]] = None + + +class EnhancedJSONRPCResponse: + """增强版 JSON-RPC 2.0 Response handler""" + + def __init__(self, request_id: Optional[Union[str, int]] = None): + # 如果没有提供request_id，生成一个UUID + if request_id is None: + self.request_id = str(uuid.uuid4()) + else: + self.request_id = request_id + + def success(self, result: Any) -> None: + """发送成功响应""" + response = JSONRPC20Response(result=result, _id=self.request_id) + self._send_response(response.data) + + def error(self, code: int, message: str, data: Any = None) -> None: + """发送错误响应""" + error = JSONRPCError(code=code, message=message, data=data) + response = JSONRPC20Response(error=error, _id=self.request_id) + self._send_response(response.data) + + def progress(self, step: str, progress: int, message: str, + level: ProgressLevel = ProgressLevel.INFO, data: Optional[Dict[str, Any]] = None) -> None: + """发送进度通知""" + progress_data = ProgressUpdate( + step=step, + type="progress", + progress=progress, + message=message, + timestamp=time.time(), + level=level, + data=data + ) + + self.notification(asdict(progress_data)) + + def notification(self, params: Any = None) -> None: + """发送通知（无需响应）""" + response = JSONRPC20Response(result=params, _id=self.request_id) + self._send_response(response.data) + + def _send_response(self, response: Dict[str, Any]) -> None: + """发送响应到标准输出""" + json_str = json.dumps(response, ensure_ascii=False, separators=(',', ':')) + print(f"JSONRPC:{json_str}", file=sys.stdout, flush=True) + +class EnhancedProgressReporter: + """增强版进度报告器""" + + def __init__(self, total: int = 0): + self.response = EnhancedJSONRPCResponse() + self.step = 0 + self.total = total + + def update(self, message: str, step: Optional[int] = None, data: Optional[Dict[str, Any]] = None) -> None: + """更新进度""" + if step is not None: + self.step = step + else: + self.step += 1 + + progress = int((self.step / self.total * 100)) if self.total > 0 else -1 + self.response.progress("update", progress, message, ProgressLevel.INFO, data) + + def info(self, message: str, data: Optional[Dict[str, Any]] = None) -> None: + """信息消息""" + self.response.progress("info", -1, message, ProgressLevel.INFO, data) + + def success(self, message: str, data: Optional[Dict[str, Any]] = None) -> None: + """成功消息""" + self.response.progress("success", -1, message, ProgressLevel.SUCCESS, data) + + def warning(self, message: str, data: Optional[Dict[str, Any]] = None) -> None: + """警告消息""" + self.response.progress("warning", -1, message, ProgressLevel.WARNING, data) + + def error(self, message: str, data: Optional[Dict[str, Any]] = None) -> None: + """错误消息""" + self.response.progress("error", -1, message, ProgressLevel.ERROR, data) + + def debug(self, message: str, data: Optional[Dict[str, Any]] = None) -> None: + """调试消息""" + self.response.progress("debug", -1, message, ProgressLevel.DEBUG, data) + + +class JSONRPCMethodRegistry: + """JSON-RPC 方法注册器""" + + def __init__(self): + self.methods: Dict[str, Callable] = {} + self.dispatcher = dispatcher + + def register(self, name: Optional[str] = None): + """注册方法装饰器""" + def decorator(func: Callable): + method_name = name or func.__name__ + self.methods[method_name] = func + self.dispatcher.add_method(func, method_name) + return func + return decorator + + def register_function(self, func: Callable, name: Optional[str] = None) -> None: + """直接注册函数""" + method_name = name or func.__name__ + self.methods[method_name] = func + self.dispatcher.add_method(func, method_name) + + + def handle_request(self, request_data: str) -> str: + """处理JSON-RPC请求""" + response = JSONRPCResponseManager.handle(request_data, self.dispatcher) + return response.json + +# 全局实例 +enhanced_progress_reporter = EnhancedProgressReporter() +method_registry = JSONRPCMethodRegistry() + + +# 便捷函数 +def create_response_handler(request_id: Optional[Union[str, int]] = None) -> EnhancedJSONRPCResponse: + """创建响应处理器""" + return EnhancedJSONRPCResponse(request_id) + + +def create_progress_reporter(total: int = 0) -> EnhancedProgressReporter: + """创建进度报告器""" + return EnhancedProgressReporter(total) + + +def register_method(name: Optional[str] = None): + """注册JSON-RPC方法""" + return method_registry.register(name) + + +def handle_jsonrpc_request(request_data: str) -> str: + """处理JSON-RPC请求""" + return method_registry.handle_request(request_data) + + +# 向后兼容的别名 +JSONRPCResponse = EnhancedJSONRPCResponse +ProgressReporter = EnhancedProgressReporter diff --git a/python_core/utils/jsonrpc_server.py b/python_core/utils/jsonrpc_server.py new file mode 100644 index 0000000..db8e2f5 --- /dev/null +++ b/python_core/utils/jsonrpc_server.py @@ -0,0 +1,274 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +JSON-RPC Server Implementation +JSON-RPC 服务器实现 + +Provides HTTP and WebSocket JSON-RPC server implementations. +""" + +import json +import asyncio +from typing import Any, Dict, Optional, Callable, List +from dataclasses import dataclass +from http.server import HTTPServer, BaseHTTPRequestHandler +import threading +import logging + +try: + from jsonrpc import JSONRPCResponseManager, dispatcher + from jsonrpc.exceptions import JSONRPCError + JSON_RPC_AVAILABLE = True +except ImportError: + JSON_RPC_AVAILABLE = False + +from .jsonrpc_enhanced import JSONRPCMethodRegistry, EnhancedProgressReporter + + +@dataclass +class ServerConfig: + """服务器配置""" + host: str = "localhost" + port: int = 8080 + debug: bool = False + cors_enabled: bool = True + max_request_size: int = 1024 * 1024 # 1MB + + +class JSONRPCHTTPHandler(BaseHTTPRequestHandler): + """JSON-RPC HTTP 请求处理器""" + + def __init__(self, method_registry: JSONRPCMethodRegistry, config: ServerConfig, *args, **kwargs): + self.method_registry = method_registry + self.config = config + super().__init__(*args, **kwargs) + + def do_POST(self): + """处理POST请求""" + try: + # 检查Content-Type + content_type = self.headers.get('Content-Type', '') + if 'application/json' not in content_type: + self._send_error(400, "Content-Type must be application/json") + return + + # 读取请求体 + content_length = int(self.headers.get('Content-Length', 0)) + if content_length > self.config.max_request_size: + self._send_error(413, "Request too large") + return + + request_data = self.rfile.read(content_length).decode('utf-8') + + # 处理JSON-RPC请求 + response_data = self.method_registry.handle_request(request_data) + + # 发送响应 + self._send_json_response(response_data) + + except Exception as e: + if self.config.debug: + logging.exception("Error handling request") + self._send_error(500, f"Internal server error: {str(e)}") + + def do_OPTIONS(self): + """处理OPTIONS请求（CORS预检）""" + if self.config.cors_enabled: + self._send_cors_headers() + self.end_headers() + else: + self._send_error(405, "Method not allowed") + + def _send_json_response(self, data: str): + """发送JSON响应""" + self.send_response(200) + self.send_header('Content-Type', 'application/json') + if self.config.cors_enabled: + self._send_cors_headers() + self.end_headers() + self.wfile.write(data.encode('utf-8')) + + def _send_cors_headers(self): + """发送CORS头""" + self.send_header('Access-Control-Allow-Origin', '*') + self.send_header('Access-Control-Allow-Methods', 'POST, OPTIONS') + self.send_header('Access-Control-Allow-Headers', 'Content-Type') + + def _send_error(self, code: int, message: str): + """发送错误响应""" + self.send_response(code) + self.send_header('Content-Type', 'application/json') + if self.config.cors_enabled: + self._send_cors_headers() + self.end_headers() + + error_response = { + "jsonrpc": "2.0", + "id": None, + "error": { + "code": code, + "message": message + } + } + self.wfile.write(json.dumps(error_response).encode('utf-8')) + + def log_message(self, format, *args): + """重写日志方法""" + if self.config.debug: + super().log_message(format, *args) + + +class JSONRPCServer: + """JSON-RPC HTTP 服务器""" + + def __init__(self, config: Optional[ServerConfig] = None): + self.config = config or ServerConfig() + self.method_registry = JSONRPCMethodRegistry() + self.server: Optional[HTTPServer] = None + self.server_thread: Optional[threading.Thread] = None + self.running = False + + def register_method(self, name: Optional[str] = None): + """注册方法装饰器""" + return self.method_registry.register(name) + + def register_function(self, func: Callable, name: Optional[str] = None): + """注册函数""" + self.method_registry.register_function(func, name) + + def start(self, blocking: bool = True): + """启动服务器""" + if self.running: + raise RuntimeError("Server is already running") + + # 创建处理器工厂 + def handler_factory(*args, **kwargs): + return JSONRPCHTTPHandler(self.method_registry, self.config, *args, **kwargs) + + # 创建服务器 + self.server = HTTPServer((self.config.host, self.config.port), handler_factory) + self.running = True + + print(f"🚀 JSON-RPC Server started on http://{self.config.host}:{self.config.port}") + + if blocking: + try: + self.server.serve_forever() + except KeyboardInterrupt: + self.stop() + else: + self.server_thread = threading.Thread(target=self.server.serve_forever) + self.server_thread.daemon = True + self.server_thread.start() + + def stop(self): + """停止服务器""" + if self.server and self.running: + print("🛑 Stopping JSON-RPC Server...") + self.server.shutdown() + self.server.server_close() + self.running = False + + if self.server_thread: + self.server_thread.join(timeout=5) + + def __enter__(self): + """上下文管理器入口""" + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """上下文管理器出口""" + self.stop() + + +# 异步WebSocket服务器（需要额外依赖） +try: + import websockets + import asyncio + + class JSONRPCWebSocketServer: + """JSON-RPC WebSocket 服务器""" + + def __init__(self, config: Optional[ServerConfig] = None): + self.config = config or ServerConfig() + self.method_registry = JSONRPCMethodRegistry() + self.clients: List[Any] = [] + + def register_method(self, name: Optional[str] = None): + """注册方法装饰器""" + return self.method_registry.register(name) + + async def handle_client(self, websocket, path): + """处理WebSocket客户端""" + self.clients.append(websocket) + try: + async for message in websocket: + try: + response = self.method_registry.handle_request(message) + await websocket.send(response) + except Exception as e: + error_response = { + "jsonrpc": "2.0", + "id": None, + "error": { + "code": -32603, + "message": f"Internal error: {str(e)}" + } + } + await websocket.send(json.dumps(error_response)) + except websockets.exceptions.ConnectionClosed: + pass + finally: + if websocket in self.clients: + self.clients.remove(websocket) + + async def broadcast(self, method: str, params: Any = None): + """广播通知到所有客户端""" + if not self.clients: + return + + notification = { + "jsonrpc": "2.0", + "method": method, + "params": params + } + message = json.dumps(notification) + + # 发送到所有连接的客户端 + disconnected = [] + for client in self.clients: + try: + await client.send(message) + except websockets.exceptions.ConnectionClosed: + disconnected.append(client) + + # 清理断开的连接 + for client in disconnected: + self.clients.remove(client) + + async def start(self): + """启动WebSocket服务器""" + print(f"🚀 JSON-RPC WebSocket Server started on ws://{self.config.host}:{self.config.port}") + + async with websockets.serve( + self.handle_client, + self.config.host, + self.config.port + ): + await asyncio.Future() # 永远运行 + +except ImportError: + class JSONRPCWebSocketServer: + def __init__(self, *args, **kwargs): + raise ImportError("WebSocket server requires 'websockets' package") + + +# 便捷函数 +def create_server(config: Optional[ServerConfig] = None) -> JSONRPCServer: + """创建HTTP服务器""" + return JSONRPCServer(config) + + +def create_websocket_server(config: Optional[ServerConfig] = None) -> JSONRPCWebSocketServer: + """创建WebSocket服务器""" + return JSONRPCWebSocketServer(config) diff --git a/setup_venv.sh b/setup_venv.sh new file mode 100644 index 0000000..d9250c4 --- /dev/null +++ b/setup_venv.sh @@ -0,0 +1,21 @@ +#!/bin/bash + +# 创建虚拟环境 +echo "🔧 创建Python虚拟环境..." +python -m venv venv + +# 激活虚拟环境 +echo "🔌 激活虚拟环境..." +source venv/bin/activate + +# 安装依赖 +echo "📦 安装依赖..." +pip install -r python_core/requirements.txt + +# 特别安装langchain_anthropic +echo "🤖 安装langchain_anthropic..." +pip install langchain_anthropic + +echo "✅ 环境设置完成！" +echo "💡 使用 'source venv/bin/activate' 激活环境" +echo "💡 使用 'deactivate' 退出环境" \ No newline at end of file