diff --git a/apps/worker/video_processing/concat_engine.py b/apps/worker/video_processing/concat_engine.py index cec7e3bf2..dc58a6723 100755 --- a/apps/worker/video_processing/concat_engine.py +++ b/apps/worker/video_processing/concat_engine.py @@ -32,14 +32,14 @@ logger = logging.getLogger(__name__) # concat demuxer 要求一致的参数列表 CONCAT_DEMUXER_REQUIRED_PARAMS = [ - "codec_name", # 视频编码 - "width", # 宽度 - "height", # 高度 - "r_frame_rate", # 帧率 - "pix_fmt", # 像素格式 - "sample_rate", # 音频采样率 - "channels", # 音频声道数 - "audio_codec", # 音频编码 + "codec_name", # 视频编码 + "width", # 宽度 + "height", # 高度 + "r_frame_rate", # 帧率 + "pix_fmt", # 像素格式 + "sample_rate", # 音频采样率 + "channels", # 音频声道数 + "audio_codec", # 音频编码 ] @@ -50,10 +50,10 @@ CONCAT_DEMUXER_REQUIRED_PARAMS = [ class ConcatSegment: """单个拼接片段.""" - video_path: str # 视频文件路径 - start_time: float = 0.0 # 开始时间(秒),从视频的哪个位置开始取 - duration: float = 0.0 # 持续时长(秒),0表示取到末尾 - has_audio: bool = True # 是否包含音频 + video_path: str # 视频文件路径 + start_time: float = 0.0 # 开始时间(秒),从视频的哪个位置开始取 + duration: float = 0.0 # 持续时长(秒),0表示取到末尾 + has_audio: bool = True # 是否包含音频 @classmethod def from_dict(cls, seg: dict) -> "ConcatSegment": @@ -81,11 +81,11 @@ class ConcatConfig: """视频拼接配置.""" segments: list[ConcatSegment] = field(default_factory=list) - output_width: int = 0 # 输出宽度(0=自动取第一段) - output_height: int = 0 # 输出高度(0=自动取第一段) - output_fps: float = 0.0 # 输出帧率(0=自动取第一段) + output_width: int = 0 # 输出宽度(0=自动取第一段) + output_height: int = 0 # 输出高度(0=自动取第一段) + output_fps: float = 0.0 # 输出帧率(0=自动取第一段) force_reencode: bool = False # 强制重新编码(不用 stream copy) - transition: str = "none" # 转场效果(none/crossfade)- 预留 + transition: str = "none" # 转场效果(none/crossfade)- 预留 transition_duration: float = 0.3 # 转场时长 @classmethod @@ -253,11 +253,7 @@ class ConcatEngine: # concat demuxer 支持 inpoint/outpoint,所以有裁剪也可以用 # 但为了简单和稳定性,有裁剪时也用 filter 模式 # (inpoint/outpoint 不是所有格式都支持得好) - has_trimming = any( - seg.start_time > 0 or seg.duration > 0 - for seg in config.segments - if seg.video_path - ) + has_trimming = any(seg.start_time > 0 or seg.duration > 0 for seg in config.segments if seg.video_path) if has_trimming: return False @@ -345,20 +341,14 @@ class ConcatEngine: if seg.has_audio: if seg.duration > 0: filter_parts.append( - f"[{i}:a]atrim=start={start:.3f}:end={end:.3f}," - f"asetpts=PTS-STARTPTS[{aud_label}]" + f"[{i}:a]atrim=start={start:.3f}:end={end:.3f}," f"asetpts=PTS-STARTPTS[{aud_label}]" ) else: - filter_parts.append( - f"[{i}:a]atrim=start={start:.3f}," - f"asetpts=PTS-STARTPTS[{aud_label}]" - ) + filter_parts.append(f"[{i}:a]atrim=start={start:.3f}," f"asetpts=PTS-STARTPTS[{aud_label}]") else: # 无音频时生成静音轨 filter_parts.append( - f"[{i}:v]trim=start={start:.3f}," - f"setpts=PTS-STARTPTS, " - f"aevalsrc=0:d={0.1}[{aud_label}]" + f"[{i}:v]trim=start={start:.3f}," f"setpts=PTS-STARTPTS, " f"aevalsrc=0:d={0.1}[{aud_label}]" ) else: # 无裁剪,直接用原始标签 @@ -368,9 +358,7 @@ class ConcatEngine: dur = probe_duration(seg.video_path) except Exception: dur = 10.0 - filter_parts.append( - f"aevalsrc=0:d={dur:.3f}:s=44100[{aud_label}]" - ) + filter_parts.append(f"aevalsrc=0:d={dur:.3f}:s=44100[{aud_label}]") # 2. 缩放/帧率统一 vf_parts = [] @@ -502,10 +490,7 @@ class ConcatEngine: concat_inputs += f"[{v_label}][{a_label}]" # concat filter: 输入按 [v0][a0][v1][a1]... 顺序 - filter_parts.append( - f"{concat_inputs}" - f"concat=n={num_segments}:v=1:a=1[vout][aout]" - ) + filter_parts.append(f"{concat_inputs}" f"concat=n={num_segments}:v=1:a=1[vout][aout]") filter_complex = ";".join(filter_parts)