fix

python_core/cli/cli.py

@@ -9,6 +9,7 @@ import typer
 
 # 导入命令模块
 from python_core.cli.commands import scene_detect
+from python_core.cli.commands.template import template_app
 
 app = typer.Typer(
     name="mixvideo",
@@ -17,8 +18,8 @@ app = typer.Typer(
 
     功能完整的视频处理和管理工具套件：
     • 🎯 场景检测 - 智能识别视频场景变化
+    • 📋 模板管理 - 视频模板批量导入、列表查看、详情获取
     • 📤 媒体管理 - 上传、处理、组织视频文件
-    • 📋 模板管理 - 视频模板导入导出
     • ⚙️ 系统管理 - 配置、状态、存储管理
     """,
     rich_markup_mode="rich",
@@ -27,6 +28,7 @@ app = typer.Typer(
 
 # 添加命令组到主应用
 app.add_typer(scene_detect, name="scene")
+app.add_typer(template_app, name="template")
 
 @app.command()
 def init():

python_core/cli/commands/template.py

@@ -0,0 +1,469 @@
+"""
+模板管理CLI命令
+"""
+
+from pathlib import Path
+from typing import Optional, List
+import typer
+from rich.console import Console
+from rich.table import Table
+from rich.panel import Panel
+from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskProgressColumn
+from rich.live import Live
+import json
+
+from python_core.services.template_manager import TemplateManager, TemplateInfo
+from python_core.utils.logger import logger
+
+console = Console()
+template_app = typer.Typer(name="template", help="模板管理命令")
+
+
+@template_app.command("import")
+def batch_import(
+    source_folder: str = typer.Argument(..., help="包含模板子文件夹的源文件夹"),
+    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
+    show_progress: bool = typer.Option(True, "--progress", "-p", help="显示进度条")
+):
+    """批量导入模板"""
+    
+    try:
+        console.print(f"📦 [bold blue]批量导入模板[/bold blue]")
+        console.print(f"📁 源文件夹: {source_folder}")
+        
+        # 验证源文件夹
+        source_path = Path(source_folder)
+        if not source_path.exists():
+            console.print(f"[red]❌ 源文件夹不存在: {source_folder}[/red]")
+            raise typer.Exit(1)
+        
+        if not source_path.is_dir():
+            console.print(f"[red]❌ 路径不是文件夹: {source_folder}[/red]")
+            raise typer.Exit(1)
+        
+        # 创建模板管理器
+        manager = TemplateManager()
+        
+        # 进度回调函数
+        progress_messages = []
+        
+        def progress_callback(message: str):
+            progress_messages.append(message)
+            if verbose:
+                console.print(f"  {message}")
+        
+        # 执行批量导入
+        if show_progress and not verbose:
+            with Progress(
+                SpinnerColumn(),
+                TextColumn("[progress.description]{task.description}"),
+                BarColumn(),
+                TaskProgressColumn(),
+                console=console
+            ) as progress:
+                task = progress.add_task("正在导入模板...", total=None)
+                
+                def progress_with_bar(message: str):
+                    progress.update(task, description=message)
+                    progress_callback(message)
+                
+                result = manager.batch_import_templates(source_folder, progress_with_bar)
+        else:
+            result = manager.batch_import_templates(source_folder, progress_callback)
+        
+        # 显示结果
+        if result['status']:
+            console.print(f"\n✅ [bold green]导入完成![/bold green]")
+            console.print(f"📊 成功导入: {result['imported_count']} 个模板")
+            console.print(f"❌ 导入失败: {result['failed_count']} 个模板")
+            
+            # 显示成功导入的模板
+            if result['imported_templates'] and verbose:
+                console.print(f"\n📋 [bold green]成功导入的模板:[/bold green]")
+                for template in result['imported_templates']:
+                    if isinstance(template, dict):
+                        console.print(f"  ✅ {template.get('name', 'Unknown')} (ID: {template.get('id', 'Unknown')})")
+                    else:
+                        console.print(f"  ✅ {template.name} (ID: {template.id})")
+            
+            # 显示失败的模板
+            if result['failed_templates']:
+                console.print(f"\n❌ [bold red]导入失败的模板:[/bold red]")
+                for failed in result['failed_templates']:
+                    console.print(f"  ❌ {failed['name']}: {failed['error']}")
+        else:
+            console.print(f"[red]❌ 导入失败: {result['msg']}[/red]")
+            raise typer.Exit(1)
+        
+    except Exception as e:
+        console.print(f"[red]❌ 批量导入失败: {str(e)}[/red]")
+        raise typer.Exit(1)
+
+
+@template_app.command("list")
+def list_templates(
+    limit: int = typer.Option(20, "--limit", "-l", help="显示数量限制"),
+    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
+    format_output: str = typer.Option("table", "--format", "-f", help="输出格式: table, json")
+):
+    """列出所有模板"""
+    
+    try:
+        console.print(f"📋 [bold blue]模板列表[/bold blue]")
+        
+        # 创建模板管理器
+        manager = TemplateManager()
+        
+        # 获取模板列表
+        templates = manager.get_templates()
+        
+        if not templates:
+            console.print("📭 没有模板")
+            return
+        
+        # 限制显示数量
+        total_count = len(templates)
+        if len(templates) > limit:
+            templates = templates[:limit]
+            console.print(f"📊 显示前 {limit} 个模板（共 {total_count} 个）")
+        else:
+            console.print(f"📊 共 {total_count} 个模板")
+        
+        if format_output == "json":
+            # JSON格式输出
+            templates_data = []
+            for template in templates:
+                if isinstance(template, TemplateInfo):
+                    from dataclasses import asdict
+                    templates_data.append(asdict(template))
+                else:
+                    templates_data.append(template)
+            
+            console.print(json.dumps(templates_data, ensure_ascii=False, indent=2))
+            return
+        
+        # 表格格式输出
+        table = Table(title="模板列表")
+        table.add_column("ID", style="cyan", width=12)
+        table.add_column("名称", style="green")
+        table.add_column("描述", style="yellow")
+        table.add_column("时长", style="magenta")
+        table.add_column("素材数", style="blue")
+        table.add_column("轨道数", style="red")
+        table.add_column("创建时间", style="dim")
+        
+        if verbose:
+            table.add_column("标签", style="cyan")
+            table.add_column("资源路径", style="dim")
+        
+        for template in templates:
+            # 格式化时长
+            duration = getattr(template, 'duration', 0)
+            if duration > 60:
+                duration_str = f"{duration // 60}m{duration % 60}s"
+            else:
+                duration_str = f"{duration}s"
+            
+            # 格式化创建时间
+            created_at = getattr(template, 'created_at', '')
+            if 'T' in created_at:
+                created_at = created_at.split('T')[0] + ' ' + created_at.split('T')[1][:8]
+            
+            row = [
+                getattr(template, 'id', '')[:12],
+                getattr(template, 'name', ''),
+                getattr(template, 'description', '')[:50] + ('...' if len(getattr(template, 'description', '')) > 50 else ''),
+                duration_str,
+                str(getattr(template, 'material_count', 0)),
+                str(getattr(template, 'track_count', 0)),
+                created_at
+            ]
+            
+            if verbose:
+                tags = getattr(template, 'tags', [])
+                tags_str = ', '.join(tags) if tags else '-'
+                resources_path = getattr(template, 'resources_path', '')
+                
+                row.extend([
+                    tags_str,
+                    resources_path
+                ])
+            
+            table.add_row(*row)
+        
+        console.print(table)
+        
+    except Exception as e:
+        console.print(f"[red]❌ 获取模板列表失败: {str(e)}[/red]")
+        raise typer.Exit(1)
+
+
+@template_app.command("get")
+def get_template(
+    template_id: str = typer.Argument(..., help="模板ID"),
+    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
+    format_output: str = typer.Option("panel", "--format", "-f", help="输出格式: panel, json")
+):
+    """获取模板详情"""
+    
+    try:
+        console.print(f"🔍 [bold blue]获取模板详情[/bold blue]")
+        console.print(f"模板ID: {template_id}")
+        
+        # 创建模板管理器
+        manager = TemplateManager()
+        
+        # 获取模板
+        template = manager.get_template(template_id)
+        
+        if not template:
+            console.print(f"[red]❌ 未找到模板: {template_id}[/red]")
+            raise typer.Exit(1)
+        
+        if format_output == "json":
+            # JSON格式输出
+            if isinstance(template, TemplateInfo):
+                from dataclasses import asdict
+                template_data = asdict(template)
+            else:
+                template_data = template
+            
+            console.print(json.dumps(template_data, ensure_ascii=False, indent=2))
+            return
+        
+        # 面板格式输出
+        console.print(f"\n✅ [bold green]模板详情[/bold green]")
+        
+        # 格式化时长
+        duration = getattr(template, 'duration', 0)
+        if duration > 60:
+            duration_str = f"{duration // 60}分{duration % 60}秒"
+        else:
+            duration_str = f"{duration}秒"
+        
+        # 基本信息
+        basic_info = f"""
+🆔 模板ID: {getattr(template, 'id', '')}
+📝 名称: {getattr(template, 'name', '')}
+📄 描述: {getattr(template, 'description', '')}
+⏱️ 时长: {duration_str}
+📦 素材数: {getattr(template, 'material_count', 0)}
+🎬 轨道数: {getattr(template, 'track_count', 0)}
+📅 创建时间: {getattr(template, 'created_at', '')}
+🔄 更新时间: {getattr(template, 'updated_at', '')}
+        """
+        
+        panel = Panel(basic_info.strip(), title="模板基本信息", border_style="green")
+        console.print(panel)
+        
+        if verbose:
+            # 详细信息
+            tags = getattr(template, 'tags', [])
+            tags_str = ', '.join(tags) if tags else '无'
+            
+            detail_info = f"""
+🏷️ 标签: {tags_str}
+📁 草稿路径: {getattr(template, 'draft_content_path', '')}
+📂 资源路径: {getattr(template, 'resources_path', '')}
+🖼️ 缩略图: {getattr(template, 'thumbnail_path', '') or '无'}
+            """
+            
+            detail_panel = Panel(detail_info.strip(), title="详细信息", border_style="blue")
+            console.print(detail_panel)
+            
+            # 画布配置
+            canvas_config = getattr(template, 'canvas_config', {})
+            if canvas_config:
+                canvas_info = f"""
+📐 画布配置:
+   宽度: {canvas_config.get('width', 'Unknown')}
+   高度: {canvas_config.get('height', 'Unknown')}
+   帧率: {canvas_config.get('fps', 'Unknown')}
+                """
+                
+                canvas_panel = Panel(canvas_info.strip(), title="画布配置", border_style="yellow")
+                console.print(canvas_panel)
+        
+    except Exception as e:
+        console.print(f"[red]❌ 获取模板详情失败: {str(e)}[/red]")
+        raise typer.Exit(1)
+
+
+@template_app.command("detail")
+def get_template_detail(
+    template_id: str = typer.Argument(..., help="模板ID"),
+    format_output: str = typer.Option("json", "--format", "-f", help="输出格式: json, summary")
+):
+    """获取模板详细信息（包含轨道和片段）"""
+    
+    try:
+        console.print(f"🔍 [bold blue]获取模板详细信息[/bold blue]")
+        console.print(f"模板ID: {template_id}")
+        
+        # 创建模板管理器
+        manager = TemplateManager()
+        
+        # 获取模板详细信息
+        detail = manager.get_template_detail(template_id)
+        
+        if not detail:
+            console.print(f"[red]❌ 未找到模板详细信息: {template_id}[/red]")
+            raise typer.Exit(1)
+        
+        if format_output == "json":
+            # JSON格式输出
+            console.print(json.dumps(detail, ensure_ascii=False, indent=2))
+            return
+        
+        # 摘要格式输出
+        console.print(f"\n✅ [bold green]模板详细信息[/bold green]")
+        
+        # 基本信息
+        template_info = detail.get('template', {})
+        console.print(f"📝 名称: {template_info.get('name', 'Unknown')}")
+        console.print(f"📄 描述: {template_info.get('description', 'Unknown')}")
+        console.print(f"⏱️ 时长: {template_info.get('duration', 0)}秒")
+        
+        # 轨道信息
+        tracks = detail.get('tracks', [])
+        console.print(f"\n🎬 [bold green]轨道信息 ({len(tracks)} 个轨道)[/bold green]")
+        
+        for i, track in enumerate(tracks, 1):
+            track_type = track.get('type', 'unknown')
+            segments_count = len(track.get('segments', []))
+            console.print(f"  轨道 {i}: {track_type} ({segments_count} 个片段)")
+            
+            # 显示片段信息
+            for j, segment in enumerate(track.get('segments', []), 1):
+                segment_name = segment.get('name', f'片段{j}')
+                start_time = segment.get('start_time', 0)
+                end_time = segment.get('end_time', 0)
+                console.print(f"    片段 {j}: {segment_name} ({start_time}s - {end_time}s)")
+        
+    except Exception as e:
+        console.print(f"[red]❌ 获取模板详细信息失败: {str(e)}[/red]")
+        raise typer.Exit(1)
+
+
+@template_app.command("delete")
+def delete_template(
+    template_id: str = typer.Argument(..., help="模板ID"),
+    force: bool = typer.Option(False, "--force", "-f", help="强制删除，不询问确认")
+):
+    """删除模板"""
+    
+    try:
+        console.print(f"🗑️ [bold red]删除模板[/bold red]")
+        console.print(f"模板ID: {template_id}")
+        
+        # 创建模板管理器
+        manager = TemplateManager()
+        
+        # 获取模板信息
+        template = manager.get_template(template_id)
+        if not template:
+            console.print(f"[red]❌ 未找到模板: {template_id}[/red]")
+            raise typer.Exit(1)
+        
+        console.print(f"模板名称: {getattr(template, 'name', 'Unknown')}")
+        console.print(f"模板描述: {getattr(template, 'description', 'Unknown')}")
+        
+        # 确认删除
+        if not force:
+            confirm = typer.confirm("确定要删除这个模板吗？此操作不可恢复。")
+            if not confirm:
+                console.print("❌ 操作已取消")
+                return
+        
+        # 删除模板
+        success = manager.delete_template(template_id)
+        
+        if success:
+            console.print(f"✅ [bold green]模板删除成功[/bold green]")
+        else:
+            console.print(f"[red]❌ 模板删除失败[/red]")
+            raise typer.Exit(1)
+        
+    except Exception as e:
+        console.print(f"[red]❌ 删除模板失败: {str(e)}[/red]")
+        raise typer.Exit(1)
+
+
+@template_app.command("stats")
+def show_stats():
+    """显示模板统计信息"""
+    
+    try:
+        console.print(f"📊 [bold blue]模板统计信息[/bold blue]")
+        
+        # 创建模板管理器
+        manager = TemplateManager()
+        
+        # 获取所有模板
+        templates = manager.get_templates()
+        
+        if not templates:
+            console.print("📭 没有模板")
+            return
+        
+        # 计算统计信息
+        total_templates = len(templates)
+        total_materials = sum(getattr(template, 'material_count', 0) for template in templates)
+        total_tracks = sum(getattr(template, 'track_count', 0) for template in templates)
+        total_duration = sum(getattr(template, 'duration', 0) for template in templates)
+        
+        # 格式化总时长
+        if total_duration > 3600:
+            duration_str = f"{total_duration // 3600}小时{(total_duration % 3600) // 60}分{total_duration % 60}秒"
+        elif total_duration > 60:
+            duration_str = f"{total_duration // 60}分{total_duration % 60}秒"
+        else:
+            duration_str = f"{total_duration}秒"
+        
+        # 创建统计面板
+        stats_content = f"""
+📈 [bold green]总体统计[/bold green]
+   模板总数: {total_templates}
+   素材总数: {total_materials}
+   轨道总数: {total_tracks}
+   总时长: {duration_str}
+   平均时长: {total_duration // total_templates if total_templates > 0 else 0}秒
+        """
+        
+        stats_panel = Panel(stats_content.strip(), title="模板统计", border_style="green")
+        console.print(stats_panel)
+        
+        # 显示最近的模板
+        recent_templates = sorted(templates, key=lambda x: getattr(x, 'created_at', ''), reverse=True)[:5]
+        
+        if recent_templates:
+            console.print(f"\n🕒 [bold green]最近创建的模板[/bold green]")
+            
+            recent_table = Table()
+            recent_table.add_column("名称", style="green")
+            recent_table.add_column("时长", style="magenta")
+            recent_table.add_column("创建时间", style="dim")
+            
+            for template in recent_templates:
+                duration = getattr(template, 'duration', 0)
+                duration_str = f"{duration}s" if duration < 60 else f"{duration // 60}m{duration % 60}s"
+                
+                created_at = getattr(template, 'created_at', '')
+                if 'T' in created_at:
+                    created_at = created_at.split('T')[0] + ' ' + created_at.split('T')[1][:8]
+                
+                recent_table.add_row(
+                    getattr(template, 'name', 'Unknown'),
+                    duration_str,
+                    created_at
+                )
+            
+            console.print(recent_table)
+        
+    except Exception as e:
+        console.print(f"[red]❌ 获取统计信息失败: {str(e)}[/red]")
+        raise typer.Exit(1)
+
+
+if __name__ == "__main__":
+    template_app()

python_core/services/scene_detection/__init__.py

@@ -1,27 +0,0 @@
-#!/usr/bin/env python3
-"""
-场景检测服务模块
-"""
-
-from .types import (
-    DetectorType, SceneInfo, VideoSceneResult, 
-    BatchDetectionConfig, BatchDetectionResult, DetectionStats
-)
-from .detector import SceneDetectionService
-from .cli import SceneDetectionCommander
-
-__all__ = [
-    # 数据类型
-    "DetectorType",
-    "SceneInfo", 
-    "VideoSceneResult",
-    "BatchDetectionConfig",
-    "BatchDetectionResult",
-    "DetectionStats",
-    
-    # 服务
-    "SceneDetectionService",
-    
-    # 命令行接口
-    "SceneDetectionCommander"
-]

python_core/services/scene_detection/__main__.py

@@ -1,10 +0,0 @@
-#!/usr/bin/env python3
-"""
-场景检测服务主入口
-支持直接运行: python -m python_core.services.scene_detection
-"""
-
-from .cli import main
-
-if __name__ == "__main__":
-    main()

python_core/services/scene_detection/cli.py

@@ -1,352 +0,0 @@
-#!/usr/bin/env python3
-"""
-场景检测命令行接口
-"""
-
-import os
-from typing import Dict, Any
-from dataclasses import asdict
-
-from .types import DetectorType, BatchDetectionConfig
-from .detector import SceneDetectionService
-from python_core.utils.progress import ProgressJSONRPCCommander
-
-class SceneDetectionCommander(ProgressJSONRPCCommander):
-    """场景检测命令行接口 - 支持进度条"""
-
-    def __init__(self):
-        super().__init__("scene_detection")
-        self.service = SceneDetectionService()
-
-    def _register_commands(self) -> None:
-        """注册命令"""
-        # 单个视频检测
-        self.register_command(
-            name="detect",
-            description="检测单个视频的场景",
-            required_args=["video_path"],
-            optional_args={
-                "detector": {"type": str, "default": "content", "choices": ["content", "threshold", "adaptive"], "description": "检测器类型"},
-                "threshold": {"type": float, "default": 30.0, "description": "检测阈值"},
-                "min_scene_length": {"type": float, "default": 1.0, "description": "最小场景长度（秒）"},
-                "output": {"type": str, "description": "输出文件路径"},
-                "format": {"type": str, "default": "json", "choices": ["json", "csv", "txt"], "description": "输出格式"}
-            }
-        )
-
-        # 批量检测
-        self.register_command(
-            name="batch_detect",
-            description="批量检测目录中所有视频的场景",
-            required_args=["input_directory"],
-            optional_args={
-                "detector": {"type": str, "default": "content", "choices": ["content", "threshold", "adaptive"], "description": "检测器类型"},
-                "threshold": {"type": float, "default": 30.0, "description": "检测阈值"},
-                "min_scene_length": {"type": float, "default": 1.0, "description": "最小场景长度（秒）"},
-                "output": {"type": str, "description": "输出文件路径"},
-                "format": {"type": str, "default": "json", "choices": ["json", "csv", "txt"], "description": "输出格式"},
-                "adaptive": {"type": bool, "default": False, "description": "启用自适应阈值"},
-                "thumbnails": {"type": bool, "default": False, "description": "生成缩略图"}
-            }
-        )
-
-        # 分析结果
-        self.register_command(
-            name="analyze",
-            description="分析检测结果并生成统计信息",
-            required_args=["result_file"],
-            optional_args={
-                "output": {"type": str, "description": "统计输出文件路径"}
-            }
-        )
-
-        # 比较检测器
-        self.register_command(
-            name="compare",
-            description="比较不同检测器的效果",
-            required_args=["video_path"],
-            optional_args={
-                "thresholds": {"type": str, "default": "20,30,40", "description": "测试阈值列表（逗号分隔）"},
-                "output": {"type": str, "description": "比较结果输出文件"}
-            }
-        )
-
-    def _is_progressive_command(self, command: str) -> bool:
-        """判断是否需要进度报告的命令"""
-        # 批量操作和比较操作需要进度报告
-        return command in ["batch_detect", "compare"]
-
-    def _execute_with_progress(self, command: str, args: Dict[str, Any]) -> Any:
-        """执行带进度的命令"""
-        if command == "batch_detect":
-            return self._batch_detect_with_progress(args)
-        elif command == "compare":
-            return self._compare_with_progress(args)
-        else:
-            raise ValueError(f"Unknown progressive command: {command}")
-
-    def _execute_simple_command(self, command: str, args: Dict[str, Any]) -> Any:
-        """执行简单命令（不需要进度）"""
-        if command == "detect":
-            return self._detect_single_video(args)
-        elif command == "analyze":
-            return self._analyze_results(args)
-        else:
-            raise ValueError(f"Unknown command: {command}")
-
-    def _detect_single_video(self, args: Dict[str, Any]) -> dict:
-        """检测单个视频"""
-        config = self._create_config(args)
-        
-        result = self.service.detect_single_video(args["video_path"], config)
-        
-        # 保存结果（如果指定了输出路径）
-        if args.get("output"):
-            output_path = args["output"]
-            # 创建临时批量结果来使用保存功能
-            from .types import BatchDetectionResult
-            batch_result = BatchDetectionResult(
-                total_files=1,
-                processed_files=1 if result.success else 0,
-                failed_files=0 if result.success else 1,
-                total_scenes=result.total_scenes,
-                total_duration=result.total_duration,
-                average_scenes_per_video=result.total_scenes,
-                detection_time=result.detection_time,
-                results=[result] if result.success else [],
-                failed_list=[] if result.success else [{"filename": result.filename, "error": result.error}],
-                config=config
-            )
-            
-            self.service.save_results(batch_result, output_path)
-        
-        return asdict(result)
-
-    def _batch_detect_with_progress(self, args: Dict[str, Any]) -> dict:
-        """带进度的批量检测"""
-        config = self._create_config(args)
-        input_directory = args["input_directory"]
-        
-        # 先扫描文件数量
-        video_files = self.service._scan_video_files(input_directory)
-        
-        if not video_files:
-            return {
-                "total_files": 0,
-                "processed_files": 0,
-                "failed_files": 0,
-                "message": "No video files found in directory"
-            }
-        
-        # 使用进度任务
-        with self.create_task("批量场景检测", len(video_files)) as task:
-            def progress_callback(message: str):
-                # 从消息中提取进度信息
-                if "(" in message and "/" in message:
-                    # 提取 (x/y) 格式的进度
-                    try:
-                        progress_part = message.split("(")[1].split(")")[0]
-                        current, total = progress_part.split("/")
-                        task.update(int(current) - 1, message)
-                    except:
-                        task.update(message=message)
-                else:
-                    task.update(message=message)
-            
-            # 执行批量检测
-            result = self.service.batch_detect_scenes(
-                input_directory, config, progress_callback
-            )
-            
-            # 保存结果（如果指定了输出路径）
-            if args.get("output"):
-                self.service.save_results(result, args["output"])
-            
-            task.finish(f"批量检测完成: {result.processed_files} 成功, {result.failed_files} 失败")
-            
-        return asdict(result)
-
-    def _compare_with_progress(self, args: Dict[str, Any]) -> dict:
-        """带进度的检测器比较"""
-        video_path = args["video_path"]
-        thresholds_str = args.get("thresholds", "20,30,40")
-        
-        try:
-            thresholds = [float(t.strip()) for t in thresholds_str.split(",")]
-        except ValueError:
-            raise ValueError("Invalid thresholds format. Use comma-separated numbers like '20,30,40'")
-        
-        detectors = ["content", "threshold", "adaptive"]
-        total_tests = len(detectors) * len(thresholds)
-        
-        with self.create_task("比较检测器", total_tests) as task:
-            results = []
-            test_count = 0
-            
-            for detector in detectors:
-                for threshold in thresholds:
-                    test_count += 1
-                    task.update(test_count - 1, f"测试 {detector} 检测器 (阈值: {threshold})")
-                    
-                    config = BatchDetectionConfig(
-                        detector_type=DetectorType(detector),
-                        threshold=threshold,
-                        min_scene_length=1.0
-                    )
-                    
-                    result = self.service.detect_single_video(video_path, config)
-                    
-                    results.append({
-                        "detector": detector,
-                        "threshold": threshold,
-                        "success": result.success,
-                        "total_scenes": result.total_scenes,
-                        "detection_time": result.detection_time,
-                        "error": result.error
-                    })
-            
-            task.finish("检测器比较完成")
-        
-        # 分析比较结果
-        comparison_result = {
-            "video_path": video_path,
-            "total_tests": total_tests,
-            "results": results,
-            "summary": self._analyze_comparison(results)
-        }
-        
-        # 保存比较结果
-        if args.get("output"):
-            import json
-            with open(args["output"], 'w', encoding='utf-8') as f:
-                json.dump(comparison_result, f, indent=2, ensure_ascii=False)
-        
-        return comparison_result
-
-    def _analyze_results(self, args: Dict[str, Any]) -> dict:
-        """分析检测结果"""
-        result_file = args["result_file"]
-        
-        try:
-            import json
-            with open(result_file, 'r', encoding='utf-8') as f:
-                data = json.load(f)
-            
-            # 重构批量结果对象
-            from .types import BatchDetectionResult, VideoSceneResult, SceneInfo
-            
-            results = []
-            for video_data in data.get("results", []):
-                scenes = [
-                    SceneInfo(
-                        index=scene["index"],
-                        start_time=scene["start_time"],
-                        end_time=scene["end_time"],
-                        duration=scene["duration"],
-                        confidence=scene.get("confidence", 1.0)
-                    )
-                    for scene in video_data.get("scenes", [])
-                ]
-                
-                video_result = VideoSceneResult(
-                    video_path=video_data["video_path"],
-                    filename=video_data["filename"],
-                    success=True,
-                    total_scenes=video_data["total_scenes"],
-                    total_duration=video_data["total_duration"],
-                    scenes=scenes,
-                    detection_time=video_data["detection_time"],
-                    detector_type=data.get("config", {}).get("detector_type", "unknown"),
-                    threshold=data.get("config", {}).get("threshold", 0.0)
-                )
-                results.append(video_result)
-            
-            # 创建批量结果对象
-            batch_result = BatchDetectionResult(
-                total_files=data["summary"]["total_files"],
-                processed_files=data["summary"]["processed_files"],
-                failed_files=data["summary"]["failed_files"],
-                total_scenes=data["summary"]["total_scenes"],
-                total_duration=data["summary"]["total_duration"],
-                average_scenes_per_video=data["summary"]["average_scenes_per_video"],
-                detection_time=data["summary"]["detection_time"],
-                results=results,
-                failed_list=data.get("failed_files", []),
-                config=BatchDetectionConfig()  # 简化配置
-            )
-            
-            # 计算统计信息
-            stats = self.service.calculate_stats(batch_result)
-            
-            analysis_result = {
-                "source_file": result_file,
-                "statistics": asdict(stats),
-                "summary": data["summary"]
-            }
-            
-            # 保存分析结果
-            if args.get("output"):
-                with open(args["output"], 'w', encoding='utf-8') as f:
-                    json.dump(analysis_result, f, indent=2, ensure_ascii=False)
-            
-            return analysis_result
-            
-        except Exception as e:
-            raise ValueError(f"Failed to analyze results: {e}")
-
-    def _analyze_comparison(self, results: list) -> dict:
-        """分析比较结果"""
-        successful_results = [r for r in results if r["success"]]
-        
-        if not successful_results:
-            return {"message": "No successful detections"}
-        
-        # 按检测器分组
-        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["total_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_tests": len(successful_results),
-            "detector_analysis": detector_analysis,
-            "best_detector": best_detector,
-            "recommendation": f"推荐使用 {best_detector} 检测器"
-        }
-
-    def _create_config(self, args: Dict[str, Any]) -> BatchDetectionConfig:
-        """创建检测配置"""
-        return BatchDetectionConfig(
-            detector_type=DetectorType(args.get("detector", "content")),
-            threshold=args.get("threshold", 30.0),
-            min_scene_length=args.get("min_scene_length", 1.0),
-            adaptive_threshold=args.get("adaptive", False),
-            generate_thumbnails=args.get("thumbnails", False),
-            output_format=args.get("format", "json")
-        )
-
-def main():
-    """主函数"""
-    commander = SceneDetectionCommander()
-    commander.run()
-
-if __name__ == "__main__":
-    main()

python_core/services/scene_detection/detector.py

@@ -1,434 +0,0 @@
-#!/usr/bin/env python3
-"""
-场景检测服务
-"""
-
-import os
-import time
-import json
-import csv
-from pathlib import Path
-from typing import List, Optional, Callable
-
-from .types import (
-    DetectorType, SceneInfo, VideoSceneResult, 
-    BatchDetectionConfig, BatchDetectionResult, DetectionStats
-)
-from python_core.utils.logger import logger
-
-class SceneDetectionService:
-    """场景检测服务"""
-    
-    def __init__(self):
-        self.supported_formats = {'.mp4', '.avi', '.mov', '.mkv', '.wmv', '.flv', '.webm', '.m4v'}
-    
-    def detect_single_video(self, video_path: str, config: BatchDetectionConfig) -> VideoSceneResult:
-        """检测单个视频的场景"""
-        start_time = time.time()
-        filename = os.path.basename(video_path)
-        
-        try:
-            # 获取场景变化点
-            scene_changes = self._detect_scene_changes(video_path, config)
-            
-            # 获取视频总时长
-            total_duration = self._get_video_duration(video_path)
-            
-            # 构建场景信息
-            scenes = []
-            for i in range(len(scene_changes) - 1):
-                start_time_scene = scene_changes[i]
-                end_time_scene = scene_changes[i + 1]
-                duration = end_time_scene - start_time_scene
-                
-                # 跳过太短的场景
-                if duration < config.min_scene_length:
-                    continue
-                
-                scene = SceneInfo(
-                    index=len(scenes),
-                    start_time=start_time_scene,
-                    end_time=end_time_scene,
-                    duration=duration,
-                    confidence=1.0,  # 简化版本，固定置信度
-                    frame_count=int(duration * 25)  # 假设25fps
-                )
-                scenes.append(scene)
-            
-            detection_time = time.time() - start_time
-            
-            return VideoSceneResult(
-                video_path=video_path,
-                filename=filename,
-                success=True,
-                total_scenes=len(scenes),
-                total_duration=total_duration,
-                scenes=scenes,
-                detection_time=detection_time,
-                detector_type=config.detector_type.value,
-                threshold=config.threshold
-            )
-            
-        except Exception as e:
-            detection_time = time.time() - start_time
-            logger.error(f"Failed to detect scenes in {filename}: {e}")
-            
-            return VideoSceneResult(
-                video_path=video_path,
-                filename=filename,
-                success=False,
-                total_scenes=0,
-                total_duration=0.0,
-                scenes=[],
-                detection_time=detection_time,
-                detector_type=config.detector_type.value,
-                threshold=config.threshold,
-                error=str(e)
-            )
-    
-    def batch_detect_scenes(self, input_directory: str, config: BatchDetectionConfig,
-                           progress_callback: Optional[Callable] = None) -> BatchDetectionResult:
-        """批量检测场景"""
-        start_time = time.time()
-        
-        # 扫描视频文件
-        video_files = self._scan_video_files(input_directory)
-        
-        if not video_files:
-            return BatchDetectionResult(
-                total_files=0,
-                processed_files=0,
-                failed_files=0,
-                total_scenes=0,
-                total_duration=0.0,
-                average_scenes_per_video=0.0,
-                detection_time=0.0,
-                results=[],
-                failed_list=[],
-                config=config
-            )
-        
-        results = []
-        failed_list = []
-        total_scenes = 0
-        total_duration = 0.0
-        
-        for i, video_path in enumerate(video_files):
-            filename = os.path.basename(video_path)
-            
-            # 报告进度
-            if progress_callback:
-                progress_callback(f"检测场景: {filename} ({i+1}/{len(video_files)})")
-            
-            # 检测单个视频
-            result = self.detect_single_video(video_path, config)
-            
-            if result.success:
-                results.append(result)
-                total_scenes += result.total_scenes
-                total_duration += result.total_duration
-            else:
-                failed_list.append({
-                    'filename': filename,
-                    'path': video_path,
-                    'error': result.error
-                })
-        
-        detection_time = time.time() - start_time
-        processed_files = len(results)
-        failed_files = len(failed_list)
-        
-        average_scenes = total_scenes / processed_files if processed_files > 0 else 0.0
-        
-        return BatchDetectionResult(
-            total_files=len(video_files),
-            processed_files=processed_files,
-            failed_files=failed_files,
-            total_scenes=total_scenes,
-            total_duration=total_duration,
-            average_scenes_per_video=average_scenes,
-            detection_time=detection_time,
-            results=results,
-            failed_list=failed_list,
-            config=config
-        )
-    
-    def _detect_scene_changes(self, video_path: str, config: BatchDetectionConfig) -> List[float]:
-        """检测场景变化点"""
-        try:
-            # 优先使用PySceneDetect
-            return self._detect_with_pyscenedetect(video_path, config)
-        except Exception:
-            try:
-                # 回退到OpenCV
-                return self._detect_with_opencv(video_path, config)
-            except Exception:
-                # 最后回退：返回整个视频作为一个场景
-                duration = self._get_video_duration(video_path)
-                return [0.0, duration]
-    
-    def _detect_with_pyscenedetect(self, video_path: str, config: BatchDetectionConfig) -> List[float]:
-        """使用PySceneDetect检测场景"""
-        try:
-            from scenedetect import VideoManager, SceneManager
-            from scenedetect.detectors import ContentDetector, ThresholdDetector
-        except ImportError:
-            raise Exception("PySceneDetect not available")
-        
-        video_manager = VideoManager([video_path])
-        scene_manager = SceneManager()
-        
-        # 根据配置选择检测器
-        if config.detector_type == DetectorType.CONTENT:
-            scene_manager.add_detector(ContentDetector(threshold=config.threshold))
-        elif config.detector_type == DetectorType.THRESHOLD:
-            scene_manager.add_detector(ThresholdDetector(threshold=config.threshold))
-        else:  # ADAPTIVE
-            # 自适应：同时使用两种检测器
-            scene_manager.add_detector(ContentDetector(threshold=config.threshold))
-            scene_manager.add_detector(ThresholdDetector(threshold=config.threshold * 0.8))
-        
-        video_manager.start()
-        scene_manager.detect_scenes(frame_source=video_manager)
-        scene_list = scene_manager.get_scene_list()
-        
-        # 提取场景时间点
-        scene_changes = [0.0]
-        for scene in scene_list:
-            start_time = scene[0].get_seconds()
-            end_time = scene[1].get_seconds()
-            
-            if start_time > 0 and start_time not in scene_changes:
-                scene_changes.append(start_time)
-            if end_time not in scene_changes:
-                scene_changes.append(end_time)
-        
-        video_manager.release()
-        return sorted(scene_changes)
-    
-    def _detect_with_opencv(self, video_path: str, config: BatchDetectionConfig) -> List[float]:
-        """使用OpenCV检测场景"""
-        try:
-            import cv2
-            import numpy as np
-        except ImportError:
-            raise Exception("OpenCV not available")
-        
-        cap = cv2.VideoCapture(video_path)
-        fps = cap.get(cv2.CAP_PROP_FPS)
-        
-        if fps <= 0:
-            cap.release()
-            raise Exception(f"Invalid fps ({fps}) for video {video_path}")
-        
-        scene_changes = [0.0]
-        prev_frame = None
-        frame_count = 0
-        frame_skip = max(1, int(fps / 2))  # 每秒检测2次
-        
-        while True:
-            ret, frame = cap.read()
-            if not ret:
-                break
-            
-            if frame_count % frame_skip == 0:
-                gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
-                gray = cv2.resize(gray, (320, 240))
-                
-                if prev_frame is not None:
-                    diff = cv2.absdiff(prev_frame, gray)
-                    mean_diff = np.mean(diff)
-                    
-                    if mean_diff > config.threshold:
-                        timestamp = frame_count / fps
-                        if not scene_changes or timestamp - scene_changes[-1] > config.min_scene_length:
-                            scene_changes.append(timestamp)
-                
-                prev_frame = gray
-            
-            frame_count += 1
-        
-        # 添加视频结束时间
-        duration = frame_count / fps if fps > 0 else 0
-        if duration > 0 and (not scene_changes or duration - scene_changes[-1] > 0.5):
-            scene_changes.append(duration)
-        
-        cap.release()
-        return scene_changes
-    
-    def _get_video_duration(self, video_path: str) -> float:
-        """获取视频时长"""
-        try:
-            import cv2
-            cap = cv2.VideoCapture(video_path)
-            fps = cap.get(cv2.CAP_PROP_FPS)
-            frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
-            cap.release()
-            
-            if fps > 0:
-                return frame_count / fps
-            return 0.0
-        except Exception:
-            return 0.0
-    
-    def _scan_video_files(self, directory: str) -> List[str]:
-        """扫描目录中的视频文件"""
-        video_files = []
-        
-        for root, _, files in os.walk(directory):
-            for file in files:
-                file_ext = os.path.splitext(file)[1].lower()
-                if file_ext in self.supported_formats:
-                    video_files.append(os.path.join(root, file))
-        
-        return sorted(video_files)
-    
-    def save_results(self, result: BatchDetectionResult, output_path: str) -> bool:
-        """保存检测结果"""
-        try:
-            output_path = Path(output_path)
-            output_path.parent.mkdir(parents=True, exist_ok=True)
-            
-            if result.config.output_format == "json":
-                self._save_json_results(result, output_path)
-            elif result.config.output_format == "csv":
-                self._save_csv_results(result, output_path)
-            elif result.config.output_format == "txt":
-                self._save_txt_results(result, output_path)
-            else:
-                raise ValueError(f"Unsupported output format: {result.config.output_format}")
-            
-            logger.info(f"Results saved to {output_path}")
-            return True
-            
-        except Exception as e:
-            logger.error(f"Failed to save results: {e}")
-            return False
-    
-    def _save_json_results(self, result: BatchDetectionResult, output_path: Path):
-        """保存JSON格式结果"""
-        # 转换为可序列化的字典
-        data = {
-            "summary": {
-                "total_files": result.total_files,
-                "processed_files": result.processed_files,
-                "failed_files": result.failed_files,
-                "total_scenes": result.total_scenes,
-                "total_duration": result.total_duration,
-                "average_scenes_per_video": result.average_scenes_per_video,
-                "detection_time": result.detection_time
-            },
-            "config": {
-                "detector_type": result.config.detector_type.value,
-                "threshold": result.config.threshold,
-                "min_scene_length": result.config.min_scene_length
-            },
-            "results": [],
-            "failed_files": result.failed_list
-        }
-        
-        for video_result in result.results:
-            video_data = {
-                "filename": video_result.filename,
-                "video_path": video_result.video_path,
-                "total_scenes": video_result.total_scenes,
-                "total_duration": video_result.total_duration,
-                "detection_time": video_result.detection_time,
-                "scenes": [
-                    {
-                        "index": scene.index,
-                        "start_time": scene.start_time,
-                        "end_time": scene.end_time,
-                        "duration": scene.duration,
-                        "confidence": scene.confidence
-                    }
-                    for scene in video_result.scenes
-                ]
-            }
-            data["results"].append(video_data)
-        
-        with open(output_path, 'w', encoding='utf-8') as f:
-            json.dump(data, f, indent=2, ensure_ascii=False)
-    
-    def _save_csv_results(self, result: BatchDetectionResult, output_path: Path):
-        """保存CSV格式结果"""
-        with open(output_path, 'w', newline='', encoding='utf-8') as f:
-            writer = csv.writer(f)
-            
-            # 写入表头
-            writer.writerow([
-                'filename', 'video_path', 'scene_index', 'start_time', 
-                'end_time', 'duration', 'confidence'
-            ])
-            
-            # 写入数据
-            for video_result in result.results:
-                for scene in video_result.scenes:
-                    writer.writerow([
-                        video_result.filename,
-                        video_result.video_path,
-                        scene.index,
-                        scene.start_time,
-                        scene.end_time,
-                        scene.duration,
-                        scene.confidence
-                    ])
-    
-    def _save_txt_results(self, result: BatchDetectionResult, output_path: Path):
-        """保存文本格式结果"""
-        with open(output_path, 'w', encoding='utf-8') as f:
-            f.write("批量场景检测结果\n")
-            f.write("=" * 50 + "\n\n")
-            
-            f.write(f"总文件数: {result.total_files}\n")
-            f.write(f"处理成功: {result.processed_files}\n")
-            f.write(f"处理失败: {result.failed_files}\n")
-            f.write(f"总场景数: {result.total_scenes}\n")
-            f.write(f"总时长: {result.total_duration:.2f}秒\n")
-            f.write(f"平均场景数: {result.average_scenes_per_video:.1f}\n")
-            f.write(f"检测耗时: {result.detection_time:.2f}秒\n\n")
-            
-            for video_result in result.results:
-                f.write(f"文件: {video_result.filename}\n")
-                f.write(f"  场景数: {video_result.total_scenes}\n")
-                f.write(f"  时长: {video_result.total_duration:.2f}秒\n")
-                f.write(f"  检测时间: {video_result.detection_time:.2f}秒\n")
-                
-                for scene in video_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")
-    
-    def calculate_stats(self, result: BatchDetectionResult) -> DetectionStats:
-        """计算检测统计信息"""
-        if not result.results:
-            return DetectionStats(
-                total_videos=0,
-                total_scenes=0,
-                total_duration=0.0,
-                average_duration_per_scene=0.0,
-                shortest_scene=0.0,
-                longest_scene=0.0,
-                most_scenes_video="",
-                least_scenes_video=""
-            )
-        
-        all_scenes = []
-        for video_result in result.results:
-            all_scenes.extend(video_result.scenes)
-        
-        scene_durations = [scene.duration for scene in all_scenes]
-        
-        # 找出场景最多和最少的视频
-        most_scenes_video = max(result.results, key=lambda x: x.total_scenes)
-        least_scenes_video = min(result.results, key=lambda x: x.total_scenes)
-        
-        return DetectionStats(
-            total_videos=len(result.results),
-            total_scenes=len(all_scenes),
-            total_duration=result.total_duration,
-            average_duration_per_scene=sum(scene_durations) / len(scene_durations) if scene_durations else 0.0,
-            shortest_scene=min(scene_durations) if scene_durations else 0.0,
-            longest_scene=max(scene_durations) if scene_durations else 0.0,
-            most_scenes_video=most_scenes_video.filename,
-            least_scenes_video=least_scenes_video.filename
-        )

python_core/services/scene_detection/types.py

@@ -1,76 +0,0 @@
-#!/usr/bin/env python3
-"""
-场景检测相关的数据类型定义
-"""
-
-from dataclasses import dataclass
-from typing import List, Optional
-from enum import Enum
-
-class DetectorType(Enum):
-    """检测器类型"""
-    CONTENT = "content"
-    THRESHOLD = "threshold"
-    ADAPTIVE = "adaptive"
-
-@dataclass
-class SceneInfo:
-    """场景信息"""
-    index: int
-    start_time: float
-    end_time: float
-    duration: float
-    confidence: float = 0.0
-    frame_count: int = 0
-    thumbnail_path: Optional[str] = None
-
-@dataclass
-class VideoSceneResult:
-    """单个视频的场景检测结果"""
-    video_path: str
-    filename: str
-    success: bool
-    total_scenes: int
-    total_duration: float
-    scenes: List[SceneInfo]
-    detection_time: float
-    detector_type: str
-    threshold: float
-    error: Optional[str] = None
-
-@dataclass
-class BatchDetectionConfig:
-    """批量检测配置"""
-    detector_type: DetectorType = DetectorType.CONTENT
-    threshold: float = 30.0
-    min_scene_length: float = 1.0
-    adaptive_threshold: bool = False
-    generate_thumbnails: bool = False
-    output_format: str = "json"  # json, csv, txt
-    save_scenes: bool = False
-
-@dataclass
-class BatchDetectionResult:
-    """批量检测结果"""
-    total_files: int
-    processed_files: int
-    failed_files: int
-    total_scenes: int
-    total_duration: float
-    average_scenes_per_video: float
-    detection_time: float
-    results: List[VideoSceneResult]
-    failed_list: List[dict]
-    config: BatchDetectionConfig
-
-@dataclass
-class DetectionStats:
-    """检测统计信息"""
-    total_videos: int
-    total_scenes: int
-    total_duration: float
-    average_duration_per_scene: float
-    shortest_scene: float
-    longest_scene: float
-    most_scenes_video: str
-    least_scenes_video: str

python_core/services/video_splitter/__init__.py

@@ -1,91 +0,0 @@
-#!/usr/bin/env python3
-"""
-视频拆分服务模块
-
-这个模块提供了基于PySceneDetect的视频场景检测和拆分功能。
-
-主要组件:
-- types: 类型定义和数据结构
-- detectors: 场景检测器实现
-- validators: 视频验证器实现
-- service: 核心服务实现
-- cli: 命令行接口
-
-使用示例:
-    from python_core.services.video_splitter import VideoSplitterService, DetectionConfig
-    
-    service = VideoSplitterService()
-    config = DetectionConfig(threshold=30.0)
-    result = service.analyze_video("video.mp4", config)
-"""
-
-from .types import (
-    SceneInfo,
-    AnalysisResult,
-    DetectionConfig,
-    DetectorType,
-    ServiceError,
-    DependencyError,
-    ValidationError,
-    SceneDetector,
-    VideoValidator
-)
-
-from .detectors import PySceneDetectDetector
-from .validators import BasicVideoValidator
-from .service import VideoSplitterService
-from .cli import VideoSplitterCommander
-
-__version__ = "1.0.0"
-__author__ = "Video Splitter Team"
-
-__all__ = [
-    # 类型和异常
-    "SceneInfo",
-    "AnalysisResult", 
-    "DetectionConfig",
-    "DetectorType",
-    "ServiceError",
-    "DependencyError",
-    "ValidationError",
-    "SceneDetector",
-    "VideoValidator",
-    
-    # 实现类
-    "PySceneDetectDetector",
-    "BasicVideoValidator",
-    "VideoSplitterService",
-    "VideoSplitterCommander",
-]
-
-# 便捷函数
-def create_service(output_base_dir: str = None) -> VideoSplitterService:
-    """
-    创建视频拆分服务实例
-    
-    Args:
-        output_base_dir: 输出基础目录
-        
-    Returns:
-        VideoSplitterService实例
-    """
-    return VideoSplitterService(output_base_dir=output_base_dir)
-
-def analyze_video(video_path: str, threshold: float = 30.0, detector_type: str = "content") -> AnalysisResult:
-    """
-    快速分析视频的便捷函数
-    
-    Args:
-        video_path: 视频路径
-        threshold: 检测阈值
-        detector_type: 检测器类型
-        
-    Returns:
-        分析结果
-    """
-    service = create_service()
-    config = DetectionConfig(
-        threshold=threshold,
-        detector_type=DetectorType(detector_type)
-    )
-    return service.analyze_video(video_path, config)

python_core/services/video_splitter/__main__.py

@@ -1,11 +0,0 @@
-#!/usr/bin/env python3
-"""
-视频拆分服务命令行入口点
-
-支持通过 python -m python_core.services.video_splitter 运行
-"""
-
-from .cli import main
-
-if __name__ == "__main__":
-    main()

python_core/services/video_splitter/cli.py

@@ -1,101 +0,0 @@
-#!/usr/bin/env python3
-"""
-视频拆分服务命令行接口
-"""
-
-from typing import Dict, Any
-from dataclasses import asdict
-
-from .types import DetectionConfig, DetectorType
-from .service import VideoSplitterService
-from python_core.utils.commander import JSONRPCCommander
-
-
-class VideoSplitterCommander(JSONRPCCommander):
-    """视频拆分服务命令行接口"""
-
-    def __init__(self):
-        self.service = None
-        super().__init__("video_splitter")
-
-    def _register_commands(self) -> None:
-        """注册命令"""
-        # 注册analyze命令
-        self.register_command(
-            name="analyze",
-            description="分析视频场景",
-            required_args=["video_path"],
-            optional_args={
-                "threshold": {"type": float, "default": 30.0, "description": "检测阈值"},
-                "detector": {"type": str, "default": "content", "choices": ["content", "threshold"], "description": "检测器类型"},
-                "min-scene-length": {"type": float, "default": 1.0, "description": "最小场景长度（秒）"},
-                "output-base": {"type": str, "default": None, "description": "输出基础目录"}
-            }
-        )
-
-        # 注册detect_scenes命令
-        self.register_command(
-            name="detect_scenes",
-            description="检测视频场景（仅返回场景信息）",
-            required_args=["video_path"],
-            optional_args={
-                "threshold": {"type": float, "default": 30.0, "description": "检测阈值"},
-                "detector": {"type": str, "default": "content", "choices": ["content", "threshold"], "description": "检测器类型"},
-                "min-scene-length": {"type": float, "default": 1.0, "description": "最小场景长度（秒）"},
-                "output-base": {"type": str, "default": None, "description": "输出基础目录"}
-            }
-        )
-
-    def _setup_service(self, output_base: str = None) -> None:
-        """设置服务"""
-        if self.service is None:
-            self.service = VideoSplitterService(output_base_dir=output_base)
-
-    def execute_command(self, command: str, args: Dict[str, Any]) -> Any:
-        """执行命令"""
-        # 设置服务
-        self._setup_service(args.get("output_base"))
-
-        # 创建配置
-        config = DetectionConfig(
-            threshold=args.get("threshold", 30.0),
-            detector_type=DetectorType(args.get("detector", "content")),
-            min_scene_length=args.get("min_scene_length", 1.0)
-        )
-
-        video_path = args["video_path"]
-
-        if command == "analyze":
-            # 完整的视频分析
-            result = self.service.analyze_video(video_path, config)
-            return result.to_dict()
-
-        elif command == "detect_scenes":
-            # 仅检测场景
-            result = self.service.analyze_video(video_path, config)
-
-            # 只返回场景信息
-            scenes_result = {
-                "success": result.success,
-                "video_path": result.video_path,
-                "total_scenes": result.total_scenes,
-                "scenes": [asdict(scene) for scene in result.scenes],
-                "detection_settings": asdict(config),
-                "detection_time": result.analysis_time
-            }
-
-            if not result.success:
-                scenes_result["error"] = result.error
-
-            return scenes_result
-
-        else:
-            raise ValueError(f"Unknown command: {command}")
-
-def main():
-    """主函数"""
-    commander = VideoSplitterCommander()
-    commander.run()
-
-if __name__ == "__main__":
-    main()

python_core/services/video_splitter/detectors.py

@@ -1,104 +0,0 @@
-#!/usr/bin/env python3
-"""
-视频场景检测器实现
-"""
-
-from contextlib import contextmanager
-from typing import List
-
-from scenedetect import VideoManager, SceneManager
-from scenedetect.detectors import ContentDetector, ThresholdDetector
-
-from .types import SceneInfo, DetectionConfig, DetectorType
-from python_core.utils.logger import logger
-
-class PySceneDetectDetector:
-    """PySceneDetect场景检测器实现"""
-
-    def __init__(self):
-        logger.info("PySceneDetect detector initialized")
-    
-    @contextmanager
-    def _video_manager(self, video_path: str):
-        """视频管理器上下文管理器"""
-        video_manager = VideoManager([video_path])
-        try:
-            video_manager.start()
-            yield video_manager
-        finally:
-            video_manager.release()
-    
-    def detect_scenes(self, video_path: str, config: DetectionConfig) -> List[SceneInfo]:
-        """检测场景"""
-        logger.info(f"Detecting scenes: {video_path}, threshold: {config.threshold}")
-        
-        with self._video_manager(video_path) as video_manager:
-            scene_manager = SceneManager()
-            
-            # 添加检测器
-            if config.detector_type == DetectorType.CONTENT:
-                scene_manager.add_detector(ContentDetector(threshold=config.threshold))
-            else:
-                scene_manager.add_detector(ThresholdDetector(threshold=config.threshold))
-            
-            # 执行检测
-            scene_manager.detect_scenes(frame_source=video_manager)
-            scene_list = scene_manager.get_scene_list()
-            
-            # 转换结果
-            scenes = self._convert_scenes(scene_list, config)
-            
-            if not scenes:
-                # 创建单个场景
-                scenes = self._create_single_scene(video_manager)
-            
-            logger.info(f"Detected {len(scenes)} scenes")
-            return scenes
-    
-    def _convert_scenes(self, scene_list: List, config: DetectionConfig) -> List[SceneInfo]:
-        """转换场景列表"""
-        scenes = []
-        for i, (start_time, end_time) in enumerate(scene_list):
-            duration = end_time.get_seconds() - start_time.get_seconds()
-            
-            # 过滤太短的场景
-            if duration < config.min_scene_length:
-                logger.debug(f"Skipping short scene {i+1}: {duration:.2f}s")
-                continue
-            
-            scene_info = SceneInfo(
-                scene_number=len(scenes) + 1,  # 重新编号
-                start_time=start_time.get_seconds(),
-                end_time=end_time.get_seconds(),
-                duration=duration,
-                start_frame=start_time.get_frames(),
-                end_frame=end_time.get_frames()
-            )
-            scenes.append(scene_info)
-        
-        return scenes
-    
-    def _create_single_scene(self, video_manager) -> List[SceneInfo]:
-        """创建单个场景"""
-        try:
-            duration_info = video_manager.get_duration()
-            fps = video_manager.get_framerate()
-            
-            if isinstance(duration_info, tuple):
-                total_frames, fps = duration_info
-                total_duration = total_frames / fps if fps > 0 else 0
-            else:
-                total_duration = duration_info.get_seconds() if hasattr(duration_info, 'get_seconds') else float(duration_info)
-                total_frames = int(total_duration * fps) if fps > 0 else 0
-            
-            return [SceneInfo(
-                scene_number=1,
-                start_time=0.0,
-                end_time=total_duration,
-                duration=total_duration,
-                start_frame=0,
-                end_frame=total_frames
-            )]
-        except Exception as e:
-            logger.warning(f"Failed to create single scene: {e}")
-            return []

python_core/services/video_splitter/service.py

@@ -1,79 +0,0 @@
-#!/usr/bin/env python3
-"""
-视频拆分服务核心实现
-"""
-
-from pathlib import Path
-from typing import Optional
-
-from .types import SceneDetector, VideoValidator, AnalysisResult, DetectionConfig
-from .detectors import PySceneDetectDetector
-from .validators import BasicVideoValidator
-from python_core.utils.command_utils import PerformanceUtils
-from python_core.utils.logger import logger
-
-
-
-class VideoSplitterService:
-    """高质量的视频拆分服务"""
-    
-    def __init__(self, 
-                 detector: Optional[SceneDetector] = None,
-                 validator: Optional[VideoValidator] = None,
-                 output_base_dir: Optional[str] = None):
-        """
-        初始化服务
-        
-        Args:
-            detector: 场景检测器
-            validator: 视频验证器
-            output_base_dir: 输出基础目录
-        """
-        self.detector = detector or PySceneDetectDetector()
-        self.validator = validator or BasicVideoValidator()
-        self.output_base_dir = Path(output_base_dir) if output_base_dir else Path("./video_splits")
-        self.output_base_dir.mkdir(parents=True, exist_ok=True)
-    
-    def analyze_video(self, video_path: str, config: Optional[DetectionConfig] = None) -> AnalysisResult:
-        """
-        分析视频
-        
-        Args:
-            video_path: 视频路径
-            config: 检测配置
-            
-        Returns:
-            分析结果
-        """
-        config = config or DetectionConfig()
-        
-        try:
-            # 验证输入
-            self.validator.validate(video_path)
-            
-            # 执行检测
-            scenes, execution_time = PerformanceUtils.time_operation(
-                self.detector.detect_scenes, video_path, config
-            )
-            
-            # 计算统计信息
-            total_duration = sum(scene.duration for scene in scenes)
-            average_duration = total_duration / len(scenes) if scenes else 0
-            
-            return AnalysisResult(
-                success=True,
-                video_path=video_path,
-                total_scenes=len(scenes),
-                total_duration=total_duration,
-                average_scene_duration=average_duration,
-                scenes=scenes,
-                analysis_time=execution_time
-            )
-            
-        except Exception as e:
-            logger.error(f"Video analysis failed: {e}")
-            return AnalysisResult(
-                success=False,
-                video_path=video_path,
-                error=str(e)
-            )

python_core/services/video_splitter/types.py

@@ -1,100 +0,0 @@
-#!/usr/bin/env python3
-"""
-视频拆分服务的类型定义和数据结构
-"""
-
-from abc import ABC, abstractmethod
-from pathlib import Path
-from typing import List, Dict, Optional, Protocol, Union, Any
-from dataclasses import dataclass, asdict, field
-from datetime import datetime
-from enum import Enum
-
-# 类型定义
-class DetectorType(Enum):
-    """检测器类型枚举"""
-    CONTENT = "content"
-    THRESHOLD = "threshold"
-
-class ServiceError(Exception):
-    """服务基础异常"""
-    def __init__(self, message: str, error_code: str = "UNKNOWN_ERROR"):
-        super().__init__(message)
-        self.error_code = error_code
-        self.message = message
-
-class DependencyError(ServiceError):
-    """依赖缺失异常"""
-    def __init__(self, dependency: str):
-        super().__init__(f"Required dependency not available: {dependency}", "DEPENDENCY_ERROR")
-
-class ValidationError(ServiceError):
-    """验证错误异常"""
-    def __init__(self, message: str):
-        super().__init__(message, "VALIDATION_ERROR")
-
-@dataclass(frozen=True)
-class SceneInfo:
-    """场景信息 - 不可变数据类"""
-    scene_number: int
-    start_time: float
-    end_time: float
-    duration: float
-    start_frame: int
-    end_frame: int
-    
-    def __post_init__(self):
-        """数据验证"""
-        if self.scene_number <= 0:
-            raise ValidationError("Scene number must be positive")
-        if self.start_time < 0 or self.end_time < 0:
-            raise ValidationError("Time values must be non-negative")
-        if self.start_time >= self.end_time:
-            raise ValidationError("Start time must be less than end time")
-        if abs(self.duration - (self.end_time - self.start_time)) > 0.01:
-            raise ValidationError("Duration must match time difference")
-
-@dataclass
-class AnalysisResult:
-    """分析结果"""
-    success: bool
-    video_path: str
-    total_scenes: int = 0
-    total_duration: float = 0.0
-    average_scene_duration: float = 0.0
-    scenes: List[SceneInfo] = field(default_factory=list)
-    analysis_time: float = 0.0
-    error: Optional[str] = None
-    
-    def to_dict(self) -> Dict[str, Any]:
-        """转换为字典"""
-        result = asdict(self)
-        result['scenes'] = [asdict(scene) for scene in self.scenes]
-        return result
-
-@dataclass
-class DetectionConfig:
-    """检测配置"""
-    threshold: float = 30.0
-    detector_type: DetectorType = DetectorType.CONTENT
-    min_scene_length: float = 1.0  # 最小场景长度（秒）
-    
-    def __post_init__(self):
-        """配置验证"""
-        if not 0 < self.threshold <= 100:
-            raise ValidationError("Threshold must be between 0 and 100")
-        if self.min_scene_length < 0:
-            raise ValidationError("Minimum scene length must be non-negative")
-
-# 协议定义
-class SceneDetector(Protocol):
-    """场景检测器协议"""
-    def detect_scenes(self, video_path: str, config: DetectionConfig) -> List[SceneInfo]:
-        """检测场景"""
-        ...
-
-class VideoValidator(Protocol):
-    """视频验证器协议"""
-    def validate(self, video_path: str) -> bool:
-        """验证视频文件"""
-        ...

python_core/services/video_splitter/validators.py

@@ -1,37 +0,0 @@
-#!/usr/bin/env python3
-"""
-视频验证器实现
-"""
-
-import logging
-from pathlib import Path
-from .types import ValidationError
-
-logger = logging.getLogger(__name__)
-
-class BasicVideoValidator:
-    """基础视频验证器"""
-    
-    SUPPORTED_EXTENSIONS = {'.mp4', '.avi', '.mov', '.mkv', '.wmv', '.flv', '.webm'}
-    
-    def validate(self, video_path: str) -> bool:
-        """验证视频文件"""
-        path = Path(video_path)
-        
-        # 检查文件存在
-        if not path.exists():
-            raise ValidationError(f"Video file not found: {video_path}")
-        
-        # 检查是否为文件
-        if not path.is_file():
-            raise ValidationError(f"Path is not a file: {video_path}")
-        
-        # 检查扩展名
-        if path.suffix.lower() not in self.SUPPORTED_EXTENSIONS:
-            logger.warning(f"Unsupported video extension: {path.suffix}")
-        
-        # 检查文件大小
-        if path.stat().st_size == 0:
-            raise ValidationError(f"Video file is empty: {video_path}")
-        
-        return True

python_core/storage/__init__.py

@@ -3,15 +3,4 @@
 存储层模块
 提供统一的存储接口，支持多种存储后端
 """
-
+__all__ = []
-from .base import StorageInterface, StorageConfig
-from .json_storage import JSONStorage
-from .factory import StorageFactory, get_storage
-
-__all__ = [
-    "StorageInterface",
-    "StorageConfig",
-    "JSONStorage",
-    "StorageFactory",
-    "get_storage"
-]

python_core/storage/base.py

@@ -1,199 +0,0 @@
-#!/usr/bin/env python3
-"""
-存储接口基类
-"""
-
-from abc import ABC, abstractmethod
-from typing import Any, List, Dict, Optional
-from dataclasses import dataclass
-from enum import Enum
-
-class StorageType(Enum):
-    """存储类型"""
-    JSON = "json"
-    DATABASE = "database"
-    MONGODB = "mongodb"
-    REDIS = "redis"
-
-@dataclass
-class StorageConfig:
-    """存储配置"""
-    storage_type: StorageType
-    connection_string: Optional[str] = None
-    base_path: Optional[str] = None
-    database_name: Optional[str] = None
-    table_prefix: Optional[str] = None
-    
-    # JSON存储配置
-    json_indent: int = 2
-    json_ensure_ascii: bool = False
-    
-    # 数据库配置
-    pool_size: int = 10
-    max_overflow: int = 20
-    
-    # MongoDB配置
-    collection_prefix: str = ""
-    
-    # Redis配置
-    redis_db: int = 0
-    redis_expire: Optional[int] = None
-
-class StorageInterface(ABC):
-    """存储接口基类"""
-    
-    def __init__(self, config: StorageConfig):
-        self.config = config
-    
-    @abstractmethod
-    def save(self, collection: str, key: str, data: Any) -> bool:
-        """保存数据
-        
-        Args:
-            collection: 集合/表名
-            key: 数据键
-            data: 要保存的数据
-            
-        Returns:
-            bool: 保存是否成功
-        """
-        pass
-    
-    @abstractmethod
-    def load(self, collection: str, key: str) -> Any:
-        """加载数据
-        
-        Args:
-            collection: 集合/表名
-            key: 数据键
-            
-        Returns:
-            Any: 加载的数据，不存在时返回None
-        """
-        pass
-    
-    @abstractmethod
-    def delete(self, collection: str, key: str) -> bool:
-        """删除数据
-        
-        Args:
-            collection: 集合/表名
-            key: 数据键
-            
-        Returns:
-            bool: 删除是否成功
-        """
-        pass
-    
-    @abstractmethod
-    def exists(self, collection: str, key: str) -> bool:
-        """检查数据是否存在
-        
-        Args:
-            collection: 集合/表名
-            key: 数据键
-            
-        Returns:
-            bool: 数据是否存在
-        """
-        pass
-    
-    @abstractmethod
-    def list_keys(self, collection: str, pattern: str = "*") -> List[str]:
-        """列出所有键
-        
-        Args:
-            collection: 集合/表名
-            pattern: 键的模式匹配
-            
-        Returns:
-            List[str]: 键列表
-        """
-        pass
-    
-    @abstractmethod
-    def save_batch(self, collection: str, data: Dict[str, Any]) -> bool:
-        """批量保存数据
-        
-        Args:
-            collection: 集合/表名
-            data: 键值对数据
-            
-        Returns:
-            bool: 保存是否成功
-        """
-        pass
-    
-    @abstractmethod
-    def load_batch(self, collection: str, keys: List[str]) -> Dict[str, Any]:
-        """批量加载数据
-        
-        Args:
-            collection: 集合/表名
-            keys: 键列表
-            
-        Returns:
-            Dict[str, Any]: 键值对数据
-        """
-        pass
-    
-    @abstractmethod
-    def clear_collection(self, collection: str) -> bool:
-        """清空集合
-        
-        Args:
-            collection: 集合/表名
-            
-        Returns:
-            bool: 清空是否成功
-        """
-        pass
-    
-    @abstractmethod
-    def get_collections(self) -> List[str]:
-        """获取所有集合名称
-        
-        Returns:
-            List[str]: 集合名称列表
-        """
-        pass
-    
-    @abstractmethod
-    def get_stats(self, collection: str) -> Dict[str, Any]:
-        """获取集合统计信息
-        
-        Args:
-            collection: 集合/表名
-            
-        Returns:
-            Dict[str, Any]: 统计信息
-        """
-        pass
-    
-    def close(self):
-        """关闭存储连接"""
-        pass
-    
-    def __enter__(self):
-        """上下文管理器入口"""
-        return self
-    
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        """上下文管理器出口"""
-        self.close()
-
-class StorageException(Exception):
-    """存储异常基类"""
-    pass
-
-class StorageConnectionError(StorageException):
-    """存储连接错误"""
-    pass
-
-class StorageOperationError(StorageException):
-    """存储操作错误"""
-    pass
-
-class StorageNotFoundError(StorageException):
-    """存储数据未找到错误"""
-    pass

python_core/storage/factory.py

@@ -1,155 +0,0 @@
-#!/usr/bin/env python3
-"""
-存储工厂类
-"""
-
-from typing import Type, Dict
-from .base import StorageInterface, StorageConfig, StorageType, StorageException
-from .json_storage import JSONStorage
-from python_core.utils.logger import logger
-
-class StorageFactory:
-    """存储工厂类"""
-    
-    _storage_classes: Dict[StorageType, Type[StorageInterface]] = {
-        StorageType.JSON: JSONStorage,
-    }
-    
-    @classmethod
-    def register_storage(cls, storage_type: StorageType, storage_class: Type[StorageInterface]):
-        """注册存储实现
-        
-        Args:
-            storage_type: 存储类型
-            storage_class: 存储实现类
-        """
-        cls._storage_classes[storage_type] = storage_class
-        logger.info(f"Registered storage type: {storage_type.value}")
-    
-    @classmethod
-    def create_storage(cls, config: StorageConfig) -> StorageInterface:
-        """创建存储实例
-        
-        Args:
-            config: 存储配置
-            
-        Returns:
-            StorageInterface: 存储实例
-            
-        Raises:
-            StorageException: 当存储类型不支持时
-        """
-        storage_type = config.storage_type
-        
-        if storage_type not in cls._storage_classes:
-            available_types = list(cls._storage_classes.keys())
-            raise StorageException(
-                f"Unsupported storage type: {storage_type.value}. "
-                f"Available types: {[t.value for t in available_types]}"
-            )
-        
-        storage_class = cls._storage_classes[storage_type]
-        
-        try:
-            storage = storage_class(config)
-            logger.info(f"Created storage instance: {storage_type.value}")
-            return storage
-        except Exception as e:
-            logger.error(f"Failed to create storage instance {storage_type.value}: {e}")
-            raise StorageException(f"Failed to create storage: {e}")
-    
-    @classmethod
-    def get_available_types(cls) -> list[StorageType]:
-        """获取可用的存储类型
-        
-        Returns:
-            List[StorageType]: 可用的存储类型列表
-        """
-        return list(cls._storage_classes.keys())
-    
-    @classmethod
-    def create_json_storage(cls, base_path: str = None) -> StorageInterface:
-        """创建JSON存储实例（便捷方法）
-        
-        Args:
-            base_path: 存储基础路径
-            
-        Returns:
-            StorageInterface: JSON存储实例
-        """
-        config = StorageConfig(
-            storage_type=StorageType.JSON,
-            base_path=base_path
-        )
-        return cls.create_storage(config)
-
-# 全局存储实例缓存
-_storage_instances: Dict[str, StorageInterface] = {}
-
-def get_storage(storage_key: str = "default", config: StorageConfig = None) -> StorageInterface:
-    """获取存储实例（单例模式）
-    
-    Args:
-        storage_key: 存储实例键
-        config: 存储配置（首次创建时需要）
-        
-    Returns:
-        StorageInterface: 存储实例
-        
-    Raises:
-        StorageException: 当配置缺失时
-    """
-    if storage_key not in _storage_instances:
-        if config is None:
-            # 使用默认JSON存储配置
-            config = StorageConfig(storage_type=StorageType.JSON)
-        
-        _storage_instances[storage_key] = StorageFactory.create_storage(config)
-    
-    return _storage_instances[storage_key]
-
-def close_all_storages():
-    """关闭所有存储实例"""
-    for storage_key, storage in _storage_instances.items():
-        try:
-            storage.close()
-            logger.info(f"Closed storage instance: {storage_key}")
-        except Exception as e:
-            logger.error(f"Failed to close storage {storage_key}: {e}")
-    
-    _storage_instances.clear()
-
-# 注册未来的存储实现
-def register_database_storage():
-    """注册数据库存储（未来实现）"""
-    try:
-        from .database_storage import DatabaseStorage
-        StorageFactory.register_storage(StorageType.DATABASE, DatabaseStorage)
-    except ImportError:
-        logger.debug("Database storage not available")
-
-def register_mongodb_storage():
-    """注册MongoDB存储（未来实现）"""
-    try:
-        from .mongodb_storage import MongoDBStorage
-        StorageFactory.register_storage(StorageType.MONGODB, MongoDBStorage)
-    except ImportError:
-        logger.debug("MongoDB storage not available")
-
-def register_redis_storage():
-    """注册Redis存储（未来实现）"""
-    try:
-        from .redis_storage import RedisStorage
-        StorageFactory.register_storage(StorageType.REDIS, RedisStorage)
-    except ImportError:
-        logger.debug("Redis storage not available")
-
-# 自动注册可用的存储类型
-def auto_register_storages():
-    """自动注册所有可用的存储类型"""
-    register_database_storage()
-    register_mongodb_storage()
-    register_redis_storage()
-
-# 模块加载时自动注册
-auto_register_storages()

python_core/storage/json_storage.py

@@ -1,263 +0,0 @@
-#!/usr/bin/env python3
-"""
-JSON文件存储实现
-"""
-
-import json
-import os
-import glob
-from pathlib import Path
-from typing import Any, List, Dict
-from datetime import datetime
-
-from .base import StorageInterface, StorageConfig, StorageException, StorageOperationError
-from python_core.utils.logger import logger
-
-class JSONStorage(StorageInterface):
-    """JSON文件存储实现"""
-    
-    def __init__(self, config: StorageConfig):
-        super().__init__(config)
-        
-        # 设置基础路径
-        if config.base_path:
-            self.base_path = Path(config.base_path)
-        else:
-            from python_core.config import settings
-            self.base_path = settings.temp_dir / "storage"
-        
-        # 确保目录存在
-        self.base_path.mkdir(parents=True, exist_ok=True)
-        
-        logger.info(f"JSON storage initialized at: {self.base_path}")
-    
-    def _get_collection_path(self, collection: str) -> Path:
-        """获取集合目录路径"""
-        return self.base_path / collection
-    
-    def _get_file_path(self, collection: str, key: str) -> Path:
-        """获取文件路径"""
-        collection_path = self._get_collection_path(collection)
-        collection_path.mkdir(parents=True, exist_ok=True)
-        return collection_path / f"{key}.json"
-    
-    def _ensure_serializable(self, data: Any) -> Any:
-        """确保数据可序列化"""
-        if hasattr(data, '__dict__'):
-            # 处理dataclass或自定义对象
-            if hasattr(data, '__dataclass_fields__'):
-                from dataclasses import asdict
-                return asdict(data)
-            else:
-                return data.__dict__
-        elif isinstance(data, (list, tuple)):
-            return [self._ensure_serializable(item) for item in data]
-        elif isinstance(data, dict):
-            return {k: self._ensure_serializable(v) for k, v in data.items()}
-        else:
-            return data
-    
-    def save(self, collection: str, key: str, data: Any) -> bool:
-        """保存数据到JSON文件"""
-        try:
-            file_path = self._get_file_path(collection, key)
-            
-            # 准备保存的数据
-            save_data = {
-                "key": key,
-                "data": self._ensure_serializable(data),
-                "created_at": datetime.now().isoformat(),
-                "updated_at": datetime.now().isoformat()
-            }
-            
-            # 如果文件已存在，保留创建时间
-            if file_path.exists():
-                try:
-                    with open(file_path, 'r', encoding='utf-8') as f:
-                        existing_data = json.load(f)
-                        save_data["created_at"] = existing_data.get("created_at", save_data["created_at"])
-                except Exception:
-                    pass  # 如果读取失败，使用新的创建时间
-            
-            # 保存文件
-            with open(file_path, 'w', encoding='utf-8') as f:
-                json.dump(save_data, f, 
-                         indent=self.config.json_indent,
-                         ensure_ascii=self.config.json_ensure_ascii)
-            
-            logger.debug(f"Saved data to {file_path}")
-            return True
-            
-        except Exception as e:
-            logger.error(f"Failed to save data to {collection}/{key}: {e}")
-            raise StorageOperationError(f"Failed to save data: {e}")
-    
-    def load(self, collection: str, key: str) -> Any:
-        """从JSON文件加载数据"""
-        try:
-            file_path = self._get_file_path(collection, key)
-            
-            if not file_path.exists():
-                return None
-            
-            with open(file_path, 'r', encoding='utf-8') as f:
-                file_data = json.load(f)
-                return file_data.get("data")
-                
-        except Exception as e:
-            logger.error(f"Failed to load data from {collection}/{key}: {e}")
-            raise StorageOperationError(f"Failed to load data: {e}")
-    
-    def delete(self, collection: str, key: str) -> bool:
-        """删除JSON文件"""
-        try:
-            file_path = self._get_file_path(collection, key)
-            
-            if file_path.exists():
-                file_path.unlink()
-                logger.debug(f"Deleted file {file_path}")
-                return True
-            
-            return False
-            
-        except Exception as e:
-            logger.error(f"Failed to delete data from {collection}/{key}: {e}")
-            raise StorageOperationError(f"Failed to delete data: {e}")
-    
-    def exists(self, collection: str, key: str) -> bool:
-        """检查JSON文件是否存在"""
-        file_path = self._get_file_path(collection, key)
-        return file_path.exists()
-    
-    def list_keys(self, collection: str, pattern: str = "*") -> List[str]:
-        """列出集合中的所有键"""
-        try:
-            collection_path = self._get_collection_path(collection)
-            
-            if not collection_path.exists():
-                return []
-            
-            # 使用glob模式匹配
-            if pattern == "*":
-                pattern = "*.json"
-            elif not pattern.endswith(".json"):
-                pattern = f"{pattern}.json"
-            
-            files = glob.glob(str(collection_path / pattern))
-            keys = [Path(f).stem for f in files]
-            
-            return sorted(keys)
-            
-        except Exception as e:
-            logger.error(f"Failed to list keys in {collection}: {e}")
-            raise StorageOperationError(f"Failed to list keys: {e}")
-    
-    def save_batch(self, collection: str, data: Dict[str, Any]) -> bool:
-        """批量保存数据"""
-        try:
-            success_count = 0
-            for key, value in data.items():
-                if self.save(collection, key, value):
-                    success_count += 1
-            
-            logger.info(f"Batch saved {success_count}/{len(data)} items to {collection}")
-            return success_count == len(data)
-            
-        except Exception as e:
-            logger.error(f"Failed to batch save data to {collection}: {e}")
-            raise StorageOperationError(f"Failed to batch save data: {e}")
-    
-    def load_batch(self, collection: str, keys: List[str]) -> Dict[str, Any]:
-        """批量加载数据"""
-        try:
-            result = {}
-            for key in keys:
-                data = self.load(collection, key)
-                if data is not None:
-                    result[key] = data
-            
-            logger.debug(f"Batch loaded {len(result)}/{len(keys)} items from {collection}")
-            return result
-            
-        except Exception as e:
-            logger.error(f"Failed to batch load data from {collection}: {e}")
-            raise StorageOperationError(f"Failed to batch load data: {e}")
-    
-    def clear_collection(self, collection: str) -> bool:
-        """清空集合（删除所有文件）"""
-        try:
-            collection_path = self._get_collection_path(collection)
-            
-            if not collection_path.exists():
-                return True
-            
-            # 删除所有JSON文件
-            json_files = list(collection_path.glob("*.json"))
-            for file_path in json_files:
-                file_path.unlink()
-            
-            logger.info(f"Cleared {len(json_files)} files from collection {collection}")
-            return True
-            
-        except Exception as e:
-            logger.error(f"Failed to clear collection {collection}: {e}")
-            raise StorageOperationError(f"Failed to clear collection: {e}")
-    
-    def get_collections(self) -> List[str]:
-        """获取所有集合名称"""
-        try:
-            if not self.base_path.exists():
-                return []
-            
-            collections = []
-            for item in self.base_path.iterdir():
-                if item.is_dir():
-                    collections.append(item.name)
-            
-            return sorted(collections)
-            
-        except Exception as e:
-            logger.error(f"Failed to get collections: {e}")
-            raise StorageOperationError(f"Failed to get collections: {e}")
-    
-    def get_stats(self, collection: str) -> Dict[str, Any]:
-        """获取集合统计信息"""
-        try:
-            collection_path = self._get_collection_path(collection)
-            
-            if not collection_path.exists():
-                return {
-                    "collection": collection,
-                    "exists": False,
-                    "file_count": 0,
-                    "total_size": 0
-                }
-            
-            json_files = list(collection_path.glob("*.json"))
-            total_size = sum(f.stat().st_size for f in json_files)
-            
-            # 获取最新和最旧的文件时间
-            if json_files:
-                file_times = [f.stat().st_mtime for f in json_files]
-                oldest_file = min(file_times)
-                newest_file = max(file_times)
-            else:
-                oldest_file = newest_file = None
-            
-            return {
-                "collection": collection,
-                "exists": True,
-                "file_count": len(json_files),
-                "total_size": total_size,
-                "oldest_file": datetime.fromtimestamp(oldest_file).isoformat() if oldest_file else None,
-                "newest_file": datetime.fromtimestamp(newest_file).isoformat() if newest_file else None,
-                "path": str(collection_path)
-            }
-            
-        except Exception as e:
-            logger.error(f"Failed to get stats for collection {collection}: {e}")
-            raise StorageOperationError(f"Failed to get stats: {e}")
-    
-    def close(self):
-        """关闭存储连接（JSON存储无需特殊关闭操作）"""
-        logger.debug("JSON storage closed")

python_core/storage/kv.py

@@ -0,0 +1,87 @@
+
+from python_core.config import settings
+import httpx
+from python_core.utils.logger import setup_logger
+from typing import Dict
+
+logger = setup_logger(__name__)
+
+class Kv:
+    kv: Dict[str, str]
+    def __init__(self):
+        self.cf_account_id = settings.cloudflare_account_id
+        self.cf_kv_api_token = settings.cloudflare_api_key
+        self.cf_kv_id = settings.cloudflare_kv_id
+        
+    def get(self, key: str):
+        ...
+        
+    def sets(self, caches: Dict[str, str]):
+        try:
+            with httpx.Client() as client:
+                response = client.put(
+                    f"https://api.cloudflare.com/client/v4/accounts/{self.cf_account_id}/storage/kv/namespaces/{self.cf_kv_id}/bulk",
+                    headers={"Authorization": f"Bearer {self.cf_kv_api_token}"},
+                    json=[
+                        {
+                            "based64": False,
+                            "key": key,
+                            "value": value,
+                        }
+                        for (key, value) in caches.items()
+                    ]
+                )
+                response.raise_for_status()
+        except httpx.RequestError as e:
+            logger.error(f"An error occurred while put kv to cloudflare")
+            raise e
+        except httpx.HTTPStatusError as e:
+            logger.error(f"HTTP error occurred while get kv from cloudflare {str(e)}")
+            raise e
+        except Exception as e:
+            logger.error(f"An unexpected error occurred: {str(e)}")
+            raise e
+        
+    def set(self, key: str, value: str):
+        try:
+            with httpx.Client() as client:
+                response = client.put(
+                    f"https://api.cloudflare.com/client/v4/accounts/{self.cf_account_id}/storage/kv/namespaces/{self.cf_kv_id}/bulk",
+                    headers={"Authorization": f"Bearer {self.cf_kv_api_token}"},
+                    json=[
+                        {
+                            "based64": False,
+                            "key": key,
+                            "value": value,
+                        }
+                    ]
+                )
+                response.raise_for_status()
+        except httpx.RequestError as e:
+            logger.error(f"An error occurred while put kv to cloudflare")
+            raise e
+        except httpx.HTTPStatusError as e:
+            logger.error(f"HTTP error occurred while get kv from cloudflare {str(e)}")
+            raise e
+        except Exception as e:
+            logger.error(f"An unexpected error occurred: {str(e)}")
+            raise e
+        
+    def remove(self, keys: list[str]):
+        with httpx.Client() as client:
+            try:
+                response = client.post(
+                    f"https://api.cloudflare.com/client/v4/accounts/{self.cf_account_id}/storage/kv/namespaces/{self.cf_kv_id}/bulk/delete",
+                    headers={"Authorization": f"Bearer {self.cf_kv_api_token}"},
+                    json=keys
+                )
+                response.raise_for_status()
+            except httpx.RequestError as e:
+                logger.error(f"An error occurred while put kv to cloudflare")
+                raise e
+            except httpx.HTTPStatusError as e:
+                logger.error(f"HTTP error occurred while get kv from cloudflare {str(e)}")
+                raise e
+            except Exception as e:
+                logger.error(f"An unexpected error occurred: {str(e)}")
+                raise e