fix
This commit is contained in:
root
319
docs/modular-commander-architecture.md
Normal file
319
docs/modular-commander-architecture.md
Normal file
@@ -0,0 +1,319 @@
|
||||
# 模块化Commander架构
|
||||
|
||||
## 🎯 重构目标
|
||||
|
||||
将原来的大文件拆分成更小、更专注的模块，提高代码的可维护性和可重用性。
|
||||
|
||||
## 📊 **重构结果**
|
||||
```
|
||||
🎉 所有模块化测试通过！
|
||||
|
||||
✅ 模块化优势:
|
||||
1. 代码组织 - 每个模块职责单一明确
|
||||
2. 易于维护 - 修改某个功能只需改对应模块
|
||||
3. 可重用性 - 模块可以独立使用
|
||||
4. 测试友好 - 可以单独测试每个模块
|
||||
5. 扩展性强 - 新功能可以独立添加
|
||||
```
|
||||
|
||||
## 📁 新的模块结构
|
||||
|
||||
### **Commander模块** (`python_core/utils/commander/`)
|
||||
|
||||
```
|
||||
commander/
|
||||
├── __init__.py - 统一导入接口
|
||||
├── types.py - 数据类型定义
|
||||
├── parser.py - 参数解析器
|
||||
├── base.py - 基础Commander类
|
||||
└── simple.py - 简化Commander类
|
||||
```
|
||||
|
||||
#### **1. types.py** - 数据类型定义
|
||||
```python
|
||||
@dataclass
|
||||
class CommandConfig:
|
||||
"""命令配置"""
|
||||
name: str
|
||||
description: str
|
||||
required_args: List[str]
|
||||
optional_args: Dict[str, Dict[str, Any]]
|
||||
```
|
||||
|
||||
#### **2. parser.py** - 参数解析器
|
||||
```python
|
||||
class ArgumentParser:
|
||||
"""命令行参数解析器"""
|
||||
|
||||
def parse_arguments(self, args: List[str]) -> Tuple[str, Dict[str, Any]]:
|
||||
# 解析命令行参数，支持类型转换和验证
|
||||
```
|
||||
|
||||
#### **3. base.py** - 基础Commander类
|
||||
```python
|
||||
class JSONRPCCommander(ABC):
|
||||
"""JSON-RPC Commander 基类"""
|
||||
|
||||
@abstractmethod
|
||||
def _register_commands(self) -> None:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def execute_command(self, command: str, args: Dict[str, Any]) -> Any:
|
||||
pass
|
||||
```
|
||||
|
||||
#### **4. simple.py** - 简化Commander类
|
||||
```python
|
||||
class SimpleJSONRPCCommander(JSONRPCCommander):
|
||||
"""简化的JSON-RPC Commander，用于快速创建命令行工具"""
|
||||
|
||||
def add_command(self, name: str, handler: Callable, ...):
|
||||
# 动态添加命令处理器
|
||||
```
|
||||
|
||||
### **Progress模块** (`python_core/utils/progress/`)
|
||||
|
||||
```
|
||||
progress/
|
||||
├── __init__.py - 统一导入接口
|
||||
├── types.py - 进度相关类型
|
||||
├── task.py - 任务管理
|
||||
├── reporter.py - 进度报告
|
||||
├── generator.py - 进度生成器
|
||||
├── decorators.py - 装饰器
|
||||
└── commander.py - 进度Commander
|
||||
```
|
||||
|
||||
#### **1. types.py** - 进度相关类型
|
||||
```python
|
||||
@dataclass
|
||||
class ProgressInfo:
|
||||
current: int
|
||||
total: int
|
||||
message: str = ""
|
||||
percentage: float = 0.0
|
||||
elapsed_time: float = 0.0
|
||||
estimated_remaining: float = 0.0
|
||||
|
||||
@dataclass
|
||||
class TaskResult:
|
||||
success: bool
|
||||
result: Any = None
|
||||
error: str = None
|
||||
total_time: float = 0.0
|
||||
```
|
||||
|
||||
#### **2. task.py** - 任务管理
|
||||
```python
|
||||
class ProgressiveTask:
|
||||
"""渐进式任务包装器"""
|
||||
|
||||
def start(self):
|
||||
# 开始任务
|
||||
|
||||
def update(self, step: int = None, message: str = ""):
|
||||
# 更新进度
|
||||
|
||||
def finish(self, message: str = "任务完成"):
|
||||
# 完成任务
|
||||
```
|
||||
|
||||
#### **3. reporter.py** - 进度报告
|
||||
```python
|
||||
class ProgressReporter:
|
||||
"""进度报告器"""
|
||||
|
||||
def report_progress(self, progress: ProgressInfo) -> None:
|
||||
# 报告进度到JSON-RPC或控制台
|
||||
```
|
||||
|
||||
#### **4. generator.py** - 进度生成器
|
||||
```python
|
||||
class ProgressGenerator:
|
||||
"""进度生成器工具类"""
|
||||
|
||||
@staticmethod
|
||||
def for_iterable(iterable, task, description="处理中"):
|
||||
# 为可迭代对象添加进度报告
|
||||
|
||||
@staticmethod
|
||||
def for_range(start, end, task, description="处理中"):
|
||||
# 为范围添加进度报告
|
||||
```
|
||||
|
||||
#### **5. decorators.py** - 装饰器
|
||||
```python
|
||||
def with_progress(total_steps: int = 100, task_name: str = None):
|
||||
"""为函数添加进度报告的装饰器"""
|
||||
```
|
||||
|
||||
#### **6. commander.py** - 进度Commander
|
||||
```python
|
||||
class ProgressJSONRPCCommander(JSONRPCCommander):
|
||||
"""带进度条的JSON-RPC Commander基类"""
|
||||
|
||||
@contextmanager
|
||||
def create_task(self, task_name: str, total_steps: int = 100):
|
||||
# 创建带进度的任务上下文
|
||||
```
|
||||
|
||||
## 🔧 使用方式
|
||||
|
||||
### **1. 基础Commander**
|
||||
```python
|
||||
from python_core.utils.commander import JSONRPCCommander
|
||||
|
||||
class MyCommander(JSONRPCCommander):
|
||||
def _register_commands(self):
|
||||
self.register_command("process", "处理数据", ["input"])
|
||||
|
||||
def execute_command(self, command, args):
|
||||
return {"result": "processed"}
|
||||
```
|
||||
|
||||
### **2. 简单Commander**
|
||||
```python
|
||||
from python_core.utils.commander import create_simple_commander
|
||||
|
||||
commander = create_simple_commander("my_service")
|
||||
commander.add_command("hello", lambda name="World": f"Hello, {name}!", "打招呼")
|
||||
commander.run()
|
||||
```
|
||||
|
||||
### **3. 进度Commander**
|
||||
```python
|
||||
from python_core.utils.progress import ProgressJSONRPCCommander
|
||||
|
||||
class MyProgressCommander(ProgressJSONRPCCommander):
|
||||
def _execute_with_progress(self, command, args):
|
||||
with self.create_task("处理中", 100) as task:
|
||||
for i in range(100):
|
||||
# 处理工作
|
||||
task.update(i, f"处理步骤 {i}")
|
||||
return {"completed": True}
|
||||
```
|
||||
|
||||
### **4. 统一导入**
|
||||
```python
|
||||
# Commander模块
|
||||
from python_core.utils.commander import (
|
||||
JSONRPCCommander, SimpleJSONRPCCommander, create_simple_commander
|
||||
)
|
||||
|
||||
# Progress模块
|
||||
from python_core.utils.progress import (
|
||||
ProgressJSONRPCCommander, ProgressiveTask, with_progress
|
||||
)
|
||||
```
|
||||
|
||||
## 📈 重构对比
|
||||
|
||||
### **重构前的问题**
|
||||
| 文件 | 行数 | 问题 |
|
||||
|------|------|------|
|
||||
| `jsonrpc_commander.py` | 301行 | 功能混杂，难以维护 |
|
||||
| `progress_commander.py` | 350行 | 代码重复，职责不清 |
|
||||
|
||||
### **重构后的优势**
|
||||
| 模块 | 文件数 | 平均行数 | 优势 |
|
||||
|------|--------|----------|------|
|
||||
| `commander/` | 5个 | ~60行 | 职责单一，易于理解 |
|
||||
| `progress/` | 7个 | ~50行 | 模块化，可独立使用 |
|
||||
|
||||
## 🎯 架构优势
|
||||
|
||||
### **1. 单一职责原则**
|
||||
- 每个模块只负责一个特定功能
|
||||
- 降低模块间的耦合度
|
||||
- 提高代码的可读性
|
||||
|
||||
### **2. 开放封闭原则**
|
||||
- 对扩展开放：可以轻松添加新模块
|
||||
- 对修改封闭：修改一个模块不影响其他模块
|
||||
|
||||
### **3. 依赖倒置原则**
|
||||
- 高层模块不依赖低层模块
|
||||
- 通过抽象接口进行交互
|
||||
|
||||
### **4. 接口隔离原则**
|
||||
- 客户端不应该依赖它不需要的接口
|
||||
- 每个模块提供最小化的接口
|
||||
|
||||
## 🚀 扩展性
|
||||
|
||||
### **添加新功能**
|
||||
```python
|
||||
# 1. 在commander/下添加新模块
|
||||
# python_core/utils/commander/advanced.py
|
||||
class AdvancedCommander(JSONRPCCommander):
|
||||
# 高级功能实现
|
||||
|
||||
# 2. 在progress/下添加新功能
|
||||
# python_core/utils/progress/scheduler.py
|
||||
class TaskScheduler:
|
||||
# 任务调度功能
|
||||
|
||||
# 3. 更新__init__.py导入
|
||||
```
|
||||
|
||||
### **独立使用模块**
|
||||
```python
|
||||
# 只使用参数解析器
|
||||
from python_core.utils.commander.parser import ArgumentParser
|
||||
|
||||
# 只使用进度任务
|
||||
from python_core.utils.progress.task import ProgressiveTask
|
||||
|
||||
# 只使用进度装饰器
|
||||
from python_core.utils.progress.decorators import with_progress
|
||||
```
|
||||
|
||||
## 🧪 测试友好
|
||||
|
||||
### **单元测试**
|
||||
```python
|
||||
# 测试单个模块
|
||||
def test_argument_parser():
|
||||
from python_core.utils.commander.parser import ArgumentParser
|
||||
# 只测试解析器功能
|
||||
|
||||
def test_progress_task():
|
||||
from python_core.utils.progress.task import ProgressiveTask
|
||||
# 只测试任务管理功能
|
||||
```
|
||||
|
||||
### **集成测试**
|
||||
```python
|
||||
# 测试模块组合
|
||||
def test_commander_with_progress():
|
||||
from python_core.utils.progress import ProgressJSONRPCCommander
|
||||
# 测试完整功能
|
||||
```
|
||||
|
||||
## 🎉 总结
|
||||
|
||||
### **重构成果**
|
||||
- ✅ **代码行数减少**: 从651行拆分为12个小文件
|
||||
- ✅ **职责明确**: 每个模块功能单一
|
||||
- ✅ **易于维护**: 修改影响范围小
|
||||
- ✅ **可重用性**: 模块可独立使用
|
||||
- ✅ **测试友好**: 可单独测试每个模块
|
||||
|
||||
### **架构原则**
|
||||
- 🎯 **单一职责** - 每个模块只做一件事
|
||||
- 🔧 **开放封闭** - 易于扩展，稳定修改
|
||||
- 📦 **模块化** - 高内聚，低耦合
|
||||
- 🧪 **可测试** - 独立测试，集成验证
|
||||
|
||||
### **使用体验**
|
||||
- 💡 **简单易用** - 清晰的API设计
|
||||
- 🚀 **快速开发** - 便捷的创建函数
|
||||
- 📈 **渐进增强** - 从简单到复杂的使用路径
|
||||
- 🔄 **向后兼容** - 保持原有接口不变
|
||||
|
||||
通过模块化重构，我们不仅提高了代码质量，还为未来的功能扩展奠定了坚实的基础！
|
||||
|
||||
---
|
||||
|
||||
*模块化架构 - 让代码更清晰、更易维护、更具扩展性！*
|
||||
@@ -17,7 +17,7 @@ def test_commander_import():
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
from python_core.utils.jsonrpc_commander import (
|
||||
from python_core.utils.commander import (
|
||||
JSONRPCCommander, SimpleJSONRPCCommander, create_simple_commander
|
||||
)
|
||||
print("✅ JSON-RPC Commander导入成功")
|
||||
@@ -41,7 +41,7 @@ def test_simple_commander():
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
from python_core.utils.jsonrpc_commander import create_simple_commander
|
||||
from python_core.utils.commander import create_simple_commander
|
||||
|
||||
# 创建Commander
|
||||
commander = create_simple_commander("test_service")
|
||||
@@ -204,7 +204,7 @@ def test_jsonrpc_output():
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
from python_core.utils.jsonrpc_commander import create_simple_commander
|
||||
from python_core.utils.commander import create_simple_commander
|
||||
import io
|
||||
import contextlib
|
||||
|
||||
|
||||
387
scripts/test_modular_commander.py
Normal file
387
scripts/test_modular_commander.py
Normal file
@@ -0,0 +1,387 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
测试模块化的Commander
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# 添加项目根目录到Python路径
|
||||
project_root = Path(__file__).parent.parent
|
||||
sys.path.insert(0, str(project_root))
|
||||
|
||||
def test_modular_imports():
|
||||
"""测试模块化导入"""
|
||||
print("🔍 测试模块化导入")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
# 测试基础模块导入
|
||||
from python_core.utils.commander.types import CommandConfig
|
||||
from python_core.utils.commander.parser import ArgumentParser
|
||||
from python_core.utils.commander.base import JSONRPCCommander
|
||||
from python_core.utils.commander.simple import SimpleJSONRPCCommander, create_simple_commander
|
||||
print("✅ 基础模块导入成功")
|
||||
|
||||
# 测试统一导入
|
||||
from python_core.utils.commander import (
|
||||
CommandConfig, ArgumentParser, JSONRPCCommander,
|
||||
SimpleJSONRPCCommander, create_simple_commander
|
||||
)
|
||||
print("✅ 统一导入成功")
|
||||
|
||||
# 测试进度模块导入
|
||||
from python_core.utils.progress import (
|
||||
ProgressInfo, TaskResult, ProgressiveTask,
|
||||
ProgressJSONRPCCommander, create_progress_commander
|
||||
)
|
||||
print("✅ 进度模块导入成功")
|
||||
|
||||
return True
|
||||
|
||||
except ImportError as e:
|
||||
print(f"❌ 导入失败: {e}")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"❌ 测试失败: {e}")
|
||||
return False
|
||||
|
||||
def test_command_config():
|
||||
"""测试命令配置"""
|
||||
print("\n📋 测试命令配置")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
from python_core.utils.commander.types import CommandConfig
|
||||
|
||||
# 创建命令配置
|
||||
config = CommandConfig(
|
||||
name="test_command",
|
||||
description="测试命令",
|
||||
required_args=["input"],
|
||||
optional_args={
|
||||
"output": {"type": str, "default": "output.txt"},
|
||||
"verbose": {"type": bool, "default": False}
|
||||
}
|
||||
)
|
||||
|
||||
print(f"✅ 命令配置创建成功: {config.name}")
|
||||
print(f" 描述: {config.description}")
|
||||
print(f" 必需参数: {config.required_args}")
|
||||
print(f" 可选参数: {list(config.optional_args.keys())}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 命令配置测试失败: {e}")
|
||||
return False
|
||||
|
||||
def test_argument_parser():
|
||||
"""测试参数解析器"""
|
||||
print("\n🔧 测试参数解析器")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
from python_core.utils.commander.types import CommandConfig
|
||||
from python_core.utils.commander.parser import ArgumentParser
|
||||
|
||||
# 创建命令配置
|
||||
commands = {
|
||||
"process": CommandConfig(
|
||||
name="process",
|
||||
description="处理文件",
|
||||
required_args=["input_file"],
|
||||
optional_args={
|
||||
"output": {"type": str, "default": "output.txt"},
|
||||
"format": {"type": str, "default": "json", "choices": ["json", "xml"]},
|
||||
"verbose": {"type": bool, "default": False}
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
# 创建解析器
|
||||
parser = ArgumentParser(commands)
|
||||
|
||||
# 测试解析
|
||||
test_cases = [
|
||||
(["process", "input.txt"], True),
|
||||
(["process", "input.txt", "--output", "result.txt"], True),
|
||||
(["process", "input.txt", "--format", "xml", "--verbose"], True),
|
||||
(["unknown"], False), # 未知命令
|
||||
(["process"], False), # 缺少必需参数
|
||||
]
|
||||
|
||||
for args, should_succeed in test_cases:
|
||||
try:
|
||||
command, parsed_args = parser.parse_arguments(args)
|
||||
if should_succeed:
|
||||
print(f"✅ 解析成功: {args} -> {command}, {parsed_args}")
|
||||
else:
|
||||
print(f"⚠️ 预期失败但成功了: {args}")
|
||||
except ValueError as e:
|
||||
if not should_succeed:
|
||||
print(f"✅ 预期失败: {args} -> {e}")
|
||||
else:
|
||||
print(f"❌ 意外失败: {args} -> {e}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 参数解析器测试失败: {e}")
|
||||
return False
|
||||
|
||||
def test_simple_commander():
|
||||
"""测试简单Commander"""
|
||||
print("\n🚀 测试简单Commander")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
from python_core.utils.commander import create_simple_commander
|
||||
|
||||
# 创建Commander
|
||||
commander = create_simple_commander("test_service")
|
||||
|
||||
# 定义处理函数
|
||||
def hello_handler(name: str = "World"):
|
||||
return {"message": f"Hello, {name}!"}
|
||||
|
||||
def add_handler(a: str, b: str):
|
||||
return {"result": float(a) + float(b)}
|
||||
|
||||
# 添加命令
|
||||
commander.add_command(
|
||||
name="hello",
|
||||
handler=hello_handler,
|
||||
description="打招呼",
|
||||
optional_args={
|
||||
"name": {"type": str, "default": "World"}
|
||||
}
|
||||
)
|
||||
|
||||
commander.add_command(
|
||||
name="add",
|
||||
handler=add_handler,
|
||||
description="加法运算",
|
||||
required_args=["a", "b"]
|
||||
)
|
||||
|
||||
print("✅ 命令注册成功")
|
||||
|
||||
# 测试命令执行
|
||||
test_cases = [
|
||||
(["hello"], {"message": "Hello, World!"}),
|
||||
(["hello", "--name", "Alice"], {"message": "Hello, Alice!"}),
|
||||
(["add", "5", "3"], {"result": 8.0}),
|
||||
]
|
||||
|
||||
for args, expected in test_cases:
|
||||
try:
|
||||
command, parsed_args = commander.parse_arguments(args)
|
||||
result = commander.execute_command(command, parsed_args)
|
||||
|
||||
if result == expected:
|
||||
print(f"✅ 命令测试成功: {args} -> {result}")
|
||||
else:
|
||||
print(f"❌ 结果不匹配: {args} -> 期望 {expected}, 得到 {result}")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 命令执行失败: {args} -> {e}")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 简单Commander测试失败: {e}")
|
||||
return False
|
||||
|
||||
def test_video_splitter_integration():
|
||||
"""测试视频拆分服务集成"""
|
||||
print("\n🎬 测试视频拆分服务集成")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
# 检查依赖
|
||||
try:
|
||||
import scenedetect
|
||||
print(f"✅ PySceneDetect {scenedetect.__version__} 可用")
|
||||
except ImportError:
|
||||
print("⚠️ PySceneDetect不可用，跳过视频拆分集成测试")
|
||||
return True
|
||||
|
||||
from python_core.services.video_splitter.cli import VideoSplitterCommander
|
||||
|
||||
# 创建Commander
|
||||
commander = VideoSplitterCommander()
|
||||
print("✅ 视频拆分Commander创建成功")
|
||||
|
||||
# 检查注册的命令
|
||||
commands = list(commander.commands.keys())
|
||||
expected_commands = ["analyze", "detect_scenes"]
|
||||
|
||||
for cmd in expected_commands:
|
||||
if cmd in commands:
|
||||
print(f"✅ 命令 '{cmd}' 已注册")
|
||||
else:
|
||||
print(f"❌ 命令 '{cmd}' 未注册")
|
||||
return False
|
||||
|
||||
# 测试参数解析
|
||||
test_args = ["analyze", "test.mp4", "--threshold", "30.0"]
|
||||
|
||||
try:
|
||||
command, parsed_args = commander.parse_arguments(test_args)
|
||||
print(f"✅ 参数解析成功: {command}, {parsed_args}")
|
||||
except Exception as e:
|
||||
print(f"❌ 参数解析失败: {e}")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 视频拆分集成测试失败: {e}")
|
||||
return False
|
||||
|
||||
def test_file_structure():
|
||||
"""测试文件结构"""
|
||||
print("\n📁 测试文件结构")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
# 检查Commander模块文件
|
||||
commander_files = [
|
||||
"python_core/utils/commander/__init__.py",
|
||||
"python_core/utils/commander/types.py",
|
||||
"python_core/utils/commander/parser.py",
|
||||
"python_core/utils/commander/base.py",
|
||||
"python_core/utils/commander/simple.py"
|
||||
]
|
||||
|
||||
for file_path in commander_files:
|
||||
full_path = project_root / file_path
|
||||
if full_path.exists():
|
||||
print(f"✅ {file_path} 存在")
|
||||
else:
|
||||
print(f"❌ {file_path} 不存在")
|
||||
return False
|
||||
|
||||
# 检查Progress模块文件
|
||||
progress_files = [
|
||||
"python_core/utils/progress/__init__.py",
|
||||
"python_core/utils/progress/types.py",
|
||||
"python_core/utils/progress/task.py",
|
||||
"python_core/utils/progress/reporter.py",
|
||||
"python_core/utils/progress/generator.py",
|
||||
"python_core/utils/progress/decorators.py",
|
||||
"python_core/utils/progress/commander.py"
|
||||
]
|
||||
|
||||
for file_path in progress_files:
|
||||
full_path = project_root / file_path
|
||||
if full_path.exists():
|
||||
print(f"✅ {file_path} 存在")
|
||||
else:
|
||||
print(f"❌ {file_path} 不存在")
|
||||
return False
|
||||
|
||||
# 检查原文件是否已删除
|
||||
old_files = [
|
||||
"python_core/utils/jsonrpc_commander.py",
|
||||
"python_core/utils/progress_commander.py"
|
||||
]
|
||||
|
||||
for file_path in old_files:
|
||||
full_path = project_root / file_path
|
||||
if not full_path.exists():
|
||||
print(f"✅ {file_path} 已删除")
|
||||
else:
|
||||
print(f"⚠️ {file_path} 仍然存在")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 文件结构测试失败: {e}")
|
||||
return False
|
||||
|
||||
def main():
|
||||
"""主函数"""
|
||||
print("🚀 测试模块化Commander")
|
||||
|
||||
try:
|
||||
# 运行所有测试
|
||||
tests = [
|
||||
test_modular_imports,
|
||||
test_command_config,
|
||||
test_argument_parser,
|
||||
test_simple_commander,
|
||||
test_video_splitter_integration,
|
||||
test_file_structure
|
||||
]
|
||||
|
||||
results = []
|
||||
for test in tests:
|
||||
try:
|
||||
result = test()
|
||||
results.append(result)
|
||||
except Exception as e:
|
||||
print(f"❌ 测试 {test.__name__} 异常: {e}")
|
||||
results.append(False)
|
||||
|
||||
# 总结
|
||||
print("\n" + "=" * 60)
|
||||
print("📊 模块化Commander测试总结")
|
||||
print("=" * 60)
|
||||
|
||||
passed = sum(results)
|
||||
total = len(results)
|
||||
|
||||
print(f"通过测试: {passed}/{total}")
|
||||
|
||||
if passed == total:
|
||||
print("🎉 所有模块化测试通过！")
|
||||
print("\n✅ 模块化优势:")
|
||||
print(" 1. 代码组织 - 每个模块职责单一明确")
|
||||
print(" 2. 易于维护 - 修改某个功能只需改对应模块")
|
||||
print(" 3. 可重用性 - 模块可以独立使用")
|
||||
print(" 4. 测试友好 - 可以单独测试每个模块")
|
||||
print(" 5. 扩展性强 - 新功能可以独立添加")
|
||||
|
||||
print("\n📁 模块结构:")
|
||||
print(" commander/")
|
||||
print(" ├── types.py - 数据类型定义")
|
||||
print(" ├── parser.py - 参数解析器")
|
||||
print(" ├── base.py - 基础Commander类")
|
||||
print(" ├── simple.py - 简化Commander类")
|
||||
print(" └── __init__.py - 统一导入接口")
|
||||
print(" ")
|
||||
print(" progress/")
|
||||
print(" ├── types.py - 进度相关类型")
|
||||
print(" ├── task.py - 任务管理")
|
||||
print(" ├── reporter.py - 进度报告")
|
||||
print(" ├── generator.py - 进度生成器")
|
||||
print(" ├── decorators.py - 装饰器")
|
||||
print(" ├── commander.py - 进度Commander")
|
||||
print(" └── __init__.py - 统一导入接口")
|
||||
|
||||
print("\n🎯 使用方式:")
|
||||
print(" # 基础Commander")
|
||||
print(" from python_core.utils.commander import JSONRPCCommander")
|
||||
print(" # 简单Commander")
|
||||
print(" from python_core.utils.commander import create_simple_commander")
|
||||
print(" # 进度Commander")
|
||||
print(" from python_core.utils.progress import ProgressJSONRPCCommander")
|
||||
|
||||
return 0
|
||||
else:
|
||||
print("⚠️ 部分模块化测试失败")
|
||||
return 1
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 测试过程中出错: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return 1
|
||||
|
||||
if __name__ == "__main__":
|
||||
exit_code = main()
|
||||
sys.exit(exit_code)
|
||||
Reference in New Issue
Block a user