You've already forked FrameTour-RenderWorker
Jerry Yan
9c6186ecd3
- 在 TASK_TYPES 中新增 COMPOSE_TRANSITION 类型 - 定义 TRANSITION_TYPES 常量支持多种转场效果 - 在 TaskType 枚举中添加 COMPOSE_TRANSITION - 创建 TransitionConfig 数据类处理转场配置 - 为 RenderSpec 添加 transition_in 和 transition_out 属性 - 在 Task 类中添加转场相关的方法 - 新增 ComposeTransitionHandler 处理转场合成任务 - 修改 PackageSegmentTsHandler 支持转场分片封装 - 修改 RenderSegmentVideoHandler 支持 overlap 区域生成 - 在 TaskExecutor 中注册转场处理器
305 lines
10 KiB
Python
305 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
TS 分片封装处理器
|
|
|
|
处理 PACKAGE_SEGMENT_TS 任务，将视频片段和对应时间区间的音频封装为 TS 分片。
|
|
支持转场相关的 overlap 裁剪和转场分片封装。
|
|
"""
|
|
|
|
import os
|
|
import logging
|
|
from typing import List, Optional
|
|
|
|
from handlers.base import BaseHandler, VIDEO_ENCODE_ARGS
|
|
from domain.task import Task, TaskType
|
|
from domain.result import TaskResult, ErrorCode
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PackageSegmentTsHandler(BaseHandler):
|
|
"""
|
|
TS 分片封装处理器
|
|
|
|
职责：
|
|
- 下载视频片段
|
|
- 下载全局音频
|
|
- 截取对应时间区间的音频
|
|
- 封装为 TS 分片
|
|
- 上传 TS 产物
|
|
|
|
关键约束：
|
|
- TS 必须包含音视频同轨
|
|
- 使用 output_ts_offset 保证时间戳连续
|
|
- 输出 extinfDurationSec 供 m3u8 使用
|
|
|
|
转场相关：
|
|
- 普通片段 TS：需要裁剪掉 overlap 区域（已被转场分片使用）
|
|
- 转场分片 TS：直接封装转场视频产物，无需裁剪
|
|
- 无转场时：走原有逻辑，不做裁剪
|
|
|
|
精确裁剪：
|
|
- 当需要裁剪 overlap 区域时，必须使用重编码方式（-vf trim）才能精确切割
|
|
- 使用 -c copy 只能从关键帧切割，会导致不精确
|
|
"""
|
|
|
|
def get_supported_type(self) -> TaskType:
|
|
return TaskType.PACKAGE_SEGMENT_TS
|
|
|
|
def handle(self, task: Task) -> TaskResult:
|
|
"""处理 TS 封装任务"""
|
|
work_dir = self.create_work_dir(task.task_id)
|
|
|
|
try:
|
|
# 解析参数
|
|
video_url = task.get_video_url()
|
|
audio_url = task.get_audio_url()
|
|
start_time_ms = task.get_start_time_ms()
|
|
duration_ms = task.get_duration_ms()
|
|
output_spec = task.get_output_spec()
|
|
|
|
# 转场相关参数
|
|
is_transition_segment = task.is_transition_segment()
|
|
trim_head = task.should_trim_head()
|
|
trim_tail = task.should_trim_tail()
|
|
trim_head_ms = task.get_trim_head_ms()
|
|
trim_tail_ms = task.get_trim_tail_ms()
|
|
|
|
if not video_url:
|
|
return TaskResult.fail(
|
|
ErrorCode.E_SPEC_INVALID,
|
|
"Missing videoUrl"
|
|
)
|
|
|
|
if not audio_url:
|
|
return TaskResult.fail(
|
|
ErrorCode.E_SPEC_INVALID,
|
|
"Missing audioUrl"
|
|
)
|
|
|
|
# 计算时间参数
|
|
start_sec = start_time_ms / 1000.0
|
|
duration_sec = duration_ms / 1000.0
|
|
|
|
# 1. 下载视频片段
|
|
video_file = os.path.join(work_dir, 'video.mp4')
|
|
if not self.download_file(video_url, video_file):
|
|
return TaskResult.fail(
|
|
ErrorCode.E_INPUT_UNAVAILABLE,
|
|
f"Failed to download video: {video_url}"
|
|
)
|
|
|
|
# 2. 下载全局音频
|
|
audio_file = os.path.join(work_dir, 'audio.aac')
|
|
if not self.download_file(audio_url, audio_file):
|
|
return TaskResult.fail(
|
|
ErrorCode.E_INPUT_UNAVAILABLE,
|
|
f"Failed to download audio: {audio_url}"
|
|
)
|
|
|
|
# 3. 判断是否需要精确裁剪视频
|
|
needs_video_trim = not is_transition_segment and (
|
|
(trim_head and trim_head_ms > 0) or
|
|
(trim_tail and trim_tail_ms > 0)
|
|
)
|
|
|
|
# 4. 如果需要裁剪，先重编码裁剪视频
|
|
processed_video_file = video_file
|
|
if needs_video_trim:
|
|
processed_video_file = os.path.join(work_dir, 'trimmed_video.mp4')
|
|
trim_cmd = self._build_trim_command(
|
|
video_file=video_file,
|
|
output_file=processed_video_file,
|
|
trim_head_ms=trim_head_ms if trim_head else 0,
|
|
trim_tail_ms=trim_tail_ms if trim_tail else 0,
|
|
output_spec=output_spec
|
|
)
|
|
|
|
logger.info(f"[task:{task.task_id}] Trimming video: head={trim_head_ms}ms, tail={trim_tail_ms}ms")
|
|
|
|
if not self.run_ffmpeg(trim_cmd, task.task_id):
|
|
return TaskResult.fail(
|
|
ErrorCode.E_FFMPEG_FAILED,
|
|
"Video trim failed"
|
|
)
|
|
|
|
if not self.ensure_file_exists(processed_video_file, min_size=1024):
|
|
return TaskResult.fail(
|
|
ErrorCode.E_FFMPEG_FAILED,
|
|
"Trimmed video file is missing or too small"
|
|
)
|
|
|
|
# 5. 构建 TS 封装命令
|
|
output_file = os.path.join(work_dir, 'segment.ts')
|
|
cmd = self._build_package_command(
|
|
video_file=processed_video_file,
|
|
audio_file=audio_file,
|
|
output_file=output_file,
|
|
start_sec=start_sec,
|
|
duration_sec=duration_sec
|
|
)
|
|
|
|
# 6. 执行 FFmpeg
|
|
if not self.run_ffmpeg(cmd, task.task_id):
|
|
return TaskResult.fail(
|
|
ErrorCode.E_FFMPEG_FAILED,
|
|
"TS packaging failed"
|
|
)
|
|
|
|
# 7. 验证输出文件
|
|
if not self.ensure_file_exists(output_file, min_size=1024):
|
|
return TaskResult.fail(
|
|
ErrorCode.E_FFMPEG_FAILED,
|
|
"TS output file is missing or too small"
|
|
)
|
|
|
|
# 8. 获取实际时长（用于 EXTINF）
|
|
actual_duration = self.probe_duration(output_file)
|
|
extinf_duration = actual_duration if actual_duration else duration_sec
|
|
|
|
# 9. 上传产物
|
|
ts_url = self.upload_file(task.task_id, 'ts', output_file)
|
|
if not ts_url:
|
|
return TaskResult.fail(
|
|
ErrorCode.E_UPLOAD_FAILED,
|
|
"Failed to upload TS"
|
|
)
|
|
|
|
return TaskResult.ok({
|
|
'tsUrl': ts_url,
|
|
'extinfDurationSec': extinf_duration
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"[task:{task.task_id}] Unexpected error: {e}", exc_info=True)
|
|
return TaskResult.fail(ErrorCode.E_UNKNOWN, str(e))
|
|
|
|
finally:
|
|
self.cleanup_work_dir(work_dir)
|
|
|
|
def _build_trim_command(
|
|
self,
|
|
video_file: str,
|
|
output_file: str,
|
|
trim_head_ms: int,
|
|
trim_tail_ms: int,
|
|
output_spec
|
|
) -> List[str]:
|
|
"""
|
|
构建视频精确裁剪命令（重编码方式）
|
|
|
|
使用 trim 滤镜进行精确帧级裁剪，而非 -ss/-t 参数的关键帧裁剪。
|
|
|
|
Args:
|
|
video_file: 输入视频路径
|
|
output_file: 输出视频路径
|
|
trim_head_ms: 头部裁剪时长（毫秒）
|
|
trim_tail_ms: 尾部裁剪时长（毫秒）
|
|
output_spec: 输出规格
|
|
|
|
Returns:
|
|
FFmpeg 命令参数列表
|
|
"""
|
|
# 获取原视频时长
|
|
original_duration = self.probe_duration(video_file)
|
|
if not original_duration:
|
|
original_duration = 10.0 # 默认值，避免除零
|
|
|
|
trim_head_sec = trim_head_ms / 1000.0
|
|
trim_tail_sec = trim_tail_ms / 1000.0
|
|
|
|
# 计算裁剪后的起止时间
|
|
start_time = trim_head_sec
|
|
end_time = original_duration - trim_tail_sec
|
|
|
|
# 构建 trim 滤镜
|
|
vf_filter = f"trim=start={start_time}:end={end_time},setpts=PTS-STARTPTS"
|
|
|
|
cmd = [
|
|
'ffmpeg', '-y', '-hide_banner',
|
|
'-i', video_file,
|
|
'-vf', vf_filter,
|
|
]
|
|
|
|
# 编码参数
|
|
cmd.extend(VIDEO_ENCODE_ARGS)
|
|
|
|
# 帧率
|
|
fps = output_spec.fps
|
|
cmd.extend(['-r', str(fps)])
|
|
|
|
# 计算输出视频帧数，动态调整 GOP
|
|
output_duration_sec = end_time - start_time
|
|
total_frames = int(output_duration_sec * fps)
|
|
|
|
# 动态 GOP：短视频使用较小的 GOP
|
|
if total_frames <= 1:
|
|
gop_size = 1
|
|
elif total_frames < fps:
|
|
gop_size = total_frames
|
|
else:
|
|
gop_size = fps # 每秒一个关键帧
|
|
|
|
cmd.extend(['-g', str(gop_size)])
|
|
cmd.extend(['-keyint_min', str(min(gop_size, fps // 2 or 1))])
|
|
|
|
# 强制第一帧为关键帧
|
|
cmd.extend(['-force_key_frames', 'expr:eq(n,0)'])
|
|
|
|
# 无音频（音频单独处理）
|
|
cmd.append('-an')
|
|
|
|
cmd.append(output_file)
|
|
|
|
return cmd
|
|
|
|
def _build_package_command(
|
|
self,
|
|
video_file: str,
|
|
audio_file: str,
|
|
output_file: str,
|
|
start_sec: float,
|
|
duration_sec: float
|
|
) -> List[str]:
|
|
"""
|
|
构建 TS 封装命令
|
|
|
|
将视频和对应时间区间的音频封装为 TS 分片。
|
|
视频使用 copy 模式（已经过精确裁剪或无需裁剪）。
|
|
|
|
Args:
|
|
video_file: 视频文件路径（已处理）
|
|
audio_file: 音频文件路径
|
|
output_file: 输出文件路径
|
|
start_sec: 音频开始时间（秒）
|
|
duration_sec: 音频时长（秒）
|
|
|
|
Returns:
|
|
FFmpeg 命令参数列表
|
|
"""
|
|
cmd = [
|
|
'ffmpeg', '-y', '-hide_banner',
|
|
# 视频输入
|
|
'-i', video_file,
|
|
# 音频输入（从 start_sec 开始截取 duration_sec）
|
|
'-ss', str(start_sec),
|
|
'-t', str(duration_sec),
|
|
'-i', audio_file,
|
|
# 映射流
|
|
'-map', '0:v:0', # 使用第一个输入的视频流
|
|
'-map', '1:a:0', # 使用第二个输入的音频流
|
|
# 复制编码（视频已处理，无需重编码）
|
|
'-c:v', 'copy',
|
|
'-c:a', 'copy',
|
|
# 关键：时间戳偏移，保证整体连续
|
|
'-output_ts_offset', str(start_sec),
|
|
# 复用参数
|
|
'-muxdelay', '0',
|
|
'-muxpreload', '0',
|
|
# 输出格式
|
|
'-f', 'mpegts',
|
|
output_file
|
|
]
|
|
|
|
return cmd
|