fix: 封装命令行

python_core/services/media_manager/cli.py

@@ -0,0 +1,297 @@
+#!/usr/bin/env python3
+"""
+媒体管理器命令行接口
+"""
+
+from typing import Dict, Any
+from dataclasses import asdict
+
+from .manager import get_media_manager
+from python_core.utils.progress import ProgressJSONRPCCommander
+
+class MediaManagerCommander(ProgressJSONRPCCommander):
+    """媒体管理器命令行接口 - 支持进度条"""
+
+    def __init__(self):
+        super().__init__("media_manager")
+
+    def _register_commands(self) -> None:
+        """注册命令"""
+        # 上传命令
+        self.register_command(
+            name="upload",
+            description="上传单个视频文件",
+            required_args=["source_path"],
+            optional_args={
+                "filename": {"type": str, "description": "文件名"},
+                "tags": {"type": str, "description": "标签（逗号分隔）"}
+            }
+        )
+
+        self.register_command(
+            name="batch_upload",
+            description="批量上传视频文件",
+            required_args=["source_directory"],
+            optional_args={
+                "tags": {"type": str, "description": "标签（逗号分隔）"}
+            }
+        )
+
+        # 查询命令
+        self.register_command(
+            name="get_all_segments",
+            description="获取所有视频片段"
+        )
+
+        self.register_command(
+            name="get_all_videos",
+            description="获取所有原始视频"
+        )
+
+        self.register_command(
+            name="get_segments_by_video",
+            description="获取指定视频的片段",
+            required_args=["video_id"]
+        )
+
+        self.register_command(
+            name="search_segments",
+            description="搜索视频片段",
+            required_args=["keyword"]
+        )
+
+        self.register_command(
+            name="get_segments_by_tags",
+            description="根据标签获取片段",
+            required_args=["tags"],
+            optional_args={
+                "match_all": {"type": bool, "default": False, "description": "是否匹配所有标签"}
+            }
+        )
+
+        self.register_command(
+            name="get_popular_segments",
+            description="获取热门片段",
+            optional_args={
+                "limit": {"type": int, "default": 10, "description": "返回数量限制"}
+            }
+        )
+
+        # 管理命令
+        self.register_command(
+            name="add_tags",
+            description="为片段添加标签",
+            required_args=["segment_id", "tags"]
+        )
+
+        self.register_command(
+            name="increment_usage",
+            description="增加片段使用次数",
+            required_args=["segment_id"]
+        )
+
+        self.register_command(
+            name="delete_segment",
+            description="删除视频片段",
+            required_args=["segment_id"]
+        )
+
+        self.register_command(
+            name="delete_video",
+            description="删除原始视频",
+            required_args=["video_id"]
+        )
+
+    def _is_progressive_command(self, command: str) -> bool:
+        """判断是否需要进度报告的命令"""
+        # 上传操作需要进度报告
+        return command in ["upload", "batch_upload"]
+
+    def _execute_with_progress(self, command: str, args: Dict[str, Any]) -> Any:
+        """执行带进度的命令"""
+        media_manager = get_media_manager()
+
+        if command == "upload":
+            return self._upload_with_progress(
+                media_manager,
+                args["source_path"],
+                args.get("filename"),
+                self._parse_tags(args.get("tags"))
+            )
+        elif command == "batch_upload":
+            return self._batch_upload_with_progress(
+                media_manager,
+                args["source_directory"],
+                self._parse_tags(args.get("tags"))
+            )
+        else:
+            raise ValueError(f"Unknown progressive command: {command}")
+
+    def _execute_simple_command(self, command: str, args: Dict[str, Any]) -> Any:
+        """执行简单命令（不需要进度）"""
+        media_manager = get_media_manager()
+
+        if command == "upload":
+            tags = self._parse_tags(args.get("tags"))
+            result = media_manager.upload_video_file(
+                args["source_path"],
+                args.get("filename"),
+                tags
+            )
+            return asdict(result)
+
+        elif command == "get_all_segments":
+            return media_manager.get_all_segments()
+
+        elif command == "get_all_videos":
+            return media_manager.get_all_original_videos()
+
+        elif command == "get_segments_by_video":
+            return media_manager.get_segments_by_video_id(args["video_id"])
+
+        elif command == "search_segments":
+            return media_manager.search_segments(args["keyword"])
+
+        elif command == "get_segments_by_tags":
+            tags = self._parse_tags(args["tags"])
+            return media_manager.get_segments_by_tags(
+                tags,
+                args.get("match_all", False)
+            )
+
+        elif command == "get_popular_segments":
+            return media_manager.get_popular_segments(args.get("limit", 10))
+
+        elif command == "add_tags":
+            tags = self._parse_tags(args["tags"])
+            success = media_manager.add_segment_tags(args["segment_id"], tags)
+            return {"success": success}
+
+        elif command == "increment_usage":
+            success = media_manager.increment_segment_usage(args["segment_id"])
+            return {"success": success}
+
+        elif command == "delete_segment":
+            success = media_manager.delete_segment(args["segment_id"])
+            return {"success": success}
+
+        elif command == "delete_video":
+            success = media_manager.delete_original_video(args["video_id"])
+            return {"success": success}
+
+        else:
+            raise ValueError(f"Unknown command: {command}")
+
+    def _upload_with_progress(self, media_manager, source_path: str, filename: str, tags: list) -> dict:
+        """带进度的单个上传"""
+        with self.create_task("上传视频文件", 5) as task:
+            def progress_callback(message: str):
+                # 根据消息更新进度
+                if "计算文件哈希" in message:
+                    task.update(0, message)
+                elif "检查重复文件" in message:
+                    task.update(1, message)
+                elif "复制文件" in message:
+                    task.update(2, message)
+                elif "提取视频信息" in message:
+                    task.update(3, message)
+                elif "检测场景变化" in message:
+                    task.update(4, message)
+                elif "分割视频" in message:
+                    task.update(5, message)
+                elif "保存数据" in message:
+                    task.update(6, message)
+                else:
+                    task.update(message=message)
+
+            result = media_manager.upload_video_file(
+                source_path, filename, tags, progress_callback
+            )
+
+            task.finish("上传完成")
+            return asdict(result)
+
+    def _batch_upload_with_progress(self, media_manager, source_directory: str, tags: list) -> dict:
+        """带进度的批量上传"""
+        import os
+
+        # 支持的视频格式
+        video_extensions = {'.mp4', '.avi', '.mov', '.mkv', '.wmv', '.flv', '.webm', '.m4v'}
+
+        # 先扫描所有视频文件
+        video_files = []
+        for root, _, files in os.walk(source_directory):
+            for file in files:
+                file_ext = os.path.splitext(file)[1].lower()
+                if file_ext in video_extensions:
+                    video_files.append(os.path.join(root, file))
+
+        if not video_files:
+            return {
+                "total_files": 0,
+                "uploaded_files": 0,
+                "skipped_files": 0,
+                "failed_files": 0,
+                "total_segments": 0,
+                "uploaded_list": [],
+                "skipped_list": [],
+                "failed_list": [],
+                "message": "No video files found"
+            }
+
+        # 使用进度任务
+        with self.create_task("批量上传视频", len(video_files)) as task:
+            result = {
+                "total_files": len(video_files),
+                "uploaded_files": 0,
+                "skipped_files": 0,
+                "failed_files": 0,
+                "total_segments": 0,
+                "uploaded_list": [],
+                "skipped_list": [],
+                "failed_list": []
+            }
+
+            for i, file_path in enumerate(video_files):
+                filename = os.path.basename(file_path)
+                task.update(i, f"处理文件: {filename}")
+
+                try:
+                    # 尝试上传文件
+                    upload_result = media_manager.upload_video_file(file_path, filename, tags)
+
+                    if upload_result.is_duplicate:
+                        result["skipped_files"] += 1
+                        result["skipped_list"].append({
+                            'filename': filename,
+                            'reason': 'Already exists (same MD5)'
+                        })
+                    else:
+                        result["uploaded_files"] += 1
+                        result["total_segments"] += len(upload_result.segments)
+                        result["uploaded_list"].append(asdict(upload_result))
+
+                except Exception as e:
+                    result["failed_files"] += 1
+                    result["failed_list"].append({
+                        'filename': filename,
+                        'error': str(e)
+                    })
+
+            task.finish(f"批量上传完成: {result['uploaded_files']} 成功, {result['skipped_files']} 跳过, {result['failed_files']} 失败")
+
+        return result
+
+    def _parse_tags(self, tags_str: str) -> list:
+        """解析标签字符串"""
+        if not tags_str:
+            return []
+        return [tag.strip() for tag in tags_str.split(",") if tag.strip()]
+
+def main():
+    """主函数"""
+    commander = MediaManagerCommander()
+    commander.run()
+
+if __name__ == "__main__":
+    main()