mxivideo/docs/typer-development-standards.md

Typer 开发规范

🎯 核心原则

基于您选择的方案2: 纯Typer实现，制定以下开发规范，确保与现有JSON-RPC进度条架构完美集成。

📋 开发规范总览

1. 项目结构规范

python_core/services/{service_name}/
├── __init__.py          # 统一导入
├── types.py             # 数据类型定义（dataclass + enum）
├── service.py           # 业务逻辑服务
├── cli.py               # Typer命令行接口
└── __main__.py          # 模块入口

2. 命名规范

• 服务名: snake_case (如: media_manager, scene_detection)
• 类名: PascalCase (如: MediaManagerCommander, SceneDetectionService)
• 函数名: snake_case (如: upload_video, batch_detect)
• 常量: UPPER_SNAKE_CASE (如: DEFAULT_THRESHOLD, MAX_FILE_SIZE)

🔧 Typer CLI 开发规范

1. 基础结构模板

#!/usr/bin/env python3
"""
{服务名}命令行接口
"""

import typer
from typing import Optional, List
from pathlib import Path
from enum import Enum
from rich.console import Console

from .types import {相关类型}
from .service import {服务类}
from python_core.utils.progress import ProgressJSONRPCCommander

console = Console()

class {服务名}Commander(ProgressJSONRPCCommander):
    """基于Typer的{服务名}命令行接口"""
    
    def __init__(self):
        super().__init__("{service_name}")
        self.app = typer.Typer(
            name="{service_name}",
            help="🎯 {服务描述}",
            rich_markup_mode="rich",
            no_args_is_help=True
        )
        self.service = {服务类}()
        self._setup_commands()
    
    def _register_commands(self):
        """注册命令（继承要求）"""
        pass  # Typer通过装饰器自动注册
    
    def _is_progressive_command(self, command: str) -> bool:
        """判断是否需要进度报告"""
        return command in ["batch_*", "compare_*", "analyze_*"]
    
    def _execute_with_progress(self, command: str, args: dict):
        """执行带进度的命令"""
        # 通过Typer命令直接处理
        pass
    
    def _execute_simple_command(self, command: str, args: dict):
        """执行简单命令"""
        # 通过Typer命令直接处理
        pass
    
    def _setup_commands(self):
        """设置Typer命令"""
        # 在这里定义所有命令
        pass
    
    def run(self):
        """运行CLI"""
        self.app()

def main():
    """主入口函数"""
    commander = {服务名}Commander()
    commander.run()

if __name__ == "__main__":
    main()

2. 类型定义规范

枚举类型

from enum import Enum

class OutputFormat(str, Enum):
    """输出格式枚举"""
    JSON = "json"
    CSV = "csv"
    TXT = "txt"
    
    def __str__(self) -> str:
        return self.value

class ProcessingMode(str, Enum):
    """处理模式枚举"""
    FAST = "fast"
    BALANCED = "balanced"
    QUALITY = "quality"

数据类型

from dataclasses import dataclass
from typing import Optional, List
from pathlib import Path

@dataclass
class ProcessConfig:
    """处理配置"""
    input_path: Path
    output_path: Optional[Path] = None
    format: OutputFormat = OutputFormat.JSON
    mode: ProcessingMode = ProcessingMode.BALANCED
    tags: List[str] = None
    
    def __post_init__(self):
        if self.tags is None:
            self.tags = []

3. 命令定义规范

基础命令模板

@self.app.command()
def {command_name}(
    # 必需参数 - 使用Argument
    input_path: Path = typer.Argument(
        ..., 
        help="📁 输入文件/目录路径",
        exists=True
    ),
    
    # 可选参数 - 使用Option
    output_format: OutputFormat = typer.Option(
        OutputFormat.JSON,
        "--format", "-f",
        help="📄 输出格式"
    ),
    
    tags: Optional[str] = typer.Option(
        None,
        "--tags", "-t", 
        help="🏷️ 标签列表（逗号分隔）"
    ),
    
    # 布尔选项
    verbose: bool = typer.Option(
        False,
        "--verbose", "-v",
        help="📝 详细输出"
    ),
    
    # 数值选项
    threshold: float = typer.Option(
        30.0,
        "--threshold",
        min=0.0,
        max=100.0,
        help="🎚️ 处理阈值 (0-100)"
    )
):
    """
    📋 命令描述
    
    详细说明命令的功能和用法。
    """
    # 参数验证
    if not input_path.exists():
        console.print("❌ [red]输入路径不存在[/red]")
        raise typer.Exit(1)
    
    # 解析标签
    tag_list = []
    if tags:
        tag_list = [tag.strip() for tag in tags.split(",")]
    
    # 显示开始信息
    console.print(f"🚀 开始处理: [bold blue]{input_path}[/bold blue]")
    
    try:
        # 调用服务逻辑
        result = self.service.process(
            input_path=input_path,
            output_format=output_format,
            tags=tag_list,
            verbose=verbose,
            threshold=threshold
        )
        
        # 显示结果
        console.print("✅ [bold green]处理完成[/bold green]")
        if verbose:
            console.print(f"📊 结果: {result}")
            
    except Exception as e:
        console.print(f"❌ [red]处理失败: {e}[/red]")
        raise typer.Exit(1)

进度命令模板

@self.app.command()
def batch_process(
    input_directory: Path = typer.Argument(..., help="📁 输入目录"),
    output_path: Optional[Path] = typer.Option(None, help="📄 输出文件路径"),
    recursive: bool = typer.Option(False, "--recursive", "-r", help="🔄 递归处理子目录")
):
    """
    📦 批量处理命令（带进度条）
    """
    console.print(f"📦 批量处理目录: [bold blue]{input_directory}[/bold blue]")
    
    # 扫描文件
    files = self._scan_files(input_directory, recursive)
    
    if not files:
        console.print("⚠️ [yellow]未找到可处理的文件[/yellow]")
        return
    
    # 使用进度任务
    with self.create_task("批量处理", len(files)) as task:
        results = []
        
        for i, file_path in enumerate(files):
            task.update(i, f"处理文件: {file_path.name} ({i+1}/{len(files)})")
            
            try:
                result = self.service.process_single(file_path)
                results.append(result)
            except Exception as e:
                console.print(f"❌ 处理失败 {file_path.name}: {e}")
        
        task.finish(f"批量处理完成: {len(results)}/{len(files)} 成功")
    
    # 保存结果
    if output_path:
        self._save_results(results, output_path)
        console.print(f"📄 结果已保存到: {output_path}")

4. 错误处理规范

# 输入验证
def validate_input(self, input_path: Path) -> None:
    """验证输入参数"""
    if not input_path.exists():
        console.print(f"❌ [red]文件不存在: {input_path}[/red]")
        raise typer.Exit(1)
    
    if input_path.is_dir() and not any(input_path.iterdir()):
        console.print(f"⚠️ [yellow]目录为空: {input_path}[/yellow]")
        raise typer.Exit(1)

# 异常处理
try:
    result = self.service.process(input_path)
except FileNotFoundError:
    console.print("❌ [red]文件未找到[/red]")
    raise typer.Exit(1)
except PermissionError:
    console.print("❌ [red]权限不足[/red]")
    raise typer.Exit(1)
except Exception as e:
    console.print(f"❌ [red]处理失败: {e}[/red]")
    if verbose:
        console.print_exception()
    raise typer.Exit(1)

5. 输出规范

Rich输出样式

from rich.table import Table
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, TextColumn

# 状态表格
def show_status(self, data: dict):
    """显示状态表格"""
    table = Table(title="📊 处理状态")
    table.add_column("项目", style="cyan")
    table.add_column("状态", style="green")
    table.add_column("详情", style="yellow")
    
    for key, value in data.items():
        table.add_row(key, "✅ 完成", str(value))
    
    console.print(table)

# 信息面板
def show_summary(self, message: str, title: str = "📋 摘要"):
    """显示信息面板"""
    panel = Panel(
        f"[bold blue]{message}[/bold blue]",
        title=title,
        border_style="green"
    )
    console.print(panel)

# 进度条（非JSON-RPC场景）
def show_local_progress(self, items: list, description: str):
    """显示本地进度条"""
    with Progress(
        SpinnerColumn(),
        TextColumn("[progress.description]{task.description}"),
        console=console
    ) as progress:
        task = progress.add_task(description, total=len(items))
        
        for item in items:
            # 处理逻辑
            progress.advance(task)

📦 服务集成规范

1. 服务基类继承

from python_core.services.base import ProgressServiceBase

class MediaManagerService(ProgressServiceBase):
    """媒体管理服务"""
    
    def get_service_name(self) -> str:
        return "media_manager"
    
    def process_with_progress(self, items: list, operation_name: str):
        """带进度的处理"""
        self.report_progress(f"开始{operation_name}")
        
        for i, item in enumerate(items):
            self.report_progress(f"{operation_name}: {item} ({i+1}/{len(items)})")
            # 处理逻辑
        
        self.report_progress(f"完成{operation_name}")

2. 存储集成

def save_results(self, results: list, collection_type: str = "results"):
    """保存结果到存储"""
    for i, result in enumerate(results):
        key = f"result_{i}_{int(time.time())}"
        self.save_data(collection_type, key, result)
    
    return len(results)

def load_recent_results(self, collection_type: str = "results", limit: int = 10):
    """加载最近的结果"""
    keys = self.list_keys(collection_type)
    recent_keys = sorted(keys)[-limit:]
    
    return self.load_batch_data(collection_type, recent_keys)

🧪 测试规范

1. 命令测试

import pytest
from typer.testing import CliRunner

def test_upload_command():
    """测试上传命令"""
    runner = CliRunner()
    result = runner.invoke(app, ["upload", "test.mp4", "--tags", "test"])
    
    assert result.exit_code == 0
    assert "上传完成" in result.stdout

def test_batch_command():
    """测试批量命令"""
    runner = CliRunner()
    result = runner.invoke(app, ["batch-upload", "/test/dir"])
    
    assert result.exit_code == 0
    assert "批量处理完成" in result.stdout

2. 进度测试

def test_progress_integration():
    """测试进度集成"""
    commander = MediaManagerCommander()
    
    # 模拟进度回调
    progress_messages = []
    def mock_callback(message):
        progress_messages.append(message)
    
    commander.set_progress_callback(mock_callback)
    
    # 执行带进度的操作
    commander.service.process_with_progress(["item1", "item2"], "测试")
    
    assert len(progress_messages) > 0
    assert "开始测试" in progress_messages[0]

📚 文档规范

1. 命令文档

@app.command()
def upload(
    video_path: Path = typer.Argument(..., help="📹 视频文件路径")
):
    """
    📤 上传视频文件
    
    上传单个视频文件到媒体库，自动进行场景检测和元数据提取。
    
    示例:
        python -m media_manager upload video.mp4 --tags "demo,test"
        python -m media_manager upload /path/to/video.mp4 --format json
    
    注意:
        - 支持的格式: MP4, AVI, MOV, MKV
        - 文件大小限制: 2GB
        - 自动检测重复文件
    """

2. 帮助信息

# 应用级帮助
app = typer.Typer(
    name="media_manager",
    help="""
    🎬 媒体管理器
    
    功能强大的视频处理和管理工具，支持：
    • 视频上传和元数据提取
    • 自动场景检测和分割
    • 批量处理和进度跟踪
    • 多种输出格式
    
    使用 --help 查看具体命令帮助。
    """,
    rich_markup_mode="rich"
)

🎯 最佳实践

1. 性能优化

• 使用类型提示提高IDE支持
• 合理使用Rich组件，避免过度渲染
• 大文件处理时显示进度条
• 异步操作使用适当的进度反馈

2. 用户体验

• 提供清晰的错误信息
• 使用emoji和颜色增强可读性
• 重要操作前进行确认
• 提供详细的帮助文档

3. 代码质量

• 所有函数添加类型提示
• 使用dataclass定义数据结构
• 遵循PEP 8代码风格
• 编写完整的单元测试

这套规范确保了Typer实现与您现有架构的完美集成，同时提供了现代化、类型安全的开发体验！