diff --git a/python_core/cli/cli.py b/python_core/cli/cli.py index 3a4b5f6..7b8e074 100644 --- a/python_core/cli/cli.py +++ b/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(): diff --git a/python_core/cli/commands/template.py b/python_core/cli/commands/template.py new file mode 100644 index 0000000..b51e5a3 --- /dev/null +++ b/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() diff --git a/python_core/server/__init__.py b/python_core/server/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/python_core/server/__main__.py b/python_core/server/__main__.py deleted file mode 100644 index e69de29..0000000 diff --git a/python_core/server/server.py b/python_core/server/server.py deleted file mode 100644 index e69de29..0000000 diff --git a/python_core/services/scene_detection/__init__.py b/python_core/services/scene_detection/__init__.py deleted file mode 100644 index 2d0d796..0000000 --- a/python_core/services/scene_detection/__init__.py +++ /dev/null @@ -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" -] diff --git a/python_core/services/scene_detection/__main__.py b/python_core/services/scene_detection/__main__.py deleted file mode 100644 index f163834..0000000 --- a/python_core/services/scene_detection/__main__.py +++ /dev/null @@ -1,10 +0,0 @@ -#!/usr/bin/env python3 -""" -场景检测服务主入口 -支持直接运行: python -m python_core.services.scene_detection -""" - -from .cli import main - -if __name__ == "__main__": - main() diff --git a/python_core/services/scene_detection/cli.py b/python_core/services/scene_detection/cli.py deleted file mode 100644 index cfad9c4..0000000 --- a/python_core/services/scene_detection/cli.py +++ /dev/null @@ -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() diff --git a/python_core/services/scene_detection/detector.py b/python_core/services/scene_detection/detector.py deleted file mode 100644 index 5463eb8..0000000 --- a/python_core/services/scene_detection/detector.py +++ /dev/null @@ -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 - ) diff --git a/python_core/services/scene_detection/types.py b/python_core/services/scene_detection/types.py deleted file mode 100644 index a567fb6..0000000 --- a/python_core/services/scene_detection/types.py +++ /dev/null @@ -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 diff --git a/python_core/services/video_splitter/__init__.py b/python_core/services/video_splitter/__init__.py deleted file mode 100644 index afa20cd..0000000 --- a/python_core/services/video_splitter/__init__.py +++ /dev/null @@ -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) diff --git a/python_core/services/video_splitter/__main__.py b/python_core/services/video_splitter/__main__.py deleted file mode 100644 index 809446d..0000000 --- a/python_core/services/video_splitter/__main__.py +++ /dev/null @@ -1,11 +0,0 @@ -#!/usr/bin/env python3 -""" -视频拆分服务命令行入口点 - -支持通过 python -m python_core.services.video_splitter 运行 -""" - -from .cli import main - -if __name__ == "__main__": - main() diff --git a/python_core/services/video_splitter/cli.py b/python_core/services/video_splitter/cli.py deleted file mode 100644 index e901ca7..0000000 --- a/python_core/services/video_splitter/cli.py +++ /dev/null @@ -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() diff --git a/python_core/services/video_splitter/detectors.py b/python_core/services/video_splitter/detectors.py deleted file mode 100644 index 35a8437..0000000 --- a/python_core/services/video_splitter/detectors.py +++ /dev/null @@ -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 [] diff --git a/python_core/services/video_splitter/service.py b/python_core/services/video_splitter/service.py deleted file mode 100644 index c112b49..0000000 --- a/python_core/services/video_splitter/service.py +++ /dev/null @@ -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) - ) diff --git a/python_core/services/video_splitter/types.py b/python_core/services/video_splitter/types.py deleted file mode 100644 index 7714d98..0000000 --- a/python_core/services/video_splitter/types.py +++ /dev/null @@ -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: - """验证视频文件""" - ... diff --git a/python_core/services/video_splitter/validators.py b/python_core/services/video_splitter/validators.py deleted file mode 100644 index f7e7c3f..0000000 --- a/python_core/services/video_splitter/validators.py +++ /dev/null @@ -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 diff --git a/python_core/storage/__init__.py b/python_core/storage/__init__.py index 735ba9a..8580345 100644 --- a/python_core/storage/__init__.py +++ b/python_core/storage/__init__.py @@ -3,15 +3,4 @@ 存储层模块 提供统一的存储接口，支持多种存储后端 """ - -from .base import StorageInterface, StorageConfig -from .json_storage import JSONStorage -from .factory import StorageFactory, get_storage - -__all__ = [ - "StorageInterface", - "StorageConfig", - "JSONStorage", - "StorageFactory", - "get_storage" -] +__all__ = [] diff --git a/python_core/storage/base.py b/python_core/storage/base.py deleted file mode 100644 index 429cf96..0000000 --- a/python_core/storage/base.py +++ /dev/null @@ -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 diff --git a/python_core/storage/factory.py b/python_core/storage/factory.py deleted file mode 100644 index a62f78b..0000000 --- a/python_core/storage/factory.py +++ /dev/null @@ -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() diff --git a/python_core/storage/json_storage.py b/python_core/storage/json_storage.py deleted file mode 100644 index 63155e7..0000000 --- a/python_core/storage/json_storage.py +++ /dev/null @@ -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") diff --git a/python_core/storage/kv.py b/python_core/storage/kv.py new file mode 100644 index 0000000..f6af204 --- /dev/null +++ b/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 \ No newline at end of file