diff --git a/docs/multi-command-integration.md b/docs/multi-command-integration.md new file mode 100644 index 0000000..48457a0 --- /dev/null +++ b/docs/multi-command-integration.md @@ -0,0 +1,559 @@ +# 多命令集成设计方案 + +## 🎯 设计目标 + +将所有服务的命令集成到一个统一的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. 开发友好** +- 🧩 清晰的代码组织 +- 🔧 共享的工具函数 +- 📝 统一的开发规范 + +这个设计方案提供了完整的多命令集成解决方案，既保持了各服务的独立性，又提供了统一的用户体验！ diff --git a/docs/typer-development-standards.md b/docs/typer-development-standards.md new file mode 100644 index 0000000..7c56ad9 --- /dev/null +++ b/docs/typer-development-standards.md @@ -0,0 +1,485 @@ +# 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. 基础结构模板** + +```python +#!/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. 类型定义规范** + +#### **枚举类型** +```python +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" +``` + +#### **数据类型** +```python +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. 命令定义规范** + +#### **基础命令模板** +```python +@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) +``` + +#### **进度命令模板** +```python +@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. 错误处理规范** + +```python +# 输入验证 +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输出样式** +```python +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. 服务基类继承** +```python +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. 存储集成** +```python +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. 命令测试** +```python +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. 进度测试** +```python +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. 命令文档** +```python +@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. 帮助信息** +```python +# 应用级帮助 +app = typer.Typer( + name="media_manager", + help=""" + 🎬 媒体管理器 + + 功能强大的视频处理和管理工具，支持： + • 视频上传和元数据提取 + • 自动场景检测和分割 + • 批量处理和进度跟踪 + • 多种输出格式 + + 使用 --help 查看具体命令帮助。 + """, + rich_markup_mode="rich" +) +``` + +## 🎯 最佳实践 + +### **1. 性能优化** +- 使用类型提示提高IDE支持 +- 合理使用Rich组件，避免过度渲染 +- 大文件处理时显示进度条 +- 异步操作使用适当的进度反馈 + +### **2. 用户体验** +- 提供清晰的错误信息 +- 使用emoji和颜色增强可读性 +- 重要操作前进行确认 +- 提供详细的帮助文档 + +### **3. 代码质量** +- 所有函数添加类型提示 +- 使用dataclass定义数据结构 +- 遵循PEP 8代码风格 +- 编写完整的单元测试 + +这套规范确保了Typer实现与您现有架构的完美集成，同时提供了现代化、类型安全的开发体验！ diff --git a/examples/typer_media_manager.py b/examples/typer_media_manager.py new file mode 100644 index 0000000..c9e9439 --- /dev/null +++ b/examples/typer_media_manager.py @@ -0,0 +1,504 @@ +#!/usr/bin/env python3 +""" +基于Typer开发规范的媒体管理器实现示例 +""" + +import sys +import time +from pathlib import Path +from typing import Optional, List +from enum import Enum + +# 添加项目根目录到Python路径 +project_root = Path(__file__).parent.parent +sys.path.insert(0, str(project_root)) + +try: + import typer + from rich.console import Console + from rich.table import Table + from rich.panel import Panel + from rich.progress import Progress, SpinnerColumn, TextColumn +except ImportError: + print("❌ 需要安装 typer 和 rich: pip install typer rich") + sys.exit(1) + +from python_core.utils.progress import ProgressJSONRPCCommander +from python_core.services.base import ProgressServiceBase + +console = Console() + +# 1. 类型定义（遵循规范） +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" + +# 2. 服务类（遵循规范） +class MediaManagerService(ProgressServiceBase): + """媒体管理服务""" + + def get_service_name(self) -> str: + return "media_manager" + + def upload_video(self, video_path: Path, tags: List[str] = None, filename: str = None) -> dict: + """上传单个视频""" + if tags is None: + tags = [] + + # 模拟上传过程 + steps = [ + "计算文件哈希...", + "检查重复文件...", + "复制文件到存储...", + "提取视频信息...", + "检测场景变化..." + ] + + for step in steps: + self.report_progress(step) + time.sleep(0.3) # 模拟处理时间 + + result = { + "video_path": str(video_path), + "filename": filename or video_path.name, + "tags": tags, + "success": True, + "scenes_detected": 3, + "duration": 120.5 + } + + # 保存结果到存储 + self.save_data("uploads", f"upload_{int(time.time())}", result) + + return result + + def batch_upload(self, directory: Path, tags: List[str] = None, recursive: bool = False) -> dict: + """批量上传视频""" + if tags is None: + tags = [] + + # 扫描视频文件 + video_extensions = {'.mp4', '.avi', '.mov', '.mkv', '.wmv'} + video_files = [] + + if recursive: + for ext in video_extensions: + video_files.extend(directory.rglob(f"*{ext}")) + else: + for ext in video_extensions: + video_files.extend(directory.glob(f"*{ext}")) + + results = [] + for i, video_file in enumerate(video_files): + self.report_progress(f"处理文件: {video_file.name} ({i+1}/{len(video_files)})") + + try: + result = self.upload_video(video_file, tags) + results.append(result) + except Exception as e: + results.append({ + "video_path": str(video_file), + "success": False, + "error": str(e) + }) + + batch_result = { + "total_files": len(video_files), + "successful": len([r for r in results if r.get("success")]), + "failed": len([r for r in results if not r.get("success")]), + "results": results + } + + # 保存批量结果 + self.save_data("batch_uploads", f"batch_{int(time.time())}", batch_result) + + return batch_result + + def get_recent_uploads(self, limit: int = 10) -> List[dict]: + """获取最近的上传记录""" + keys = self.list_keys("uploads") + recent_keys = sorted(keys)[-limit:] + return list(self.load_batch_data("uploads", recent_keys).values()) + +# 3. Typer命令行接口（遵循规范） +class MediaManagerCommander(ProgressJSONRPCCommander): + """基于Typer的媒体管理器命令行接口""" + + def __init__(self): + super().__init__("media_manager") + self.app = typer.Typer( + name="media_manager", + help=""" + 🎬 媒体管理器 + + 功能强大的视频处理和管理工具，支持： + • 视频上传和元数据提取 + • 自动场景检测和分割 + • 批量处理和进度跟踪 + • 多种输出格式 + + 使用 --help 查看具体命令帮助。 + """, + rich_markup_mode="rich", + no_args_is_help=True + ) + self.service = MediaManagerService() + self._setup_commands() + + def _register_commands(self): + """注册命令（继承要求）""" + pass # Typer通过装饰器自动注册 + + def _is_progressive_command(self, command: str) -> bool: + """判断是否需要进度报告""" + return command in ["batch_upload", "analyze"] + + def _execute_with_progress(self, command: str, args: dict): + """执行带进度的命令""" + pass # 通过Typer命令直接处理 + + def _execute_simple_command(self, command: str, args: dict): + """执行简单命令""" + pass # 通过Typer命令直接处理 + + def _setup_commands(self): + """设置Typer命令""" + + @self.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="📝 自定义文件名" + ), + verbose: bool = typer.Option( + False, + "--verbose", "-v", + help="📝 详细输出" + ) + ): + """ + 📤 上传视频文件 + + 上传单个视频文件到媒体库，自动进行场景检测和元数据提取。 + + 示例: + media_manager upload video.mp4 --tags "demo,test" + media_manager upload /path/to/video.mp4 --filename "my_video" + + 注意: + - 支持的格式: MP4, AVI, MOV, MKV, WMV + - 自动检测重复文件 + - 自动提取视频元数据 + """ + # 参数验证 + if not video_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]{video_path}[/bold blue]") + + try: + # 设置进度回调 + def progress_callback(message: str): + console.print(f"📊 {message}") + + self.service.set_progress_callback(progress_callback) + + # 执行上传 + result = self.service.upload_video(video_path, tag_list, filename) + + # 显示结果 + console.print("✅ [bold green]上传完成[/bold green]") + + if verbose or True: # 总是显示基本信息 + self._show_upload_result(result) + + except Exception as e: + console.print(f"❌ [red]上传失败: {e}[/red]") + if verbose: + console.print_exception() + raise typer.Exit(1) + + @self.app.command() + def batch_upload( + input_directory: Path = typer.Argument( + ..., + help="📁 输入目录路径", + exists=True, + file_okay=False + ), + output_path: Optional[Path] = typer.Option( + None, + "--output", "-o", + help="📄 结果输出文件路径" + ), + tags: Optional[str] = typer.Option( + None, + "--tags", "-t", + help="🏷️ 标签列表（逗号分隔）" + ), + recursive: bool = typer.Option( + False, + "--recursive", "-r", + help="🔄 递归处理子目录" + ), + output_format: OutputFormat = typer.Option( + OutputFormat.JSON, + "--format", "-f", + help="📄 输出格式" + ) + ): + """ + 📦 批量上传视频文件（带进度条） + + 批量处理目录中的所有视频文件，支持递归扫描和进度跟踪。 + + 示例: + media_manager batch-upload /videos --tags "batch,demo" + media_manager batch-upload /videos -r --output results.json + """ + console.print(f"📦 批量上传目录: [bold blue]{input_directory}[/bold blue]") + + # 解析标签 + tag_list = [] + if tags: + tag_list = [tag.strip() for tag in tags.split(",")] + + # 扫描文件数量 + video_extensions = {'.mp4', '.avi', '.mov', '.mkv', '.wmv'} + video_files = [] + + if recursive: + for ext in video_extensions: + video_files.extend(input_directory.rglob(f"*{ext}")) + else: + for ext in video_extensions: + video_files.extend(input_directory.glob(f"*{ext}")) + + if not video_files: + console.print("⚠️ [yellow]未找到可处理的视频文件[/yellow]") + return + + console.print(f"📋 找到 {len(video_files)} 个视频文件") + + # 使用进度任务 + with self.create_task("批量上传", len(video_files)) as task: + def progress_callback(message: str): + # 从消息中提取文件信息更新任务 + if "处理文件:" in message: + task.update(message=message) + + self.service.set_progress_callback(progress_callback) + + # 执行批量上传 + result = self.service.batch_upload(input_directory, tag_list, recursive) + + task.finish(f"批量上传完成: {result['successful']}/{result['total_files']} 成功") + + # 显示结果摘要 + self._show_batch_result(result) + + # 保存结果文件 + if output_path: + self._save_batch_results(result, output_path, output_format) + console.print(f"📄 结果已保存到: {output_path}") + + @self.app.command() + def list_uploads( + limit: int = typer.Option( + 10, + "--limit", "-l", + min=1, + max=100, + help="📊 显示数量限制" + ), + verbose: bool = typer.Option( + False, + "--verbose", "-v", + help="📝 详细信息" + ) + ): + """ + 📋 列出最近的上传记录 + + 显示最近上传的视频文件信息和统计数据。 + """ + console.print("📋 最近的上传记录") + + try: + uploads = self.service.get_recent_uploads(limit) + + if not uploads: + console.print("⚠️ [yellow]暂无上传记录[/yellow]") + return + + self._show_uploads_table(uploads, verbose) + + except Exception as e: + console.print(f"❌ [red]获取记录失败: {e}[/red]") + raise typer.Exit(1) + + @self.app.command() + def status(): + """ + 📊 显示系统状态 + + 显示媒体管理器的当前状态和统计信息。 + """ + console.print("📊 媒体管理器状态") + + try: + # 获取统计信息 + uploads_stats = self.service.get_collection_stats("uploads") + batch_stats = self.service.get_collection_stats("batch_uploads") + + # 创建状态表格 + table = Table(title="系统状态") + table.add_column("组件", style="cyan") + table.add_column("状态", style="green") + table.add_column("详情", style="yellow") + + table.add_row("存储", "✅ 正常", f"JSON文件存储") + table.add_row("上传记录", "✅ 活跃", f"{uploads_stats.get('file_count', 0)} 条记录") + table.add_row("批量任务", "✅ 就绪", f"{batch_stats.get('file_count', 0)} 个任务") + table.add_row("进度系统", "✅ 就绪", "JSON-RPC协议") + + console.print(table) + + # 创建信息面板 + info_panel = Panel( + "[bold blue]媒体管理器运行正常[/bold blue]\n" + "所有组件状态良好，可以开始处理视频文件。", + title="📊 状态摘要", + border_style="green" + ) + console.print(info_panel) + + except Exception as e: + console.print(f"❌ [red]获取状态失败: {e}[/red]") + raise typer.Exit(1) + + def _show_upload_result(self, result: dict): + """显示上传结果""" + table = Table(title="📤 上传结果") + table.add_column("项目", style="cyan") + table.add_column("值", style="green") + + table.add_row("文件名", result["filename"]) + table.add_row("标签", ", ".join(result["tags"]) if result["tags"] else "无") + table.add_row("检测场景", str(result["scenes_detected"])) + table.add_row("视频时长", f"{result['duration']:.1f}秒") + + console.print(table) + + def _show_batch_result(self, result: dict): + """显示批量结果摘要""" + panel = Panel( + f"[bold green]总文件: {result['total_files']}[/bold green]\n" + f"[bold blue]成功: {result['successful']}[/bold blue]\n" + f"[bold red]失败: {result['failed']}[/bold red]", + title="📦 批量上传摘要", + border_style="green" + ) + console.print(panel) + + def _show_uploads_table(self, uploads: List[dict], verbose: bool): + """显示上传记录表格""" + table = Table(title="📋 上传记录") + table.add_column("文件名", style="cyan") + table.add_column("场景数", style="green") + table.add_column("时长", style="yellow") + + if verbose: + table.add_column("标签", style="magenta") + + for upload in uploads: + row = [ + upload["filename"], + str(upload.get("scenes_detected", "N/A")), + f"{upload.get('duration', 0):.1f}s" + ] + + if verbose: + tags = ", ".join(upload.get("tags", [])) or "无" + row.append(tags) + + table.add_row(*row) + + console.print(table) + + def _save_batch_results(self, result: dict, output_path: Path, format: OutputFormat): + """保存批量结果""" + if format == OutputFormat.JSON: + import json + with open(output_path, 'w', encoding='utf-8') as f: + json.dump(result, f, indent=2, ensure_ascii=False) + elif format == OutputFormat.CSV: + import csv + with open(output_path, 'w', newline='', encoding='utf-8') as f: + writer = csv.writer(f) + writer.writerow(['filename', 'success', 'scenes', 'duration', 'error']) + for item in result['results']: + writer.writerow([ + item.get('filename', ''), + item.get('success', False), + item.get('scenes_detected', ''), + item.get('duration', ''), + item.get('error', '') + ]) + else: # TXT + with open(output_path, 'w', encoding='utf-8') as f: + f.write(f"批量上传结果\n") + f.write(f"总文件数: {result['total_files']}\n") + f.write(f"成功: {result['successful']}\n") + f.write(f"失败: {result['failed']}\n\n") + + for item in result['results']: + f.write(f"文件: {item.get('filename', '')}\n") + f.write(f" 状态: {'成功' if item.get('success') else '失败'}\n") + if item.get('success'): + f.write(f" 场景数: {item.get('scenes_detected', 'N/A')}\n") + f.write(f" 时长: {item.get('duration', 0):.1f}秒\n") + else: + f.write(f" 错误: {item.get('error', '')}\n") + f.write("\n") + + def run(self): + """运行CLI""" + self.app() + +def main(): + """主入口函数""" + commander = MediaManagerCommander() + commander.run() + +if __name__ == "__main__": + main() diff --git a/prompt.md b/prompt.md index 2618027..3367276 100644 --- a/prompt.md +++ b/prompt.md @@ -5,6 +5,42 @@ * 核心层Python 视频处理核心逻辑 * 服务层Python 后台服务和文件管理 + +- 命令行工具 + +### **方案2: 纯Typer实现** + +#### **特点** +- 🎯 类型提示驱动 +- ✨ 现代化Python风格 +- 🤖 自动补全和验证 + +#### **代码示例** +```python +from typing import Optional +from enum import Enum + +class DetectorType(str, Enum): + content = "content" + threshold = "threshold" + adaptive = "adaptive" + +@app.command() +def detect( + video_path: Path = typer.Argument(..., help="视频文件路径"), + detector: DetectorType = typer.Option(DetectorType.content, help="检测器类型"), + threshold: float = typer.Option(30.0, help="检测阈值"), + output: Optional[Path] = typer.Option(None, help="输出文件路径") +): + """检测单个视频的场景""" + # 类型安全的处理逻辑 + pass +``` + +## 多命令集成方案 + + + - MoviePy‌ ``` diff --git a/python_core/cli/__init__.py b/python_core/cli/__init__.py new file mode 100644 index 0000000..912304a --- /dev/null +++ b/python_core/cli/__init__.py @@ -0,0 +1,8 @@ +#!/usr/bin/env python3 +""" +MixVideo 统一命令行接口 +""" + +from .cli import main + +__all__ = ["main"] diff --git a/python_core/cli/__main__.py b/python_core/cli/__main__.py new file mode 100644 index 0000000..eb35d9b --- /dev/null +++ b/python_core/cli/__main__.py @@ -0,0 +1,5 @@ +#!/usr/bin/env python3 +from .cli import main + +if __name__ == "__main__": + main() diff --git a/python_core/cli/cli.py b/python_core/cli/cli.py new file mode 100644 index 0000000..7755885 --- /dev/null +++ b/python_core/cli/cli.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python3 +""" +MixVideo 主命令行接口 +""" + +import sys +from pathlib import Path +from typing import Optional +from python_core.cli.const import progress_reporter, console, project_root +import typer + +# 导入命令模块 +from python_core.cli.commands import scene_app + +app = typer.Typer( + name="mixvideo", + help=""" + 🎬 MixVideo - 智能视频处理平台 + + 功能完整的视频处理和管理工具套件： + • 🎯 场景检测 - 智能识别视频场景变化 + • 📤 媒体管理 - 上传、处理、组织视频文件 + • 📋 模板管理 - 视频模板导入导出 + • ⚙️ 系统管理 - 配置、状态、存储管理 + + 快速开始: + mixvideo scene detect video.mp4 # 检测场景 + mixvideo scene batch-detect /videos # 批量检测 + mixvideo scene split video.mp4 # 分割视频 + mixvideo scene info video.mp4 # 视频信息 + """, + rich_markup_mode="rich", + no_args_is_help=True +) + +# 添加场景检测命令组到主应用 +app.add_typer(scene_app, name="scene") + +@app.command() +def init(): + """🚀 初始化MixVideo工作环境""" + progress_reporter.info("🚀 初始化MixVideo环境...") + # TODO: 实现初始化逻辑 + progress_reporter.success("✅ 初始化完成") + +def main(): + """主入口函数""" + try: + app() + except KeyboardInterrupt: + progress_reporter.error("\n👋 用户取消操作") + sys.exit(0) + except Exception as e: + progress_reporter.error(f"\n❌ [red]程序异常: {e}[/red]") + sys.exit(1) + +if __name__ == "__main__": + main() diff --git a/python_core/cli/commands/__init__.py b/python_core/cli/commands/__init__.py new file mode 100644 index 0000000..57cf584 --- /dev/null +++ b/python_core/cli/commands/__init__.py @@ -0,0 +1,8 @@ +#!/usr/bin/env python3 +""" +CLI 命令模块 +""" + +from .scene import scene_app + +__all__ = ["scene_app"] diff --git a/python_core/cli/commands/scene.py b/python_core/cli/commands/scene.py new file mode 100644 index 0000000..98116e4 --- /dev/null +++ b/python_core/cli/commands/scene.py @@ -0,0 +1,344 @@ +#!/usr/bin/env python3 +""" +场景检测命令模块 +""" + +from pathlib import Path +from typing import Optional +import typer + +from python_core.cli.const import progress_reporter, console +from json import dumps +# 创建场景检测命令组 +scene_app = typer.Typer(help="🎯 场景检测工具") + +@scene_app.command() +def detect( + video_path: Path = typer.Argument(..., help="📹 视频文件路径", exists=True), + detector: str = typer.Option("content", help="🔧 检测器类型 (content/threshold/adaptive)"), + threshold: float = typer.Option(30.0, help="🎚️ 检测阈值 (0-100)"), + min_scene_length: float = typer.Option(1.0, help="⏱️ 最小场景长度（秒）"), + output: Optional[Path] = typer.Option(None, "--output", "-o", help="📄 输出文件路径"), + format: str = typer.Option("json", help="📋 输出格式 (json/csv/txt)") +): + """🎯 检测单个视频的场景""" + try: + from python_core.cli.scene_detect import detector as scene_detector, DetectorType, OutputFormat + + # 验证参数 + try: + detector_type = DetectorType(detector) + except ValueError: + progress_reporter.error(f"❌ 无效的检测器类型: {detector}") + progress_reporter.info("💡 可用类型: content, threshold, adaptive") + raise typer.Exit(1) + + try: + output_format = OutputFormat(format) + except ValueError: + progress_reporter.error(f"❌ 无效的输出格式: {format}") + progress_reporter.info("💡 可用格式: json, csv, txt") + raise typer.Exit(1) + + # 执行检测 + result = scene_detector.detect_scenes( + video_path, detector_type, threshold, min_scene_length + ) + + if not result.success: + progress_reporter.error(f"❌ 检测失败: {result.error}") + raise typer.Exit(1) + + # 显示结果摘要 + console.print(f"📊 检测结果摘要:") + console.print(f" 文件: {result.filename}") + console.print(f" 检测器: {result.detector_type}") + console.print(f" 阈值: {result.threshold}") + console.print(f" 场景数: {result.total_scenes}") + console.print(f" 总时长: {result.total_duration:.2f}秒") + console.print(f" 检测时间: {result.detection_time:.2f}秒") + + # 显示场景详情 + if result.scenes: + console.print(f"\n🎬 场景列表:") + for scene in result.scenes[:10]: # 只显示前10个场景 + console.print(f" 场景 {scene.index}: {scene.start_time:.2f}s - {scene.end_time:.2f}s ({scene.duration:.2f}s)") + + if len(result.scenes) > 10: + console.print(f" ... 还有 {len(result.scenes) - 10} 个场景") + + # 保存结果 + if output: + scene_detector.save_results(result, output, output_format) + progress_reporter.success(f"📄 结果已保存到: {output}") + + return result + + except Exception as e: + progress_reporter.error(f"❌ 命令执行失败: {e}") + raise typer.Exit(1) + +@scene_app.command() +def batch_detect( + input_directory: Path = typer.Argument(..., help="📁 输入目录路径", exists=True), + detector: str = typer.Option("content", help="🔧 检测器类型 (content/threshold/adaptive)"), + threshold: float = typer.Option(30.0, help="🎚️ 检测阈值 (0-100)"), + recursive: bool = typer.Option(False, "--recursive", "-r", help="🔄 递归扫描子目录"), + output: Optional[Path] = typer.Option(None, "--output", "-o", help="📄 输出文件路径"), + format: str = typer.Option("json", help="📋 输出格式 (json/csv/txt)") +): + """📦 批量检测目录中的所有视频""" + try: + from python_core.cli.scene_detect import detector as scene_detector, DetectorType, OutputFormat + + # 验证参数 + try: + detector_type = DetectorType(detector) + output_format = OutputFormat(format) + except ValueError as e: + progress_reporter.error(f"❌ 参数错误: {e}") + raise typer.Exit(1) + + # 执行批量检测 + results = scene_detector.batch_detect( + input_directory, detector_type, threshold, recursive + ) + + if not results: + progress_reporter.warning("⚠️ 没有检测到任何视频文件") + return + + # 统计结果 + successful = len([r for r in results if r.success]) + failed = len(results) - successful + total_scenes = sum(r.total_scenes for r in results if r.success) + total_duration = sum(r.total_duration for r in results if r.success) + + console.print(f"📊 批量检测结果:") + console.print(f" 总文件数: {len(results)}") + console.print(f" 成功: {successful}") + console.print(f" 失败: {failed}") + console.print(f" 总场景数: {total_scenes}") + console.print(f" 总时长: {total_duration:.2f}秒") + + # 显示详细的场景信息 + console.print(f"\n🎬 详细场景信息:") + for result in results: + if result.success and result.scenes: + console.print(f"\n📹 {result.filename} ({result.total_scenes} 个场景):") + for scene in result.scenes: + console.print(f" 场景 {scene.index}: {scene.start_time:.2f}s - {scene.end_time:.2f}s (时长: {scene.duration:.2f}s)") + elif result.success: + console.print(f"\n📹 {result.filename}: 无场景数据") + + # 显示失败的文件 + failed_files = [r for r in results if not r.success] + if failed_files: + console.print(f"\n❌ 失败的文件:") + for result in failed_files[:5]: # 只显示前5个失败文件 + console.print(f" {result.filename}: {result.error}") + + if len(failed_files) > 5: + console.print(f" ... 还有 {len(failed_files) - 5} 个失败文件") + + # 保存结果 + if output: + scene_detector.save_results(results, output, output_format) + progress_reporter.success(f"📄 结果已保存到: {output}") + + return results + + except Exception as e: + progress_reporter.error(f"❌ 批量检测失败: {e}") + raise typer.Exit(1) + +@scene_app.command() +def compare( + video_path: Path = typer.Argument(..., help="📹 视频文件路径", exists=True), + thresholds: str = typer.Option("20,30,40", help="🎚️ 测试阈值列表（逗号分隔）"), + output: Optional[Path] = typer.Option(None, "--output", "-o", help="📄 输出文件路径") +): + """🔬 比较不同检测器的效果""" + try: + from python_core.cli.scene_detect import detector as scene_detector + + # 解析阈值列表 + try: + threshold_list = [float(t.strip()) for t in thresholds.split(",")] + except ValueError: + progress_reporter.error("❌ 无效的阈值格式，请使用逗号分隔的数字") + raise typer.Exit(1) + + # 执行比较 + result = scene_detector.compare_detectors(video_path, threshold_list) + + # 显示分析结果 + analysis = result["analysis"] + console.print(f"🔬 检测器比较结果:") + console.print(f" 视频: {Path(result['video_path']).name}") + console.print(f" 总测试数: {result['total_tests']}") + console.print(f" 成功测试数: {analysis['total_successful']}") + console.print(f" 推荐检测器: {analysis['best_detector']}") + console.print(f" 建议: {analysis['recommendation']}") + + # 显示详细分析 + console.print(f"\n📊 各检测器表现:") + for detector_name, stats in analysis["detector_analysis"].items(): + console.print(f" 🔧 {detector_name}:") + console.print(f" 平均场景数: {stats['average_scenes']:.1f}") + console.print(f" 平均检测时间: {stats['average_detection_time']:.2f}秒") + console.print(f" 测试次数: {stats['test_count']}") + + # 显示详细测试结果 + console.print(f"\n🧪 详细测试结果:") + for test_result in result["results"]: + if test_result["success"]: + console.print(f" {test_result['detector']} (阈值: {test_result['threshold']}): " + f"{test_result['scenes']} 场景, {test_result['detection_time']:.2f}s") + else: + console.print(f" {test_result['detector']} (阈值: {test_result['threshold']}): " + f"❌ {test_result['error']}") + + # 保存结果 + if output: + scene_detector.save_results(result, output) + progress_reporter.success(f"📄 结果已保存到: {output}") + + return result + + except Exception as e: + progress_reporter.error(f"❌ 比较测试失败: {e}") + raise typer.Exit(1) + +@scene_app.command() +def split( + video_path: Path = typer.Argument(..., help="📹 视频文件路径", exists=True), + detector: str = typer.Option("content", help="🔧 检测器类型 (content/threshold/adaptive)"), + threshold: float = typer.Option(30.0, help="🎚️ 检测阈值 (0-100)"), + output_dir: Optional[Path] = typer.Option(None, "--output-dir", "-d", help="📁 输出目录"), + filename_template: str = typer.Option("scene_{:03d}.mp4", help="📝 文件名模板") +): + """✂️ 根据场景检测结果分割视频""" + try: + from python_core.cli.scene_detect import detector as scene_detector, DetectorType + from scenedetect.video_splitter import split_video_ffmpeg + + # 验证参数 + try: + detector_type = DetectorType(detector) + except ValueError: + progress_reporter.error(f"❌ 无效的检测器类型: {detector}") + raise typer.Exit(1) + + # 设置输出目录 + if output_dir is None: + output_dir = video_path.parent / f"{video_path.stem}_scenes" + + output_dir.mkdir(parents=True, exist_ok=True) + + # 先检测场景 + progress_reporter.info("🎯 正在检测场景...") + result = scene_detector.detect_scenes(video_path, detector_type, threshold) + + if not result.success: + progress_reporter.error(f"❌ 场景检测失败: {result.error}") + raise typer.Exit(1) + + if not result.scenes: + progress_reporter.warning("⚠️ 未检测到任何场景") + return + + # 构建场景列表（PySceneDetect格式） + from scenedetect import FrameTimecode + scene_list = [] + + # 假设视频帧率（实际应该从视频中获取） + fps = 25.0 # 默认帧率，实际使用时应该从视频文件中获取 + + for scene in result.scenes: + start_tc = FrameTimecode(timecode=scene.start_time, fps=fps) + end_tc = FrameTimecode(timecode=scene.end_time, fps=fps) + scene_list.append((start_tc, end_tc)) + + # 分割视频 + progress_reporter.info(f"✂️ 正在分割视频到 {len(scene_list)} 个场景...") + + try: + split_video_ffmpeg( + input_video_path=str(video_path), + scene_list=scene_list, + output_file_template=str(output_dir / filename_template), + video_name=video_path.stem, + arg_override=None, + hide_progress=False + ) + + progress_reporter.success(f"✅ 视频分割完成，输出到: {output_dir}") + console.print(f"📁 输出目录: {output_dir}") + console.print(f"🎬 场景数量: {len(scene_list)}") + + # 列出生成的文件 + output_files = list(output_dir.glob("*.mp4")) + if output_files: + console.print(f"\n📄 生成的文件:") + for file_path in sorted(output_files)[:10]: + console.print(f" {file_path.name}") + + if len(output_files) > 10: + console.print(f" ... 还有 {len(output_files) - 10} 个文件") + + except Exception as e: + progress_reporter.error(f"❌ 视频分割失败: {e}") + raise typer.Exit(1) + + except Exception as e: + progress_reporter.error(f"❌ 分割命令执行失败: {e}") + raise typer.Exit(1) + +@scene_app.command() +def info( + video_path: Path = typer.Argument(..., help="📹 视频文件路径", exists=True) +): + """📋 显示视频基本信息""" + try: + import cv2 + + progress_reporter.info(f"📋 获取视频信息: {video_path.name}") + + # 使用OpenCV获取视频信息 + cap = cv2.VideoCapture(str(video_path)) + + if not cap.isOpened(): + raise Exception("无法打开视频文件") + + # 获取视频信息 + fps = cap.get(cv2.CAP_PROP_FPS) + frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) + width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) + height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) + duration = frame_count / fps if fps > 0 else 0 + resolution = (width, height) + + cap.release() + + # 显示信息 + console.print(f"📹 视频信息:") + console.print(f" 文件名: {video_path.name}") + console.print(f" 文件大小: {video_path.stat().st_size / (1024*1024):.2f} MB") + console.print(f" 分辨率: {resolution[0]}x{resolution[1]}") + console.print(f" 帧率: {fps:.2f} fps") + console.print(f" 总帧数: {frame_count}") + console.print(f" 时长: {duration:.2f}秒 ({duration//60:.0f}分{duration%60:.0f}秒)") + + progress_reporter.success(dumps({ + "filename": video_path.name, + "file_size_mb": video_path.stat().st_size / (1024*1024), + "resolution": resolution, + "fps": fps, + "frame_count": frame_count, + "duration": duration + })) + + except Exception as e: + progress_reporter.error(f"❌ 获取视频信息失败: {e}") + raise typer.Exit(1) diff --git a/python_core/cli/const.py b/python_core/cli/const.py new file mode 100644 index 0000000..44b81c3 --- /dev/null +++ b/python_core/cli/const.py @@ -0,0 +1,6 @@ +from python_core.utils.jsonrpc import create_progress_reporter +from rich.console import Console +from python_core.config import settings +console = Console() +progress_reporter = create_progress_reporter() +project_root = settings.project_root \ No newline at end of file diff --git a/python_core/cli/scene_detect.py b/python_core/cli/scene_detect.py new file mode 100644 index 0000000..0b36b78 --- /dev/null +++ b/python_core/cli/scene_detect.py @@ -0,0 +1,365 @@ +#!/usr/bin/env python3 +""" +PySceneDetect 场景检测命令行工具 +""" + +import os +import json +import time +from pathlib import Path +from typing import Optional, List +from enum import Enum +from dataclasses import dataclass, asdict + +import typer +from python_core.cli.const import progress_reporter, console, project_root + +# 检查 PySceneDetect 依赖 +from scenedetect import open_video, SceneManager +from scenedetect.detectors import ContentDetector, ThresholdDetector +from scenedetect.video_splitter import split_video_ffmpeg + +class DetectorType(str, Enum): + """检测器类型""" + CONTENT = "content" + THRESHOLD = "threshold" + ADAPTIVE = "adaptive" + +class OutputFormat(str, Enum): + """输出格式""" + JSON = "json" + CSV = "csv" + TXT = "txt" + +@dataclass +class SceneInfo: + """场景信息""" + index: int + start_time: float + end_time: float + duration: float + start_frame: int + end_frame: int + +@dataclass +class DetectionResult: + """检测结果""" + video_path: str + filename: str + detector_type: str + threshold: float + total_scenes: int + total_duration: float + detection_time: float + scenes: List[SceneInfo] + success: bool + error: Optional[str] = None + +class SceneDetector: + """场景检测器""" + + def __init__(self): + self.supported_formats = {'.mp4', '.avi', '.mov', '.mkv', '.wmv', '.flv', '.webm', '.m4v'} + + def detect_scenes(self, video_path: Path, detector_type: DetectorType = DetectorType.CONTENT, + threshold: float = 30.0, min_scene_length: float = 1.0) -> DetectionResult: + """检测单个视频的场景""" + start_time = time.time() + filename = video_path.name + + try: + progress_reporter.info(f"🎬 开始检测: {filename}") + + # 打开视频文件 + video = open_video(str(video_path)) + scene_manager = SceneManager() + + # 根据类型添加检测器 + if detector_type == DetectorType.CONTENT: + scene_manager.add_detector(ContentDetector(threshold=threshold)) + progress_reporter.info(f"📊 使用内容检测器，阈值: {threshold}") + elif detector_type == DetectorType.THRESHOLD: + scene_manager.add_detector(ThresholdDetector(threshold=threshold)) + progress_reporter.info(f"📊 使用阈值检测器，阈值: {threshold}") + elif detector_type == DetectorType.ADAPTIVE: + # 自适应：同时使用两种检测器 + scene_manager.add_detector(ContentDetector(threshold=threshold)) + scene_manager.add_detector(ThresholdDetector(threshold=threshold * 0.8)) + progress_reporter.info(f"📊 使用自适应检测器，阈值: {threshold}") + + # 开始检测 + progress_reporter.info("🔍 正在分析视频帧...") + scene_manager.detect_scenes(video) + scene_list = scene_manager.get_scene_list() + progress_reporter.info(f"✅ 检测到 {len(scene_list)} 个场景") + + # 获取视频信息 + fps = video.frame_rate + total_duration = video.duration.get_seconds() + progress_reporter.info(f"📊 视频信息: {fps:.2f}fps, {total_duration:.2f}秒") + + # 构建场景信息 + scenes = [] + + if not scene_list: + # 如果没有检测到场景变化，整个视频就是一个场景 + scene_info = SceneInfo( + index=0, + start_time=0.0, + end_time=total_duration, + duration=total_duration, + start_frame=0, + end_frame=int(total_duration * fps) if fps > 0 else 0 + ) + scenes.append(scene_info) + progress_reporter.info(f"📝 无场景变化，整个视频作为单一场景: {total_duration:.2f}秒") + else: + # 处理检测到的场景 + for start_time_scene, end_time_scene in scene_list: + start_seconds = start_time_scene.get_seconds() + end_seconds = end_time_scene.get_seconds() + duration = end_seconds - start_seconds + + # 跳过太短的场景 + if duration < min_scene_length: + continue + + scene_info = SceneInfo( + index=len(scenes), + start_time=start_seconds, + end_time=end_seconds, + duration=duration, + start_frame=start_time_scene.get_frames(), + end_frame=end_time_scene.get_frames() + ) + scenes.append(scene_info) + detection_time = time.time() - start_time + + result = DetectionResult( + video_path=str(video_path), + filename=filename, + detector_type=detector_type.value, + threshold=threshold, + total_scenes=len(scenes), + total_duration=total_duration, + detection_time=detection_time, + scenes=scenes, + success=True + ) + + progress_reporter.success(f"🎯 检测完成: {len(scenes)} 个场景，耗时 {detection_time:.2f}秒") + return result + + except Exception as e: + detection_time = time.time() - start_time + error_msg = str(e) + progress_reporter.error(f"❌ 检测失败: {error_msg}") + + return DetectionResult( + video_path=str(video_path), + filename=filename, + detector_type=detector_type.value, + threshold=threshold, + total_scenes=0, + total_duration=0.0, + detection_time=detection_time, + scenes=[], + success=False, + error=error_msg + ) + + def batch_detect(self, input_directory: Path, detector_type: DetectorType = DetectorType.CONTENT, + threshold: float = 30.0, recursive: bool = False) -> List[DetectionResult]: + """批量检测场景""" + progress_reporter.info(f"📦 开始批量检测: {input_directory}") + + # 扫描视频文件 + video_files = self._scan_video_files(input_directory, recursive) + + if not video_files: + progress_reporter.warning("⚠️ 未找到视频文件") + return [] + + progress_reporter.info(f"📋 找到 {len(video_files)} 个视频文件") + + results = [] + for i, video_file in enumerate(video_files): + progress_reporter.info(f"📊 处理进度: {i+1}/{len(video_files)} - {video_file.name}") + + result = self.detect_scenes(video_file, detector_type, threshold) + results.append(result) + + successful = len([r for r in results if r.success]) + progress_reporter.success(f"🎉 批量检测完成: {successful}/{len(results)} 成功") + + return results + + def compare_detectors(self, video_path: Path, thresholds: List[float] = None) -> dict: + """比较不同检测器效果""" + if thresholds is None: + thresholds = [20.0, 30.0, 40.0] + + progress_reporter.info(f"🔬 开始检测器比较: {video_path.name}") + + detectors = [DetectorType.CONTENT, DetectorType.THRESHOLD, DetectorType.ADAPTIVE] + results = [] + + total_tests = len(detectors) * len(thresholds) + current_test = 0 + + for detector in detectors: + for threshold in thresholds: + current_test += 1 + progress_reporter.info(f"🧪 测试 {current_test}/{total_tests}: {detector.value} (阈值: {threshold})") + + result = self.detect_scenes(video_path, detector, threshold) + results.append({ + "detector": detector.value, + "threshold": threshold, + "scenes": result.total_scenes, + "duration": result.total_duration, + "detection_time": result.detection_time, + "success": result.success, + "error": result.error + }) + + # 分析结果 + analysis = self._analyze_comparison(results) + + comparison_result = { + "video_path": str(video_path), + "total_tests": total_tests, + "results": results, + "analysis": analysis + } + + progress_reporter.success("🔬 检测器比较完成") + return comparison_result + + def _scan_video_files(self, directory: Path, recursive: bool = False) -> List[Path]: + """扫描视频文件""" + video_files = [] + + if recursive: + for ext in self.supported_formats: + video_files.extend(directory.rglob(f"*{ext}")) + else: + for ext in self.supported_formats: + video_files.extend(directory.glob(f"*{ext}")) + + return sorted(video_files) + + def _analyze_comparison(self, results: List[dict]) -> dict: + """分析比较结果""" + successful_results = [r for r in results if r["success"]] + + if not successful_results: + return {"message": "所有测试都失败了"} + + # 按检测器分组 + by_detector = {} + for result in successful_results: + detector = result["detector"] + if detector not in by_detector: + by_detector[detector] = [] + by_detector[detector].append(result) + + # 分析每个检测器 + detector_analysis = {} + for detector, detector_results in by_detector.items(): + avg_scenes = sum(r["scenes"] for r in detector_results) / len(detector_results) + avg_time = sum(r["detection_time"] for r in detector_results) / len(detector_results) + + detector_analysis[detector] = { + "average_scenes": avg_scenes, + "average_detection_time": avg_time, + "test_count": len(detector_results) + } + + # 推荐最佳检测器 + best_detector = max(detector_analysis.keys(), + key=lambda d: detector_analysis[d]["average_scenes"]) + + return { + "total_successful": len(successful_results), + "detector_analysis": detector_analysis, + "best_detector": best_detector, + "recommendation": f"推荐使用 {best_detector} 检测器" + } + + def save_results(self, results, output_path: Path, format: OutputFormat = OutputFormat.JSON): + """保存检测结果""" + output_path.parent.mkdir(parents=True, exist_ok=True) + + if format == OutputFormat.JSON: + self._save_json(results, output_path) + elif format == OutputFormat.CSV: + self._save_csv(results, output_path) + elif format == OutputFormat.TXT: + self._save_txt(results, output_path) + + progress_reporter.success(f"📄 结果已保存: {output_path}") + + def _save_json(self, results, output_path: Path): + """保存JSON格式""" + if isinstance(results, list): + # 批量结果 + data = [asdict(result) for result in results] + else: + # 单个结果或比较结果 + if hasattr(results, '__dict__'): + data = asdict(results) + else: + data = results + + with open(output_path, 'w', encoding='utf-8') as f: + json.dump(data, f, indent=2, ensure_ascii=False) + + def _save_csv(self, results, output_path: Path): + """保存CSV格式""" + import csv + + with open(output_path, 'w', newline='', encoding='utf-8') as f: + writer = csv.writer(f) + + if isinstance(results, list) and results: + # 批量结果 + writer.writerow(['filename', 'detector', 'threshold', 'scenes', 'duration', 'detection_time', 'success']) + + for result in results: + writer.writerow([ + result.filename, + result.detector_type, + result.threshold, + result.total_scenes, + result.total_duration, + result.detection_time, + result.success + ]) + + def _save_txt(self, results, output_path: Path): + """保存文本格式""" + with open(output_path, 'w', encoding='utf-8') as f: + f.write("PySceneDetect 场景检测结果\n") + f.write("=" * 50 + "\n\n") + + if isinstance(results, list): + # 批量结果 + for result in results: + f.write(f"文件: {result.filename}\n") + f.write(f" 检测器: {result.detector_type}\n") + f.write(f" 阈值: {result.threshold}\n") + f.write(f" 场景数: {result.total_scenes}\n") + f.write(f" 总时长: {result.total_duration:.2f}秒\n") + f.write(f" 检测时间: {result.detection_time:.2f}秒\n") + f.write(f" 状态: {'成功' if result.success else '失败'}\n") + + if result.scenes: + f.write(" 场景列表:\n") + for scene in result.scenes: + f.write(f" 场景 {scene.index}: {scene.start_time:.2f}s - {scene.end_time:.2f}s ({scene.duration:.2f}s)\n") + + f.write("\n") + +# 创建全局检测器实例 +detector = SceneDetector() \ No newline at end of file diff --git a/python_core/requirements.txt b/python_core/requirements.txt index 2644bf4..f37092a 100644 --- a/python_core/requirements.txt +++ b/python_core/requirements.txt @@ -24,4 +24,6 @@ loguru pydantic pyyaml pydantic_settings -scenedetect[opencv] \ No newline at end of file +scenedetect[opencv] +typer +rich \ No newline at end of file diff --git a/python_core/utils/jsonrpc.py b/python_core/utils/jsonrpc.py index cc181b3..2e1e8a1 100644 --- a/python_core/utils/jsonrpc.py +++ b/python_core/utils/jsonrpc.py @@ -144,6 +144,18 @@ class ProgressReporter: """Report error""" self.report("error", -1, message, error_details) + def info(self, message: str) -> None: + """Report info message""" + self.report("info", -1, message) + + def success(self, message: str) -> None: + """Report success message""" + self.report("success", -1, message) + + def warning(self, message: str) -> None: + """Report warning message""" + self.report("warning", -1, message) + # Error codes following JSON-RPC 2.0 specification class JSONRPCError: