mxivideo/docs/modular-commander-architecture.md

模块化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 - 数据类型定义

@dataclass
class CommandConfig:
    """命令配置"""
    name: str
    description: str
    required_args: List[str]
    optional_args: Dict[str, Dict[str, Any]]

2. parser.py - 参数解析器

class ArgumentParser:
    """命令行参数解析器"""
    
    def parse_arguments(self, args: List[str]) -> Tuple[str, Dict[str, Any]]:
        # 解析命令行参数，支持类型转换和验证

3. base.py - 基础Commander类

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类

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 - 进度相关类型

@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 - 任务管理

class ProgressiveTask:
    """渐进式任务包装器"""
    
    def start(self):
        # 开始任务
    
    def update(self, step: int = None, message: str = ""):
        # 更新进度
    
    def finish(self, message: str = "任务完成"):
        # 完成任务

3. reporter.py - 进度报告

class ProgressReporter:
    """进度报告器"""
    
    def report_progress(self, progress: ProgressInfo) -> None:
        # 报告进度到JSON-RPC或控制台

4. generator.py - 进度生成器

class ProgressGenerator:
    """进度生成器工具类"""
    
    @staticmethod
    def for_iterable(iterable, task, description="处理中"):
        # 为可迭代对象添加进度报告
    
    @staticmethod
    def for_range(start, end, task, description="处理中"):
        # 为范围添加进度报告

5. decorators.py - 装饰器

def with_progress(total_steps: int = 100, task_name: str = None):
    """为函数添加进度报告的装饰器"""

6. commander.py - 进度Commander

class ProgressJSONRPCCommander(JSONRPCCommander):
    """带进度条的JSON-RPC Commander基类"""
    
    @contextmanager
    def create_task(self, task_name: str, total_steps: int = 100):
        # 创建带进度的任务上下文

🔧 使用方式

1. 基础Commander

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

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

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. 统一导入

# 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. 接口隔离原则

• 客户端不应该依赖它不需要的接口
• 每个模块提供最小化的接口

🚀 扩展性

添加新功能

# 1. 在commander/下添加新模块
# python_core/utils/commander/advanced.py
class AdvancedCommander(JSONRPCCommander):
    # 高级功能实现

# 2. 在progress/下添加新功能
# python_core/utils/progress/scheduler.py
class TaskScheduler:
    # 任务调度功能

# 3. 更新__init__.py导入

独立使用模块

# 只使用参数解析器
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

🧪 测试友好

单元测试

# 测试单个模块
def test_argument_parser():
    from python_core.utils.commander.parser import ArgumentParser
    # 只测试解析器功能

def test_progress_task():
    from python_core.utils.progress.task import ProgressiveTask
    # 只测试任务管理功能

集成测试

# 测试模块组合
def test_commander_with_progress():
    from python_core.utils.progress import ProgressJSONRPCCommander
    # 测试完整功能

🎉 总结

重构成果

• ✅ 代码行数减少: 从651行拆分为12个小文件
• ✅ 职责明确: 每个模块功能单一
• ✅ 易于维护: 修改影响范围小
• ✅ 可重用性: 模块可独立使用
• ✅ 测试友好: 可单独测试每个模块

架构原则

• 🎯 单一职责 - 每个模块只做一件事
• 🔧 开放封闭 - 易于扩展，稳定修改
• 📦 模块化 - 高内聚，低耦合
• 🧪 可测试 - 独立测试，集成验证

使用体验

• 💡 简单易用 - 清晰的API设计
• 🚀 快速开发 - 便捷的创建函数
• 📈 渐进增强 - 从简单到复杂的使用路径
• 🔄 向后兼容 - 保持原有接口不变

通过模块化重构，我们不仅提高了代码质量，还为未来的功能扩展奠定了坚实的基础！

---

模块化架构 - 让代码更清晰、更易维护、更具扩展性！