From 493e347b03d180705a4b45983f8d1da3c6cefb14 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 12 Jul 2025 14:38:15 +0800 Subject: [PATCH] fix --- examples/test_batch_workflow.py | 226 +++++ examples/test_video_validation.py | 154 +++ python_core/cli/commands/scene_detect.py | 134 +++ python_core/models/ffmpeg_tasks/__init__.py | 0 python_core/models/ffmpeg_tasks/models.py | 302 ++++++ python_core/scene_detection/scene_detector.py | 124 +-- .../scene_detection/services/__init__.py | 6 +- .../types/batch_workflow_state.py | 163 +++ .../workflows/batch_workflow_nodes.py | 254 +++++ .../workflows/workflow_manager.py | 113 ++- python_core/services/ffmpeg_slice_service.py | 705 +++++++++++++ python_core/utils/PathUtils.py | 84 ++ python_core/utils/TimeUtils.py | 163 +++ python_core/utils/VideoUtils.py | 953 ++++++++++++++++++ 14 files changed, 3267 insertions(+), 114 deletions(-) create mode 100644 examples/test_batch_workflow.py create mode 100644 examples/test_video_validation.py create mode 100644 python_core/models/ffmpeg_tasks/__init__.py create mode 100644 python_core/models/ffmpeg_tasks/models.py create mode 100644 python_core/scene_detection/types/batch_workflow_state.py create mode 100644 python_core/scene_detection/workflows/batch_workflow_nodes.py create mode 100644 python_core/services/ffmpeg_slice_service.py create mode 100644 python_core/utils/PathUtils.py create mode 100644 python_core/utils/TimeUtils.py create mode 100644 python_core/utils/VideoUtils.py diff --git a/examples/test_batch_workflow.py b/examples/test_batch_workflow.py new file mode 100644 index 0000000..503258c --- /dev/null +++ b/examples/test_batch_workflow.py @@ -0,0 +1,226 @@ +#!/usr/bin/env python3 +""" +Test Batch Scene Detection and Splitting Workflow +测试批量场景检测和切分工作流 +""" + +import sys +import json +import tempfile +from pathlib import Path + +# 添加项目根目录到Python路径 +project_root = Path(__file__).parent.parent +sys.path.insert(0, str(project_root)) + +from python_core.scene_detection import SceneDetector, DetectorType, OutputFormat + + +def test_batch_workflow(): + """测试批量工作流""" + print("🚀 测试批量场景检测和切分工作流") + print("=" * 60) + + try: + # 创建检测器 + detector = SceneDetector() + + # 准备测试视频 + test_videos = [ + Path("assets/1/1752032011698.mp4") + ] + + # 检查测试视频是否存在 + existing_videos = [v for v in test_videos if v.exists()] + if not existing_videos: + print("❌ 没有找到测试视频文件") + print("💡 请确保 assets/1/1752032011698.mp4 文件存在") + return False + + print(f"📹 找到 {len(existing_videos)} 个测试视频:") + for video in existing_videos: + print(f" • {video}") + + # 创建临时输出目录 + with tempfile.TemporaryDirectory() as temp_dir: + output_dir = Path(temp_dir) / "batch_output" + + print(f"\n📂 输出目录: {output_dir}") + + # 执行批量处理 + print("\n🔄 开始批量处理...") + result = detector.batch_detect_and_split( + video_paths=existing_videos, + output_base_dir=output_dir, + detector_type=DetectorType.CONTENT, + threshold=30.0, + min_scene_length=1.0, + output_format=OutputFormat.JSON, + enable_ai_analysis=False, + enable_video_splitting=True, + max_concurrent=1, # 使用单线程避免资源竞争 + continue_on_error=True + ) + + # 检查结果 + if result.get("workflow_state") == "completed": + print("✅ 批量处理完成!") + + # 显示摘要 + batch_results = result.get("batch_results", {}) + print(f"\n📊 处理摘要:") + print(f" 总视频数: {batch_results.get('total_videos', 0)}") + print(f" 成功处理: {batch_results.get('completed_videos', 0)}") + print(f" 处理失败: {batch_results.get('failed_videos', 0)}") + print(f" 成功率: {batch_results.get('success_rate', 0):.1f}%") + + # 显示每个任务的详细信息 + tasks = batch_results.get('tasks', []) + print(f"\n📋 任务详情:") + for i, task in enumerate(tasks, 1): + video_name = Path(task['video_path']).name + status = task['status'] + scenes = task.get('total_scenes', 0) + splits = task.get('split_count', 0) + proc_time = task.get('processing_time', 0) + + print(f" 任务 {i}: {video_name}") + print(f" 状态: {status}") + print(f" 场景数: {scenes}") + print(f" 切分数: {splits}") + print(f" 处理时间: {proc_time:.2f}s") + + if task.get('error'): + print(f" 错误: {task['error']}") + + # 检查输出文件 + if task.get('output_dir'): + output_path = Path(task['output_dir']) + if output_path.exists(): + print(f" 输出目录: {output_path}") + + # 检查场景检测结果文件 + scenes_file = output_path / "scenes.json" + if scenes_file.exists(): + print(f" ✅ 场景检测结果: {scenes_file}") + + # 检查切分目录 + scenes_dir = output_path / "scenes" + if scenes_dir.exists(): + split_files = list(scenes_dir.glob("*.mp4")) + print(f" ✅ 切分文件: {len(split_files)} 个") + for split_file in split_files[:3]: # 只显示前3个 + size_mb = split_file.stat().st_size / (1024 * 1024) + print(f" • {split_file.name} ({size_mb:.1f}MB)") + if len(split_files) > 3: + print(f" ... 还有 {len(split_files) - 3} 个文件") + + # 检查切分摘要 + summary_file = output_path / "split_summary.json" + if summary_file.exists(): + print(f" ✅ 切分摘要: {summary_file}") + try: + with open(summary_file, 'r', encoding='utf-8') as f: + summary_data = json.load(f) + print(f" 成功切分: {summary_data.get('successful_splits', 0)}") + print(f" 失败切分: {summary_data.get('failed_splits', 0)}") + print(f" 总输出大小: {summary_data.get('total_output_size', 0):,} bytes") + except Exception as e: + print(f" ⚠️ 读取摘要失败: {e}") + + return True + else: + print("❌ 批量处理失败") + errors = result.get("errors", []) + if errors: + print("错误信息:") + for error in errors: + print(f" • {error}") + return False + + except Exception as e: + print(f"❌ 测试异常: {e}") + import traceback + print(f"详细错误: {traceback.format_exc()}") + return False + + +def test_cli_batch_command(): + """测试CLI批量命令""" + print("\n🧪 测试CLI批量命令") + print("=" * 60) + + try: + # 检查CLI帮助 + import subprocess + result = subprocess.run( + ['python3', '-m', 'python_core.cli', 'scene', 'batch', '--help'], + capture_output=True, + text=True, + cwd=project_root + ) + + if result.returncode == 0: + print("✅ CLI批量命令帮助正常") + print("📋 命令帮助预览:") + help_lines = result.stdout.split('\n')[:10] # 只显示前10行 + for line in help_lines: + if line.strip(): + print(f" {line}") + if len(result.stdout.split('\n')) > 10: + print(" ...") + return True + else: + print("❌ CLI批量命令帮助失败") + print(f"错误: {result.stderr}") + return False + + except Exception as e: + print(f"❌ CLI测试异常: {e}") + return False + + +def main(): + """主函数""" + print("🚀 开始测试批量场景检测和切分工作流") + print("=" * 80) + + tests = [ + ("批量工作流功能", test_batch_workflow), + ("CLI批量命令", test_cli_batch_command), + ] + + passed = 0 + total = len(tests) + + for test_name, test_func in tests: + try: + if test_func(): + passed += 1 + print(f"✅ {test_name} - 通过") + else: + print(f"❌ {test_name} - 失败") + except Exception as e: + print(f"❌ {test_name} - 异常: {e}") + + print("\n" + "=" * 80) + print(f"🎉 测试完成: {passed}/{total} 通过") + + if passed == total: + print("🎊 所有测试都通过了！批量工作流开发成功！") + print("\n💡 使用方法:") + print(" # 批量处理目录中的所有视频") + print(" python3 -m python_core.cli scene batch input_dir output_dir") + print(" ") + print(" # 自定义参数") + print(" python3 -m python_core.cli scene batch input_dir output_dir \\") + print(" --threshold 15.0 --concurrent 4 --no-ai --split") + else: + print("⚠️ 部分测试失败，需要进一步调试") + + return passed == total + + +if __name__ == "__main__": + success = main() + sys.exit(0 if success else 1) diff --git a/examples/test_video_validation.py b/examples/test_video_validation.py new file mode 100644 index 0000000..7a59b3c --- /dev/null +++ b/examples/test_video_validation.py @@ -0,0 +1,154 @@ +#!/usr/bin/env python3 +""" +Video Validation Tool +视频验证工具 + +验证切分后的视频文件是否有效 +""" + +import sys +import subprocess +from pathlib import Path + + +def validate_video_file(video_path: Path) -> dict: + """验证单个视频文件""" + try: + # 使用ffprobe获取视频信息 + cmd = [ + 'ffprobe', + '-v', 'quiet', + '-print_format', 'json', + '-show_format', + '-show_streams', + str(video_path) + ] + + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=30 + ) + + if result.returncode == 0: + import json + data = json.loads(result.stdout) + + # 提取基本信息 + format_info = data.get('format', {}) + streams = data.get('streams', []) + + video_stream = None + audio_stream = None + + for stream in streams: + if stream.get('codec_type') == 'video': + video_stream = stream + elif stream.get('codec_type') == 'audio': + audio_stream = stream + + return { + 'valid': True, + 'file_size': int(format_info.get('size', 0)), + 'duration': float(format_info.get('duration', 0)), + 'bitrate': int(format_info.get('bit_rate', 0)), + 'video_codec': video_stream.get('codec_name') if video_stream else None, + 'video_resolution': f"{video_stream.get('width')}x{video_stream.get('height')}" if video_stream else None, + 'video_fps': eval(video_stream.get('r_frame_rate', '0/1')) if video_stream else 0, + 'audio_codec': audio_stream.get('codec_name') if audio_stream else None, + 'audio_sample_rate': int(audio_stream.get('sample_rate', 0)) if audio_stream else 0, + 'error': None + } + else: + return { + 'valid': False, + 'error': f"ffprobe failed: {result.stderr}" + } + + except Exception as e: + return { + 'valid': False, + 'error': str(e) + } + + +def validate_split_videos(scenes_dir: Path): + """验证切分目录中的所有视频""" + print(f"🔍 验证目录: {scenes_dir}") + + if not scenes_dir.exists(): + print(f"❌ 目录不存在: {scenes_dir}") + return False + + # 查找所有mp4文件 + video_files = list(scenes_dir.glob("*.mp4")) + + if not video_files: + print(f"❌ 目录中没有找到mp4文件") + return False + + print(f"📹 找到 {len(video_files)} 个视频文件") + + valid_count = 0 + total_size = 0 + total_duration = 0 + + for video_file in sorted(video_files): + print(f"\n🎬 验证: {video_file.name}") + + validation = validate_video_file(video_file) + + if validation['valid']: + valid_count += 1 + file_size = validation['file_size'] + duration = validation['duration'] + total_size += file_size + total_duration += duration + + print(f" ✅ 有效") + print(f" 📏 大小: {file_size:,} bytes ({file_size/1024/1024:.1f} MB)") + print(f" ⏱️ 时长: {duration:.2f}秒") + print(f" 🎥 视频: {validation['video_codec']} {validation['video_resolution']} {validation['video_fps']:.1f}fps") + if validation['audio_codec']: + print(f" 🔊 音频: {validation['audio_codec']} {validation['audio_sample_rate']}Hz") + print(f" 📊 码率: {validation['bitrate']:,} bps") + else: + print(f" ❌ 无效: {validation['error']}") + + print(f"\n📊 验证摘要:") + print(f" 有效文件: {valid_count}/{len(video_files)}") + print(f" 成功率: {valid_count/len(video_files)*100:.1f}%") + print(f" 总大小: {total_size:,} bytes ({total_size/1024/1024:.1f} MB)") + print(f" 总时长: {total_duration:.2f}秒") + + return valid_count == len(video_files) + + +def main(): + """主函数""" + print("🚀 视频验证工具") + print("=" * 50) + + # 验证最新的切分结果 + test_scenes_dir = Path("/tmp/test_batch_output/1752032011698/scenes") + + if test_scenes_dir.exists(): + print("🎯 验证最新的批量切分结果") + success = validate_split_videos(test_scenes_dir) + + if success: + print("\n🎉 所有视频文件都有效！切分成功！") + return 0 + else: + print("\n⚠️ 部分视频文件无效") + return 1 + else: + print("❌ 没有找到测试切分结果") + print("💡 请先运行批量切分命令:") + print(" python3 -m python_core.cli scene batch /tmp/test_batch_input /tmp/test_batch_output --verbose --no-ai") + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/python_core/cli/commands/scene_detect.py b/python_core/cli/commands/scene_detect.py index 697a897..d9092ed 100644 --- a/python_core/cli/commands/scene_detect.py +++ b/python_core/cli/commands/scene_detect.py @@ -95,5 +95,139 @@ def _display_scenes_table(scenes): console.print(table) +@scene_detect.command("batch") +def batch_detect_and_split( + input_dir: Path = typer.Argument(..., help="包含视频文件的输入目录"), + output_dir: Path = typer.Argument(..., help="输出目录"), + detector_type: DetectorType = typer.Option(DetectorType.CONTENT, "--detector", "-d", help="检测器类型"), + threshold: float = typer.Option(30.0, "--threshold", "-t", help="检测阈值"), + min_scene_length: float = typer.Option(1.0, "--min-length", "-m", help="最小场景长度(秒)"), + output_format: OutputFormat = typer.Option(OutputFormat.JSON, "--format", "-f", help="输出格式"), + ai_analysis: bool = typer.Option(False, "--ai/--no-ai", help="启用/禁用AI分析"), + video_splitting: bool = typer.Option(True, "--split/--no-split", help="启用/禁用视频切分"), + max_concurrent: int = typer.Option(2, "--concurrent", "-c", help="最大并发数"), + continue_on_error: bool = typer.Option(True, "--continue/--stop-on-error", help="遇到错误时继续/停止"), + file_pattern: str = typer.Option("*.mp4", "--pattern", "-p", 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)"), + verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出") +): + """批量场景检测和视频切分""" + console.print(f"🔄 批量处理目录: [bold blue]{input_dir}[/bold blue]") + console.print(f"📂 输出目录: [bold blue]{output_dir}[/bold blue]") + + try: + # 检查输入目录 + if not input_dir.exists() or not input_dir.is_dir(): + console.print(f"❌ 输入目录不存在或不是目录: [bold red]{input_dir}[/bold red]") + raise typer.Exit(1) + + # 查找视频文件 + video_extensions = ['*.mp4', '*.avi', '*.mov', '*.mkv', '*.wmv', '*.flv', '*.webm', '*.m4v'] + + video_files = [] + if file_pattern in video_extensions: + # 使用指定的模式 + video_files = list(input_dir.glob(file_pattern)) + else: + # 使用自定义模式 + video_files = list(input_dir.glob(file_pattern)) + # 如果自定义模式没找到文件，尝试所有支持的格式 + if not video_files: + for pattern in video_extensions: + video_files.extend(input_dir.glob(pattern)) + + if not video_files: + console.print(f"❌ 在目录中未找到视频文件: [bold red]{input_dir}[/bold red]") + console.print(f"💡 尝试的模式: {file_pattern}") + raise typer.Exit(1) + + console.print(f"📹 找到 {len(video_files)} 个视频文件") + + # 创建检测器 + detector = SceneDetector() + + # 执行批量处理 + result = detector.batch_detect_and_split( + video_paths=video_files, + output_base_dir=output_dir, + detector_type=detector_type, + threshold=threshold, + min_scene_length=min_scene_length, + output_format=output_format, + enable_ai_analysis=ai_analysis, + enable_video_splitting=video_splitting, + max_concurrent=max_concurrent, + continue_on_error=continue_on_error, + use_advanced_split=use_advanced_split, + split_quality=split_quality, + split_preset=split_preset + ) + + # 显示结果 + if result.get("workflow_state") == "completed": + summary = result.get("batch_results", {}) + console.print(f"\n✅ 批量处理完成!") + console.print(f"📊 处理统计:") + console.print(f" 总视频数: {summary.get('total_videos', 0)}") + console.print(f" 成功处理: {summary.get('completed_videos', 0)}") + console.print(f" 处理失败: {summary.get('failed_videos', 0)}") + console.print(f" 成功率: {summary.get('success_rate', 0):.1f}%") + + if video_splitting: + tasks_data = summary.get('tasks', []) + if tasks_data: + total_scenes = sum(task.get('total_scenes', 0) for task in tasks_data) + total_splits = sum(task.get('split_count', 0) for task in tasks_data) + console.print(f" 总场景数: {total_scenes}") + console.print(f" 切分片段: {total_splits}") + else: + console.print(" ⚠️ 无任务数据") + + # 显示详细结果 + if verbose: + tasks = summary.get('tasks', []) + if tasks: + _display_batch_results_table(tasks) + else: + console.print(" ⚠️ 无详细任务数据可显示") + else: + console.print(f"❌ 批量处理失败") + errors = result.get("errors", []) + if errors: + for error in errors: + console.print(f" • {error}") + raise typer.Exit(1) + + except Exception as e: + console.print(f"❌ 执行失败: [bold red]{str(e)}[/bold red]") + raise typer.Exit(1) + + +def _display_batch_results_table(tasks): + """显示批量处理结果表格""" + 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") + table.add_column("错误", style="red") + + for task in tasks: + video_name = Path(task["video_path"]).name + status = "✅ 成功" if task["status"] == "completed" else "❌ 失败" + scenes = str(task.get("total_scenes", 0)) + splits = str(task.get("split_count", 0)) + proc_time = f"{task.get('processing_time', 0):.1f}s" + error_text = task.get("error") or "" + error = error_text[:50] + "..." if len(error_text) > 50 else error_text + + table.add_row(video_name, status, scenes, splits, proc_time, error) + + console.print(table) + + if __name__ == "__main__": scene_detect() \ No newline at end of file diff --git a/python_core/models/ffmpeg_tasks/__init__.py b/python_core/models/ffmpeg_tasks/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/python_core/models/ffmpeg_tasks/models.py b/python_core/models/ffmpeg_tasks/models.py new file mode 100644 index 0000000..b6cf7ff --- /dev/null +++ b/python_core/models/ffmpeg_tasks/models.py @@ -0,0 +1,302 @@ +import json +from typing import Union, Any, Optional, List, Dict + +from loguru import logger +from pydantic import BaseModel, Field, computed_field, field_validator, model_validator, ConfigDict, HttpUrl +from pydantic.json_schema import JsonSchemaValue +from datetime import timedelta, datetime + + +class TimeDelta(timedelta): + @classmethod + def from_timedelta(cls, delta: timedelta) -> "TimeDelta": + return cls(days=delta.days, seconds=delta.seconds, microseconds=delta.microseconds) + + @classmethod + def from_format_string(cls, format_string: str) -> "TimeDelta": + formated_time = datetime.strptime(format_string, "%H:%M:%S.%f") + return cls(hours=formated_time.hour, minutes=formated_time.minute, seconds=formated_time.second, + microseconds=formated_time.microsecond) + + def toFormatStr(self) -> str: + return (datetime(year=2000, month=1, day=1) + self).strftime("%H:%M:%S.%f")[:-3] + + +class FFMpegSliceSegment(BaseModel): + start: TimeDelta = Field( + description="视频切割的开始时间点秒数, 可为浮点小数(精确到小数点后3位,毫秒级)或者为标准格式的时间戳") + end: TimeDelta = Field( + description="视频切割的结束时间点秒数, 可为浮点小数(精确到小数点后3位,毫秒级)或者标准格式的时间戳") + + @computed_field + @property + def duration(self) -> TimeDelta: + return self.end - self.start + + @field_validator('start', mode='before') + @classmethod + def parse_start(cls, v: Union[float, str, TimeDelta]): + if isinstance(v, float): + if v < 0.0: + raise ValueError("开始时间点不能小于0") + return TimeDelta(seconds=v) + elif isinstance(v, int): + if v < 0: + raise ValueError("开始时间点不能小于0") + return TimeDelta(seconds=v) + elif isinstance(v, str): + timedelta = TimeDelta.from_format_string(v) + if timedelta.total_seconds() < 0.0: + raise ValueError("开始时间点不能小于0") + return timedelta + elif isinstance(v, TimeDelta): + if v.total_seconds() < 0.0: + raise ValueError("开始时间点不能小于0") + return v + else: + raise TypeError(v) + + @field_validator('end', mode='before') + @classmethod + def parse_end(cls, v: Union[float, TimeDelta]): + if isinstance(v, float): + if v < 0.0: + raise ValueError("结束时间点不能小于0") + return TimeDelta(seconds=v) + elif isinstance(v, int): + if v < 0: + raise ValueError("结束时间点不能小于0") + return TimeDelta(seconds=v) + elif isinstance(v, str): + timedelta = TimeDelta.from_format_string(v) + if timedelta.total_seconds() < 0.0: + raise ValueError("结束时间点不能小于0") + return timedelta + elif isinstance(v, TimeDelta): + if v.total_seconds() < 0.0: + raise ValueError("结束时间点不能小于0") + return v + else: + raise TypeError(v) + + @model_validator(mode='after') + def validate_end_after_start(self) -> 'FFMpegSliceSegment': + if self.end <= self.start: + raise ValueError("end time must be greater than start time") + return self + + @classmethod + def __get_pydantic_json_schema__(cls, core_schema: Any, handler: Any) -> JsonSchemaValue: + # Override the schema to represent it as a string + return { + "type": "object", + "properties": { + "start": { + "type": "number", + "examples": [5, 10.5, '00:00:10.500'], + }, + "end": { + "type": "number", + "examples": [8, 12.5, '00:00:12.500'], + } + }, + "required": [ + "start", + "end" + ] + } + + model_config = { + "arbitrary_types_allowed": True + } + + +class FFMPEGSliceOptions(BaseModel): + limit_size: Optional[int] = Field(default=None, description="不超过指定文件(字节)大小, 默认为空不限制输出大小") + bit_rate: Optional[int] = Field(default=None, description="指定输出视频的比特率, 不能与limit_size同时设置") + + crf: int = Field(default=16, description="输出视频的质量") + fps: int = Field(default=30, description="输出视频的FPS") + width: int = Field(default=None, description="输出视频的宽(像素)") + height: int = Field(default=None, description="输出视频的高(像素)") + + @computed_field(description="解析为字符表达式的文件大小") + @property + def pretty_limit_size(self) -> Optional[str]: + if self.limit_size is not None: + for unit in ['B', 'KB', 'MB', 'GB', 'TB']: + if self.limit_size < 1024.0: + return f"{self.limit_size:.2f}{unit}" + self.limit_size /= 1024.0 + return f"{self.limit_size:.2f}PB" + else: + return None + + @computed_field(description="解析为字符表达式的比特率") + @property + def pretty_bit_rate(self) -> Optional[str]: + if self.bit_rate is not None: + return f"{self.bit_rate}k" + + +class MediaStream(BaseModel): + duration: float = Field(0, description="时长") + codec_name: str + tags: Optional[Any] = Field(None) + + model_config = ConfigDict(extra='allow') + + +class HLSMediaVideoStream(BaseModel): + stream_type: str = "video" + codec_name: str + codec_type: str + width: int + height: int + avg_frame_rate: str + tags: Optional[Any] = Field(None) + duration: Optional[float] = Field(None) + + @computed_field + @property + def video_frame_rate(self) -> float: + numerator, denominator = map(int, self.avg_frame_rate.split('/')) + if denominator != 0: + return numerator / denominator + return 0 + + +class HLSMediaAudioStream(BaseModel): + stream_type: str = "audio" + sample_rate: str + channels: int + channel_layout: str + start_time: str + tags: Optional[Any] = Field(None) + + +class AudioStream(MediaStream): + stream_type: str = Field("audio") + codec_type: str + sample_rate: str + channels: int + tags: Optional[Any] = Field(None) + + model_config = ConfigDict(extra='allow') + + +class VideoStream(MediaStream): + stream_type: str = "video" + width: int + height: int + bit_rate: int + avg_frame_rate: str + + @computed_field + @property + def video_bitrate(self) -> str: + return str(int(self.bit_rate / 1000)) + 'k' + + @computed_field + @property + def video_frame_rate(self) -> float: + numerator, denominator = map(int, self.avg_frame_rate.split('/')) + if denominator != 0: + return numerator / denominator + return 0 + + model_config = ConfigDict(extra='allow') + + +class ImageStream(MediaStream): + stream_type: str = "image" + width: int + height: int + + model_config = ConfigDict(extra='allow') + + +class SubtitleStreamTags(BaseModel): + language: str = Field(description="内嵌字幕的语言") + + model_config = ConfigDict(extra='allow') + + +class SubtitleStream(MediaStream): + stream_type: str = "subtitle" + tags: SubtitleStreamTags + + model_config = ConfigDict(extra='allow') + + +class VideoFormat(BaseModel): + filename: str + format_name: str + start_time: float = Field(0, description="起始时间") + size: int + bit_rate: Optional[int] = Field(None, description="文件比特率") + duration: float = Field(3600 * 12, description="文件时长") + + +class VideoMetadata(BaseModel): + streams: List[ + Union[ImageStream, AudioStream, VideoStream, HLSMediaAudioStream, HLSMediaVideoStream, SubtitleStream]] = Field( + description="媒体包含的数据轨道") + format: Optional[VideoFormat] = Field(None) + + @field_validator('streams', mode='before') + def parse_streams(cls, value): + streams = [] + if isinstance(value, List): + for stream in value: + if isinstance(stream, Dict): + logger.info(f"Parsing stream : {json.dumps(stream, ensure_ascii=False)}") + if stream.get("codec_type") == 'audio': + if stream.get("duration") is None: + logger.info("Parsing audio stream") + hls_audio = HLSMediaAudioStream.model_validate(stream) + streams.append(hls_audio) + else: + logger.info("Parsing hls audio stream") + audio = AudioStream.model_validate(stream) + streams.append(audio) + elif stream.get("codec_type") == 'video': + if stream.get("codec_name") in ("gif", "png", "mjpg", "mjpeg", "webp"): + logger.info("Parsing image stream") + image = ImageStream.model_validate(stream) + streams.append(image) + else: + if stream.get("duration") is None: + logger.info("Parsing hls video stream") + hls_video = HLSMediaVideoStream.model_validate(stream) + streams.append(hls_video) + else: + logger.info("Parsing video stream") + video = VideoStream.model_validate(stream) + streams.append(video) + elif stream.get("codec_type") == 'subtitle': + logger.info("Parsing subtitle stream") + subtitle_stream = SubtitleStream.model_validate(stream) + streams.append(subtitle_stream) + return streams + else: + raise TypeError + + model_config = ConfigDict(extra='allow') + + +class SentryTransactionHeader(BaseModel): + x_trace_id: Optional[str] = Field(description="Sentry Transaction ID", default=None) + x_baggage: Optional[str] = Field(description="Sentry Transaction baggage", default=None) + + +class SentryTransactionInfo(BaseModel): + x_trace_id: str = Field(description="Sentry Transaction ID") + x_baggage: str = Field(description="Sentry Transaction baggage") + + +class FFMPEGResult(BaseModel): + urn: str = Field(description="FFMPEG任务结果urn") + content_length: int = Field(description="媒体资源文件字节大小(Byte)") + metadata: VideoMetadata = Field(description="媒体元数据") + diff --git a/python_core/scene_detection/scene_detector.py b/python_core/scene_detection/scene_detector.py index 17cf1c1..a4a6314 100644 --- a/python_core/scene_detection/scene_detector.py +++ b/python_core/scene_detection/scene_detector.py @@ -114,115 +114,17 @@ class SceneDetector: return results - # JSON-RPC方法 - def jsonrpc_detect_scenes(self, video_path: str, detector_type: str = "content", - threshold: float = 30.0, min_scene_length: float = 1.0) -> Dict[str, Any]: - """JSON-RPC方法：基础场景检测""" - try: - result = self.detect_scenes( - Path(video_path), - DetectorType(detector_type), - threshold, - min_scene_length - ) - - return asdict(result) - - except Exception as e: - return { - "success": False, - "error": str(e) - } + def batch_detect_and_split(self, video_paths: List[Path], output_base_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, + max_concurrent: int = 2, continue_on_error: bool = True, + use_advanced_split: bool = True, split_quality: int = 23, split_preset: str = "fast", + request_id: Optional[str] = None) -> Dict[str, Any]: + """批量场景检测和视频切分""" + return self.workflow_manager.batch_detect_and_split( + video_paths, output_base_dir, detector_type, threshold, min_scene_length, + output_format, enable_ai_analysis, enable_video_splitting, + max_concurrent, continue_on_error, use_advanced_split, split_quality, split_preset, request_id + ) - def jsonrpc_get_video_info(self, video_path: str) -> Dict[str, Any]: - """JSON-RPC方法：获取视频信息""" - try: - return self.get_video_info(Path(video_path)) - except Exception as e: - return { - "success": False, - "error": str(e) - } - - def jsonrpc_detect_with_workflow(self, video_path: str, detector_type: str = "content", - threshold: float = 30.0, min_scene_length: float = 1.0, - output_path: Optional[str] = None, output_format: str = "json", - enable_ai_analysis: bool = True, request_id: Optional[str] = None) -> Optional[Dict[str, Any]]: - """JSON-RPC方法：工作流场景检测""" - try: - output_path_obj = Path(output_path) if output_path else None - - result = self.detect_with_workflow( - Path(video_path), - DetectorType(detector_type), - threshold, - min_scene_length, - output_path_obj, - OutputFormat(output_format), - enable_ai_analysis, - request_id - ) - - # 检查是否是JSON-RPC模式 - if result.get("jsonrpc_mode"): - # JSON-RPC模式：结果已经通过工作流发送，返回None - return None - else: - # 非JSON-RPC模式：序列化并返回结果 - serialized_result = {} - for key, value in result.items(): - if key == "detection_result" and value: - serialized_result[key] = asdict(value) - else: - serialized_result[key] = value - return serialized_result - - except Exception as e: - # 如果有request_id，错误已经在detect_with_workflow中发送 - if request_id: - return None - else: - return { - "success": False, - "error": str(e) - } - - def jsonrpc_batch_detect(self, video_paths: List[str], detector_type: str = "content", - threshold: float = 30.0, min_scene_length: float = 1.0, - output_dir: Optional[str] = None, output_format: str = "json") -> Dict[str, Any]: - """JSON-RPC方法：批量检测""" - try: - paths = [Path(p) for p in video_paths] - output_dir_obj = Path(output_dir) if output_dir else None - - results = self.batch_detect( - paths, - DetectorType(detector_type), - threshold, - min_scene_length, - output_dir_obj, - OutputFormat(output_format) - ) - - return { - "success": True, - "results": results, - "total_videos": len(video_paths), - "successful_detections": sum(1 for r in results if r["success"]) - } - - except Exception as e: - return { - "success": False, - "error": str(e) - } - - def register_jsonrpc_methods(self): - """注册JSON-RPC方法到全局注册器""" - from python_core.utils.jsonrpc_enhanced import method_registry - - # 直接注册方法 - method_registry.register_function(self.jsonrpc_detect_scenes, "scene.detect") - method_registry.register_function(self.jsonrpc_detect_with_workflow, "scene.detect_workflow") - method_registry.register_function(self.jsonrpc_get_video_info, "scene.get_video_info") - method_registry.register_function(self.jsonrpc_batch_detect, "scene.batch_detect") diff --git a/python_core/scene_detection/services/__init__.py b/python_core/scene_detection/services/__init__.py index 01e4456..9d1b532 100644 --- a/python_core/scene_detection/services/__init__.py +++ b/python_core/scene_detection/services/__init__.py @@ -9,9 +9,11 @@ Scene Detection Services from .detector_service import SceneDetectorService from .ai_analysis_service import AIAnalysisService from .video_info_service import VideoInfoService +from .video_splitter_service import VideoSplitterService __all__ = [ "SceneDetectorService", - "AIAnalysisService", - "VideoInfoService" + "AIAnalysisService", + "VideoInfoService", + "VideoSplitterService" ] diff --git a/python_core/scene_detection/types/batch_workflow_state.py b/python_core/scene_detection/types/batch_workflow_state.py new file mode 100644 index 0000000..da250cd --- /dev/null +++ b/python_core/scene_detection/types/batch_workflow_state.py @@ -0,0 +1,163 @@ +#!/usr/bin/env python3 +""" +Batch Workflow State +批量工作流状态定义 +""" + +from dataclasses import dataclass, field +from typing import Dict, Any, List, Optional +from pathlib import Path +from .models import SceneInfo, DetectionResult +from python_core.utils.jsonrpc_enhanced import EnhancedJSONRPCResponse, ProgressLevel + + +@dataclass +class BatchVideoTask: + """批量视频处理任务""" + video_path: Path + output_dir: Optional[Path] = None + status: str = "pending" # pending, processing, completed, failed + detection_result: Optional[DetectionResult] = None + split_results: List[Dict[str, Any]] = field(default_factory=list) + error: Optional[str] = None + start_time: Optional[float] = None + end_time: Optional[float] = None + + +@dataclass +class BatchSceneDetectionWorkflowState: + """批量场景检测工作流状态""" + # 输入参数 + video_paths: List[Path] = field(default_factory=list) + output_base_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_concurrent: int = 2 + continue_on_error: bool = True + + # 工作流状态 + current_stage: str = "init" + current_task_index: int = 0 + total_tasks: int = 0 + completed_tasks: int = 0 + failed_tasks: int = 0 + + # JSON-RPC支持 + request_id: Optional[str] = None + enable_jsonrpc: bool = False + + # 任务列表 + tasks: List[BatchVideoTask] = field(default_factory=list) + + # 全局结果 + batch_results: Dict[str, Any] = field(default_factory=dict) + global_errors: List[str] = field(default_factory=list) + + def __post_init__(self): + if not self.tasks and self.video_paths: + # 从视频路径创建任务 + self.tasks = [ + BatchVideoTask( + video_path=video_path, + output_dir=self.output_base_dir / video_path.stem if self.output_base_dir else None + ) + for video_path in self.video_paths + ] + self.total_tasks = len(self.tasks) + + def get_jsonrpc_handler(self) -> Optional[EnhancedJSONRPCResponse]: + """获取JSON-RPC响应处理器""" + if self.enable_jsonrpc and self.request_id: + return EnhancedJSONRPCResponse(self.request_id) + return None + + def send_progress(self, step: str, message: str, level: ProgressLevel = ProgressLevel.INFO, + data: Optional[Dict[str, Any]] = None) -> None: + """发送进度更新""" + handler = self.get_jsonrpc_handler() + if handler: + # 计算总体进度 + if self.total_tasks > 0: + progress_percent = int((self.completed_tasks / self.total_tasks * 100)) + else: + progress_percent = -1 + + # 添加批量处理特定数据 + progress_data = { + "current_task": self.current_task_index + 1, + "total_tasks": self.total_tasks, + "completed_tasks": self.completed_tasks, + "failed_tasks": self.failed_tasks, + "current_stage": self.current_stage + } + + if data: + progress_data.update(data) + + handler.progress(step, progress_percent, message, level, progress_data) + + def send_final_result(self, result: Dict[str, Any]) -> None: + """发送最终结果""" + handler = self.get_jsonrpc_handler() + if handler: + handler.success(result) + + def send_error_result(self, error_code: int, error_message: str, error_data: Any = None) -> None: + """发送错误结果""" + handler = self.get_jsonrpc_handler() + if handler: + handler.error(error_code, error_message, error_data) + + def get_current_task(self) -> Optional[BatchVideoTask]: + """获取当前任务""" + if 0 <= self.current_task_index < len(self.tasks): + return self.tasks[self.current_task_index] + return None + + def mark_task_completed(self, task_index: int, result: DetectionResult, split_results: List[Dict[str, Any]] = None): + """标记任务完成""" + if 0 <= task_index < len(self.tasks): + task = self.tasks[task_index] + task.status = "completed" + task.detection_result = result + task.split_results = split_results or [] + self.completed_tasks += 1 + + def mark_task_failed(self, task_index: int, error: str): + """标记任务失败""" + if 0 <= task_index < len(self.tasks): + task = self.tasks[task_index] + task.status = "failed" + task.error = error + self.failed_tasks += 1 + + def get_summary(self) -> Dict[str, Any]: + """获取批量处理摘要""" + return { + "total_tasks": self.total_tasks, + "completed_tasks": self.completed_tasks, + "failed_tasks": self.failed_tasks, + "success_rate": (self.completed_tasks / self.total_tasks * 100) if self.total_tasks > 0 else 0, + "current_stage": self.current_stage, + "tasks": [ + { + "video_path": str(task.video_path), + "status": task.status, + "total_scenes": task.detection_result.total_scenes if task.detection_result else 0, + "split_count": len(task.split_results), + "error": task.error + } + for task in self.tasks + ] + } diff --git a/python_core/scene_detection/workflows/batch_workflow_nodes.py b/python_core/scene_detection/workflows/batch_workflow_nodes.py new file mode 100644 index 0000000..39a3380 --- /dev/null +++ b/python_core/scene_detection/workflows/batch_workflow_nodes.py @@ -0,0 +1,254 @@ +#!/usr/bin/env python3 +""" +Batch Workflow Nodes +批量工作流节点 +""" + +import time +import asyncio +from pathlib import Path +from typing import Dict, Any, List +from concurrent.futures import ThreadPoolExecutor, as_completed + +from python_core.utils.logger import logger +from python_core.utils.jsonrpc_enhanced import ProgressLevel +from ..types.batch_workflow_state import BatchSceneDetectionWorkflowState, BatchVideoTask +from ..types.enums import DetectorType, OutputFormat +from ..services.detector_service import SceneDetectorService +from ..services.video_info_service import VideoInfoService +from ..services.ai_analysis_service import AIAnalysisService +from python_core.services.ffmpeg_slice_service import FfmpegSliceService, SliceOptions +from ..utils.result_saver import ResultSaver + + +class BatchWorkflowNodes: + """批量工作流节点""" + + def __init__(self): + # 初始化服务 + self.detector_service = SceneDetectorService() + self.video_info_service = VideoInfoService() + self.ai_analysis_service = AIAnalysisService() + self.splitter_service = FfmpegSliceService() + self.result_saver = ResultSaver() + + def validate_batch_input(self, state: BatchSceneDetectionWorkflowState) -> Dict[str, Any]: + """验证批量输入""" + state.current_stage = "validation" + state.send_progress("validate_input", "验证批量输入参数", ProgressLevel.INFO) + + errors = [] + + # 检查视频文件 + valid_videos = [] + for video_path in state.video_paths: + if not video_path.exists(): + errors.append(f"视频文件不存在: {video_path}") + continue + + if video_path.suffix.lower() not in {'.mp4', '.avi', '.mov', '.mkv', '.wmv', '.flv', '.webm', '.m4v'}: + errors.append(f"不支持的视频格式: {video_path}") + continue + + valid_videos.append(video_path) + + if not valid_videos: + errors.append("没有有效的视频文件") + + # 检查输出目录 + if state.output_base_dir: + try: + state.output_base_dir.mkdir(parents=True, exist_ok=True) + except Exception as e: + errors.append(f"无法创建输出目录: {e}") + + # 检查FFmpeg（如果需要切分） + if state.enable_video_splitting: + if not self.splitter_service.check_ffmpeg_available(): + errors.append("FFmpeg不可用，无法进行视频切分") + + if errors: + state.global_errors.extend(errors) + return { + "workflow_state": "failed", + "errors": errors, + "valid_videos": valid_videos + } + + # 更新有效视频列表 + state.video_paths = valid_videos + state.tasks = [ + BatchVideoTask( + video_path=video_path, + output_dir=state.output_base_dir / video_path.stem if state.output_base_dir else None + ) + for video_path in valid_videos + ] + state.total_tasks = len(state.tasks) + + state.send_progress("validate_input", f"验证完成，找到 {len(valid_videos)} 个有效视频", ProgressLevel.SUCCESS) + + return { + "workflow_state": "validated", + "valid_videos": valid_videos, + "total_tasks": state.total_tasks + } + + def process_videos_batch(self, state: BatchSceneDetectionWorkflowState) -> Dict[str, Any]: + """批量处理视频""" + state.current_stage = "batch_processing" + state.send_progress("batch_process", "开始批量处理视频", ProgressLevel.INFO) + + # 使用线程池进行并发处理 + with ThreadPoolExecutor(max_workers=state.max_concurrent) as executor: + # 提交所有任务 + future_to_index = {} + for i, task in enumerate(state.tasks): + future = executor.submit(self._process_single_video, state, i, task) + future_to_index[future] = i + + # 处理完成的任务 + for future in as_completed(future_to_index): + task_index = future_to_index[future] + try: + result = future.result() + if result["success"]: + state.mark_task_completed( + task_index, + result["detection_result"], + result.get("split_results", []) + ) + state.send_progress( + "task_completed", + f"任务 {task_index + 1}/{state.total_tasks} 完成: {state.tasks[task_index].video_path.name}", + ProgressLevel.SUCCESS, + {"task_index": task_index + 1, "video_name": state.tasks[task_index].video_path.name} + ) + else: + state.mark_task_failed(task_index, result["error"]) + state.send_progress( + "task_failed", + f"任务 {task_index + 1}/{state.total_tasks} 失败: {result['error']}", + ProgressLevel.ERROR, + {"task_index": task_index + 1, "error": result["error"]} + ) + except Exception as e: + state.mark_task_failed(task_index, str(e)) + state.send_progress( + "task_error", + f"任务 {task_index + 1}/{state.total_tasks} 异常: {str(e)}", + ProgressLevel.ERROR, + {"task_index": task_index + 1, "error": str(e)} + ) + + # 生成批量处理结果 + summary = state.get_summary() + + state.send_progress( + "batch_complete", + f"批量处理完成: {state.completed_tasks}/{state.total_tasks} 成功", + ProgressLevel.SUCCESS, + summary + ) + + return { + "workflow_state": "completed", + "summary": summary, + "completed_tasks": state.completed_tasks, + "failed_tasks": state.failed_tasks, + "total_tasks": state.total_tasks + } + + def _process_single_video(self, state: BatchSceneDetectionWorkflowState, + task_index: int, task: BatchVideoTask) -> Dict[str, Any]: + """处理单个视频""" + try: + task.status = "processing" + task.start_time = time.time() + + logger.info(f"🎬 处理视频 {task_index + 1}/{state.total_tasks}: {task.video_path.name}") + + # 1. 场景检测 + detection_result = self.detector_service.detect_scenes( + task.video_path, + DetectorType(state.detector_type), + state.threshold, + state.min_scene_length + ) + + if not detection_result.success: + return { + "success": False, + "error": f"场景检测失败: {detection_result.error}" + } + + # 2. 保存检测结果 + if task.output_dir: + task.output_dir.mkdir(parents=True, exist_ok=True) + result_file = task.output_dir / f"scenes.{state.output_format}" + self.result_saver.save_results( + detection_result, + result_file, + OutputFormat(state.output_format) + ) + + # 3. 视频切分（如果启用） + split_results = [] + if state.enable_video_splitting and task.output_dir: + try: + # 创建切分选项 + slice_options = SliceOptions( + crf=state.split_quality, + preset=state.split_preset + ) + + split_results_raw = self.splitter_service.split_video_by_scenes( + task.video_path, + detection_result.scenes, + task.output_dir / "scenes", + use_advanced=state.use_advanced_split, + options=slice_options + ) + + # 转换为字典格式 + split_results = [ + { + "scene_index": r.scene_index + 1, + "output_path": str(r.output_path), + "start_time": r.start_time, + "end_time": r.end_time, + "duration": r.duration, + "file_size": r.file_size, + "success": r.success, + "error": r.error + } + for r in split_results_raw + ] + + # 保存切分摘要 + split_summary = self.splitter_service.create_split_summary( + task.video_path, split_results_raw + ) + summary_file = task.output_dir / "split_summary.json" + import json + with open(summary_file, 'w', encoding='utf-8') as f: + json.dump(split_summary, f, indent=2, ensure_ascii=False) + + except Exception as e: + logger.error(f"视频切分失败: {e}") + # 切分失败不影响整体任务成功 + + task.end_time = time.time() + + return { + "success": True, + "detection_result": detection_result, + "split_results": split_results + } + + except Exception as e: + task.end_time = time.time() + return { + "success": False, + "error": str(e) + } diff --git a/python_core/scene_detection/workflows/workflow_manager.py b/python_core/scene_detection/workflows/workflow_manager.py index 397a6a0..2d0a838 100644 --- a/python_core/scene_detection/workflows/workflow_manager.py +++ b/python_core/scene_detection/workflows/workflow_manager.py @@ -6,12 +6,14 @@ Workflow Manager import time from pathlib import Path -from typing import Dict, Any, Optional, Literal +from typing import Dict, Any, Optional, List, Literal from python_core.utils.logger import logger from python_core.utils.jsonrpc_enhanced import EnhancedJSONRPCResponse from ..types import SceneDetectionWorkflowState, DetectorType, OutputFormat from .workflow_nodes import WorkflowNodes +from .batch_workflow_nodes import BatchWorkflowNodes +from ..types.batch_workflow_state import BatchSceneDetectionWorkflowState class SceneDetectionWorkflowManager: @@ -19,6 +21,7 @@ class SceneDetectionWorkflowManager: def __init__(self): self.nodes = WorkflowNodes() + self.batch_nodes = BatchWorkflowNodes() self.workflow = None def create_detection_workflow(self): @@ -174,3 +177,111 @@ class SceneDetectionWorkflowManager: # 非JSON-RPC模式：抛出异常 logger.error(f"❌ {error_msg}") raise + + def batch_detect_and_split(self, video_paths: List[Path], output_base_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, + max_concurrent: int = 2, continue_on_error: bool = True, + use_advanced_split: bool = True, split_quality: int = 23, split_preset: str = "fast", + request_id: Optional[str] = None) -> Dict[str, Any]: + """批量场景检测和视频切分""" + + # 创建批量工作流状态 + state = BatchSceneDetectionWorkflowState( + video_paths=video_paths, + output_base_dir=output_base_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_concurrent=max_concurrent, + continue_on_error=continue_on_error, + request_id=request_id, + enable_jsonrpc=request_id is not None + ) + + try: + logger.info(f"🚀 开始批量场景检测和切分") + logger.info(f"📁 视频数量: {len(video_paths)}") + logger.info(f"📂 输出目录: {output_base_dir}") + logger.info(f"🎯 检测器: {detector_type.value}, 阈值: {threshold}") + logger.info(f"✂️ 视频切分: {'启用' if enable_video_splitting else '禁用'}") + logger.info(f"🔄 并发数: {max_concurrent}") + + # 1. 验证输入 + validation_result = self.batch_nodes.validate_batch_input(state) + if validation_result["workflow_state"] == "failed": + error_msg = f"批量输入验证失败: {'; '.join(validation_result['errors'])}" + if request_id: + state.send_error_result(-32602, error_msg, validation_result["errors"]) + return { + "jsonrpc_mode": True, + "request_id": request_id, + "error_sent": True, + "error": error_msg + } + else: + raise ValueError(error_msg) + + # 2. 批量处理 + processing_result = self.batch_nodes.process_videos_batch(state) + + # 3. 构建最终结果 + final_result = { + "workflow_state": processing_result["workflow_state"], + "summary": processing_result["summary"], + "batch_results": { + "total_videos": state.total_tasks, + "completed_videos": state.completed_tasks, + "failed_videos": state.failed_tasks, + "success_rate": (state.completed_tasks / state.total_tasks * 100) if state.total_tasks > 0 else 0, + "output_base_dir": str(output_base_dir) if output_base_dir else None, + "enable_video_splitting": enable_video_splitting, + "tasks": [ + { + "video_path": str(task.video_path), + "status": task.status, + "output_dir": str(task.output_dir) if task.output_dir else None, + "total_scenes": task.detection_result.total_scenes if task.detection_result else 0, + "detection_time": task.detection_result.detection_time if task.detection_result else 0, + "split_count": len(task.split_results) if task.split_results else 0, + "error": task.error, + "processing_time": (task.end_time - task.start_time) if task.start_time and task.end_time else 0 + } + for task in state.tasks + ] + } + } + + if request_id: + state.send_final_result(final_result) + return { + "jsonrpc_mode": True, + "request_id": request_id, + "result_sent": True, + "result": final_result + } + else: + return final_result + + except Exception as e: + error_msg = f"批量工作流执行失败: {str(e)}" + logger.error(error_msg) + + if request_id: + state.send_error_result(-32603, error_msg, str(e)) + return { + "jsonrpc_mode": True, + "request_id": request_id, + "error_sent": True, + "error": error_msg + } + else: + logger.error(f"❌ {error_msg}") + raise diff --git a/python_core/services/ffmpeg_slice_service.py b/python_core/services/ffmpeg_slice_service.py new file mode 100644 index 0000000..ce6a6e5 --- /dev/null +++ b/python_core/services/ffmpeg_slice_service.py @@ -0,0 +1,705 @@ +""" +FFmpeg视频切片服务 + +基于demo.py的VideoUtils.ffmpeg_slice_media方法封装的专业视频切片服务。 +""" + +import asyncio +import os +import json +import math +from typing import Dict, List, Any, Optional, Tuple +from pathlib import Path +from datetime import timedelta +from dataclasses import dataclass +from ffmpeg.asyncio import FFmpeg as AsyncFFmpeg +from loguru import logger + + +@dataclass +class SliceSegment: + """切片段配置""" + start: float # 开始时间（秒） + end: float # 结束时间（秒） + + @property + def duration(self) -> float: + """片段时长""" + return self.end - self.start + + def to_timedelta(self) -> Tuple[timedelta, timedelta]: + """转换为timedelta格式""" + return ( + timedelta(seconds=self.start), + timedelta(seconds=self.end) + ) + + +@dataclass +class SliceOptions: + """切片输出选项""" + width: Optional[int] = None # 输出宽度 + height: Optional[int] = None # 输出高度 + crf: int = 23 # 视频质量 (18-28, 越小质量越好) + fps: int = 30 # 输出帧率 + bit_rate: Optional[str] = None # 比特率 (如 "2M") + limit_size: Optional[str] = None # 文件大小限制 (如 "10M") + preset: str = "medium" # 编码预设 (ultrafast, fast, medium, slow, veryslow) + + @property + def pretty_bit_rate(self) -> str: + """格式化的比特率""" + return self.bit_rate or "2M" + + +@dataclass +class VideoMetadata: + """视频元数据""" + duration: float + width: int + height: int + fps: float + format_name: str + size: int + codec_name: str + audio_codec: str + + @classmethod + def from_ffprobe(cls, metadata: dict) -> 'VideoMetadata': + """从ffprobe结果创建元数据""" + format_info = metadata.get('format', {}) + video_stream = None + audio_stream = None + + for stream in metadata.get('streams', []): + if stream.get('codec_type') == 'video': + video_stream = stream + elif stream.get('codec_type') == 'audio': + audio_stream = stream + + if not video_stream: + raise ValueError("No video stream found") + + # 解析帧率 + r_frame_rate = video_stream.get('r_frame_rate', '30/1') + if '/' in r_frame_rate: + num, den = map(int, r_frame_rate.split('/')) + fps = num / den if den != 0 else 30.0 + else: + fps = float(r_frame_rate) + + # 获取音频编码器信息 + audio_codec = audio_stream.get('codec_name', '') if audio_stream else '' + + return cls( + duration=float(format_info.get('duration', 0)), + width=int(video_stream.get('width', 0)), + height=int(video_stream.get('height', 0)), + fps=fps, + format_name=format_info.get('format_name', 'unknown'), + size=int(format_info.get('size', 0)), + codec_name=video_stream.get('codec_name', 'unknown'), + audio_codec=audio_codec + ) + +class FfmpegSliceService: + """ + FFmpeg视频切片服务 + + 提供专业的视频切片功能，支持按时间段切割、质量控制、批量处理等。 + """ + + def __init__(self): + self.logger = logger + + def _create_async_ffmpeg_cmd(self, quiet: bool = False) -> Optional[Any]: + """创建异步FFmpeg命令对象""" + # 使用隐藏窗口的FFmpeg包装器 + ffmpeg_cmd = AsyncFFmpeg(executable="ffmpeg").option('y').option('hide_banner') + + @ffmpeg_cmd.on("start") + def on_start(arguments: list[str]): + try: + filter_index = arguments.index("-filter_complex") + filter_content = arguments[filter_index + 1] + arguments[filter_index + 1] = f'"{filter_content}"' + args = " ".join(arguments) + arguments[filter_index + 1] = filter_content + except ValueError: + args = " ".join(arguments) + logger.info(f"FFmpeg command: {args}") + + @ffmpeg_cmd.on("progress") + def on_progress(progress): + if not quiet: + logger.info(f"处理进度: {progress}") + + @ffmpeg_cmd.on("completed") + def on_completed(result=None): + logger.info(f"FFmpeg task completed.") + + @ffmpeg_cmd.on("stderr") + def on_stderr(line: str): + if line.startswith('Error') and ".m3u8" not in line: + raise RuntimeError(line) + elif "Output file is empty" in line: + raise RuntimeError("输出是空文件") + else: + ... + + return ffmpeg_cmd + + async def get_video_metadata(self, media_path: str) -> VideoMetadata: + """ + 获取视频元数据 + + Args: + media_path: 视频文件路径 + + Returns: + VideoMetadata: 视频元数据对象 + """ + ffprobe = AsyncFFmpeg(executable='ffprobe') + # 配置FFprobe参数 + ffprobe.option("v", "quiet") + ffprobe.option("print_format", "json") + ffprobe.option("show_streams", None) # 明确指定None值 + ffprobe.option("show_format", None) # 明确指定None值 + + # 首先验证文件是否存在和有效 + media_file = Path(media_path) + if not media_file.exists(): + raise FileNotFoundError(f"视频文件不存在: {media_path}") + + if media_file.stat().st_size == 0: + raise RuntimeError(f"视频文件为空: {media_path}") + + ffprobe.input(media_path) + logger.info(f"开始获取视频元数据: {media_path} (大小: {media_file.stat().st_size} 字节)") + + try: + result = await ffprobe.execute() + + # 详细的错误信息 + if result.returncode != 0: + stderr_text = result.stderr.decode() if result.stderr else 'No error output' + stdout_text = result.stdout.decode() if result.stdout else 'No output' + logger.error(f"FFprobe failed with return code {result.returncode}") + logger.error(f"FFprobe stderr: {stderr_text}") + logger.error(f"FFprobe stdout: {stdout_text}") + + # 如果是空JSON，说明文件可能不是有效的视频文件 + if stdout_text.strip() in ['{}', '']: + raise RuntimeError(f"文件不是有效的视频文件或已损坏: {media_path}") + + raise RuntimeError(f"FFprobe failed (code {result.returncode}): {stderr_text}") + + # 检查输出是否为空 + if not result.stdout: + raise RuntimeError("FFprobe returned empty output") + + stdout_text = result.stdout.decode() + if not stdout_text.strip(): + raise RuntimeError("FFprobe returned empty stdout") + + # 解析JSON + try: + metadata_json = json.loads(stdout_text) + + # 检查JSON是否包含有效数据 + if not metadata_json or (not metadata_json.get('streams') and not metadata_json.get('format')): + raise RuntimeError(f"FFprobe返回空的元数据，文件可能已损坏: {media_path}") + + except json.JSONDecodeError as e: + logger.error(f"Failed to parse FFprobe JSON output: {e}") + logger.error(f"Raw output: {stdout_text[:500]}...") + raise RuntimeError(f"Invalid JSON from FFprobe: {e}") + + logger.info(f"成功获取视频元数据: {media_path}") + return VideoMetadata.from_ffprobe(metadata_json) + + except Exception as e: + logger.error(f"Failed to get video metadata for {media_path}: {e}") + raise + + def _validate_segments(self, segments: List[SliceSegment], video_duration: float) -> None: + """ + 验证切片段配置 + + Args: + segments: 切片段列表 + video_duration: 视频总时长 + """ + diff_tolerance = 0.001 + + for i, segment in enumerate(segments): + if segment.start > video_duration or segment.start < 0: + raise ValueError( + f"第{i}个切割点起始点{segment.start}s超出视频时长[0-{video_duration}s]范围" + ) + + if segment.end > video_duration or segment.end < 0: + if segment.end > 0 and math.isclose(segment.end, video_duration, rel_tol=diff_tolerance): + # 允许小的误差 + segment.end = video_duration + logger.warning( + f"第{i}个切割点结束点{segment.end}s接近视频时长，已调整为{video_duration}s" + ) + else: + raise ValueError( + f"第{i}个切割点结束点{segment.end}s超出视频时长[0-{video_duration}s]范围" + ) + + if segment.start >= segment.end: + raise ValueError( + f"第{i}个切割点起始时间{segment.start}s必须小于结束时间{segment.end}s" + ) + + def _generate_output_path(self, base_path: str, index: int, extension: str = "mp4") -> str: + """生成输出文件路径""" + base = Path(base_path) + return str(base.parent / f"{base.stem}_{index:03d}.{extension}") + + async def slice_video(self, + media_path: str, + segments: List[SliceSegment], + options: SliceOptions, + output_path: Optional[str] = None) -> List[Tuple[str, VideoMetadata]]: + """ + 使用本地视频文件按时间段切割出分段视频 + + Args: + media_path: 本地视频路径 + segments: 分段起始结束时间标记列表 + options: 输出切割质量选项 + output_path: 最终输出文件路径，片段会根据指定路径附加_001.mp4等片段编号 + + Returns: + List[Tuple[str, VideoMetadata]]: 输出片段的本地路径和元数据 + """ + if not segments: + raise ValueError("No segments provided") + + # 获取视频元数据 + metadata = await self.get_video_metadata(media_path) + logger.info(f"视频信息: {metadata.width}x{metadata.height}, {metadata.duration:.2f}s, {metadata.fps}fps") + + # 验证切片段 + self._validate_segments(segments, metadata.duration) + + # 准备输出路径 + if not output_path: + output_path = str(Path(media_path).with_suffix('_slice.mp4')) + + os.makedirs(os.path.dirname(output_path), exist_ok=True) + + # 创建FFmpeg命令 + ffmpeg_cmd = self._create_async_ffmpeg_cmd() + if not ffmpeg_cmd: + raise RuntimeError("Failed to create FFmpeg command") + + ffmpeg_cmd.input(media_path) + + # 检查是否有音频流 + has_audio = metadata.audio_codec is not None and metadata.audio_codec != "" + + # 构建filter_complex + filter_complex = [] + temp_outputs = [] + + for index, segment in enumerate(segments): + start = segment.start + end = segment.end + + # 处理指定的输出分辨率 + if options.width and options.height: + filter_complex.append(f"[v:0]trim=start={start}:end={end},scale={options.width}:{options.height},setpts=PTS-STARTPTS[cut{index}]") + if has_audio: + filter_complex.append(f"[a:0]atrim=start={start}:end={end},asetpts=PTS-STARTPTS[acut{index}]") + else: + filter_complex.append(f"[v:0]trim=start={start}:end={end},setpts=PTS-STARTPTS[cut{index}]") + if has_audio: + filter_complex.append(f"[a:0]atrim=start={start}:end={end},asetpts=PTS-STARTPTS[acut{index}]") + + ffmpeg_cmd.option('filter_complex', ';'.join(filter_complex)) + + # 为每个片段配置输出 + for i, segment in enumerate(segments): + segment_output_path = self._generate_output_path(output_path, i) + + # 根据是否有音频流配置映射 + if has_audio: + map_options = [f"[cut{i}]", f"[acut{i}]"] + else: + map_options = [f"[cut{i}]"] + + ffmpeg_options = { + "map": map_options, + "reset_timestamps": "1", + "sc_threshold": "0", + "g": "1", + "force_key_frames": "expr:gte(t,n_forced*1)", + "vcodec": "libx264", + "crf": options.crf, + "r": options.fps, + "preset": options.preset + } + + # 只有在有音频时才添加音频编码器 + if has_audio: + ffmpeg_options["acodec"] = "aac" + + if options.limit_size: + ffmpeg_options["fs"] = options.limit_size + elif options.bit_rate: + ffmpeg_options["b:v"] = options.pretty_bit_rate + + ffmpeg_cmd.output(segment_output_path, options=ffmpeg_options) + temp_outputs.append(segment_output_path) + + # 执行切片 + try: + logger.info(f"开始执行FFmpeg切片命令...") + result = await ffmpeg_cmd.execute() + + # 检查执行结果 + if result.returncode != 0: + stderr_text = result.stderr.decode() if result.stderr else 'No error output' + stdout_text = result.stdout.decode() if result.stdout else 'No output' + logger.error(f"FFmpeg切片失败，返回码: {result.returncode}") + logger.error(f"FFmpeg stderr: {stderr_text}") + logger.error(f"FFmpeg stdout: {stdout_text}") + raise RuntimeError(f"FFmpeg切片失败 (返回码 {result.returncode}): {stderr_text}") + + logger.info(f"FFmpeg执行成功，检查输出文件...") + + # 验证输出文件是否真的被创建 + missing_files = [] + for output_file in temp_outputs: + if not Path(output_file).exists(): + missing_files.append(output_file) + + if missing_files: + logger.error(f"FFmpeg执行成功但输出文件未创建: {len(missing_files)} 个文件缺失") + for missing_file in missing_files[:5]: # 只显示前5个 + logger.error(f" 缺失文件: {missing_file}") + raise RuntimeError(f"FFmpeg执行成功但未生成预期的输出文件，缺失 {len(missing_files)} 个文件") + + logger.info(f"视频切片完成: 生成{len(temp_outputs)}个片段") + + except Exception as e: + logger.error(f"视频切片失败: {e}") + raise + + # 获取输出文件的元数据 + outputs = [] + for output_file in temp_outputs: + try: + output_metadata = await self.get_video_metadata(output_file) + outputs.append((output_file, output_metadata)) + except Exception as e: + logger.warning(f"无法获取输出文件元数据 {output_file}: {e}") + # 创建一个基本的元数据对象 + basic_metadata = VideoMetadata( + duration=0, width=0, height=0, fps=0, + format_name="mp4", size=0, codec_name="h264", + audio_codec="aac" # 添加缺少的audio_codec参数 + ) + outputs.append((output_file, basic_metadata)) + + return outputs + + async def slice_video_by_duration(self, + media_path: str, + segment_duration: float, + options: SliceOptions, + output_dir: Optional[str] = None, + overlap: float = 0.0) -> List[Tuple[str, VideoMetadata]]: + """ + 按固定时长切割视频 + + Args: + media_path: 输入视频路径 + segment_duration: 每段时长（秒） + options: 输出选项 + output_dir: 输出目录 + overlap: 片段重叠时间（秒） + + Returns: + 输出片段列表 + """ + metadata = await self.get_video_metadata(media_path) + + if not output_dir: + output_dir = str(Path(media_path).parent / f"{Path(media_path).stem}_segments") + + # 计算切片段 + segments = [] + current_start = 0.0 + + while current_start < metadata.duration: + end_time = min(current_start + segment_duration, metadata.duration) + + if end_time - current_start >= 1.0: # 至少1秒的片段 + segments.append(SliceSegment(start=current_start, end=end_time)) + + current_start += segment_duration - overlap + + if current_start >= metadata.duration: + break + + logger.info(f"按{segment_duration}s时长切割，生成{len(segments)}个片段") + + # 使用基础切片方法 + output_path = os.path.join(output_dir, f"{Path(media_path).stem}_segment.mp4") + return await self.slice_video(media_path, segments, options, output_path) + + async def slice_video_by_count(self, + media_path: str, + segment_count: int, + options: SliceOptions, + output_dir: Optional[str] = None) -> List[Tuple[str, VideoMetadata]]: + """ + 按片段数量平均切割视频 + + Args: + media_path: 输入视频路径 + segment_count: 片段数量 + options: 输出选项 + output_dir: 输出目录 + + Returns: + 输出片段列表 + """ + if segment_count <= 0: + raise ValueError("Segment count must be positive") + + metadata = await self.get_video_metadata(media_path) + segment_duration = metadata.duration / segment_count + + if not output_dir: + output_dir = str(Path(media_path).parent / f"{Path(media_path).stem}_segments") + + # 计算切片段 + segments = [] + for i in range(segment_count): + start_time = i * segment_duration + end_time = min((i + 1) * segment_duration, metadata.duration) + + if end_time - start_time >= 0.5: # 至少0.5秒的片段 + segments.append(SliceSegment(start=start_time, end=end_time)) + + logger.info(f"平均切割为{len(segments)}个片段，每段约{segment_duration:.2f}s") + + # 使用基础切片方法 + output_path = os.path.join(output_dir, f"{Path(media_path).stem}_segment.mp4") + return await self.slice_video(media_path, segments, options, output_path) + + async def batch_slice_videos(self, + video_files: List[str], + segment_duration: float, + options: SliceOptions, + output_base_dir: str, + max_concurrent: int = 3) -> Dict[str, Any]: + """ + 批量切割多个视频 + + Args: + video_files: 视频文件列表 + segment_duration: 每段时长 + options: 输出选项 + output_base_dir: 输出基础目录 + max_concurrent: 最大并发数 + + Returns: + 批处理结果 + """ + os.makedirs(output_base_dir, exist_ok=True) + + # 创建任务 + async def process_single_video(video_file: str): + try: + file_name = Path(video_file).stem + output_dir = os.path.join(output_base_dir, file_name) + + results = await self.slice_video_by_duration( + media_path=video_file, + segment_duration=segment_duration, + options=options, + output_dir=output_dir + ) + + return { + "file": video_file, + "success": True, + "segments": len(results), + "outputs": [path for path, _ in results] + } + except Exception as e: + logger.error(f"处理视频失败 {video_file}: {e}") + return { + "file": video_file, + "success": False, + "error": str(e) + } + + # 并发执行 + semaphore = asyncio.Semaphore(max_concurrent) + + async def limited_task(video_file): + async with semaphore: + return await process_single_video(video_file) + + logger.info(f"开始批量处理{len(video_files)}个视频文件") + results = await asyncio.gather( + *[limited_task(video_file) for video_file in video_files], + return_exceptions=True + ) + + # 统计结果 + success_count = 0 + error_count = 0 + total_segments = 0 + errors = [] + + for result in results: + if isinstance(result, Exception): + error_count += 1 + errors.append({"error": str(result)}) + elif result["success"]: + success_count += 1 + total_segments += result["segments"] + else: + error_count += 1 + errors.append(result) + + batch_result = { + "total_files": len(video_files), + "success_count": success_count, + "error_count": error_count, + "total_segments": total_segments, + "results": results, + "errors": errors + } + + logger.info(f"批量处理完成: 成功{success_count}, 失败{error_count}, 总片段{total_segments}") + return batch_result + + def create_slice_options(self, + quality: str = "medium", + width: Optional[int] = None, + height: Optional[int] = None, + fps: int = 30, + bit_rate: Optional[str] = None, + limit_size: Optional[str] = None) -> SliceOptions: + """ + 创建切片选项的便捷方法 + + Args: + quality: 质量预设 (low, medium, high) + width: 输出宽度 + height: 输出高度 + fps: 帧率 + bit_rate: 比特率 + limit_size: 文件大小限制 + + Returns: + SliceOptions对象 + """ + quality_presets = { + "low": {"crf": 28, "preset": "fast"}, + "medium": {"crf": 23, "preset": "medium"}, + "high": {"crf": 18, "preset": "slow"}, + "ultra": {"crf": 15, "preset": "veryslow"} + } + + preset_config = quality_presets.get(quality, quality_presets["medium"]) + + return SliceOptions( + width=width, + height=height, + crf=preset_config["crf"], + fps=fps, + bit_rate=bit_rate, + limit_size=limit_size, + preset=preset_config["preset"] + ) + + async def _slice_video_fallback(self, media_path: str, segments: List[SliceSegment], output_path: str = None) -> List[Tuple[str, VideoMetadata]]: + """ + 备用的视频切片实现（当 FFmpeg Python 不可用时） + + Args: + media_path: 输入视频路径 + segments: 要切片的段落列表 + output_path: 输出路径（可选） + + Returns: + List[Tuple[str, VideoMetadata]]: 输出片段的本地路径和元数据 + """ + logger.warning("FFmpeg Python 不可用，使用备用实现（仅返回原视频信息）") + + try: + # 获取视频基本信息 + import os + from pathlib import Path + + if not os.path.exists(media_path): + raise FileNotFoundError(f"视频文件不存在: {media_path}") + + # 创建基本的视频元数据 + file_size = os.path.getsize(media_path) + file_name = Path(media_path).name + + # 创建简化的元数据 + metadata = VideoMetadata( + duration=60.0, # 默认值 + width=1920, # 默认值 + height=1080, # 默认值 + fps=30.0, # 默认值 + format_name="mp4", # 默认值 + size=file_size, + codec_name="h264", # 默认值 + audio_codec="aac" # 默认值 + ) + + # 为每个段落创建结果（实际上返回原视频） + results = [] + for i, segment in enumerate(segments): + # 创建输出文件名 + output_dir = Path(media_path).parent / "segments" + output_dir.mkdir(exist_ok=True) + + segment_filename = f"{Path(media_path).stem}_segment_{i+1}_{segment.start_time:.1f}s-{segment.end_time:.1f}s.mp4" + segment_path = output_dir / segment_filename + + # 在备用模式下，我们只是复制原文件（或创建一个占位符） + try: + import shutil + shutil.copy2(media_path, segment_path) + logger.info(f"备用模式：复制原视频到 {segment_path}") + except Exception as e: + logger.warning(f"复制文件失败: {e}") + # 创建一个空文件作为占位符 + segment_path.touch() + + # 创建段落元数据 + segment_metadata = VideoMetadata( + width=metadata.width, + height=metadata.height, + duration=segment.end_time - segment.start_time, + fps=metadata.fps, + bitrate=metadata.bitrate, + codec=metadata.codec, + file_size=file_size, # 简化处理 + format=metadata.format + ) + + results.append((str(segment_path), segment_metadata)) + + logger.info(f"备用模式完成，生成了 {len(results)} 个段落") + return results + + except Exception as e: + logger.error(f"备用视频切片失败: {e}") + raise RuntimeError(f"视频切片失败: {e}") diff --git a/python_core/utils/PathUtils.py b/python_core/utils/PathUtils.py new file mode 100644 index 0000000..348f9d1 --- /dev/null +++ b/python_core/utils/PathUtils.py @@ -0,0 +1,84 @@ +import os +from pathlib import Path +from typing import List + + +class FileUtils: + @staticmethod + def file_path_extend(media_path: str, extend: str) -> str: + """ + 基于现有文件路径添加后缀, 例如 extend = "def" : 123/abc.txt -> 123/abc_def.txt + :param media_path: 现有文件路径 + :param extend: 后缀名 + + :return: 处理过的新文件路径 + """ + media_filename = os.path.basename(media_path) + media_dir = os.path.dirname(media_path) + '/' + filenames = media_filename.split('.') + filenames[0] = f"{filenames[0]}_{extend}" + extend_filename = '.'.join(filenames) + return os.path.join(media_dir, extend_filename) + + @staticmethod + def replace_root_by_depth(media_path: str, root: str, depth: int = 1) -> str: + """ + 使用根目录名替换路径起始位置, 例如: + + prefix="pre",depth=0 : ./abc/def.txt -> pre\\.\\abc\\def.txt + + prefix="pre",depth=1 : ./abc/def.txt -> pre\\abc\\def.txt + + prefix="pre",depth=2 : ./abc/def.txt -> pre\\def.txt + + prefix="pre",depth=3 : IndexError("Depth is out of range") + + :param media_path: 现有文件路径 + :param root: 替换用的根目录名 + :param depth: root所占的路径深度层级 + + :return: 处理后的文件路径 + :exception IndexError: 根目录深度超过实际路径深度 + """ + media_dirs = media_path.split('/') + if depth >= len(media_dirs): + raise IndexError("Depth is out of range") + media_dir_prefix = os.path.join(root, *media_dirs[depth:]) + return media_dir_prefix + + @staticmethod + def file_path_change_extension(media_path: str, extension: str) -> str: + media_filename = os.path.basename(media_path) + media_dir = os.path.dirname(media_path) + "/" + filenames = media_filename.split(".") + filenames[-1] = extension + filename = ".".join(filenames) + return os.path.join(media_dir, filename) + + @staticmethod + def get_folder_size(folder_path: str) -> int: + """ + + Args: + folder_path: + + Returns: + + """ + total_size = 0 + for path in Path(folder_path).rglob('*'): + if path.is_file(): + total_size += path.stat().st_size + return total_size + + @staticmethod + def get_file_size(file_path: str) -> int: + return os.path.getsize(file_path) + + @staticmethod + def get_files_size(files: List[str]) -> int: + total_size = 0 + for file in files: + total_size += os.path.getsize(file) + return total_size + diff --git a/python_core/utils/TimeUtils.py b/python_core/utils/TimeUtils.py new file mode 100644 index 0000000..7e97802 --- /dev/null +++ b/python_core/utils/TimeUtils.py @@ -0,0 +1,163 @@ +import re +from datetime import timedelta, datetime + + + + +def parse_time(time_str): + """解析时间字符串为datetime对象""" + + parsed_time = None + if re.match(r"^\d{2}:\d{2}:\d{2}\.\d{3}$", time_str): + # 先尝试完整格式 HH:MM:SS.fff + parsed_time = datetime.strptime(time_str, '%H:%M:%S.%f') + elif re.match(r"^\d{2}:\d{2}:\d{2}:\d{3}$", time_str): + # 如果失败，尝试 HH:MM:SS:fff 格式 + parsed_time = datetime.strptime(time_str, '%H:%M:%S:%f') + elif re.match(r"^\d{2}:\d{2}\.\d{3}$", time_str): + # 如果失败，尝试 MM:SS.fff 格式 + dt = datetime.strptime(time_str, '%M:%S.%f') + # 将小时设为0，只保留分钟和秒 + parsed_time = datetime.combine(dt.date(), dt.time().replace(hour=0)) + elif re.match(r"^\d{2}\.\d{2}:\d{3}$", time_str): + # 如果失败，尝试 MM.SS:fff 格式 + dt = datetime.strptime(time_str, '%M.%S:%f') + # 将小时设为0，只保留分钟和秒 + parsed_time = datetime.combine(dt.date(), dt.time().replace(hour=0)) + elif re.match(r"^\d{2}\.\d{2}\.\d{3}$", time_str): + # 如果失败，尝试 MM.SS.fff 格式 + dt = datetime.strptime(time_str, '%M.%S.%f') + # 将小时设为0，只保留分钟和秒 + parsed_time = datetime.combine(dt.date(), dt.time().replace(hour=0)) + elif re.match(r"^\d{2}:\d{2}:\d{3}$", time_str): + # 如果失败，尝试 MM:SS:fff 格式 + dt = datetime.strptime(time_str, '%M:%S:%f') + # 将小时设为0，只保留分钟和秒 + parsed_time = datetime.combine(dt.date(), dt.time().replace(hour=0)) + elif re.match(r"^\d{2}:\d{2}$", time_str): + # 如果失败，尝试 MM:SS:fff 格式 + dt = datetime.strptime(time_str, '%M:%S') + # 将小时设为0，只保留分钟和秒 + parsed_time = datetime.combine(dt.date(), dt.time().replace(hour=0)) + elif re.match(r"^\d{2}\.\d{2}$", time_str): + # 如果失败，尝试 MM:SS:fff 格式 + dt = datetime.strptime(time_str, '%M.%S') + # 将小时设为0，只保留分钟和秒 + parsed_time = datetime.combine(dt.date(), dt.time().replace(hour=0)) + else: + raise RuntimeError(f"转换时间格式失败 {time_str}") + return parsed_time + + +def format_time(dt): + """将datetime对象格式化为时间字符串""" + return dt.strftime('%H:%M:%S.%f')[:-3] # 保留3位小数 + + +def parse_timeline_item(item): + """解析时间线项目，提取开始和结束时间""" + time_range, _ = item.split(' (', 1) + start_time_str, end_time_str = time_range.split(' - ') + return parse_time(start_time_str), parse_time(end_time_str) + + +def format_timeline_item(start_time, end_time, activity): + """格式化时间线项目""" + return f"{format_time(start_time)} - {format_time(end_time)} ({activity})" + + +def merge_timeline_items(items, merge_diff=5): + """合并相邻的时间线项目，保留不同的活动描述""" + if not items: + return [] + + # 解析所有项目 + parsed_items = [] + for item in items: + start_time, end_time = parse_timeline_item(item) + activity = item.split(' (', 1)[1].rstrip(')') + parsed_items.append((start_time, end_time, activity)) + + # 按开始时间排序 + parsed_items.sort(key=lambda x: x[0]) + + # 合并相邻的时间段 + merged_items = [parsed_items[0]] + for current in parsed_items[1:]: + last = merged_items[-1] + + # 合法时间段 + if current[1] > current[0]: + # 如果当前项目的开始时间与上一个项目的结束时间相邻或重叠 活动描述相同，直接合并 + if current[0] <= last[1] + timedelta(seconds=merge_diff) and last[2] == current[2]: + # 更新结束时间为两个结束时间的最大值 + new_end_time = max(last[1], current[1]) + merged_items[-1] = (last[0], new_end_time, last[2]) + else: + # 不相邻，添加新项目 + merged_items.append(current) + + # 格式化回原始字符串格式 + return [format_timeline_item(start, end, activity) for start, end, activity in merged_items] + + +def convert_time(time_str): + # 去除秒字段并转换为标准时间 + parts = time_str.split(':') + if len(parts) == 3: + new_time_str = f"00:{parts[0]}:{parts[1]}.{parts[2].split('.')[1]}" + return new_time_str + return time_str + + +def merge_product_data(data, start_time_str, end_time_str, merge_diff=5): + """合并相同产品的数据""" + start_time = parse_time(start_time_str) + end_time = parse_time(end_time_str) + duration = end_time - start_time + + # 手动格式化时间差 + total_seconds = duration.total_seconds() + hours, remainder = divmod(total_seconds, 3600) + minutes, seconds = divmod(remainder, 60) + microseconds = duration.microseconds + max_time_str = f"{int(hours):02d}:{int(minutes):02d}:{int(seconds):02d}.{microseconds // 1000:03d}" + + product_dict = {} + + # 按产品名称分组 + for item in data: + product = item["product"] + if product not in product_dict: + product_dict[product] = {"product": product, "timeline": []} + product_dict[product]["timeline"].extend(item["timeline"]) + + # 合并每个产品的时间线 + for product in product_dict: + timeline = product_dict[product]["timeline"] + new_timeline = [] + for item in timeline: + start, end = parse_timeline_item(item) + # 比较起始时间与时间差 + start_str = format_time(start) + if (start - datetime.strptime("00:00:00.000", '%H:%M:%S.%f')) > duration and not start_str.startswith("00"): + new_start_str = convert_time(start_str) + else: + new_start_str = start_str + if (parse_time(new_start_str) - datetime.strptime("00:00:00.000", '%H:%M:%S.%f')) > duration: + new_start_str = max_time_str + + end_str = format_time(end) + if (end - datetime.strptime("00:00:00.000", '%H:%M:%S.%f')) > duration and not end_str.startswith("00"): + new_end_str = convert_time(end_str) + else: + new_end_str = end_str + if (parse_time(new_end_str) - datetime.strptime("00:00:00.000", '%H:%M:%S.%f')) > duration: + new_end_str = max_time_str + activity = item.split(' (', 1)[1].rstrip(')') + new_item = f"{new_start_str} - {new_end_str} ({activity})" + new_timeline.append(new_item) + product_dict[product]["timeline"] = merge_timeline_items(new_timeline, merge_diff=merge_diff) + + # 返回合并后的列表 + return list(product_dict.values()) diff --git a/python_core/utils/VideoUtils.py b/python_core/utils/VideoUtils.py new file mode 100644 index 0000000..030b299 --- /dev/null +++ b/python_core/utils/VideoUtils.py @@ -0,0 +1,953 @@ +import asyncio +import shutil +import tempfile +from datetime import datetime, timedelta + +import aiofiles +import aiohttp +from typing import List, Tuple, Optional, Any + +import m3u8 +import numpy as np +import json, os +import math +from aiohttp import ClientTimeout + +from .TimeUtils import TimeDelta +from ffmpeg import FFmpeg +from ffmpeg.asyncio import FFmpeg as AsyncFFmpeg +import soundfile as sf +import pyloudnorm as pyln +import noisereduce as nr +from pedalboard import ( + Pedalboard, + Compressor, + Limiter, + HighpassFilter, + LowpassFilter, + Gain, + Reverb, + Chorus, + Distortion, +) +from pedalboard.io import AudioFile + +from loguru import logger +from .PathUtils import FileUtils +from ..models.ffmpeg_tasks.models import FFMpegSliceSegment, FFMPEGSliceOptions, VideoStream, VideoMetadata + + + +class VideoUtils: + """ + python-ffmpeg package docs : https://python-ffmpeg.readthedocs.io/en/stable/ + """ + + @staticmethod + def ffprobe_video_format(media_path: str) -> VideoStream: + ffprobe = FFmpeg(executable="ffprobe").input( + media_path, print_format="json", + show_streams=None, show_format=None + ) + video_metadata = VideoMetadata.model_validate_json(ffprobe.execute()) + return video_metadata.streams[0] + + @staticmethod + def ffprobe_media_metadata(media_path: str) -> VideoMetadata: + ffprobe = FFmpeg(executable="ffprobe").input( + media_path, print_format="json", + show_streams=None, show_format=None + ) + metadata_bytes = ffprobe.execute() + metadata_json = json.loads(metadata_bytes) + logger.info(f"metadata = {json.dumps(metadata_json, ensure_ascii=False)}") + video_metadata = VideoMetadata.model_validate_json(metadata_bytes) + return video_metadata + + @staticmethod + def ffprobe_video_duration(media_path: str) -> TimeDelta: + ffprobe_cmd = VideoUtils.ffmpeg_init(use_ffprobe=True) + ffprobe_cmd.input( + media_path, print_format="json", + show_streams=None, show_format=None + ) + metadata_json = ffprobe_cmd.execute() + metadata = VideoMetadata.model_validate_json(metadata_json) + return TimeDelta(seconds=metadata.streams[0].duration) + + @staticmethod + def ffprobe_audio_duration(media_path: str) -> TimeDelta: + ffprobe_cmd = VideoUtils.ffmpeg_init(use_ffprobe=True) + ffprobe_cmd.input( + media_path, print_format="json", + show_streams=None, show_format=None + ) + metadata_json = ffprobe_cmd.execute() + metadata = VideoMetadata.model_validate_json(metadata_json) + return TimeDelta(seconds=metadata.streams[-1].duration) + + @staticmethod + async def ffprobe_video_format_async(media_path: str) -> VideoStream: + ffprobe = AsyncFFmpeg(executable="ffprobe").input( + media_path, print_format="json", show_streams=None, show_format=None + ) + video_metadata = VideoMetadata.model_validate_json(await ffprobe.execute()) + return video_metadata.streams[0] + + @staticmethod + def ffprobe_video_size(media_path: str) -> Tuple[int, int]: + """ + :param media_path: local path to video + :return: video_width, video_height + """ + ffprobe = FFmpeg(executable="ffprobe").input( + media_path, print_format="json", show_streams=None, show_format=None + ) + video_metadata = VideoMetadata.model_validate_json(ffprobe.execute()) + return video_metadata.streams[0].width, video_metadata.streams[0].height + + @staticmethod + async def ffprobe_video_size_async(media_path: str) -> Tuple[int, int]: + ffprobe = AsyncFFmpeg(executable="ffprobe").input( + media_path, print_format="json", show_streams=None, show_format=None + ) + video_metadata = VideoMetadata.model_validate_json(await ffprobe.execute()) + return video_metadata.streams[0].width, video_metadata.streams[0].height + + @staticmethod + def noise_reduce(audio_path: str, noise_sample_path: Optional[str] = None, + output_path: Optional[str] = None) -> str: + samplerate = 44100 + with AudioFile(audio_path).resampled_to(float(samplerate)) as f: + audio = f.read(f.frames) + + if noise_sample_path: + with AudioFile(noise_sample_path).resampled_to(float(samplerate)) as f: + noise_sample = f.read(f.frames) + else: + # 获取前2秒作为噪声样本 + noise_sample_length = min(int(2 * samplerate), audio.shape[0]) + noise_sample = audio[:noise_sample_length] + + if not output_path: + output_path = FileUtils.file_path_extend(audio_path, "nr") + + reduced_noise = nr.reduce_noise(y=audio, y_noise=noise_sample, sr=samplerate, + stationary=True, prop_decrease=0.75, n_std_thresh_stationary=1.5, + n_fft=2048, win_length=1024, hop_length=512, n_jobs=1) + + board = Pedalboard( + [ + HighpassFilter(cutoff_frequency_hz=150), + LowpassFilter(cutoff_frequency_hz=8000), + Reverb(room_size=0.08, damping=0.7, wet_level=0.08, + dry_level=0.92, width=0.4), + Chorus(rate_hz=0.7, depth=0.12, centre_delay_ms=3.0, mix=0.10), + Distortion(drive_db=3.0), + Compressor(threshold_db=-30, ratio=1.8, attack_ms=20, release_ms=200), + Compressor(threshold_db=-24, ratio=2.2, attack_ms=15, release_ms=180), + Compressor(threshold_db=-18, ratio=1.5, attack_ms=10, release_ms=150), + Gain(gain_db=4), + Limiter(threshold_db=-6, release_ms=200), + ] + ) + # Convert to float32 if not already + reduced_noise = reduced_noise.astype(np.float32) + # Ensure audio is in the correct range (-1.0 to 1.0) + if np.abs(reduced_noise).max() > 1.0: + reduced_noise = reduced_noise / np.abs(reduced_noise).max() + + processed_audio = board(reduced_noise, samplerate) + # 格式处理 + if len(processed_audio.shape) == 1: + processed_audio = processed_audio.reshape(-1, 1) + elif len(processed_audio.shape) == 2: + if processed_audio.shape[0] < processed_audio.shape[1]: + processed_audio = processed_audio.T + if processed_audio.shape[1] > 2: + processed_audio = processed_audio[:, :2] + # 响度标准化 + meter = pyln.Meter(samplerate) + min_samples = int(0.4 * samplerate) + + if processed_audio.shape[0] < min_samples: + normalized_audio = processed_audio + else: + loudness = meter.integrated_loudness(processed_audio) + safety_factor = 0.7 + processed_audio = processed_audio * safety_factor + normalized_audio = pyln.normalize.loudness( + processed_audio, loudness, -16.0 + ) + + max_peak = np.max(np.abs(normalized_audio)) + if max_peak > 0.85: + additional_safety_factor = 0.85 / max_peak + normalized_audio = normalized_audio * additional_safety_factor + + sf.write( + output_path, + normalized_audio, + samplerate, + format="WAV", + subtype="PCM_16", + ) + return output_path + + @staticmethod + def async_ffmpeg_init(use_ffprobe: bool = False, quiet: bool = False) -> AsyncFFmpeg: + if use_ffprobe: + ffmpeg_cmd = AsyncFFmpeg('ffprobe') + else: + ffmpeg_cmd = AsyncFFmpeg().option('y').option('hide_banner') + + @ffmpeg_cmd.on("start") + def on_start(arguments: list[str]): + try: + filter_index = arguments.index("-filter_complex") + filter_content = arguments[filter_index + 1] + arguments[filter_index + 1] = f'"{filter_content}"' + args = " ".join(arguments) + logger.info(f"FFmpeg command:{args}") + arguments[filter_index + 1] = filter_content + except ValueError: + args = " ".join(arguments) + logger.info(f"FFmpeg command:{args}") + + @ffmpeg_cmd.on("progress") + def on_progress(progress): + if not quiet: + logger.info(f"处理进度: {progress}") + + @ffmpeg_cmd.on("completed") + def on_completed(): + logger.info(f"FFMpeg task completed.") + + @ffmpeg_cmd.on("stderr") + def on_stderr(line: str): + if line.startswith('Error') and ".m3u8" not in line: + logger.error(line) + raise RuntimeError(line) + elif "Output file is empty" in line: + raise RuntimeError("输出是空文件") + else: + if not quiet: + if "Skip" not in line: + logger.warning(line) + + return ffmpeg_cmd + + @staticmethod + def ffmpeg_init(use_ffprobe: bool = False) -> FFmpeg: + if use_ffprobe: + ffmpeg_cmd = FFmpeg('ffprobe') + else: + ffmpeg_cmd = FFmpeg().option('y').option('hide_banner') + + @ffmpeg_cmd.on("start") + def on_start(arguments: list[str]): + try: + filter_index = arguments.index("-filter_complex") + filter_content = arguments[filter_index + 1] + arguments[filter_index + 1] = f'"{filter_content}"' + args = " ".join(arguments) + logger.info(f"FFmpeg command:{args}") + arguments[filter_index + 1] = filter_content + except ValueError: + args = " ".join(arguments) + logger.info(f"FFmpeg command:{args}") + + @ffmpeg_cmd.on("progress") + def on_progress(progress): + logger.info(f"处理进度: {progress}") + + @ffmpeg_cmd.on("completed") + def on_completed(): + logger.info(f"FFMpeg task completed.") + + @ffmpeg_cmd.on("stderr") + def on_stderr(line: str): + if line.startswith('Error'): + logger.error(line) + raise RuntimeError(line) + else: + logger.warning(line) + + return ffmpeg_cmd + + @staticmethod + async def ffmpeg_slice_media(media_path: str, media_markers: List[FFMpegSliceSegment], + options: FFMPEGSliceOptions, is_streams: bool = False, + output_path: Optional[str] = None) -> List[Tuple[str, VideoMetadata]]: + """ + 使用本地视频文件按时间段切割出分段视频, 如果是直播流则按时间分段切割HLS视频流_预先多线程下载所有ts + :param media_path: 本地视频路径 + :param media_markers: 分段起始结束时间标记 + :param options: 输出切割质量选项 + :param is_streams: 输入是否为直播流 + :param output_path: 最终输出文件路径, 片段会根据指定路径附加_1.mp4， _2.mp4等片段编号 + :return: 输出片段的本地路径 + """ + ffmpeg_cmd = VideoUtils.async_ffmpeg_init() + + if not is_streams: + ffmpeg_cmd.input(media_path) + seek_head = 0 + else: + seek_head = media_markers[0].start.total_seconds() + seek_tail = media_markers[-1].end.total_seconds() + duration = seek_tail - seek_head + logger.info(f"Only using {seek_head}s --> {seek_tail}s = {duration}s") + local_m3u8_path, temp_dir, diff = await VideoUtils.convert_m3u8_to_local_source(media_path, head=seek_head, + tail=seek_tail) + logger.info(f"local_playlist: {local_m3u8_path}") + + for segment in media_markers: + segment.start = segment.start - timedelta(seconds=seek_head) + diff + segment.end = segment.end - timedelta(seconds=seek_head) + diff + logger.info(f"Only using {seek_head}s --> {seek_tail}s = {duration}s") + ffmpeg_cmd.input(local_m3u8_path, + t=duration, + protocol_whitelist="file,http,https,tcp,tls") + + filter_complex: List[str] = [] + temp_outputs: List[str] = [] + if not output_path: + output_path = FileUtils.file_path_extend(media_path, "slice") + os.makedirs(os.path.dirname(output_path), exist_ok=True) + + metadata = VideoUtils.ffprobe_media_metadata(media_path) + for index, marker in enumerate(media_markers): + start = marker.start.total_seconds() + end = marker.end.total_seconds() + + # 处理指定的输出分辨率 + if options.width and options.height: + filter_complex.extend( + [ + f"[v:0]trim=start={start}:end={end},scale={options.width}:{options.height},setpts=PTS-STARTPTS[cut{index}]", + f"[a:0]atrim=start={start}:end={end},asetpts=PTS-STARTPTS[acut{index}]", + ] + ) + else: + filter_complex.extend( + [ + f"[v:0]trim=start={start}:end={end},setpts=PTS-STARTPTS[cut{index}]", + f"[a:0]atrim=start={start}:end={end},asetpts=PTS-STARTPTS[acut{index}]", + ] + ) + ffmpeg_cmd.option('filter_complex', ';'.join(filter_complex)) + + diff_tolerance = 0.001 + + for i, marker in enumerate(media_markers): + if marker.start.total_seconds() > metadata.format.duration or marker.start.total_seconds() < 0: + raise ValueError( + f"第{i}个切割点起始点{marker.start.total_seconds()}s超出视频时长[0-{metadata.format.duration}s]范围") + if marker.end.total_seconds() > metadata.format.duration or marker.end.total_seconds() < 0: + if marker.end.total_seconds() > 0 and math.isclose(marker.end.total_seconds(), metadata.format.duration, + rel_tol=diff_tolerance): + marker.end = TimeDelta(seconds=metadata.format.duration) + logger.warning( + f"第{i}个切割点结束点{marker.end.total_seconds()}s接近视频时长[0-{metadata.format.duration}s]范围") + else: + raise ValueError( + f"第{i}个切割点结束点{marker.end.total_seconds()}s超出视频时长[0-{metadata.format.duration}s]范围") + segment_output_path = FileUtils.file_path_extend(output_path, str(i)) + ffmpeg_options = { + "map": [f"[cut{i}]", f"[acut{i}]"], + "reset_timestamps": "1", + "sc_threshold": "0", + "g": "1", + "force_key_frames": "expr:gte(t,n_forced*1)", + "vcodec": "libx264", + "acodec": "aac", + "crf": options.crf, + "r": options.fps + } + if options.limit_size: + ffmpeg_options["fs"] = options.limit_size + elif options.bit_rate: + ffmpeg_options["b:v"] = options.pretty_bit_rate + + ffmpeg_cmd.output(segment_output_path, options=ffmpeg_options) + temp_outputs.append(segment_output_path) + + await ffmpeg_cmd.execute() + outputs: List[Tuple[str, VideoMetadata]] = [(output, VideoUtils.ffprobe_media_metadata(output)) for output in + temp_outputs] + return outputs + + @staticmethod + async def async_download_file(url: str, output_path: Optional[str] = None) -> str | None | Any: + t = 10 + while t > 0: + try: + logger.info(f"Downloading {url} to {output_path}") + async with aiohttp.ClientSession(timeout=ClientTimeout(total=60)) as session: + async with session.get(url) as response: + if response.status != 200: + raise Exception(f"Failed to download {url}, status code: {response.status}") + + if output_path: + async with aiofiles.open(output_path, 'wb') as f: + await f.write(await response.read()) + return output_path + else: + return await response.text() + except: + t -= 1 + logger.warning(f"Retrying downloading {url} to {output_path} Remain Times: {t}") + + @staticmethod + async def convert_m3u8_to_local_source(media_stream_url: str, + head: float = 0, + tail: float = 86400, # 使用24H时长替代♾️ + temp_dir: str = None) -> tuple[str, str, TimeDelta]: + """ + 转换m3u8为本地来源 + """ + # 创建临时目录存储TS片段 + if temp_dir: + os.makedirs(temp_dir, exist_ok=True) + else: + temp_dir = tempfile.mkdtemp() + from m3u8 import SegmentList, Segment + try: + # 1. 下载m3u8文件 + playlist = m3u8.load(media_stream_url) + # duration = (tail - head) if head else None + origin_time: datetime = playlist.segments[0].current_program_date_time + logger.info(f"Start Timestamp: {origin_time}") + # 2. 解析TS片段URL + ts_urls: SegmentList[Segment] = SegmentList() + duration = 0 + min_head = origin_time + timedelta(seconds=head) + max_head = origin_time + timedelta(seconds=tail) + logger.info(f"min: {min_head}, max: {max_head}") + for segment in playlist.segments: + if min_head - timedelta(seconds=segment.duration) <= segment.current_program_date_time <= max_head: + logger.info(f"duration: {segment.duration}, head: {segment.current_program_date_time}") + duration += segment.duration + ts_urls.append(segment) + if len(ts_urls) > 0: + delta = min_head - ts_urls[0].current_program_date_time + diff = TimeDelta.from_timedelta(delta) + else: + diff = TimeDelta(seconds=0) + logger.info(f"diff = {diff.total_seconds()}") + # 3. 并行下载TS片段 + tasks = [] + playlist.segments = ts_urls + duration_delta = TimeDelta(seconds=duration) + logger.info(f"count : {len(playlist.segments)}, duration = {duration_delta.toFormatStr()}") + playlist.is_endlist = True + for url in ts_urls: + tasks.append(VideoUtils.async_download_file(url.absolute_uri, f"{temp_dir}/{url.uri}")) + await asyncio.gather(*tasks) + # 4. 修改m3u8文件指向本地TS片段 + local_m3u8_path = os.path.join(temp_dir, "local.m3u8") + playlist.dump(local_m3u8_path) + return local_m3u8_path, temp_dir, diff + except Exception as e: + logger.exception(e) + raise Exception(f"下载TS转换M3U8失败 {e}") + + @staticmethod + def purge_temp_ts_dir(temp_dir: str) -> None: + # 6. 删除临时文件和目录 + try: + shutil.rmtree(temp_dir) + except Exception as e: + logger.exception(e) + + @staticmethod + async def ffmpeg_convert_stream_media(media_stream_url: str, options: FFMPEGSliceOptions, + output_path: Optional[str] = None) -> tuple[ + str, VideoMetadata] | None: + if not output_path: + output_path = FileUtils.file_path_extend(media_stream_url, "convert") + if not output_path.endswith(".mp4"): + output_path = output_path + ".mp4" + os.makedirs(os.path.dirname(output_path), exist_ok=True) + + try: + local_m3u8_path, temp_dir, diff = await VideoUtils.convert_m3u8_to_local_source( + media_stream_url=media_stream_url) + # 使用ffmpeg合并TS片段 + ffmpeg_cmd = VideoUtils.async_ffmpeg_init() + ffmpeg_cmd.input(local_m3u8_path, + protocol_whitelist="file,http,https,tcp,tls") + ffmpeg_options = { + "reset_timestamps": "1", + "sc_threshold": "0", + "g": "1", + "force_key_frames": "expr:gte(t,n_forced*1)", + "vcodec": "libx264", + "b:v": options.pretty_bit_rate, + "acodec": "aac", + "crf": options.crf, + "r": options.fps, + } + + ffmpeg_cmd.output(output_path, options=ffmpeg_options) + await ffmpeg_cmd.execute() + VideoUtils.purge_temp_ts_dir(temp_dir) + except Exception as e: + logger.exception(f"合并TS失败 {e}") + output: Tuple[str, VideoMetadata] = (output_path, VideoUtils.ffprobe_media_metadata(output_path)) + return output + + @staticmethod + async def ffmpeg_concat_medias(media_paths: List[str], + target_width: int = 1080, + target_height: int = 1920, + output_path: Optional[str] = None) -> Tuple[str, VideoMetadata]: + """ + 将待处理的视频合并为一个视频 + :param media_paths: 待合并的多个视频文件路径 + :param target_width: 输出的视频分辨率宽 + :param target_height: 输出的视频分辨率高 + :param output_path: 指定输出视频路径 + :return: 最终合并结果路径，最终合并结果时长 + """ + + total_videos = len(media_paths) + if total_videos == 0: + raise ValueError("没有可以合并的视频源") + if not output_path: + output_path = FileUtils.file_path_extend(media_paths[0], "concat") + os.makedirs(os.path.dirname(output_path), exist_ok=True) + ffmpeg_cmd = VideoUtils.async_ffmpeg_init() + filter_complex = [] + for input_path in media_paths: + ffmpeg_cmd.input(input_path) + # 2. 统一所有视频的格式、分辨率和帧率 + for i in range(total_videos): + filter_complex.extend( + [ + # 先缩放到统一分辨率，然后设置帧率和格式 + f"[{i}:v]scale={target_width}:{target_height}:force_original_aspect_ratio=decrease," + f"pad={target_width}:{target_height}:(ow-iw)/2:(oh-ih)/2," + f"setsar=1:1," # 新增强制设置SAR + f"fps=30,format=yuv420p[v{i}]", + # 修改音频过滤器，确保输出为AAC兼容格式 + # f"[{i}:a]aformat=sample_fmts=fltp:sample_rates=44100:channel_layouts=stereo[a{i}]", + f"[{i}:a]aformat=sample_fmts=s16:sample_rates=44100:channel_layouts=stereo[a{i}]", + ] + ) + # 3. 准备处理后的视频流和音频流的连接字符串 + video_streams = "".join(f"[v{i}]" for i in range(total_videos)) + audio_streams = "".join(f"[a{i}]" for i in range(total_videos)) + + # 4. 使用concat过滤器合并视频和音频 + filter_complex.extend( + [ + f"{video_streams}concat=n={total_videos}:v=1:a=0[vconcated]", + f"{audio_streams}concat=n={total_videos}:v=0:a=1[aconcated]", + ] + ) + + ffmpeg_cmd.output( + output_path, + { + "filter_complex": ";".join(filter_complex), + "map": ["[vconcated]", "[aconcated]"], + "vcodec": "libx264", + "crf": 16, + "r": 30, + "acodec": "aac", + "ar": 44100, + "ac": 2, + "ab": "192k", + }, + ) + await ffmpeg_cmd.execute() + video_metadata = VideoUtils.ffprobe_media_metadata(output_path) + return output_path, video_metadata + + @staticmethod + async def ffmpeg_extract_audio_async(media_path: str, output_path: Optional[str] = None) -> Tuple[ + str, VideoMetadata]: + """ + 提取源视频的音频 + :param media_path: 待处理的源视频 + :param output_path: 指定输出的音频文件路径（可选） + :return: 最终输出音频文件路径，音频文件时长 + """ + if not output_path: + output_path = FileUtils.file_path_change_extension(output_path, 'wav') + os.makedirs(os.path.dirname(output_path), exist_ok=True) + ffprobe_cmd = VideoUtils.async_ffmpeg_init(use_ffprobe=True) + ffprobe_cmd.input(media_path, + v="quiet", + print_format="json", + select_streams="a", + show_entries="stream=codec_type") + audio_check_bytes = await ffprobe_cmd.execute() + audio_check = json.loads(audio_check_bytes) + logger.info(audio_check) + if len(audio_check['streams']) == 0: + raise RuntimeError(f"Media has no audio streams.") + # output_path = f"{output_path_prefix}/extract_audio/outputs/{fn_id}/output.wav" + os.makedirs(os.path.dirname(output_path), exist_ok=True) + ffmpeg_cmd = VideoUtils.async_ffmpeg_init() + ffmpeg_cmd.input(media_path).output(output_path, + map="0:a", + acodec="pcm_s16le", + ar=44100, + ac=1) + await ffmpeg_cmd.execute() + video_metadata = VideoUtils.ffprobe_media_metadata(output_path) + return output_path, video_metadata + + @staticmethod + async def ffmpeg_mix_bgm(origin_audio_path: str, bgm_audio_path: str, video_volume: float = 1.4, + music_volume: float = 0.1, output_path: Optional[str] = None) -> Tuple[str, VideoMetadata]: + """ + 给待处理视频混合BGM + :param origin_audio_path: 待处理的源视频 + :param bgm_audio_path: 需要混合的BGM + :param video_volume: 最终输出视频的音量系数 + :param music_volume: BGM在源视频音量内占比的音量系数 + :param output_path: 指定最终输出的视频路径（可选） + :return: 最终输出视频文件路径，最终输出视频时长 + """ + if not output_path: + output_path = FileUtils.file_path_extend(origin_audio_path, "bgm") + os.makedirs(os.path.dirname(output_path), exist_ok=True) + origin_audio_duration = VideoUtils.ffprobe_audio_duration(origin_audio_path) + bgm_duration = VideoUtils.ffprobe_audio_duration(bgm_audio_path) + loops_needed = math.ceil(origin_audio_duration.total_seconds() / bgm_duration.total_seconds()) + + ffmpeg_cmd = VideoUtils.async_ffmpeg_init(use_ffprobe=True) + ffmpeg_cmd.input(origin_audio_path) + ffmpeg_cmd.input(bgm_audio_path) + filter_complex = [ + f"[0:a]volume={video_volume}[a1]", + f"[1:a]aloop=loop={loops_needed}:size={bgm_duration.total_seconds()},volume={music_volume}[a2]" + "[a1][a2]amix=inputs=2:duration=first[audio]" + ] + ffmpeg_cmd.output(output_path, + options={"filter_complex": ";".join(filter_complex), }, + map="[audio]", + acodec='libmp3lame', # 音频编码器 + ar=48000, # 音频采样率 + ab='192k', # 音频码率 + ac=2, # 音频通道数 + ) + await ffmpeg_cmd.execute() + video_metadata = VideoUtils.ffprobe_media_metadata(output_path) + return output_path, video_metadata + + @staticmethod + async def ffmpeg_mix_bgm_with_noise_reduce(media_path: str, + bgm_audio_path: str, + video_volume: float = 1.4, + music_volume: float = 0.1, + noise_sample_path: Optional[str] = None, + temp_audio_path: Optional[str] = None, + output_path: Optional[str] = None) -> Tuple[str, VideoMetadata]: + """ + 先对待处理的视频音轨降噪，再将降噪后的结果添加BGM，最终输出降噪过且混合BGM的视频； + 由于最终视频画面和音轨是同步混合+合成视频，所以处理速度会比分步降噪, 加BGM快； + :param media_path: 待处理的原始视频路径 + :param bgm_audio_path: 待处理的BGM音频路径 + :param video_volume: 最终输出的视频音量系数 + :param music_volume: 最终输出的BGM音量系数 + :param noise_sample_path: 降噪使用的噪音样本，如不指定将使用源视频的前2秒作为样本（可选） + :param temp_audio_path: 指定暂存音频的路径（可选） + :param output_path: 指定输出视频的路径（可选） + :return: 最终输出视频的路径， 最终输出视频时长 + """ + + if not output_path: + output_path = FileUtils.file_path_extend(media_path, "bgm_nr") + os.makedirs(os.path.dirname(output_path), exist_ok=True) + if not temp_audio_path: + temp_audio_path = FileUtils.file_path_extend(media_path, "temp") + temp_audio_path = FileUtils.file_path_change_extension(temp_audio_path, "wav") + media_audio, metadata = await VideoUtils.ffmpeg_extract_audio_async(media_path=media_path, + output_path=temp_audio_path) + logger.info(f"media_audio = {media_audio}, metadata = {metadata}") + nr_audio_path = VideoUtils.noise_reduce(audio_path=media_audio, noise_sample_path=noise_sample_path) + logger.info(f"nr_audio_path = {nr_audio_path}") + video_metadata = VideoUtils.ffprobe_video_format(media_path) + origin_audio_duration = VideoUtils.ffprobe_audio_duration(nr_audio_path) + bgm_duration = VideoUtils.ffprobe_audio_duration(bgm_audio_path) + loops_needed = math.ceil(origin_audio_duration.total_seconds() / bgm_duration.total_seconds()) + logger.info( + f"{bgm_duration.total_seconds()}s的BGM循环{loops_needed}次, 填充{origin_audio_duration.total_seconds()}s的视频长度") + ffmpeg_cmd = VideoUtils.async_ffmpeg_init() + ffmpeg_cmd.input(media_path) + ffmpeg_cmd.input(nr_audio_path) + ffmpeg_cmd.input(bgm_audio_path) + filter_complex = [ + f"[1:a]volume={video_volume}[a1]", + f"[2:a]aloop=loop={loops_needed}:size={bgm_duration.total_seconds()},volume={music_volume}[a2]", + "[a1][a2]amix=inputs=2:duration=first[audio]" + ] + ffmpeg_cmd.output(output_path, + options={"filter_complex": ";".join(filter_complex), }, + map=["0:v", "[audio]"], + crf=16, + vcodec='libx264', + b=video_metadata.video_bitrate, # 视频码率 + r=video_metadata.video_frame_rate, # 帧率 + acodec='libmp3lame', + ar=48000, ab='192k', ac=2, + ) + await ffmpeg_cmd.execute() + video_metadata = VideoUtils.ffprobe_media_metadata(output_path) + return output_path, video_metadata + + @staticmethod + async def ffmpeg_overlay_gif(media_path: str, overlay_gif_path: str, output_path: Optional[str] = None) -> Tuple[ + str, VideoMetadata]: + """ + 将GIF特效叠加到视频上，如果视频较长则循环播放GIF + :param media_path: 输入视频路径 + :param overlay_gif_path: GIF特效文件路径 + :param output_path: 指定输出路径 + :return: 输出视频路径， 最终输出视频时长 + """ + if not output_path: + output_path = FileUtils.file_path_extend(media_path, "overlay") + os.makedirs(os.path.dirname(output_path), exist_ok=True) + video_metadata = VideoUtils.ffprobe_video_format(media_path) + filter_complex = [ + # 确保GIF正确解码并循环 + "[1:v]fps=30,format=rgba[gif]", # 强制设置30fps + # 叠加GIF到视频上，保持透明通道 + "[0:v][gif]overlay=shortest=1:format=auto,setpts=PTS-STARTPTS[v]", + ] + ffmpeg_cmd = VideoUtils.async_ffmpeg_init() + ffmpeg_cmd.input(media_path) + ffmpeg_cmd.input(overlay_gif_path, stream_loop=-1) # 使用stream_loop让GIF循环直到视频结束 + ffmpeg_cmd.output(output_path, + options={"filter_complex": ";".join(filter_complex), }, + map=["[v]", "0:a"], + crf=16, + vcodec='libx264', + b=video_metadata.video_bitrate, # 视频码率 + r=video_metadata.video_frame_rate, # 帧率 + ) + await ffmpeg_cmd.execute() + video_metadata = VideoUtils.ffprobe_media_metadata(output_path) + return output_path, video_metadata + + @staticmethod + async def ffmpeg_zoom_loop(media_path: str, duration: float = 6.0, zoom: float = 0.1, + output_path: Optional[str] = None) -> Tuple[str, VideoMetadata]: + """ + 视频放大缩小循环特效 + :param media_path: 待处理的视频文件路径 + :param duration: 视频特效循环时间长度 + :param zoom: 视频特效放大缩小系数 + :param output_path: 指定输出视频地址（可选） + :return: 最终输出视频地址， 最终输出视频时长 + """ + + if not output_path: + output_path = FileUtils.file_path_extend(media_path, 'zoomed') + os.makedirs(os.path.dirname(output_path), exist_ok=True) + video_metadata = VideoUtils.ffprobe_video_format(media_path) + # abs(sin())表达式会导致实际的往复频率为2倍 + duration = duration * 2 + ffmpeg_cmd = VideoUtils.async_ffmpeg_init() + ffmpeg_cmd.input(media_path).output(output_path, + options={ + "vf": f"scale={4 * video_metadata.width}x{4 * video_metadata.height},fps=30," + f"zoompan=z='1+{zoom}*abs(sin(2*PI*time/{duration}))':" + "x='trunc(iw/2*(1-1/zoom))':" + "y='trunc(ih/2*(1-1/zoom))':" + f"d=1:s={video_metadata.width}x{video_metadata.height}:fps=30" + }, + vcodec="libx264", + acodec="copy", + crf=16, + b=video_metadata.video_bitrate, # 视频码率 + r=video_metadata.video_frame_rate, # 帧率 + ) + await ffmpeg_cmd.execute() + video_metadata = VideoUtils.ffprobe_media_metadata(output_path) + return output_path, video_metadata + + @staticmethod + async def ffmpeg_corner_mirror(media_path: str, mirror_scale_down_size: int = 6, + mirror_from_right: bool = True, mirror_position: tuple[float, float] = (40, 40), + output_path: Optional[str] = None) -> Tuple[str, VideoMetadata]: + """ + 对源视频添加镜像小窗特效 + :param media_path: 待处理的源视频 + :param mirror_scale_down_size: 源视频画面缩放系数 + :param mirror_from_right: 小窗原点是否使用右下角 + :param mirror_position: 小窗基于原点坐标轴的偏移量 + :param output_path: 指定的输出视频路径（可选） + :return: 返回最终输出视频的路径， 最终输出视频时长 + """ + if not output_path: + output_path = FileUtils.file_path_extend(media_path, 'mir') + os.makedirs(os.path.dirname(output_path), exist_ok=True) + + ffmpeg_cmd = VideoUtils.async_ffmpeg_init() + + mirror_x = ( + f"main_w-overlay_w-{mirror_position[0]}" + if mirror_from_right + else str(mirror_position[0]) + ) + filter_complex = [ + "[0:v]split[original][mirror]", + f"[mirror]hflip,scale=iw/{mirror_scale_down_size}:-1,format=rgba[flipped]", + "[flipped]split[fm1][fm2]", + f"[fm2]format=gray,geq=lum='255*(1-pow(min(1,2*sqrt(pow(X/W-0.5,2)+pow(Y/H-0.5,2))),1.5))':a='if(lt(pow(X/W-0.5,2)+pow(Y/H-0.5,2),0.15),(1-pow(2*sqrt(pow(X/W-0.5,2)+pow(Y/H-0.5,2)),1.5))*255,0)'[fm2Blur]", + "[fm1][fm2Blur]alphamerge[flipped_blured]", + f"[original][flipped_blured]overlay=x={mirror_x}:y=main_h-overlay_h-{mirror_position[1]}[video]", + ] + video_metadata = VideoUtils.ffprobe_video_format(media_path) + ffmpeg_cmd.input(media_path).output(output_path, + options={"filter_complex": ";".join(filter_complex)}, + map=["[video]", "0:a"], + vcodec="libx264", + crf=16, + b=video_metadata.video_bitrate, # 视频码率 + r=video_metadata.video_frame_rate # 帧率 + ) + await ffmpeg_cmd.execute() + video_metadata = VideoUtils.ffprobe_media_metadata(output_path) + return output_path, video_metadata + + @staticmethod + async def ffmpeg_subtitle_apply(media_path: str, subtitle_path: Optional[str], embed_subtitle_path: Optional[str], + font_dir: Optional[str], output_path: Optional[str] = None) -> Tuple[ + str, VideoMetadata]: + """ + 给视频画面叠加字幕，需要确保字幕文件为ass字幕，并且subtitle文件内设置的字体存在与font_dir文件夹内 + :param media_path: 待处理的源视频 + :param subtitle_path: ass渲染字幕文件路径 + :param embed_subtitle_path: ass/vtt/srt内嵌字幕文件路径 + :param font_dir: 字体文件目录路径 + :param output_path: 指定输出文件路径（可选） + :return: 返回最终输出视频路径， 最终输出视频时长 + """ + if not output_path: + output_path = FileUtils.file_path_extend(media_path, 'sub') + + video_metadata = VideoUtils.ffprobe_video_format(media_path) + ffmpeg_cmd = VideoUtils.async_ffmpeg_init() + ffmpeg_cmd.input(media_path) + if embed_subtitle_path: + ffmpeg_cmd.input(embed_subtitle_path) + ffmpeg_options = { + "vcodec": "libx264", + "acodec": "copy", + "crf": 16, + "b": video_metadata.video_bitrate, + "r": video_metadata.video_frame_rate, + "map": ["0:v", "0:a"] + } + if subtitle_path and font_dir: + ffmpeg_options["vf"] = f"subtitles=filename={subtitle_path}:fontsdir={font_dir}" + if embed_subtitle_path: + ffmpeg_options['map'].append("1") + ffmpeg_options['c:s'] = "mov_text" + ffmpeg_options['metadata:s:s:0'] = "language=chi" + ffmpeg_cmd.output(output_path, options=ffmpeg_options) + await ffmpeg_cmd.execute() + video_metadata = VideoUtils.ffprobe_media_metadata(output_path) + return output_path, video_metadata + + @staticmethod + async def ffmpeg_fill_longest(video_path: str, audio_path: str, output_path: Optional[str] = None) -> Tuple[ + str, VideoMetadata]: + """ + 用视频循环对齐音频时长，如果短于音频时长则循环填满音频时长，如短于音频时长则裁剪结尾 + :param video_path: 使用的视频文件路径 + :param audio_path: 匹配的音频文件路径 + :param output_path: 指定输出文件地址 + :return: 最终输出的文件路径， 最终输出视频详细信息 + """ + video_metadata = VideoUtils.ffprobe_video_format(video_path) + audio_duration = VideoUtils.ffprobe_audio_duration(audio_path) + loop_times = 0 if video_metadata.duration > audio_duration.total_seconds() else int( + math.ceil(audio_duration.total_seconds() / video_metadata.duration)) + logger.info( + f"视频长度 = {video_metadata.duration}, 音频长度 = {audio_duration.total_seconds()}, 重复 = {loop_times}") + + if not output_path: + output_path = FileUtils.file_path_extend(video_path, 'fill') + os.makedirs(os.path.dirname(output_path), exist_ok=True) + + ffmpeg_cmd = VideoUtils.async_ffmpeg_init() + ffmpeg_cmd.input(video_path, stream_loop=str(loop_times)) + ffmpeg_cmd.input(audio_path) + + ffmpeg_cmd.output(output_path, + map=["0:v", "1:a"], + vcodec="copy", + acodec="aac", + shortest=None, + ) + await ffmpeg_cmd.execute() + video_metadata = VideoUtils.ffprobe_media_metadata(output_path) + return output_path, video_metadata + + @staticmethod + async def ffmpeg_extract_frame_image(video_path: str, frame_index: int, seek_time: Optional[TimeDelta] = None, + output_path: Optional[str] = None) -> Tuple[ + str, VideoMetadata]: + """ + 获取视频的第n帧输出为图片, 并返回图片相关的元数据 + """ + if not output_path: + output_path = FileUtils.file_path_extend(video_path, 'cover') + output_path = FileUtils.file_path_change_extension(output_path, 'jpg') + os.makedirs(os.path.dirname(output_path), exist_ok=True) + ffmpeg_cmd = VideoUtils.async_ffmpeg_init() + if seek_time: + ffmpeg_cmd.input(video_path, ss=seek_time.total_seconds()) + else: + ffmpeg_cmd.input(video_path) + ffmpeg_cmd.output(output_path, vframes=frame_index) + await ffmpeg_cmd.execute() + image_metadata = VideoUtils.ffprobe_media_metadata(output_path) + return output_path, image_metadata + + @staticmethod + async def ffmpeg_stream_record_as_hls(stream_url: str, + segments_output_dir: str, + playlist_output_dir: str, + first_segment_duration: float = 2.0, + segment_duration: float = 5.0, + stream_content_timeout: int = 300, + stream_monitor_timeout: int = 36000, + output_file_pattern: str = "%10d.ts"): + os.makedirs(segments_output_dir, exist_ok=True) + ffmpeg_cmd = VideoUtils.async_ffmpeg_init(quiet=True) + # ffmpeg_cmd.option("loglevel", "debug") + ffmpeg_cmd.option("t", stream_monitor_timeout) + ffmpeg_cmd.input(stream_url, + protocol_whitelist="file,http,https,tcp,tls", # 使用flv + reconnect="1", # 自动重连 + reconnect_at_eof="1", + reconnect_streamed="1", + reconnect_delay_max="5") + output_playlist = f"{playlist_output_dir}/playlist.m3u8" + ffmpeg_cmd.output( + output_playlist, + f="hls", + hls_init_time=first_segment_duration, + hls_time=segment_duration, + hls_segment_filename=f"{segments_output_dir}/{output_file_pattern}", + hls_segment_type="mpegts", + hls_flags="append_list+independent_segments+program_date_time+split_by_time+discont_start", + hls_playlist_type="event", + hls_list_size=0, + hls_start_number_source="epoch_us", + timeout=stream_content_timeout, + c="copy", + ) + await ffmpeg_cmd.execute() + logger.info(f'停止录制') + return output_playlist