fix: 视频切分
This commit is contained in:
root
@@ -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
|
||||
|
||||
# 工作流状态
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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}")
|
||||
504
python_core/services/ffmpeg_slice_service_sync.py
Normal file
504
python_core/services/ffmpeg_slice_service_sync.py
Normal file
@@ -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)
|
||||
}
|
||||
Reference in New Issue
Block a user