diff --git a/python_core/cli/commands/project_material.py b/python_core/cli/commands/project_material.py new file mode 100644 index 0000000..c032271 --- /dev/null +++ b/python_core/cli/commands/project_material.py @@ -0,0 +1,292 @@ +""" +项目素材管理CLI命令 +""" + +from pathlib import Path +from typing import Optional, List +import typer +from rich.console import Console +from rich.table import Table + +from python_core.services.project_material_service import ProjectMaterialService +from python_core.scene_detection.single_scene_detector import SingleSceneDetector +from python_core.scene_detection.types.enums import DetectorType + +console = Console() +material_app = typer.Typer(name="material", help="项目素材管理命令") + + +@material_app.command("import") +def import_video( + video_path: str = typer.Argument(..., help="视频文件路径"), + project_id: str = typer.Argument(..., help="项目ID"), + project_directory: str = typer.Argument(..., help="项目目录路径"), + tags: Optional[str] = typer.Option(None, "--tags", "-t", help="素材标签，用逗号分隔"), + detector_type: str = typer.Option("content", "--detector", "-d", help="检测器类型"), + threshold: float = typer.Option(30.0, "--threshold", help="检测阈值"), + min_scene_length: float = typer.Option(1.0, "--min-length", help="最小场景长度（秒）"), + split_quality: int = typer.Option(23, "--quality", "-q", help="切分质量"), + split_preset: str = typer.Option("fast", "--preset", help="编码预设"), + max_duration: float = typer.Option(2.0, "--max-duration", "-m", help="最大视频时长限制（秒）"), + verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出") +): + """导入单个视频到项目素材库""" + + try: + console.print(f"📦 [bold blue]导入视频到项目素材库[/bold blue]") + console.print(f"📁 视频文件: {video_path}") + console.print(f"🎯 项目ID: {project_id}") + console.print(f"📂 项目目录: {project_directory}") + + # 解析标签 + material_tags = [] + if tags: + material_tags = [tag.strip() for tag in tags.split(",") if tag.strip()] + console.print(f"🏷️ 素材标签: {', '.join(material_tags)}") + + # 验证输入参数 + try: + detector_type_enum = DetectorType(detector_type) + except ValueError as e: + console.print(f"[red]❌ 参数错误: {e}[/red]") + raise typer.Exit(1) + + # 创建检测器 + detector = SingleSceneDetector() + + # 验证视频文件 + validation_result = detector.validate_video(video_path) + if not validation_result["valid"]: + console.print(f"[red]❌ 视频文件验证失败: {validation_result['error']}[/red]") + raise typer.Exit(1) + + # 验证项目目录 + project_dir_path = Path(project_directory) + if not project_dir_path.exists(): + console.print(f"[red]❌ 项目目录不存在: {project_directory}[/red]") + raise typer.Exit(1) + + console.print(f"✅ 验证通过，开始导入...") + + # 执行导入 + result = detector.import_to_project( + video_path=video_path, + project_id=project_id, + project_directory=project_directory, + material_tags=material_tags, + detector_type=detector_type_enum, + threshold=threshold, + min_scene_length=min_scene_length, + split_quality=split_quality, + split_preset=split_preset, + max_video_duration=max_duration + ) + + # 显示结果 + if result["success"]: + console.print(f"\n✅ [bold green]导入完成![/bold green]") + console.print(f"📊 导入统计:") + console.print(f" 视频文件: {Path(result['video_path']).name}") + console.print(f" 处理时间: {result['processing_time']:.1f}s") + console.print(f" 场景数量: {result['total_scenes']}") + console.print(f" 导入片段: {result['total_segments']}") + console.print(f" 项目目录: {result['output_dir']}") + + # 显示详细结果 + if result.get("split_results") and verbose: + table = Table(title="导入素材详情") + table.add_column("片段", style="cyan") + table.add_column("时长", style="yellow") + table.add_column("文件大小", style="blue") + table.add_column("状态", style="magenta") + + for split_result in result["split_results"]: + status = "✅ 已导入" if split_result["success"] else f"❌ {split_result.get('error', '失败')}" + file_size = f"{split_result['file_size']:,} B" if split_result['file_size'] > 0 else "0 B" + duration = f"{split_result['duration']:.2f}s" + + table.add_row( + str(split_result['scene_index'] + 1), + duration, + file_size, + status + ) + + console.print(table) + else: + console.print(f"\n[red]❌ 导入失败: {result.get('error', '未知错误')}[/red]") + raise typer.Exit(1) + + except Exception as e: + console.print(f"\n[red]❌ 项目导入失败: {str(e)}[/red]") + raise typer.Exit(1) + + +@material_app.command("list") +def list_materials( + project_directory: str = typer.Argument(..., help="项目目录路径"), + tags: Optional[str] = typer.Option(None, "--tags", "-t", help="按标签过滤，用逗号分隔"), + limit: int = typer.Option(20, "--limit", "-l", help="显示数量限制"), + verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出") +): + """列出项目素材""" + + try: + console.print(f"📋 [bold blue]项目素材列表[/bold blue]") + console.print(f"📂 项目目录: {project_directory}") + + # 验证项目目录 + project_dir_path = Path(project_directory) + if not project_dir_path.exists(): + console.print(f"[red]❌ 项目目录不存在: {project_directory}[/red]") + raise typer.Exit(1) + + # 创建素材服务 + material_service = ProjectMaterialService() + + # 获取素材列表 + materials = material_service.get_project_materials(project_dir_path) + + if not materials: + console.print("📭 项目中没有素材") + return + + # 按标签过滤 + if tags: + filter_tags = [tag.strip() for tag in tags.split(",") if tag.strip()] + materials = [ + material for material in materials + if any(tag in material.get('tags', []) for tag in filter_tags) + ] + console.print(f"🏷️ 按标签过滤: {', '.join(filter_tags)}") + + # 限制显示数量 + if len(materials) > limit: + materials = materials[:limit] + console.print(f"📊 显示前 {limit} 个素材（共 {len(materials)} 个）") + + # 创建表格 + table = Table(title="项目素材列表") + table.add_column("ID", style="cyan", width=8) + table.add_column("原始文件名", style="green") + table.add_column("时长", style="yellow") + table.add_column("文件大小", style="blue") + table.add_column("标签", style="magenta") + table.add_column("使用次数", style="red") + + if verbose: + table.add_column("创建时间", style="dim") + + for material in materials: + file_size = f"{material.get('file_size', 0):,} B" + duration = f"{material.get('duration', 0):.2f}s" + tags_str = ", ".join(material.get('tags', [])) + use_count = str(material.get('use_count', 0)) + + row = [ + material.get('id', '')[:8], + material.get('original_filename', ''), + duration, + file_size, + tags_str, + use_count + ] + + if verbose: + created_at = material.get('created_at', '')[:19].replace('T', ' ') + row.append(created_at) + + table.add_row(*row) + + console.print(table) + + except Exception as e: + console.print(f"\n[red]❌ 获取素材列表失败: {str(e)}[/red]") + raise typer.Exit(1) + + +@material_app.command("stats") +def show_stats( + project_directory: str = typer.Argument(..., help="项目目录路径") +): + """显示项目素材统计信息""" + + try: + console.print(f"📊 [bold blue]项目素材统计[/bold blue]") + console.print(f"📂 项目目录: {project_directory}") + + # 验证项目目录 + project_dir_path = Path(project_directory) + if not project_dir_path.exists(): + console.print(f"[red]❌ 项目目录不存在: {project_directory}[/red]") + raise typer.Exit(1) + + # 创建素材服务 + material_service = ProjectMaterialService() + + # 获取统计信息 + stats = material_service.get_material_stats(project_dir_path) + + console.print(f"\n📈 [bold green]统计信息[/bold green]") + console.print(f" 总素材数: {stats['total_count']}") + console.print(f" 总文件大小: {stats['total_size']:,} 字节") + console.print(f" 总时长: {stats['total_duration']:.2f} 秒") + console.print(f" 已使用: {stats['used_count']}") + console.print(f" 未使用: {stats['unused_count']}") + + # 标签统计 + if stats['tag_stats']: + console.print(f"\n🏷️ [bold green]标签统计[/bold green]") + for tag, count in sorted(stats['tag_stats'].items(), key=lambda x: x[1], reverse=True): + console.print(f" {tag}: {count}") + + except Exception as e: + console.print(f"\n[red]❌ 获取统计信息失败: {str(e)}[/red]") + raise typer.Exit(1) + + +@material_app.command("remove") +def remove_material( + project_directory: str = typer.Argument(..., help="项目目录路径"), + material_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"📂 项目目录: {project_directory}") + console.print(f"🎯 素材ID: {material_id}") + + # 验证项目目录 + project_dir_path = Path(project_directory) + if not project_dir_path.exists(): + console.print(f"[red]❌ 项目目录不存在: {project_directory}[/red]") + raise typer.Exit(1) + + # 创建素材服务 + material_service = ProjectMaterialService() + + # 确认删除 + if not force: + confirm = typer.confirm("确定要删除这个素材吗？此操作不可撤销。") + if not confirm: + console.print("❌ 操作已取消") + return + + # 删除素材 + success = material_service.remove_material(project_dir_path, material_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"\n[red]❌ 删除素材失败: {str(e)}[/red]") + raise typer.Exit(1) + + +if __name__ == "__main__": + material_app() diff --git a/python_core/cli/commands/scene_detect.py b/python_core/cli/commands/scene_detect.py index e5e89c0..6308879 100644 --- a/python_core/cli/commands/scene_detect.py +++ b/python_core/cli/commands/scene_detect.py @@ -13,10 +13,11 @@ from rich.console import Console from rich.table import Table from python_core.scene_detection import ( - SceneDetector, - DetectorType, + SceneDetector, + DetectorType, OutputFormat ) +from python_core.scene_detection.single_scene_detector import SingleSceneDetector from python_core.utils.logger import logger scene_detect = typer.Typer(help="场景检测工具 - 重构版") @@ -231,5 +232,212 @@ def _display_batch_results_table(tasks): console.print(table) +@scene_detect.command("single") +def single_detect_and_split( + video_path: str = typer.Argument(..., help="视频文件路径"), + output_dir: Optional[str] = typer.Option(None, "--output", "-o", help="输出目录"), + detector_type: str = typer.Option("content", "--detector", "-d", help="检测器类型 (content/threshold/adaptive)"), + threshold: float = typer.Option(30.0, "--threshold", "-t", help="检测阈值 (0-100)"), + min_scene_length: float = typer.Option(1.0, "--min-length", "-l", help="最小场景长度（秒）"), + output_format: str = typer.Option("json", "--format", "-f", help="输出格式 (json/csv/txt)"), + ai_analysis: bool = typer.Option(False, "--ai/--no-ai", help="启用/禁用AI分析"), + video_splitting: bool = typer.Option(True, "--split/--no-split", help="启用/禁用视频切分"), + use_advanced_split: bool = typer.Option(True, "--advanced/--traditional", help="使用高效批量切分/传统逐个切分"), + split_quality: int = typer.Option(23, "--quality", "-q", help="切分质量 (CRF值, 18-28)"), + split_preset: str = typer.Option("fast", "--preset", help="编码预设 (ultrafast/fast/medium/slow)"), + max_duration: float = typer.Option(2.0, "--max-duration", "-m", help="最大视频时长限制（秒），超过将二次切分"), + verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出") +): + """单个视频文件的场景检测和切分""" + + try: + console.print(f"🎬 [bold blue]单个视频场景检测和切分[/bold blue]") + console.print(f"📁 视频文件: {video_path}") + + # 验证输入参数 + try: + detector_type_enum = DetectorType(detector_type) + output_format_enum = OutputFormat(output_format) + except ValueError as e: + console.print(f"[red]❌ 参数错误: {e}[/red]") + raise typer.Exit(1) + + # 创建检测器 + detector = SingleSceneDetector() + + # 验证视频文件 + validation_result = detector.validate_video(video_path) + if not validation_result["valid"]: + console.print(f"[red]❌ 视频文件验证失败: {validation_result['error']}[/red]") + raise typer.Exit(1) + + console.print(f"✅ 视频文件验证通过") + console.print(f"📊 文件大小: {validation_result['file_size']:,} 字节") + + # 执行处理 + result = detector.detect_and_split( + video_path=video_path, + output_dir=output_dir, + detector_type=detector_type_enum, + threshold=threshold, + min_scene_length=min_scene_length, + output_format=output_format_enum, + enable_ai_analysis=ai_analysis, + enable_video_splitting=video_splitting, + use_advanced_split=use_advanced_split, + split_quality=split_quality, + split_preset=split_preset, + max_video_duration=max_duration + ) + + # 显示结果 + if result["success"]: + console.print(f"\n✅ [bold green]处理完成![/bold green]") + console.print(f"📊 处理统计:") + console.print(f" 视频文件: {Path(result['video_path']).name}") + console.print(f" 输出目录: {result['output_dir']}") + console.print(f" 处理时间: {result['processing_time']:.1f}s") + console.print(f" 场景数量: {result['total_scenes']}") + console.print(f" 切分片段: {result['total_segments']}") + + # 显示详细结果表格 + if result.get("split_results") and verbose: + table = Table(title="切分结果详情") + table.add_column("片段", style="cyan") + table.add_column("输出文件", style="green") + table.add_column("时长", style="yellow") + table.add_column("文件大小", style="blue") + table.add_column("状态", style="magenta") + + for split_result in result["split_results"]: + status = "✅ 成功" if split_result["success"] else f"❌ {split_result.get('error', '失败')}" + file_size = f"{split_result['file_size']:,} B" if split_result['file_size'] > 0 else "0 B" + duration = f"{split_result['duration']:.2f}s" + filename = Path(split_result['output_path']).name + + table.add_row( + str(split_result['scene_index'] + 1), + filename, + duration, + file_size, + status + ) + + console.print(table) + else: + console.print(f"\n[red]❌ 处理失败: {result.get('error', '未知错误')}[/red]") + raise typer.Exit(1) + + except Exception as e: + console.print(f"\n[red]❌ 单个视频处理失败: {str(e)}[/red]") + raise typer.Exit(1) + + +@scene_detect.command("import") +def import_to_project( + video_path: str = typer.Argument(..., help="视频文件路径"), + project_id: str = typer.Argument(..., help="项目ID"), + project_directory: str = typer.Argument(..., help="项目目录路径"), + tags: Optional[str] = typer.Option(None, "--tags", "-t", help="素材标签，用逗号分隔"), + detector_type: str = typer.Option("content", "--detector", "-d", help="检测器类型"), + threshold: float = typer.Option(30.0, "--threshold", help="检测阈值"), + min_scene_length: float = typer.Option(1.0, "--min-length", help="最小场景长度（秒）"), + split_quality: int = typer.Option(23, "--quality", "-q", help="切分质量"), + split_preset: str = typer.Option("fast", "--preset", help="编码预设"), + max_duration: float = typer.Option(2.0, "--max-duration", "-m", help="最大视频时长限制（秒）"), + verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出") +): + """导入单个视频到项目素材库""" + + try: + console.print(f"📦 [bold blue]导入视频到项目素材库[/bold blue]") + console.print(f"📁 视频文件: {video_path}") + console.print(f"🎯 项目ID: {project_id}") + console.print(f"📂 项目目录: {project_directory}") + + # 解析标签 + material_tags = [] + if tags: + material_tags = [tag.strip() for tag in tags.split(",") if tag.strip()] + console.print(f"🏷️ 素材标签: {', '.join(material_tags)}") + + # 验证输入参数 + try: + detector_type_enum = DetectorType(detector_type) + except ValueError as e: + console.print(f"[red]❌ 参数错误: {e}[/red]") + raise typer.Exit(1) + + # 创建检测器 + detector = SingleSceneDetector() + + # 验证视频文件 + validation_result = detector.validate_video(video_path) + if not validation_result["valid"]: + console.print(f"[red]❌ 视频文件验证失败: {validation_result['error']}[/red]") + raise typer.Exit(1) + + # 验证项目目录 + project_dir_path = Path(project_directory) + if not project_dir_path.exists(): + console.print(f"[red]❌ 项目目录不存在: {project_directory}[/red]") + raise typer.Exit(1) + + console.print(f"✅ 验证通过，开始导入...") + + # 执行导入 + result = detector.import_to_project( + video_path=video_path, + project_id=project_id, + project_directory=project_directory, + material_tags=material_tags, + detector_type=detector_type_enum, + threshold=threshold, + min_scene_length=min_scene_length, + split_quality=split_quality, + split_preset=split_preset, + max_video_duration=max_duration + ) + + # 显示结果 + if result["success"]: + console.print(f"\n✅ [bold green]导入完成![/bold green]") + console.print(f"📊 导入统计:") + console.print(f" 视频文件: {Path(result['video_path']).name}") + console.print(f" 处理时间: {result['processing_time']:.1f}s") + console.print(f" 场景数量: {result['total_scenes']}") + console.print(f" 导入片段: {result['total_segments']}") + console.print(f" 项目目录: {result['output_dir']}") + + # 显示详细结果 + if result.get("split_results") and verbose: + table = Table(title="导入素材详情") + table.add_column("片段", style="cyan") + table.add_column("时长", style="yellow") + table.add_column("文件大小", style="blue") + table.add_column("状态", style="magenta") + + for split_result in result["split_results"]: + status = "✅ 已导入" if split_result["success"] else f"❌ {split_result.get('error', '失败')}" + file_size = f"{split_result['file_size']:,} B" if split_result['file_size'] > 0 else "0 B" + duration = f"{split_result['duration']:.2f}s" + + table.add_row( + str(split_result['scene_index'] + 1), + duration, + file_size, + status + ) + + console.print(table) + else: + console.print(f"\n[red]❌ 导入失败: {result.get('error', '未知错误')}[/red]") + raise typer.Exit(1) + + except Exception as e: + console.print(f"\n[red]❌ 项目导入失败: {str(e)}[/red]") + raise typer.Exit(1) + + if __name__ == "__main__": scene_detect() \ No newline at end of file diff --git a/python_core/project/__Init__.py b/python_core/project/__Init__.py new file mode 100644 index 0000000..e69de29 diff --git a/python_core/project/commands/__init__.py b/python_core/project/commands/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/python_core/project/commands/create.py b/python_core/project/commands/create.py new file mode 100644 index 0000000..22d9737 --- /dev/null +++ b/python_core/project/commands/create.py @@ -0,0 +1,4 @@ +# 创建项目 +# 不需要工作流 + +# {config.projects_dir}/{project.name} \ No newline at end of file diff --git a/python_core/project/commands/del.py b/python_core/project/commands/del.py new file mode 100644 index 0000000..1533280 --- /dev/null +++ b/python_core/project/commands/del.py @@ -0,0 +1,5 @@ + + +# 删除项目及项目下的资源 +# 不需要工作流 +# {config.projects_dir}/{project.name} \ No newline at end of file diff --git a/python_core/project/commands/detail.py b/python_core/project/commands/detail.py new file mode 100644 index 0000000..e07a25e --- /dev/null +++ b/python_core/project/commands/detail.py @@ -0,0 +1,8 @@ + +# 获取项目详情 +# 不需要工作流 +# 项目详情：{config.projects_dir}/{project.name}/project.json +# 素材：{config.projects_dir}/{project.name}/material.json +# 模板：{config.projects_dir}/{project.name}/template.json +# 成品：{config.projects_dir}/{project.name}/product.json + diff --git a/python_core/project/commands/export.py b/python_core/project/commands/export.py new file mode 100644 index 0000000..7f3ebe8 --- /dev/null +++ b/python_core/project/commands/export.py @@ -0,0 +1,4 @@ + + +# 控制剪映 导出项目 +# 需要工作流 diff --git a/python_core/project/commands/import.py b/python_core/project/commands/import.py new file mode 100644 index 0000000..8cdc22e --- /dev/null +++ b/python_core/project/commands/import.py @@ -0,0 +1,5 @@ + +# 导入素材 +# 需要工作流 +# 1. 导入文件 +# 2. 导入文件夹 \ No newline at end of file diff --git a/python_core/project/commands/list.py b/python_core/project/commands/list.py new file mode 100644 index 0000000..583e46a --- /dev/null +++ b/python_core/project/commands/list.py @@ -0,0 +1,3 @@ +# 获取所有项目 +# 不需要工作流 +# {config.projects_dir}/projects.json \ No newline at end of file diff --git a/python_core/project/commands/mix.py b/python_core/project/commands/mix.py new file mode 100644 index 0000000..e4b8b4d --- /dev/null +++ b/python_core/project/commands/mix.py @@ -0,0 +1,2 @@ +# 混剪 +# 需要工作流 diff --git a/python_core/project/services/__init__.py b/python_core/project/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/python_core/project/types/__init__.py b/python_core/project/types/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/python_core/project/utils/__init__.py b/python_core/project/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/python_core/project/workflows/__init__.py b/python_core/project/workflows/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/python_core/project/workflows/import_dir.py b/python_core/project/workflows/import_dir.py new file mode 100644 index 0000000..d775c83 --- /dev/null +++ b/python_core/project/workflows/import_dir.py @@ -0,0 +1,3 @@ + + +# 导入文件夹 diff --git a/python_core/project/workflows/import_file.py b/python_core/project/workflows/import_file.py new file mode 100644 index 0000000..447eb2d --- /dev/null +++ b/python_core/project/workflows/import_file.py @@ -0,0 +1 @@ +# 导入文件 \ No newline at end of file diff --git a/python_core/scene_detection/single_scene_detector.py b/python_core/scene_detection/single_scene_detector.py new file mode 100644 index 0000000..b97e0f5 --- /dev/null +++ b/python_core/scene_detection/single_scene_detector.py @@ -0,0 +1,212 @@ +""" +单个文件场景检测器 +提供单个视频文件的场景检测和切分功能 +""" + +from pathlib import Path +from typing import Dict, Any, Optional, List + +from python_core.utils.logger import logger +from .workflows.single_workflow_manager import SingleWorkflowManager +from .types.enums import DetectorType, OutputFormat + + +class SingleSceneDetector: + """单个文件场景检测器""" + + def __init__(self): + self.workflow_manager = SingleWorkflowManager() + logger.info("SingleSceneDetector 初始化完成") + + def detect_and_split( + self, + video_path: str | Path, + output_dir: Optional[str | Path] = None, + detector_type: DetectorType = DetectorType.CONTENT, + threshold: float = 30.0, + min_scene_length: float = 1.0, + output_format: OutputFormat = OutputFormat.JSON, + enable_ai_analysis: bool = False, + enable_video_splitting: bool = True, + use_advanced_split: bool = True, + split_quality: int = 23, + split_preset: str = "fast", + max_video_duration: float = 60.0, + request_id: Optional[str] = None + ) -> Dict[str, Any]: + """ + 单个视频文件的场景检测和切分 + + Args: + video_path: 视频文件路径 + output_dir: 输出目录，如果为None则使用视频文件同目录 + detector_type: 检测器类型 (content/threshold/adaptive) + threshold: 检测阈值 (0-100) + min_scene_length: 最小场景长度（秒） + output_format: 输出格式 (json/csv/txt) + enable_ai_analysis: 是否启用AI分析 + enable_video_splitting: 是否启用视频切分 + use_advanced_split: 是否使用高级切分 + split_quality: 切分质量 (CRF值, 18-28) + split_preset: 编码预设 (ultrafast/fast/medium/slow) + max_video_duration: 最大视频时长限制（秒） + request_id: 请求ID（用于JSON-RPC进度报告） + + Returns: + Dict: 处理结果 + """ + + # 转换路径 + video_path = Path(video_path) + output_dir = Path(output_dir) if output_dir else None + + return self.workflow_manager.detect_and_split_single_video( + video_path=video_path, + output_dir=output_dir, + detector_type=detector_type, + threshold=threshold, + min_scene_length=min_scene_length, + output_format=output_format, + enable_ai_analysis=enable_ai_analysis, + enable_video_splitting=enable_video_splitting, + use_advanced_split=use_advanced_split, + split_quality=split_quality, + split_preset=split_preset, + max_video_duration=max_video_duration, + request_id=request_id + ) + + def import_to_project( + self, + video_path: str | Path, + project_id: str, + project_directory: str | Path, + material_tags: Optional[List[str]] = None, + detector_type: DetectorType = DetectorType.CONTENT, + threshold: float = 30.0, + min_scene_length: float = 1.0, + split_quality: int = 23, + split_preset: str = "fast", + max_video_duration: float = 2.0, + request_id: Optional[str] = None + ) -> Dict[str, Any]: + """ + 导入单个视频到项目素材库 + + Args: + video_path: 视频文件路径 + project_id: 项目ID + project_directory: 项目目录 + material_tags: 素材标签列表 + detector_type: 检测器类型 + threshold: 检测阈值 + min_scene_length: 最小场景长度 + split_quality: 切分质量 + split_preset: 编码预设 + max_video_duration: 最大视频时长限制 + request_id: 请求ID + + Returns: + Dict: 导入结果 + """ + + # 转换路径 + video_path = Path(video_path) + project_directory = Path(project_directory) + + return self.workflow_manager.detect_and_split_for_project( + video_path=video_path, + project_id=project_id, + project_directory=project_directory, + material_tags=material_tags, + detector_type=detector_type, + threshold=threshold, + min_scene_length=min_scene_length, + split_quality=split_quality, + split_preset=split_preset, + max_video_duration=max_video_duration, + request_id=request_id + ) + + def validate_video(self, video_path: str | Path) -> Dict[str, Any]: + """ + 验证视频文件 + + Args: + video_path: 视频文件路径 + + Returns: + Dict: 验证结果 + """ + video_path = Path(video_path) + + result = { + "valid": False, + "path": str(video_path), + "exists": False, + "is_file": False, + "supported_format": False, + "file_size": 0, + "error": None + } + + try: + # 检查文件是否存在 + result["exists"] = video_path.exists() + if not result["exists"]: + result["error"] = "文件不存在" + return result + + # 检查是否为文件 + result["is_file"] = video_path.is_file() + if not result["is_file"]: + result["error"] = "路径不是文件" + return result + + # 检查文件格式 + supported_formats = self.workflow_manager.get_supported_formats() + result["supported_format"] = video_path.suffix.lower() in supported_formats + if not result["supported_format"]: + result["error"] = f"不支持的格式: {video_path.suffix}" + return result + + # 获取文件大小 + result["file_size"] = video_path.stat().st_size + + # 验证通过 + result["valid"] = True + + except Exception as e: + result["error"] = str(e) + + return result + + def get_supported_formats(self) -> List[str]: + """获取支持的视频格式""" + return self.workflow_manager.get_supported_formats() + + def get_detector_types(self) -> List[str]: + """获取支持的检测器类型""" + return [dt.value for dt in DetectorType] + + def get_output_formats(self) -> List[str]: + """获取支持的输出格式""" + return [of.value for of in OutputFormat] + + def get_default_config(self) -> Dict[str, Any]: + """获取默认配置""" + return { + "detector_type": DetectorType.CONTENT.value, + "threshold": 30.0, + "min_scene_length": 1.0, + "output_format": OutputFormat.JSON.value, + "enable_ai_analysis": False, + "enable_video_splitting": True, + "use_advanced_split": True, + "split_quality": 23, + "split_preset": "fast", + "max_video_duration": 60.0, + "supported_formats": self.get_supported_formats(), + "detector_types": self.get_detector_types(), + "output_formats": self.get_output_formats() + } diff --git a/python_core/scene_detection/types/single_workflow_state.py b/python_core/scene_detection/types/single_workflow_state.py new file mode 100644 index 0000000..79f540c --- /dev/null +++ b/python_core/scene_detection/types/single_workflow_state.py @@ -0,0 +1,185 @@ +""" +单个文件场景检测工作流状态 +""" + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional, List, Dict, Any +from datetime import datetime + +from .models import DetectionResult +from python_core.utils.jsonrpc_enhanced import ProgressLevel + + +@dataclass +class SingleVideoTask: + """单个视频处理任务""" + video_path: Path + output_dir: Optional[Path] = None + + # 任务状态 + status: str = "pending" # pending, processing, completed, failed + start_time: Optional[datetime] = None + end_time: Optional[datetime] = None + error: Optional[str] = None + + # 处理结果 + detection_result: Optional[DetectionResult] = None + split_results: List[Any] = field(default_factory=list) + + +@dataclass +class SingleSceneDetectionWorkflowState: + """单个文件场景检测工作流状态""" + + # 基础配置 + video_path: Path + output_dir: Optional[Path] = None + + # 检测配置 + detector_type: str = "content" + threshold: float = 30.0 + min_scene_length: float = 1.0 + output_format: str = "json" + + # 功能开关 + enable_ai_analysis: bool = False + enable_video_splitting: bool = True + + # 视频切分配置 + use_advanced_split: bool = True + split_quality: int = 23 + split_preset: str = "fast" + max_video_duration: float = 60.0 # 最大视频时长（秒） + + # 项目配置（用于项目素材导入） + project_id: Optional[str] = None + project_directory: Optional[Path] = None + material_tags: List[str] = field(default_factory=list) + + # 工作流状态 + task: Optional[SingleVideoTask] = None + current_step: str = "initialized" + progress: float = 0.0 + + # JSON-RPC 支持 + request_id: Optional[str] = None + enable_jsonrpc: bool = False + + def __post_init__(self): + """初始化后处理""" + if self.task is None: + self.task = SingleVideoTask( + video_path=self.video_path, + output_dir=self.output_dir + ) + + def update_progress(self, step: str, progress: float, message: str = ""): + """更新进度""" + self.current_step = step + self.progress = progress + + if self.enable_jsonrpc and self.request_id: + self.send_progress(step, message, ProgressLevel.INFO, { + "progress": progress, + "step": step + }) + + def send_progress(self, event_type: str, message: str, level: ProgressLevel, data: Dict[str, Any] = None): + """发送进度消息（JSON-RPC）""" + if not self.enable_jsonrpc or not self.request_id: + return + + try: + from python_core.utils.jsonrpc_enhanced import create_response_handler + handler = create_response_handler(self.request_id) + handler.progress( + step=event_type, + progress=-1, # 不确定进度时使用-1 + message=message, + level=level, + data=data or {} + ) + except Exception as e: + print(f"Failed to send progress: {e}") + + def mark_task_started(self): + """标记任务开始""" + if self.task: + self.task.status = "processing" + self.task.start_time = datetime.now() + + def mark_task_completed(self, detection_result: DetectionResult, split_results: List[Any] = None): + """标记任务完成""" + if self.task: + self.task.status = "completed" + self.task.end_time = datetime.now() + self.task.detection_result = detection_result + self.task.split_results = split_results or [] + + def mark_task_failed(self, error: str): + """标记任务失败""" + if self.task: + self.task.status = "failed" + self.task.end_time = datetime.now() + self.task.error = error + + @property + def is_completed(self) -> bool: + """是否已完成""" + return self.task and self.task.status == "completed" + + @property + def is_failed(self) -> bool: + """是否失败""" + return self.task and self.task.status == "failed" + + @property + def processing_time(self) -> Optional[float]: + """处理时间（秒）""" + if not self.task or not self.task.start_time: + return None + + end_time = self.task.end_time or datetime.now() + return (end_time - self.task.start_time).total_seconds() + + @property + def total_scenes(self) -> int: + """总场景数""" + if self.task and self.task.detection_result: + return self.task.detection_result.total_scenes + return 0 + + @property + def total_split_segments(self) -> int: + """总切分片段数""" + if self.task and self.task.split_results: + return len(self.task.split_results) + return 0 + + def to_dict(self) -> Dict[str, Any]: + """转换为字典""" + return { + "video_path": str(self.video_path), + "output_dir": str(self.output_dir) if self.output_dir else None, + "detector_type": self.detector_type, + "threshold": self.threshold, + "min_scene_length": self.min_scene_length, + "output_format": self.output_format, + "enable_ai_analysis": self.enable_ai_analysis, + "enable_video_splitting": self.enable_video_splitting, + "use_advanced_split": self.use_advanced_split, + "split_quality": self.split_quality, + "split_preset": self.split_preset, + "max_video_duration": self.max_video_duration, + "project_id": self.project_id, + "project_directory": str(self.project_directory) if self.project_directory else None, + "material_tags": self.material_tags, + "current_step": self.current_step, + "progress": self.progress, + "task_status": self.task.status if self.task else "pending", + "processing_time": self.processing_time, + "total_scenes": self.total_scenes, + "total_split_segments": self.total_split_segments, + "request_id": self.request_id + } diff --git a/python_core/scene_detection/workflows/single_workflow_manager.py b/python_core/scene_detection/workflows/single_workflow_manager.py new file mode 100644 index 0000000..95ca5aa --- /dev/null +++ b/python_core/scene_detection/workflows/single_workflow_manager.py @@ -0,0 +1,238 @@ +""" +单个文件场景检测工作流管理器 +""" + +from pathlib import Path +from typing import Dict, Any, Optional, List + +from python_core.utils.logger import logger +from .single_workflow_nodes import SingleWorkflowNodes +from ..types.single_workflow_state import SingleSceneDetectionWorkflowState +from ..types.enums import DetectorType, OutputFormat + + +class SingleWorkflowManager: + """单个文件场景检测工作流管理器""" + + def __init__(self): + self.workflow_nodes = SingleWorkflowNodes() + + def detect_and_split_single_video( + self, + video_path: Path, + output_dir: Optional[Path] = None, + detector_type: DetectorType = DetectorType.CONTENT, + threshold: float = 30.0, + min_scene_length: float = 1.0, + output_format: OutputFormat = OutputFormat.JSON, + enable_ai_analysis: bool = False, + enable_video_splitting: bool = True, + use_advanced_split: bool = True, + split_quality: int = 23, + split_preset: str = "fast", + max_video_duration: float = 60.0, + project_id: Optional[str] = None, + project_directory: Optional[Path] = None, + material_tags: Optional[List[str]] = None, + request_id: Optional[str] = None + ) -> Dict[str, Any]: + """ + 单个视频文件的场景检测和切分 + + Args: + video_path: 视频文件路径 + output_dir: 输出目录 + detector_type: 检测器类型 + threshold: 检测阈值 + min_scene_length: 最小场景长度 + output_format: 输出格式 + enable_ai_analysis: 是否启用AI分析 + enable_video_splitting: 是否启用视频切分 + use_advanced_split: 是否使用高级切分 + split_quality: 切分质量 + split_preset: 切分预设 + max_video_duration: 最大视频时长 + project_id: 项目ID（用于项目素材导入） + project_directory: 项目目录（用于项目素材导入） + material_tags: 素材标签 + request_id: 请求ID（用于JSON-RPC进度报告） + + Returns: + Dict: 处理结果 + """ + + # 验证输入 + if not video_path.exists(): + raise FileNotFoundError(f"视频文件不存在: {video_path}") + + if not video_path.suffix.lower() in {'.mp4', '.avi', '.mov', '.mkv', '.wmv', '.flv', '.webm', '.m4v'}: + raise ValueError(f"不支持的视频格式: {video_path.suffix}") + + # 设置默认输出目录 + if output_dir is None: + output_dir = video_path.parent / f"{video_path.stem}_scenes" + + # 确保输出目录存在 + output_dir.mkdir(parents=True, exist_ok=True) + + logger.info(f"🚀 开始单个视频场景检测和切分") + logger.info(f"📁 视频文件: {video_path}") + logger.info(f"📂 输出目录: {output_dir}") + logger.info(f"🎯 检测器: {detector_type.value}, 阈值: {threshold}") + logger.info(f"✂️ 视频切分: {'启用' if enable_video_splitting else '禁用'}") + + if project_id: + logger.info(f"📦 项目导入: {project_id}") + + # 创建工作流状态 + state = SingleSceneDetectionWorkflowState( + video_path=video_path, + output_dir=output_dir, + detector_type=detector_type.value, + threshold=threshold, + min_scene_length=min_scene_length, + output_format=output_format.value, + enable_ai_analysis=enable_ai_analysis, + enable_video_splitting=enable_video_splitting, + use_advanced_split=use_advanced_split, + split_quality=split_quality, + split_preset=split_preset, + max_video_duration=max_video_duration, + project_id=project_id, + project_directory=project_directory, + material_tags=material_tags or [], + request_id=request_id, + enable_jsonrpc=request_id is not None + ) + + try: + # 执行工作流 + result = self.workflow_nodes.process_single_video(state) + + # 构建返回结果 + response = { + "success": result["success"], + "video_path": str(video_path), + "output_dir": str(output_dir), + "processing_time": result.get("processing_time", 0), + "total_scenes": result.get("total_scenes", 0), + "total_segments": result.get("total_segments", 0), + "state": state.to_dict() + } + + if result["success"]: + # 转换检测结果为字典 + detection_result_dict = None + if result.get("detection_result"): + from dataclasses import asdict + detection_result_dict = asdict(result["detection_result"]) + + response.update({ + "detection_result": detection_result_dict, + "split_results": [ + { + "scene_index": sr.scene_index, + "output_path": str(sr.output_path), + "start_time": sr.start_time, + "end_time": sr.end_time, + "duration": sr.duration, + "file_size": sr.file_size, + "success": sr.success, + "error": sr.error + } + for sr in result.get("split_results", []) + ] + }) + + logger.info(f"✅ 单个视频处理成功!") + logger.info(f"📊 处理统计:") + logger.info(f" 视频文件: {video_path.name}") + logger.info(f" 处理时间: {result.get('processing_time', 0):.1f}s") + logger.info(f" 场景数量: {result.get('total_scenes', 0)}") + logger.info(f" 切分片段: {result.get('total_segments', 0)}") + + else: + response["error"] = result.get("error", "未知错误") + logger.error(f"❌ 单个视频处理失败: {response['error']}") + + return response + + except Exception as e: + error_msg = f"工作流执行失败: {str(e)}" + logger.error(f"❌ {error_msg}") + + return { + "success": False, + "error": error_msg, + "video_path": str(video_path), + "output_dir": str(output_dir), + "processing_time": state.processing_time or 0, + "state": state.to_dict() + } + + def detect_and_split_for_project( + self, + video_path: Path, + project_id: str, + project_directory: Path, + material_tags: Optional[List[str]] = None, + detector_type: DetectorType = DetectorType.CONTENT, + threshold: float = 30.0, + min_scene_length: float = 1.0, + split_quality: int = 23, + split_preset: str = "fast", + max_video_duration: float = 2.0, + request_id: Optional[str] = None + ) -> Dict[str, Any]: + """ + 为项目导入单个视频素材 + + Args: + video_path: 视频文件路径 + project_id: 项目ID + project_directory: 项目目录 + material_tags: 素材标签 + 其他参数: 检测和切分配置 + + Returns: + Dict: 处理结果 + """ + + # 创建临时输出目录 + temp_output_dir = project_directory / "temp" / f"{video_path.stem}_processing" + + return self.detect_and_split_single_video( + video_path=video_path, + output_dir=temp_output_dir, + detector_type=detector_type, + threshold=threshold, + min_scene_length=min_scene_length, + output_format=OutputFormat.JSON, + enable_ai_analysis=False, + enable_video_splitting=True, + use_advanced_split=True, + split_quality=split_quality, + split_preset=split_preset, + max_video_duration=max_video_duration, + project_id=project_id, + project_directory=project_directory, + material_tags=material_tags, + request_id=request_id + ) + + def get_supported_formats(self) -> List[str]: + """获取支持的视频格式""" + return ['.mp4', '.avi', '.mov', '.mkv', '.wmv', '.flv', '.webm', '.m4v'] + + def validate_video_file(self, video_path: Path) -> bool: + """验证视频文件""" + if not video_path.exists(): + return False + + if not video_path.is_file(): + return False + + if video_path.suffix.lower() not in self.get_supported_formats(): + return False + + return True diff --git a/python_core/scene_detection/workflows/single_workflow_nodes.py b/python_core/scene_detection/workflows/single_workflow_nodes.py new file mode 100644 index 0000000..41201e8 --- /dev/null +++ b/python_core/scene_detection/workflows/single_workflow_nodes.py @@ -0,0 +1,315 @@ +""" +单个文件场景检测工作流节点 +""" + +import os +import json +import hashlib +from pathlib import Path +from typing import Dict, Any +from datetime import datetime + +from python_core.utils.logger import logger +from python_core.utils.jsonrpc_enhanced import ProgressLevel + +from ..services.detector_service import SceneDetectorService +from ..services.ai_analysis_service import AIAnalysisService +from ..utils.result_saver import ResultSaver +from ..types.single_workflow_state import SingleSceneDetectionWorkflowState, SingleVideoTask +from python_core.services.ffmpeg_slice_service_sync import FfmpegSliceService, SliceOptions, SliceSegment +from python_core.services.project_material_service import ProjectMaterialService + + +class SingleWorkflowNodes: + """单个文件场景检测工作流节点""" + + def __init__(self): + self.detector_service = SceneDetectorService() + self.ai_service = AIAnalysisService() + self.result_saver = ResultSaver() + self.splitter_service = FfmpegSliceService() + self.material_service = ProjectMaterialService() + + def process_single_video(self, state: SingleSceneDetectionWorkflowState) -> Dict[str, Any]: + """处理单个视频文件""" + try: + state.mark_task_started() + state.update_progress("started", 0.0, f"开始处理视频: {state.video_path.name}") + + logger.info(f"🎬 开始处理单个视频: {state.video_path.name}") + + # 1. 场景检测 + state.update_progress("detecting", 20.0, "正在进行场景检测...") + + # 转换检测器类型 + from ..types.enums import DetectorType + detector_type_enum = DetectorType(state.detector_type) + + detection_result = self.detector_service.detect_scenes( + video_path=state.video_path, + detector_type=detector_type_enum, + threshold=state.threshold, + min_scene_length=state.min_scene_length + ) + + logger.info(f"✅ 场景检测完成: 检测到 {detection_result.total_scenes} 个场景") + + # 2. 保存检测结果 + if state.output_dir: + state.update_progress("saving", 40.0, "保存检测结果...") + + # 转换输出格式 + from ..types.enums import OutputFormat + output_format_enum = OutputFormat(state.output_format) + + # 创建结果文件路径 + result_filename = f"{state.video_path.stem}_scenes.{output_format_enum.value}" + result_path = state.output_dir / result_filename + + self.result_saver.save_results( + detection_result, + result_path, + output_format_enum + ) + + # 3. AI分析（如果启用） + if state.enable_ai_analysis: + state.update_progress("ai_analysis", 60.0, "正在进行AI分析...") + try: + ai_result = self.ai_service.analyze_scenes(detection_result) + detection_result.ai_analysis = ai_result + logger.info("✅ AI分析完成") + except Exception as e: + logger.warning(f"⚠️ AI分析失败: {e}") + + # 4. 视频切分（如果启用） + split_results = [] + if state.enable_video_splitting and state.output_dir: + state.update_progress("splitting", 70.0, "正在进行视频切分...") + split_results = self._handle_video_splitting(state, detection_result) + + # 5. 项目素材管理（如果是项目导入） + if state.project_id and state.project_directory: + state.update_progress("project_import", 90.0, "导入到项目素材库...") + self._handle_project_material_import(state, detection_result, split_results) + + # 标记完成 + state.mark_task_completed(detection_result, split_results) + state.update_progress("completed", 100.0, "处理完成") + + logger.info(f"🎉 视频处理完成: {state.video_path.name}") + + return { + "success": True, + "detection_result": detection_result, + "split_results": split_results, + "processing_time": state.processing_time, + "total_scenes": state.total_scenes, + "total_segments": state.total_split_segments + } + + except Exception as e: + error_msg = f"处理视频失败: {str(e)}" + logger.error(f"❌ {error_msg}") + state.mark_task_failed(error_msg) + state.update_progress("failed", 0.0, error_msg) + + return { + "success": False, + "error": error_msg, + "processing_time": state.processing_time + } + + def _handle_video_splitting(self, state: SingleSceneDetectionWorkflowState, detection_result) -> list: + """处理视频切分""" + try: + # 创建切分选项 + slice_options = SliceOptions( + crf=state.split_quality, + preset=state.split_preset + ) + + # 转换场景为SliceSegment + segments = [ + SliceSegment(start=scene.start_time, end=scene.end_time) + for scene in detection_result.scenes + ] + + # 创建输出目录 + scenes_dir = state.output_dir / "scenes" + scenes_dir.mkdir(parents=True, exist_ok=True) + + # 检查是否无场景，如果是则跳过切分 + if len(segments) <= 0: + logger.info(f"🔄 检测到 0 个场景切换，跳过切分，直接使用源文件") + + # 直接复制源文件作为结果 + import shutil + source_file = state.video_path + target_file = scenes_dir / f"{state.video_path.stem}_scene_001.mp4" + + try: + shutil.copy2(source_file, target_file) + logger.info(f"✅ 源文件已复制到: {target_file}") + + # 获取源文件元数据 + metadata = self.splitter_service.get_video_metadata(str(source_file)) + + # 创建模拟的切分结果 + slice_results = [(str(target_file), metadata)] + + # 修正场景统计：无场景切换 = 1个场景（整个视频） + detection_result.total_scenes = 1 + + except Exception as e: + logger.error(f"❌ 复制源文件失败: {e}") + slice_results = [] + else: + logger.info(f"🎬 检测到 {len(segments)} 个场景，开始切分") + + # 生成输出路径 + base_output_path = str(scenes_dir / f"{state.video_path.stem}_scene") + + # 使用同步方法进行切分 + slice_results = self.splitter_service.slice_video( + media_path=str(state.video_path), + segments=segments, + options=slice_options, + output_path=base_output_path + ) + + # 转换为兼容格式 + split_results_raw = [] + + # 创建一个简单的结果对象 + class SplitResult: + def __init__(self, scene_index, output_path, start_time, end_time, duration, file_size, success, error=None): + self.scene_index = scene_index + self.output_path = Path(output_path) + self.start_time = start_time + self.end_time = end_time + self.duration = duration + self.file_size = file_size + self.success = success + self.error = error + + for i, (output_path, metadata) in enumerate(slice_results): + if len(detection_result.scenes) == 0: + # 无场景切换的情况：整个视频作为一个场景 + split_result = SplitResult( + scene_index=0, + output_path=output_path, + start_time=0.0, + end_time=metadata.duration, + duration=metadata.duration, + file_size=metadata.size, + success=Path(output_path).exists(), + error=None if Path(output_path).exists() else "文件不存在" + ) + split_results_raw.append(split_result) + else: + # 有场景切换的情况 + scene = detection_result.scenes[i] if i < len(detection_result.scenes) else None + if scene: + split_result = SplitResult( + scene_index=i, + output_path=output_path, + start_time=scene.start_time, + end_time=scene.end_time, + duration=scene.duration, + file_size=metadata.size, + success=Path(output_path).exists(), + error=None if Path(output_path).exists() else "文件不存在" + ) + split_results_raw.append(split_result) + + # 4. 添加视频时长检查，如果时长大于最大视频时长那么就要进行二次切分 + final_split_results = [] + + for split_result in split_results_raw: + try: + # 检查切分后的视频时长 + video_path = str(split_result.output_path) + + if split_result.duration > state.max_video_duration: + logger.info(f"⚠️ 片段 {split_result.scene_index + 1} 时长 {split_result.duration:.2f}s 超过限制 {state.max_video_duration:.2f}s，进行二次切分") + + # 进行二次切分 + secondary_results = self.splitter_service.check_and_split_by_duration( + video_path=video_path, + max_duration=state.max_video_duration, + options=slice_options, + output_dir=str(split_result.output_path.parent) + ) + + # 处理二次切分结果 + for i, (secondary_path, secondary_metadata) in enumerate(secondary_results): + if len(secondary_results) > 1: + # 如果进行了二次切分，创建新的结果对象 + secondary_split_result = SplitResult( + scene_index=split_result.scene_index, + output_path=secondary_path, + start_time=split_result.start_time + (i * state.max_video_duration), + end_time=min(split_result.start_time + ((i + 1) * state.max_video_duration), split_result.end_time), + duration=secondary_metadata.duration, + file_size=secondary_metadata.size, + success=Path(secondary_path).exists(), + error=None if Path(secondary_path).exists() else "二次切分文件不存在" + ) + final_split_results.append(secondary_split_result) + + # 删除原始的过长片段（如果二次切分成功） + if Path(video_path).exists() and len(secondary_results) > 1: + try: + Path(video_path).unlink() + logger.info(f"🗑️ 已删除过长的原始片段: {Path(video_path).name}") + except Exception as e: + logger.warning(f"⚠️ 删除原始片段失败: {e}") + else: + # 如果没有进行二次切分（时长检查通过），保持原结果 + final_split_results.append(split_result) + else: + # 时长未超过限制，直接添加到最终结果 + final_split_results.append(split_result) + + except Exception as e: + logger.error(f"❌ 处理片段 {split_result.scene_index + 1} 的时长检查时出错: {e}") + # 出错时保持原结果 + final_split_results.append(split_result) + + # 使用最终的切分结果 + split_results_raw = final_split_results + + # 更新统计信息 + logger.info(f"📊 最终生成 {len(split_results_raw)} 个视频片段（包含二次切分）") + + return split_results_raw + + except Exception as e: + logger.error(f"❌ 视频切分失败: {e}") + return [] + + def _handle_project_material_import(self, state: SingleSceneDetectionWorkflowState, detection_result, split_results): + """处理项目素材导入""" + try: + if not state.project_directory: + logger.warning("⚠️ 项目目录未设置，跳过素材导入") + return + + # 使用项目素材管理服务 + result = self.material_service.import_video_materials( + video_segments=split_results, + project_id=state.project_id, + project_directory=state.project_directory, + source_video_path=str(state.video_path), + material_tags=state.material_tags + ) + + if result["success"]: + logger.info(f"📝 项目素材导入完成: 新增 {result['imported_count']} 个素材，跳过 {result['skipped_count']} 个重复素材") + else: + logger.error(f"❌ 项目素材导入失败: {result.get('error', '未知错误')}") + + except Exception as e: + logger.error(f"❌ 项目素材导入失败: {e}") + diff --git a/python_core/services/project_material_service.py b/python_core/services/project_material_service.py new file mode 100644 index 0000000..ab3c9a7 --- /dev/null +++ b/python_core/services/project_material_service.py @@ -0,0 +1,298 @@ +""" +项目素材管理服务 +""" + +import json +import hashlib +from pathlib import Path +from typing import List, Dict, Any, Optional +from datetime import datetime + +from python_core.utils.logger import logger + + +class ProjectMaterialService: + """项目素材管理服务""" + + def __init__(self): + logger.info("ProjectMaterialService 初始化完成") + + def import_video_materials( + self, + video_segments: List[Any], + project_id: str, + project_directory: Path, + source_video_path: str, + material_tags: Optional[List[str]] = None + ) -> Dict[str, Any]: + """ + 导入视频素材到项目 + + Args: + video_segments: 视频片段列表 + project_id: 项目ID + project_directory: 项目目录 + source_video_path: 源视频路径 + material_tags: 素材标签 + + Returns: + Dict: 导入结果 + """ + + try: + # 创建项目素材目录 + uncategorized_dir = project_directory / "未分类文件夹" + uncategorized_dir.mkdir(parents=True, exist_ok=True) + + # 加载现有的项目素材列表 + material_list_file = project_directory / "project_material.json" + existing_materials = self._load_material_list(material_list_file) + + # 处理视频片段 + new_materials = [] + imported_count = 0 + skipped_count = 0 + + for segment in video_segments: + if not segment.success: + continue + + try: + result = self._import_single_segment( + segment=segment, + uncategorized_dir=uncategorized_dir, + existing_materials=existing_materials, + project_id=project_id, + source_video_path=source_video_path, + material_tags=material_tags or [] + ) + + if result["imported"]: + new_materials.append(result["material_info"]) + imported_count += 1 + logger.info(f"✅ 素材已导入: {result['original_filename']} -> {result['target_filename']}") + else: + skipped_count += 1 + logger.info(f"📋 素材已存在，跳过: {result['original_filename']}") + + except Exception as e: + logger.error(f"❌ 导入素材失败 {segment.output_path.name}: {e}") + skipped_count += 1 + + # 更新项目素材列表 + if new_materials: + existing_materials.extend(new_materials) + self._save_material_list(material_list_file, existing_materials) + logger.info(f"📝 项目素材列表已更新: 新增 {len(new_materials)} 个素材") + + return { + "success": True, + "imported_count": imported_count, + "skipped_count": skipped_count, + "total_count": len(video_segments), + "new_materials": new_materials + } + + except Exception as e: + logger.error(f"❌ 项目素材导入失败: {e}") + return { + "success": False, + "error": str(e), + "imported_count": 0, + "skipped_count": 0, + "total_count": len(video_segments) + } + + def _import_single_segment( + self, + segment: Any, + uncategorized_dir: Path, + existing_materials: List[Dict], + project_id: str, + source_video_path: str, + material_tags: List[str] + ) -> Dict[str, Any]: + """导入单个视频片段""" + + # 复制文件到项目目录 + import shutil + temp_target_path = uncategorized_dir / f"temp_{segment.output_path.name}" + shutil.copy2(segment.output_path, temp_target_path) + + # 计算复制后文件的MD5 + file_md5 = self._calculate_file_md5(temp_target_path) + + # 检查是否已存在相同MD5的素材 + if any(material.get('md5') == file_md5 for material in existing_materials): + # 删除临时文件 + temp_target_path.unlink() + return { + "imported": False, + "original_filename": segment.output_path.name, + "reason": "duplicate_md5" + } + + # 重命名为最终文件名 + target_filename = f"{file_md5}.mp4" + target_path = uncategorized_dir / target_filename + temp_target_path.rename(target_path) + + # 创建素材信息 + material_info = { + "id": file_md5, + "md5": file_md5, + "original_filename": segment.output_path.name, + "filename": target_filename, + "file_path": str(target_path), + "relative_path": f"未分类文件夹/{target_filename}", + "file_size": segment.file_size, + "duration": segment.duration, + "start_time": segment.start_time, + "end_time": segment.end_time, + "scene_index": segment.scene_index, + "tags": material_tags.copy(), + "created_at": datetime.now().isoformat(), + "use_count": 0, + "source_video": source_video_path, + "project_id": project_id + } + + return { + "imported": True, + "material_info": material_info, + "original_filename": segment.output_path.name, + "target_filename": target_filename + } + + def _load_material_list(self, material_list_file: Path) -> List[Dict]: + """加载项目素材列表""" + if not material_list_file.exists(): + return [] + + try: + with open(material_list_file, 'r', encoding='utf-8') as f: + return json.load(f) + except Exception as e: + logger.warning(f"⚠️ 读取现有素材列表失败: {e}") + return [] + + def _save_material_list(self, material_list_file: Path, materials: List[Dict]): + """保存项目素材列表""" + try: + with open(material_list_file, 'w', encoding='utf-8') as f: + json.dump(materials, f, ensure_ascii=False, indent=2) + except Exception as e: + logger.error(f"❌ 保存项目素材列表失败: {e}") + raise + + def _calculate_file_md5(self, file_path: Path) -> str: + """计算文件MD5值""" + hash_md5 = hashlib.md5() + try: + with open(file_path, "rb") as f: + for chunk in iter(lambda: f.read(4096), b""): + hash_md5.update(chunk) + return hash_md5.hexdigest() + except Exception as e: + logger.error(f"❌ 计算MD5失败 {file_path}: {e}") + # 如果计算MD5失败，使用文件名和大小作为备用标识 + return hashlib.md5(f"{file_path.name}_{file_path.stat().st_size}".encode()).hexdigest() + + def get_project_materials(self, project_directory: Path) -> List[Dict]: + """获取项目素材列表""" + material_list_file = project_directory / "project_material.json" + return self._load_material_list(material_list_file) + + def add_material_tags(self, project_directory: Path, material_id: str, tags: List[str]) -> bool: + """为素材添加标签""" + try: + material_list_file = project_directory / "project_material.json" + materials = self._load_material_list(material_list_file) + + for material in materials: + if material.get('id') == material_id: + existing_tags = set(material.get('tags', [])) + new_tags = existing_tags.union(set(tags)) + material['tags'] = list(new_tags) + + self._save_material_list(material_list_file, materials) + logger.info(f"✅ 素材标签已更新: {material_id}") + return True + + logger.warning(f"⚠️ 未找到素材: {material_id}") + return False + + except Exception as e: + logger.error(f"❌ 添加素材标签失败: {e}") + return False + + def remove_material(self, project_directory: Path, material_id: str) -> bool: + """删除项目素材""" + try: + material_list_file = project_directory / "project_material.json" + materials = self._load_material_list(material_list_file) + + # 查找要删除的素材 + material_to_remove = None + for i, material in enumerate(materials): + if material.get('id') == material_id: + material_to_remove = materials.pop(i) + break + + if material_to_remove: + # 删除文件 + file_path = Path(material_to_remove['file_path']) + if file_path.exists(): + file_path.unlink() + logger.info(f"🗑️ 已删除素材文件: {file_path.name}") + + # 更新素材列表 + self._save_material_list(material_list_file, materials) + logger.info(f"✅ 素材已删除: {material_id}") + return True + else: + logger.warning(f"⚠️ 未找到素材: {material_id}") + return False + + except Exception as e: + logger.error(f"❌ 删除素材失败: {e}") + return False + + def get_material_stats(self, project_directory: Path) -> Dict[str, Any]: + """获取项目素材统计信息""" + try: + materials = self.get_project_materials(project_directory) + + total_count = len(materials) + total_size = sum(material.get('file_size', 0) for material in materials) + total_duration = sum(material.get('duration', 0) for material in materials) + + # 按标签统计 + tag_stats = {} + for material in materials: + for tag in material.get('tags', []): + tag_stats[tag] = tag_stats.get(tag, 0) + 1 + + # 按使用次数统计 + used_count = sum(1 for material in materials if material.get('use_count', 0) > 0) + unused_count = total_count - used_count + + return { + "total_count": total_count, + "total_size": total_size, + "total_duration": total_duration, + "used_count": used_count, + "unused_count": unused_count, + "tag_stats": tag_stats + } + + except Exception as e: + logger.error(f"❌ 获取素材统计失败: {e}") + return { + "total_count": 0, + "total_size": 0, + "total_duration": 0, + "used_count": 0, + "unused_count": 0, + "tag_stats": {} + }