From 5659bb34f35bcf917ee2cc5d49328428d51042b3 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 12 Jul 2025 15:10:38 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E8=A7=86=E9=A2=91=E5=88=87=E5=88=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../types/batch_workflow_state.py | 3 +- .../workflows/batch_workflow_nodes.py | 124 ++- python_core/services/ffmpeg_slice_service.py | 705 ------------------ .../services/ffmpeg_slice_service_sync.py | 504 +++++++++++++ 4 files changed, 616 insertions(+), 720 deletions(-) delete mode 100644 python_core/services/ffmpeg_slice_service.py create mode 100644 python_core/services/ffmpeg_slice_service_sync.py diff --git a/python_core/scene_detection/types/batch_workflow_state.py b/python_core/scene_detection/types/batch_workflow_state.py index da250cd..9a1de7b 100644 --- a/python_core/scene_detection/types/batch_workflow_state.py +++ b/python_core/scene_detection/types/batch_workflow_state.py @@ -41,9 +41,10 @@ class BatchSceneDetectionWorkflowState: use_advanced_split: bool = True split_quality: int = 23 split_preset: str = "fast" + max_video_duration: float = 60.0 # 最大视频时长（秒），默认60秒 # 批量处理配置 - max_concurrent: int = 2 + max_concurrent: int = 4 continue_on_error: bool = True # 工作流状态 diff --git a/python_core/scene_detection/workflows/batch_workflow_nodes.py b/python_core/scene_detection/workflows/batch_workflow_nodes.py index 39a3380..064d675 100644 --- a/python_core/scene_detection/workflows/batch_workflow_nodes.py +++ b/python_core/scene_detection/workflows/batch_workflow_nodes.py @@ -5,9 +5,8 @@ Batch Workflow Nodes """ import time -import asyncio from pathlib import Path -from typing import Dict, Any, List +from typing import Dict, Any from concurrent.futures import ThreadPoolExecutor, as_completed from python_core.utils.logger import logger @@ -17,7 +16,7 @@ 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 python_core.services.ffmpeg_slice_service_sync import FfmpegSliceService, SliceOptions, SliceSegment from ..utils.result_saver import ResultSaver @@ -202,13 +201,100 @@ class BatchWorkflowNodes: 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 - ) + # 转换场景为SliceSegment + segments = [ + SliceSegment(start=scene.start_time, end=scene.end_time) + for scene in detection_result.scenes + ] + + # 创建输出目录 + scenes_dir = task.output_dir / "scenes" + scenes_dir.mkdir(parents=True, exist_ok=True) + + # 生成输出路径 + base_output_path = str(scenes_dir / f"{task.video_path.stem}_scene") + + # 使用同步方法进行切分 + # 检查是否无场景，如果是则跳过切分 + if len(segments) <= 0: + logger.info(f"🔄 检测到 0 个场景切换，跳过切分，直接使用源文件") + # 创建输出目录 + scenes_dir.mkdir(parents=True, exist_ok=True) + + # 直接复制源文件作为结果 + import shutil + source_file = task.video_path + target_file = scenes_dir / f"{task.video_path.stem}_scene_001.mp4" + + try: + shutil.copy2(source_file, target_file) + logger.info(f"✅ 源文件已复制到: {target_file}") + + # 获取源文件元数据 + metadata = self.splitter_service.get_video_metadata(str(source_file)) + + # 创建模拟的切分结果 + slice_results = [(str(target_file), metadata)] + + # 修正场景统计：无场景切换 = 1个场景（整个视频） + detection_result.total_scenes = 1 + + except Exception as e: + logger.error(f"❌ 复制源文件失败: {e}") + slice_results = [] + else: + logger.info(f"🎬 检测到 {len(segments)} 个场景，开始切分") + slice_results = self.splitter_service.slice_video( + media_path=str(task.video_path), + segments=segments, + options=slice_options, + output_path=base_output_path + ) + + # 转换为兼容格式 + split_results_raw = [] + + # 创建一个简单的结果对象 + class SplitResult: + def __init__(self, scene_index, output_path, start_time, end_time, duration, file_size, success, error=None): + self.scene_index = scene_index + self.output_path = Path(output_path) + self.start_time = start_time + self.end_time = end_time + self.duration = duration + self.file_size = file_size + self.success = success + self.error = error + + for i, (output_path, metadata) in enumerate(slice_results): + if len(detection_result.scenes) == 0: + # 无场景切换的情况：整个视频作为一个场景 + split_result = SplitResult( + scene_index=0, + output_path=output_path, + start_time=0.0, + end_time=metadata.duration, + duration=metadata.duration, + file_size=metadata.size, + success=Path(output_path).exists(), + error=None if Path(output_path).exists() else "文件不存在" + ) + split_results_raw.append(split_result) + else: + # 有场景切换的情况 + scene = detection_result.scenes[i] if i < len(detection_result.scenes) else None + if scene: + split_result = SplitResult( + scene_index=i, + output_path=output_path, + start_time=scene.start_time, + end_time=scene.end_time, + duration=scene.duration, + file_size=metadata.size, + success=Path(output_path).exists(), + error=None if Path(output_path).exists() else "文件不存在" + ) + split_results_raw.append(split_result) # 转换为字典格式 split_results = [ @@ -225,10 +311,18 @@ class BatchWorkflowNodes: for r in split_results_raw ] - # 保存切分摘要 - split_summary = self.splitter_service.create_split_summary( - task.video_path, split_results_raw - ) + # 创建切分摘要 + successful_results = [r for r in split_results_raw if r.success] + failed_results = [r for r in split_results_raw if not r.success] + + split_summary = { + "video_path": str(task.video_path), + "total_scenes": len(split_results_raw), + "successful_splits": len(successful_results), + "failed_splits": len(failed_results), + "success_rate": len(successful_results) / len(split_results_raw) * 100 if split_results_raw else 0, + "total_output_size": sum(r.file_size for r in successful_results) + } summary_file = task.output_dir / "split_summary.json" import json with open(summary_file, 'w', encoding='utf-8') as f: @@ -238,6 +332,8 @@ class BatchWorkflowNodes: logger.error(f"视频切分失败: {e}") # 切分失败不影响整体任务成功 + # 4. 添加视频时长检查，如果时长大于 最大视频时长 那么就要进行二次切分 确保 视频不大于 最大视频时长 + task.end_time = time.time() return { diff --git a/python_core/services/ffmpeg_slice_service.py b/python_core/services/ffmpeg_slice_service.py deleted file mode 100644 index ce6a6e5..0000000 --- a/python_core/services/ffmpeg_slice_service.py +++ /dev/null @@ -1,705 +0,0 @@ -""" -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/services/ffmpeg_slice_service_sync.py b/python_core/services/ffmpeg_slice_service_sync.py new file mode 100644 index 0000000..d3aee6e --- /dev/null +++ b/python_core/services/ffmpeg_slice_service_sync.py @@ -0,0 +1,504 @@ +""" +FFmpeg视频切片服务 - 同步版本 + +基于原有的FfmpegSliceService，但使用同步方法而不是异步。 +""" + +import subprocess +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 loguru import logger +from ffmpeg.ffmpeg import FFmpeg + + +@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, ffprobe_data: Dict[str, Any]) -> 'VideoMetadata': + """从ffprobe数据创建VideoMetadata对象""" + format_info = ffprobe_data.get('format', {}) + streams = ffprobe_data.get('streams', []) + + video_stream = None + audio_stream = None + + for stream in streams: + if stream.get('codec_type') == 'video' and not video_stream: + video_stream = stream + elif stream.get('codec_type') == 'audio' and not audio_stream: + audio_stream = stream + + if not video_stream: + raise ValueError("No video stream found in the media file") + + # 解析帧率 + fps = 0.0 + if video_stream.get('r_frame_rate'): + try: + fps_str = video_stream['r_frame_rate'] + if '/' in fps_str: + num, den = fps_str.split('/') + fps = float(num) / float(den) if float(den) != 0 else 0.0 + else: + fps = float(fps_str) + except (ValueError, ZeroDivisionError): + fps = 0.0 + + 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_stream.get('codec_name', '') if audio_stream else '' + ) + + +class FfmpegSliceService: + """FFmpeg视频切片服务 - 同步版本""" + + def __init__(self): + """初始化服务""" + self.temp_dir = None + logger.info("FfmpegSliceService (同步版本) 初始化完成") + + def get_video_metadata(self, media_path: str) -> VideoMetadata: + """ + 获取视频元数据 + + Args: + media_path: 视频文件路径 + + Returns: + VideoMetadata: 视频元数据对象 + """ + # 首先验证文件是否存在和有效 + 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}") + + logger.info(f"开始获取视频元数据: {media_path} (大小: {media_file.stat().st_size} 字节)") + + try: + # 构建ffprobe命令 + cmd = [ + 'ffprobe', + '-v', 'quiet', + '-print_format', 'json', + '-show_streams', + '-show_format', + str(media_path) + ] + + # 执行ffprobe命令 + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=30 + ) + + # 详细的错误信息 + if result.returncode != 0: + stderr_text = result.stderr if result.stderr else 'No error output' + stdout_text = result.stdout 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}") + + stdout_text = result.stdout + 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 subprocess.TimeoutExpired: + logger.error(f"FFprobe timeout for {media_path}") + raise RuntimeError(f"FFprobe timeout for {media_path}") + 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 < 0: + raise ValueError(f"Segment {i+1}: start time cannot be negative ({segment.start})") + + if segment.end <= segment.start: + raise ValueError(f"Segment {i+1}: end time ({segment.end}) must be greater than start time ({segment.start})") + + if segment.start > video_duration: + raise ValueError(f"Segment {i+1}: start time ({segment.start}) exceeds video duration ({video_duration})") + + if segment.end > video_duration + diff_tolerance: + raise ValueError(f"Segment {i+1}: end time ({segment.end}) exceeds video duration ({video_duration})") + + def _generate_output_path(self, base_path: str, index: int) -> str: + """ + 生成输出文件路径 + + Args: + base_path: 基础路径 + index: 片段索引 + + Returns: + str: 输出文件路径 + """ + base = Path(base_path) + # 确保有扩展名 + suffix = base.suffix if base.suffix else '.mp4' + return str(base.parent / f"{base.stem}_{index+1:03d}{suffix}") + + 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 = 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) + + # 检查是否有音频流 + 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"[0:v]trim=start={start}:end={end},scale={options.width}:{options.height},setpts=PTS-STARTPTS[cut{index}]") + if has_audio: + filter_complex.append(f"[0:a]atrim=start={start}:end={end},asetpts=PTS-STARTPTS[acut{index}]") + else: + filter_complex.append(f"[0:v]trim=start={start}:end={end},setpts=PTS-STARTPTS[cut{index}]") + if has_audio: + filter_complex.append(f"[0:a]atrim=start={start}:end={end},asetpts=PTS-STARTPTS[acut{index}]") + + segment_output_path = self._generate_output_path(output_path, index) + temp_outputs.append(segment_output_path) + + # 构建完整的FFmpeg命令 + cmd = [ + 'ffmpeg', + '-i', media_path, + '-filter_complex', ';'.join(filter_complex) + ] + + # 为每个片段添加映射和编码选项 + for i, segment in enumerate(segments): + segment_output_path = temp_outputs[i] + + # 根据是否有音频流配置映射 + if has_audio: + cmd.extend(['-map', f'[cut{i}]', '-map', f'[acut{i}]']) + else: + cmd.extend(['-map', f'[cut{i}]']) + + # 编码选项 + cmd.extend([ + '-c:v', 'libx264', + '-preset', options.preset, + '-crf', str(options.crf), + '-r', str(options.fps), + '-reset_timestamps', '1', + '-sc_threshold', '0', + '-g', '1', + '-force_key_frames', 'expr:gte(t,n_forced*1)' + ]) + + # 只有在有音频时才添加音频编码器 + if has_audio: + cmd.extend(['-c:a', 'aac']) + + if options.limit_size: + cmd.extend(['-fs', options.limit_size]) + elif options.bit_rate: + cmd.extend(['-b:v', options.pretty_bit_rate]) + + cmd.extend(['-y', segment_output_path]) + + # 执行切片 + try: + logger.info(f"开始执行FFmpeg切片命令...") + logger.debug(f"FFmpeg命令: {' '.join(cmd[:10])}...") # 只显示前10个参数 + + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=600 # 10分钟超时 + ) + + # 检查执行结果 + if result.returncode != 0: + stderr_text = result.stderr if result.stderr else 'No error output' + stdout_text = result.stdout 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 subprocess.TimeoutExpired: + logger.error("FFmpeg切片超时") + # 清理可能创建的部分文件 + for output_file in temp_outputs: + try: + if Path(output_file).exists(): + Path(output_file).unlink() + logger.info(f"清理临时文件: {output_file}") + except Exception as cleanup_error: + logger.warning(f"清理文件失败 {output_file}: {cleanup_error}") + raise RuntimeError("FFmpeg切片超时") + except Exception as e: + logger.error(f"FFmpeg切片过程中发生错误: {e}") + # 清理可能创建的部分文件 + for output_file in temp_outputs: + try: + if Path(output_file).exists(): + Path(output_file).unlink() + logger.info(f"清理临时文件: {output_file}") + except Exception as cleanup_error: + logger.warning(f"清理文件失败 {output_file}: {cleanup_error}") + raise + + # 收集输出文件和元数据 + outputs = [] + for output_file in temp_outputs: + try: + output_metadata = 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.0, + width=metadata.width, + height=metadata.height, + fps=metadata.fps, + format_name="mp4", + size=Path(output_file).stat().st_size if Path(output_file).exists() else 0, + codec_name="h264", + audio_codec=metadata.audio_codec if has_audio else "" + ) + outputs.append((output_file, basic_metadata)) + + logger.info(f"切片完成，生成 {len(outputs)} 个文件") + return outputs + + def check_and_split_by_duration(self, video_path: str, max_duration: float, + options: SliceOptions, output_dir: str = None) -> List[Tuple[str, VideoMetadata]]: + """ + 检查视频时长，如果超过最大时长则进行二次切分 + + Args: + video_path: 视频文件路径 + max_duration: 最大允许时长（秒） + options: 切分选项 + output_dir: 输出目录，如果为None则使用视频文件所在目录 + + Returns: + List[Tuple[str, VideoMetadata]]: 切分后的文件路径和元数据列表 + """ + # 获取视频元数据 + metadata = self.get_video_metadata(video_path) + + if metadata.duration <= max_duration: + # 时长未超过限制，直接返回原文件 + logger.info(f"📏 视频时长 {metadata.duration:.2f}s 未超过限制 {max_duration:.2f}s，无需二次切分") + return [(video_path, metadata)] + + logger.info(f"⚠️ 视频时长 {metadata.duration:.2f}s 超过限制 {max_duration:.2f}s，开始二次切分") + + # 计算需要切分的段数 + num_segments = int(metadata.duration / max_duration) + 1 + segment_duration = metadata.duration / num_segments + + logger.info(f"🔄 将切分为 {num_segments} 段，每段约 {segment_duration:.2f}s") + + # 创建切分段 + segments = [] + for i in range(num_segments): + start_time = i * segment_duration + end_time = min((i + 1) * segment_duration, metadata.duration) + segments.append(SliceSegment(start=start_time, end=end_time)) + + # 准备输出路径 + if output_dir is None: + output_dir = str(Path(video_path).parent) + + video_name = Path(video_path).stem + base_output_path = str(Path(output_dir) / f"{video_name}_split") + + # 执行切分 + try: + results = self.slice_video( + media_path=video_path, + segments=segments, + options=options, + output_path=base_output_path + ) + + logger.info(f"✅ 二次切分完成，生成 {len(results)} 个文件") + return results + + except Exception as e: + logger.error(f"❌ 二次切分失败: {e}") + # 如果二次切分失败，返回原文件 + return [(video_path, metadata)] + + def check_ffmpeg_available(self) -> bool: + """检查FFmpeg是否可用""" + try: + result = subprocess.run( + ['ffmpeg', '-version'], + capture_output=True, + text=True, + timeout=10 + ) + return result.returncode == 0 + except (subprocess.TimeoutExpired, FileNotFoundError): + return False + + def create_split_summary(self, video_path, results) -> dict: + """创建切分摘要 - 兼容性方法""" + successful_results = [r for r in results if r.success] + failed_results = [r for r in results if not r.success] + + return { + "video_path": str(video_path), + "total_scenes": len(results), + "successful_splits": len(successful_results), + "failed_splits": len(failed_results), + "success_rate": len(successful_results) / len(results) * 100 if results else 0, + "total_output_size": sum(r.file_size for r in successful_results) + }