374 lines
9.7 KiB
Markdown
374 lines
9.7 KiB
Markdown
# JSON-RPC Commander 基类使用指南
|
||
|
||
## 🎯 概述
|
||
|
||
JSON-RPC Commander 基类为命令行工具提供了统一的JSON-RPC通信接口，简化了命令行工具的开发和集成。
|
||
|
||
## 📊 **测试结果**
|
||
```
|
||
🎉 所有JSON-RPC Commander测试通过！
|
||
|
||
✅ 基类功能验证:
|
||
1. 命令注册和解析 - ✅
|
||
2. 参数类型转换 - ✅
|
||
3. 错误处理 - ✅
|
||
4. JSON-RPC输出 - ✅
|
||
5. 视频拆分集成 - ✅
|
||
```
|
||
|
||
## 🔧 核心特性
|
||
|
||
### **1. 统一的命令行接口**
|
||
- 自动参数解析和类型转换
|
||
- 标准化的错误处理
|
||
- JSON-RPC 2.0 协议支持
|
||
- 灵活的命令注册机制
|
||
|
||
### **2. 两种使用方式**
|
||
- **继承方式**: 适合复杂的命令行工具
|
||
- **组合方式**: 适合简单的快速开发
|
||
|
||
## 🚀 使用方法
|
||
|
||
### **方式一：继承 JSONRPCCommander**
|
||
|
||
```python
|
||
from python_core.utils.jsonrpc_commander import JSONRPCCommander
|
||
from typing import Dict, Any
|
||
|
||
class MyServiceCommander(JSONRPCCommander):
|
||
"""自定义服务Commander"""
|
||
|
||
def __init__(self):
|
||
super().__init__("my_service")
|
||
|
||
def _register_commands(self) -> None:
|
||
"""注册命令"""
|
||
self.register_command(
|
||
name="process",
|
||
description="处理数据",
|
||
required_args=["input_file"],
|
||
optional_args={
|
||
"output": {"type": str, "default": "output.txt", "description": "输出文件"},
|
||
"format": {"type": str, "default": "json", "choices": ["json", "xml"], "description": "输出格式"},
|
||
"verbose": {"type": bool, "default": False, "description": "详细输出"}
|
||
}
|
||
)
|
||
|
||
def execute_command(self, command: str, args: Dict[str, Any]) -> Any:
|
||
"""执行命令"""
|
||
if command == "process":
|
||
return self._process_data(
|
||
input_file=args["input_file"],
|
||
output=args["output"],
|
||
format=args["format"],
|
||
verbose=args["verbose"]
|
||
)
|
||
else:
|
||
raise ValueError(f"Unknown command: {command}")
|
||
|
||
def _process_data(self, input_file: str, output: str, format: str, verbose: bool) -> Dict[str, Any]:
|
||
"""处理数据的具体实现"""
|
||
# 实际的业务逻辑
|
||
return {
|
||
"success": True,
|
||
"input_file": input_file,
|
||
"output_file": output,
|
||
"format": format,
|
||
"processed_items": 100
|
||
}
|
||
|
||
# 使用
|
||
def main():
|
||
commander = MyServiceCommander()
|
||
commander.run()
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
```
|
||
|
||
### **方式二：使用 SimpleJSONRPCCommander**
|
||
|
||
```python
|
||
from python_core.utils.jsonrpc_commander import create_simple_commander
|
||
|
||
# 创建Commander
|
||
commander = create_simple_commander("my_service")
|
||
|
||
# 定义命令处理器
|
||
def hello_handler(name: str = "World", count: int = 1):
|
||
"""打招呼命令"""
|
||
return {
|
||
"message": f"Hello, {name}!",
|
||
"count": count,
|
||
"repeated": [f"Hello, {name}!" for _ in range(count)]
|
||
}
|
||
|
||
def calculate_handler(operation: str, a: str, b: str):
|
||
"""计算命令"""
|
||
num_a, num_b = float(a), float(b)
|
||
|
||
if operation == "add":
|
||
result = num_a + num_b
|
||
elif operation == "multiply":
|
||
result = num_a * num_b
|
||
else:
|
||
raise ValueError(f"Unknown operation: {operation}")
|
||
|
||
return {
|
||
"operation": operation,
|
||
"operands": [num_a, num_b],
|
||
"result": result
|
||
}
|
||
|
||
# 注册命令
|
||
commander.add_command(
|
||
name="hello",
|
||
handler=hello_handler,
|
||
description="打招呼命令",
|
||
optional_args={
|
||
"name": {"type": str, "default": "World", "description": "名称"},
|
||
"count": {"type": int, "default": 1, "description": "重复次数"}
|
||
}
|
||
)
|
||
|
||
commander.add_command(
|
||
name="calc",
|
||
handler=calculate_handler,
|
||
description="计算命令",
|
||
required_args=["operation", "a", "b"]
|
||
)
|
||
|
||
# 运行
|
||
if __name__ == "__main__":
|
||
commander.run()
|
||
```
|
||
|
||
## 📡 JSON-RPC 输出格式
|
||
|
||
### **成功响应**
|
||
```json
|
||
{
|
||
"jsonrpc": "2.0",
|
||
"id": null,
|
||
"result": {
|
||
"success": true,
|
||
"data": "处理结果"
|
||
}
|
||
}
|
||
```
|
||
|
||
### **错误响应**
|
||
```json
|
||
{
|
||
"jsonrpc": "2.0",
|
||
"id": null,
|
||
"error": {
|
||
"code": "INVALID_COMMAND",
|
||
"message": "Unknown command: invalid_cmd"
|
||
}
|
||
}
|
||
```
|
||
|
||
### **标准错误代码**
|
||
- `INVALID_COMMAND`: 未知命令
|
||
- `MISSING_ARGS`: 缺少必需参数
|
||
- `MISSING_VALUE`: 参数缺少值
|
||
- `INVALID_VALUE`: 参数值无效
|
||
- `INTERRUPTED`: 用户中断
|
||
- `INTERNAL_ERROR`: 内部错误
|
||
|
||
## 🎬 实际应用：视频拆分服务
|
||
|
||
### **重构前的问题**
|
||
```python
|
||
# 复杂的参数解析
|
||
def parse_arguments(self) -> tuple:
|
||
if len(sys.argv) < 3:
|
||
print("Usage: ...")
|
||
sys.exit(1)
|
||
|
||
command = sys.argv[1]
|
||
video_path = sys.argv[2]
|
||
|
||
# 手动解析可选参数...
|
||
arg_definitions = {...}
|
||
parsed_args = CommandLineParser.parse_command_args(...)
|
||
# 复杂的类型转换和验证...
|
||
|
||
# 复杂的响应处理
|
||
def handle_response(self, result, error_code):
|
||
if self.rpc_handler:
|
||
JSONRPCHandler.handle_command_response(...)
|
||
else:
|
||
print(json.dumps(...))
|
||
```
|
||
|
||
### **重构后的简洁实现**
|
||
```python
|
||
class VideoSplitterCommander(JSONRPCCommander):
|
||
"""视频拆分服务命令行接口"""
|
||
|
||
def _register_commands(self) -> None:
|
||
"""注册命令"""
|
||
self.register_command(
|
||
name="analyze",
|
||
description="分析视频场景",
|
||
required_args=["video_path"],
|
||
optional_args={
|
||
"threshold": {"type": float, "default": 30.0},
|
||
"detector": {"type": str, "default": "content", "choices": ["content", "threshold"]},
|
||
"min-scene-length": {"type": float, "default": 1.0}
|
||
}
|
||
)
|
||
|
||
def execute_command(self, command: str, args: Dict[str, Any]) -> Any:
|
||
"""执行命令"""
|
||
# 创建配置
|
||
config = DetectionConfig(
|
||
threshold=args.get("threshold", 30.0),
|
||
detector_type=DetectorType(args.get("detector", "content")),
|
||
min_scene_length=args.get("min_scene_length", 1.0)
|
||
)
|
||
|
||
# 执行分析
|
||
result = self.service.analyze_video(args["video_path"], config)
|
||
return result.to_dict()
|
||
```
|
||
|
||
## 🔧 高级功能
|
||
|
||
### **1. 参数验证**
|
||
```python
|
||
optional_args={
|
||
"threshold": {
|
||
"type": float,
|
||
"default": 30.0,
|
||
"description": "检测阈值"
|
||
},
|
||
"format": {
|
||
"type": str,
|
||
"default": "json",
|
||
"choices": ["json", "xml", "yaml"], # 限制选择范围
|
||
"description": "输出格式"
|
||
},
|
||
"verbose": {
|
||
"type": bool,
|
||
"default": False,
|
||
"description": "详细输出"
|
||
}
|
||
}
|
||
```
|
||
|
||
### **2. 错误处理**
|
||
```python
|
||
def execute_command(self, command: str, args: Dict[str, Any]) -> Any:
|
||
try:
|
||
# 业务逻辑
|
||
return self._do_work(args)
|
||
except FileNotFoundError as e:
|
||
# 自定义错误会自动转换为JSON-RPC错误响应
|
||
raise ValueError(f"File not found: {e}")
|
||
except Exception as e:
|
||
# 所有异常都会被捕获并转换为INTERNAL_ERROR
|
||
raise
|
||
```
|
||
|
||
### **3. 使用帮助**
|
||
```bash
|
||
# 不提供参数时自动显示帮助
|
||
python my_service.py
|
||
|
||
# 输出：
|
||
{
|
||
"service": "my_service",
|
||
"usage": "python -m my_service <command> [args...]",
|
||
"commands": {
|
||
"process": {
|
||
"description": "处理数据",
|
||
"required_args": ["input_file"],
|
||
"optional_args": {
|
||
"output": {
|
||
"type": "str",
|
||
"default": "output.txt",
|
||
"description": "输出文件"
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
## 📈 优势对比
|
||
|
||
### **使用基类前**
|
||
| 方面 | 手动实现 | 问题 |
|
||
|------|----------|------|
|
||
| 参数解析 | 50行代码 | 重复、易错 |
|
||
| 类型转换 | 手动处理 | 不一致 |
|
||
| 错误处理 | 分散逻辑 | 格式不统一 |
|
||
| JSON-RPC | 手动实现 | 协议不标准 |
|
||
| 维护成本 | 高 | 每个工具都要重复 |
|
||
|
||
### **使用基类后**
|
||
| 方面 | 基类实现 | 优势 |
|
||
|------|----------|------|
|
||
| 参数解析 | 自动化 | 声明式配置 |
|
||
| 类型转换 | 自动化 | 统一处理 |
|
||
| 错误处理 | 标准化 | 一致的格式 |
|
||
| JSON-RPC | 内置支持 | 标准协议 |
|
||
| 维护成本 | 低 | 一次实现，处处使用 |
|
||
|
||
## 🎯 最佳实践
|
||
|
||
### **1. 命令设计**
|
||
- 使用动词作为命令名：`analyze`, `process`, `convert`
|
||
- 保持命令名简洁明了
|
||
- 提供清晰的描述信息
|
||
|
||
### **2. 参数设计**
|
||
- 必需参数放在前面
|
||
- 提供合理的默认值
|
||
- 使用描述性的参数名
|
||
- 为枚举类型提供choices
|
||
|
||
### **3. 错误处理**
|
||
- 抛出有意义的异常
|
||
- 包含足够的上下文信息
|
||
- 使用标准的错误代码
|
||
|
||
### **4. 返回值设计**
|
||
- 返回结构化的数据
|
||
- 包含操作状态信息
|
||
- 提供足够的调试信息
|
||
|
||
## 🚀 扩展应用
|
||
|
||
### **可以使用此基类的场景**
|
||
1. **AI服务命令行工具** - 文本生成、图像处理等
|
||
2. **数据处理工具** - ETL、格式转换等
|
||
3. **系统管理工具** - 配置管理、监控等
|
||
4. **开发工具** - 代码生成、测试等
|
||
|
||
### **集成建议**
|
||
1. **统一标准** - 所有命令行工具使用相同基类
|
||
2. **文档生成** - 自动生成API文档
|
||
3. **测试框架** - 统一的测试方法
|
||
4. **监控集成** - 标准化的日志和指标
|
||
|
||
## 🎉 总结
|
||
|
||
JSON-RPC Commander 基类提供了：
|
||
|
||
- ✅ **统一接口** - 标准化的命令行工具开发
|
||
- ✅ **自动化处理** - 参数解析、类型转换、错误处理
|
||
- ✅ **JSON-RPC支持** - 标准化的通信协议
|
||
- ✅ **易于使用** - 简洁的API设计
|
||
- ✅ **高度可扩展** - 支持复杂的业务逻辑
|
||
|
||
通过使用这个基类，可以大大简化命令行工具的开发，提高代码质量和一致性！
|
||
|
||
---
|
||
|
||
*JSON-RPC Commander - 让命令行工具开发更简单、更标准！*
|