560 lines
16 KiB
Markdown
560 lines
16 KiB
Markdown
# 多命令集成设计方案
|
||
|
||
## 🎯 设计目标
|
||
|
||
将所有服务的命令集成到一个统一的CLI中，提供一致的用户体验和管理界面。
|
||
|
||
## 🏗️ 架构设计
|
||
|
||
### **方案1: 主CLI + 子命令组**
|
||
```
|
||
mixvideo
|
||
├── media # 媒体管理命令组
|
||
│ ├── upload
|
||
│ ├── batch-upload
|
||
│ ├── list
|
||
│ └── status
|
||
├── scene # 场景检测命令组
|
||
│ ├── detect
|
||
│ ├── batch-detect
|
||
│ ├── compare
|
||
│ └── analyze
|
||
├── template # 模板管理命令组
|
||
│ ├── import
|
||
│ ├── export
|
||
│ ├── list
|
||
│ └── validate
|
||
└── system # 系统管理命令组
|
||
├── status
|
||
├── config
|
||
├── storage
|
||
└── logs
|
||
```
|
||
|
||
### **方案2: 扁平化命令**
|
||
```
|
||
mixvideo
|
||
├── media-upload
|
||
├── media-batch-upload
|
||
├── scene-detect
|
||
├── scene-batch-detect
|
||
├── template-import
|
||
├── template-export
|
||
├── system-status
|
||
└── system-config
|
||
```
|
||
|
||
### **方案3: 混合方案（推荐）**
|
||
```
|
||
mixvideo
|
||
├── upload # 常用命令提升到顶层
|
||
├── batch-upload
|
||
├── detect
|
||
├── batch-detect
|
||
├── media # 完整的媒体管理命令组
|
||
├── scene # 完整的场景检测命令组
|
||
├── template # 模板管理命令组
|
||
└── system # 系统管理命令组
|
||
```
|
||
|
||
## 📁 文件结构设计
|
||
|
||
```
|
||
python_core/
|
||
├── cli/ # 统一CLI模块
|
||
│ ├── __init__.py
|
||
│ ├── main.py # 主CLI入口
|
||
│ ├── commands/ # 命令组织
|
||
│ │ ├── __init__.py
|
||
│ │ ├── media.py # 媒体管理命令组
|
||
│ │ ├── scene.py # 场景检测命令组
|
||
│ │ ├── template.py # 模板管理命令组
|
||
│ │ └── system.py # 系统管理命令组
|
||
│ └── utils/ # CLI工具
|
||
│ ├── __init__.py
|
||
│ ├── common.py # 通用CLI工具
|
||
│ └── decorators.py # CLI装饰器
|
||
├── services/ # 现有服务保持不变
|
||
└── __main__.py # 项目入口
|
||
```
|
||
|
||
## 🔧 实现方案
|
||
|
||
### **1. 主CLI框架**
|
||
```python
|
||
# python_core/cli/main.py
|
||
import typer
|
||
from rich.console import Console
|
||
from typing import Optional
|
||
|
||
from .commands import media, scene, template, system
|
||
from python_core.utils.progress import ProgressJSONRPCCommander
|
||
|
||
console = Console()
|
||
|
||
class MixVideoCommander(ProgressJSONRPCCommander):
|
||
"""MixVideo 统一命令行接口"""
|
||
|
||
def __init__(self):
|
||
super().__init__("mixvideo")
|
||
self.app = typer.Typer(
|
||
name="mixvideo",
|
||
help="""
|
||
🎬 MixVideo - 智能视频处理平台
|
||
|
||
功能完整的视频处理和管理工具套件：
|
||
• 📤 媒体管理 - 上传、处理、组织视频文件
|
||
• 🎯 场景检测 - 智能识别视频场景变化
|
||
• 📋 模板管理 - 视频模板导入导出
|
||
• ⚙️ 系统管理 - 配置、状态、存储管理
|
||
|
||
快速开始:
|
||
mixvideo upload video.mp4 # 上传视频
|
||
mixvideo detect video.mp4 # 检测场景
|
||
mixvideo system status # 查看状态
|
||
""",
|
||
rich_markup_mode="rich",
|
||
no_args_is_help=True
|
||
)
|
||
self._setup_commands()
|
||
|
||
def _register_commands(self):
|
||
"""注册命令（继承要求）"""
|
||
pass
|
||
|
||
def _is_progressive_command(self, command: str) -> bool:
|
||
"""判断是否需要进度报告"""
|
||
return command.startswith(("batch-", "import-", "export-", "analyze-"))
|
||
|
||
def _execute_with_progress(self, command: str, args: dict):
|
||
"""执行带进度的命令"""
|
||
pass
|
||
|
||
def _execute_simple_command(self, command: str, args: dict):
|
||
"""执行简单命令"""
|
||
pass
|
||
|
||
def _setup_commands(self):
|
||
"""设置命令结构"""
|
||
# 添加子命令组
|
||
self.app.add_typer(media.app, name="media", help="📤 媒体管理")
|
||
self.app.add_typer(scene.app, name="scene", help="🎯 场景检测")
|
||
self.app.add_typer(template.app, name="template", help="📋 模板管理")
|
||
self.app.add_typer(system.app, name="system", help="⚙️ 系统管理")
|
||
|
||
# 添加常用命令到顶层（快捷方式）
|
||
self._add_shortcuts()
|
||
|
||
def _add_shortcuts(self):
|
||
"""添加常用命令的快捷方式"""
|
||
|
||
@self.app.command()
|
||
def upload(
|
||
video_path: Path = typer.Argument(..., help="📹 视频文件路径"),
|
||
tags: Optional[str] = typer.Option(None, "--tags", "-t", help="🏷️ 标签")
|
||
):
|
||
"""📤 快速上传视频（media upload 的快捷方式）"""
|
||
# 调用媒体管理的上传命令
|
||
return media.upload_video(video_path, tags)
|
||
|
||
@self.app.command()
|
||
def detect(
|
||
video_path: Path = typer.Argument(..., help="📹 视频文件路径"),
|
||
threshold: float = typer.Option(30.0, help="🎚️ 检测阈值")
|
||
):
|
||
"""🎯 快速场景检测（scene detect 的快捷方式）"""
|
||
# 调用场景检测的检测命令
|
||
return scene.detect_scenes(video_path, threshold)
|
||
|
||
@self.app.command()
|
||
def status():
|
||
"""📊 快速状态查看（system status 的快捷方式）"""
|
||
# 调用系统状态命令
|
||
return system.show_status()
|
||
|
||
def run(self):
|
||
"""运行CLI"""
|
||
self.app()
|
||
|
||
def main():
|
||
"""主入口函数"""
|
||
commander = MixVideoCommander()
|
||
commander.run()
|
||
```
|
||
|
||
### **2. 媒体管理命令组**
|
||
```python
|
||
# python_core/cli/commands/media.py
|
||
import typer
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
from rich.console import Console
|
||
|
||
from python_core.services.media_manager import MediaManagerService
|
||
from ..utils.common import create_progress_task, parse_tags
|
||
|
||
console = Console()
|
||
app = typer.Typer(help="📤 媒体管理命令")
|
||
|
||
# 全局服务实例
|
||
media_service = MediaManagerService()
|
||
|
||
@app.command()
|
||
def upload(
|
||
video_path: Path = typer.Argument(..., help="📹 视频文件路径", exists=True),
|
||
tags: Optional[str] = typer.Option(None, "--tags", "-t", help="🏷️ 标签列表"),
|
||
filename: Optional[str] = typer.Option(None, "--filename", "-f", help="📝 自定义文件名")
|
||
):
|
||
"""📤 上传单个视频文件"""
|
||
return upload_video(video_path, tags, filename)
|
||
|
||
@app.command()
|
||
def batch_upload(
|
||
directory: Path = typer.Argument(..., help="📁 视频目录", exists=True),
|
||
tags: Optional[str] = typer.Option(None, "--tags", "-t", help="🏷️ 标签列表"),
|
||
recursive: bool = typer.Option(False, "--recursive", "-r", help="🔄 递归扫描")
|
||
):
|
||
"""📦 批量上传视频文件"""
|
||
tag_list = parse_tags(tags)
|
||
|
||
with create_progress_task("批量上传") as task:
|
||
result = media_service.batch_upload(directory, tag_list, recursive)
|
||
task.finish(f"上传完成: {result['successful']}/{result['total']} 成功")
|
||
|
||
return result
|
||
|
||
@app.command()
|
||
def list_videos(
|
||
limit: int = typer.Option(10, "--limit", "-l", help="📊 显示数量"),
|
||
tags: Optional[str] = typer.Option(None, "--tags", "-t", help="🏷️ 标签过滤")
|
||
):
|
||
"""📋 列出视频文件"""
|
||
# 实现列表逻辑
|
||
pass
|
||
|
||
@app.command()
|
||
def delete(
|
||
video_id: str = typer.Argument(..., help="🗑️ 视频ID"),
|
||
confirm: bool = typer.Option(False, "--yes", "-y", help="✅ 确认删除")
|
||
):
|
||
"""🗑️ 删除视频文件"""
|
||
if not confirm:
|
||
confirm = typer.confirm("确定要删除这个视频吗？")
|
||
|
||
if confirm:
|
||
# 实现删除逻辑
|
||
console.print("✅ 视频已删除")
|
||
else:
|
||
console.print("❌ 取消删除")
|
||
|
||
# 共享函数
|
||
def upload_video(video_path: Path, tags: str = None, filename: str = None):
|
||
"""上传视频的共享实现"""
|
||
tag_list = parse_tags(tags)
|
||
|
||
console.print(f"🚀 开始上传: [bold blue]{video_path}[/bold blue]")
|
||
|
||
result = media_service.upload_video(video_path, tag_list, filename)
|
||
|
||
console.print("✅ [bold green]上传完成[/bold green]")
|
||
return result
|
||
```
|
||
|
||
### **3. 场景检测命令组**
|
||
```python
|
||
# python_core/cli/commands/scene.py
|
||
import typer
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
from rich.console import Console
|
||
|
||
from python_core.services.scene_detection import SceneDetectionService
|
||
from ..utils.common import create_progress_task
|
||
|
||
console = Console()
|
||
app = typer.Typer(help="🎯 场景检测命令")
|
||
|
||
# 全局服务实例
|
||
scene_service = SceneDetectionService()
|
||
|
||
@app.command()
|
||
def detect(
|
||
video_path: Path = typer.Argument(..., help="📹 视频文件路径", exists=True),
|
||
threshold: float = typer.Option(30.0, help="🎚️ 检测阈值"),
|
||
output: Optional[Path] = typer.Option(None, "--output", "-o", help="📄 输出文件")
|
||
):
|
||
"""🎯 检测单个视频的场景"""
|
||
return detect_scenes(video_path, threshold, output)
|
||
|
||
@app.command()
|
||
def batch_detect(
|
||
directory: Path = typer.Argument(..., help="📁 视频目录", exists=True),
|
||
threshold: float = typer.Option(30.0, help="🎚️ 检测阈值"),
|
||
output: Optional[Path] = typer.Option(None, "--output", "-o", help="📄 输出文件")
|
||
):
|
||
"""📦 批量检测场景"""
|
||
with create_progress_task("批量检测") as task:
|
||
result = scene_service.batch_detect(directory, threshold)
|
||
task.finish(f"检测完成: {result['processed']} 个文件")
|
||
|
||
return result
|
||
|
||
@app.command()
|
||
def compare(
|
||
video_path: Path = typer.Argument(..., help="📹 视频文件路径", exists=True),
|
||
thresholds: str = typer.Option("20,30,40", help="🎚️ 阈值列表")
|
||
):
|
||
"""🔬 比较不同检测器效果"""
|
||
with create_progress_task("检测器比较") as task:
|
||
result = scene_service.compare_detectors(video_path, thresholds)
|
||
task.finish("比较完成")
|
||
|
||
return result
|
||
|
||
# 共享函数
|
||
def detect_scenes(video_path: Path, threshold: float = 30.0, output: Path = None):
|
||
"""场景检测的共享实现"""
|
||
console.print(f"🎯 开始检测: [bold blue]{video_path}[/bold blue]")
|
||
|
||
result = scene_service.detect_single(video_path, threshold)
|
||
|
||
if output:
|
||
scene_service.save_result(result, output)
|
||
console.print(f"📄 结果已保存: {output}")
|
||
|
||
console.print("✅ [bold green]检测完成[/bold green]")
|
||
return result
|
||
```
|
||
|
||
### **4. 系统管理命令组**
|
||
```python
|
||
# python_core/cli/commands/system.py
|
||
import typer
|
||
from rich.console import Console
|
||
from rich.table import Table
|
||
from rich.panel import Panel
|
||
|
||
from python_core.storage import get_storage
|
||
from python_core.config import settings
|
||
|
||
console = Console()
|
||
app = typer.Typer(help="⚙️ 系统管理命令")
|
||
|
||
@app.command()
|
||
def status():
|
||
"""📊 显示系统状态"""
|
||
return show_status()
|
||
|
||
@app.command()
|
||
def config(
|
||
show: bool = typer.Option(False, "--show", help="📋 显示配置"),
|
||
set_key: Optional[str] = typer.Option(None, "--set", help="⚙️ 设置配置项"),
|
||
value: Optional[str] = typer.Option(None, "--value", help="💾 配置值")
|
||
):
|
||
"""⚙️ 配置管理"""
|
||
if show:
|
||
show_config()
|
||
elif set_key and value:
|
||
set_config(set_key, value)
|
||
else:
|
||
console.print("❌ 请指定 --show 或 --set 和 --value")
|
||
|
||
@app.command()
|
||
def storage(
|
||
action: str = typer.Argument(..., help="📦 操作类型 (info/clean/migrate)"),
|
||
target: Optional[str] = typer.Option(None, help="🎯 目标存储类型")
|
||
):
|
||
"""📦 存储管理"""
|
||
if action == "info":
|
||
show_storage_info()
|
||
elif action == "clean":
|
||
clean_storage()
|
||
elif action == "migrate":
|
||
migrate_storage(target)
|
||
else:
|
||
console.print("❌ 未知操作类型")
|
||
|
||
@app.command()
|
||
def logs(
|
||
lines: int = typer.Option(50, "--lines", "-n", help="📄 显示行数"),
|
||
follow: bool = typer.Option(False, "--follow", "-f", help="🔄 实时跟踪")
|
||
):
|
||
"""📄 查看日志"""
|
||
# 实现日志查看逻辑
|
||
pass
|
||
|
||
# 共享函数
|
||
def show_status():
|
||
"""显示系统状态的共享实现"""
|
||
console.print("📊 MixVideo 系统状态")
|
||
|
||
# 创建状态表格
|
||
table = Table(title="系统组件状态")
|
||
table.add_column("组件", style="cyan")
|
||
table.add_column("状态", style="green")
|
||
table.add_column("详情", style="yellow")
|
||
|
||
# 检查各组件状态
|
||
storage = get_storage()
|
||
collections = storage.get_collections()
|
||
|
||
table.add_row("存储系统", "✅ 正常", f"{len(collections)} 个集合")
|
||
table.add_row("媒体管理", "✅ 就绪", "服务正常")
|
||
table.add_row("场景检测", "✅ 就绪", "服务正常")
|
||
table.add_row("模板管理", "✅ 就绪", "服务正常")
|
||
|
||
console.print(table)
|
||
|
||
# 系统信息面板
|
||
info_panel = Panel(
|
||
"[bold blue]MixVideo 运行正常[/bold blue]\n"
|
||
"所有核心组件状态良好，系统准备就绪。",
|
||
title="📊 系统摘要",
|
||
border_style="green"
|
||
)
|
||
console.print(info_panel)
|
||
|
||
def show_config():
|
||
"""显示配置信息"""
|
||
# 实现配置显示逻辑
|
||
pass
|
||
|
||
def set_config(key: str, value: str):
|
||
"""设置配置项"""
|
||
# 实现配置设置逻辑
|
||
pass
|
||
```
|
||
|
||
### **5. 通用工具**
|
||
```python
|
||
# python_core/cli/utils/common.py
|
||
from contextlib import contextmanager
|
||
from typing import List, Optional
|
||
from rich.console import Console
|
||
|
||
console = Console()
|
||
|
||
def parse_tags(tags_str: Optional[str]) -> List[str]:
|
||
"""解析标签字符串"""
|
||
if not tags_str:
|
||
return []
|
||
return [tag.strip() for tag in tags_str.split(",") if tag.strip()]
|
||
|
||
@contextmanager
|
||
def create_progress_task(name: str):
|
||
"""创建进度任务的上下文管理器"""
|
||
class MockTask:
|
||
def __init__(self, name):
|
||
self.name = name
|
||
console.print(f"🚀 开始{name}...")
|
||
|
||
def update(self, message: str):
|
||
console.print(f"📊 {message}")
|
||
|
||
def finish(self, message: str):
|
||
console.print(f"✅ {message}")
|
||
|
||
yield MockTask(name)
|
||
|
||
def confirm_action(message: str, default: bool = False) -> bool:
|
||
"""确认操作"""
|
||
import typer
|
||
return typer.confirm(message, default=default)
|
||
|
||
def show_error(message: str):
|
||
"""显示错误信息"""
|
||
console.print(f"❌ [red]{message}[/red]")
|
||
|
||
def show_success(message: str):
|
||
"""显示成功信息"""
|
||
console.print(f"✅ [green]{message}[/green]")
|
||
|
||
def show_warning(message: str):
|
||
"""显示警告信息"""
|
||
console.print(f"⚠️ [yellow]{message}[/yellow]")
|
||
```
|
||
|
||
### **6. 项目入口**
|
||
```python
|
||
# python_core/__main__.py
|
||
"""
|
||
MixVideo 项目主入口
|
||
支持: python -m python_core
|
||
"""
|
||
|
||
from .cli.main import main
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
```
|
||
|
||
## 🎯 使用示例
|
||
|
||
### **命令组方式**
|
||
```bash
|
||
# 媒体管理
|
||
mixvideo media upload video.mp4 --tags "demo,test"
|
||
mixvideo media batch-upload /videos --recursive
|
||
mixvideo media list --limit 20
|
||
|
||
# 场景检测
|
||
mixvideo scene detect video.mp4 --threshold 25
|
||
mixvideo scene batch-detect /videos --output results.json
|
||
mixvideo scene compare video.mp4 --thresholds "20,30,40"
|
||
|
||
# 模板管理
|
||
mixvideo template import /templates --batch
|
||
mixvideo template export template_id --output template.json
|
||
|
||
# 系统管理
|
||
mixvideo system status
|
||
mixvideo system config --show
|
||
mixvideo system storage info
|
||
```
|
||
|
||
### **快捷命令方式**
|
||
```bash
|
||
# 常用命令的快捷方式
|
||
mixvideo upload video.mp4 --tags "demo"
|
||
mixvideo detect video.mp4 --threshold 30
|
||
mixvideo status
|
||
```
|
||
|
||
### **帮助系统**
|
||
```bash
|
||
# 查看主帮助
|
||
mixvideo --help
|
||
|
||
# 查看命令组帮助
|
||
mixvideo media --help
|
||
mixvideo scene --help
|
||
|
||
# 查看具体命令帮助
|
||
mixvideo media upload --help
|
||
mixvideo scene detect --help
|
||
```
|
||
|
||
## 🎉 方案优势
|
||
|
||
### **1. 统一体验**
|
||
- 🎯 一个入口点，所有功能
|
||
- 📋 一致的命令风格和参数
|
||
- 🎨 统一的输出格式和样式
|
||
|
||
### **2. 灵活使用**
|
||
- ⚡ 快捷命令满足日常需求
|
||
- 🔧 完整命令组满足专业需求
|
||
- 📚 清晰的帮助系统
|
||
|
||
### **3. 易于扩展**
|
||
- 🔌 新服务易于集成
|
||
- 📦 模块化的命令组织
|
||
- 🔄 向后兼容的设计
|
||
|
||
### **4. 开发友好**
|
||
- 🧩 清晰的代码组织
|
||
- 🔧 共享的工具函数
|
||
- 📝 统一的开发规范
|
||
|
||
这个设计方案提供了完整的多命令集成解决方案，既保持了各服务的独立性，又提供了统一的用户体验！
|