CI Bot
9f153eec54
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 1m9s
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Successful in 46s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 1m4s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Validate - Code Quality (push) Failing after 2m34s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 1m45s
CI/CD Pipeline / Build Staging API Image (push) Successful in 7m18s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 7m15s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 1m21s
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Integration Tests (push) Successful in 2m30s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 2m22s
CI/CD Pipeline / Unit Tests (push) Failing after 6m18s
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 41s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 1m46s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 3m50s
CI/CD Pipeline / Canary Release to Production (push) Has been skipped
1946 lines
80 KiB
Python
Executable File
1946 lines
80 KiB
Python
Executable File
"""统一渲染引擎 — 输入 EditPlan + EditPlanClips，按时间线+图层渲染视频.
|
||
|
||
核心原则（灵应）：渲染引擎是统一的，不判断模式，只按 clip_type/config.role
|
||
分组为图层再合成。
|
||
|
||
图层分组：
|
||
main (无 config.role) → main (z=0)
|
||
main + config.role=b_roll → broll (z=0，与 main 同层替换)
|
||
overlay → overlay (z=1，画中画叠加)
|
||
background → background (z=0，全屏底图)
|
||
corner_voice → corner_voice (z=1，右上角小窗)
|
||
b_roll → broll (z=0)
|
||
intro / outro → main (z=0，按 order 排在首/尾)
|
||
|
||
合成流程：
|
||
1. 每个 clip 先 trim + scale + setpts 预处理
|
||
2. 同层 clips 按 order 用 xfade 串联
|
||
3. overlay/corner_voice 层 overlay 到主层
|
||
4. 如有独立音频轨，amix 混入
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import subprocess
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from video_processing.color_grade_engine import ColorGradeConfig, ColorGradeEngine
|
||
from video_processing.ffmpeg_utils import (
|
||
DEFAULT_FPS,
|
||
DEFAULT_OUTPUT_HEIGHT,
|
||
DEFAULT_OUTPUT_WIDTH,
|
||
DEFAULT_TRANSITION_DURATION,
|
||
FFMPEG_BIN,
|
||
probe_duration,
|
||
probe_video_info,
|
||
run_ffmpeg,
|
||
)
|
||
from video_processing.intro_outro_engine import IntroOutroConfig, IntroOutroEngine
|
||
from video_processing.pip_engine import PiPConfig, PiPEngine, PiPLayerConfig
|
||
from video_processing.render_audio import RenderContext, merge_audio_video, mix_audio
|
||
from video_processing.render_subtitles import generate_ass_subtitles
|
||
from video_processing.reverse_engine import ReverseConfig, ReverseEngine
|
||
from video_processing.speed_engine import SpeedEngine
|
||
from video_processing.sticker_engine import StickerEngine
|
||
from video_processing.subtitle_generator import generate_ass_from_timeline
|
||
from video_processing.transition_engine import TransitionEngine
|
||
from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_from_clip_config
|
||
from video_processing.tts_engine import TtsEngine
|
||
from video_processing.watermark_engine import WatermarkConfig, WatermarkEngine
|
||
|
||
from packages.domain.render_layer_utils import LAYER_Z_INDEX as _IMPORTED_LAYER_Z_INDEX
|
||
from packages.domain.render_layer_utils import can_pass_through as _can_pass_through_pure
|
||
from packages.domain.render_layer_utils import clip_adjusted_duration as _clip_adjusted_duration_pure
|
||
from packages.domain.render_layer_utils import clip_effective_duration as _clip_effective_duration_pure
|
||
from packages.domain.render_layer_utils import clip_playback_speed as _clip_playback_speed_pure
|
||
from packages.domain.render_layer_utils import estimate_total_duration as _estimate_total_duration_pure
|
||
from packages.domain.render_layer_utils import resolve_layer_role as _resolve_layer_role_pure
|
||
from packages.domain.tts_config import TtsConfig
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ── 数据结构 ──────────────────────────────────────────────────────────────────
|
||
|
||
|
||
@dataclass
|
||
class ResolvedClip:
|
||
"""已解析到本地路径的片段。"""
|
||
|
||
clip_id: str
|
||
asset_id: str
|
||
local_path: Path
|
||
clip_type: str
|
||
order: int
|
||
start_time: float = 0.0
|
||
duration: float = 0.0 # 0 表示使用素材完整时长
|
||
transition_effect: str = "cut"
|
||
transition_duration: float = 0.0 # 0 表示使用全局默认值
|
||
playback_speed: float = 1.0 # 0 或 1.0 表示原速
|
||
config: dict[str, Any] = field(default_factory=dict)
|
||
|
||
# 运行时填充
|
||
actual_duration: float = 0.0 # 素材实际时长（probe 后填充）
|
||
trim_config: TrimConfig | None = None # 解析后的裁剪配置（运行时填充）
|
||
|
||
|
||
@dataclass
|
||
class RenderLayer:
|
||
"""渲染图层。"""
|
||
|
||
role: str # "main" | "overlay" | "pip" | "background" | "corner_voice" | "broll" | "audio"
|
||
clips: list[ResolvedClip] = field(default_factory=list)
|
||
z_index: int = 0
|
||
opacity: float = 1.0
|
||
position: tuple[int, int] | None = None # (x, y) 偏移，None 表示全屏
|
||
|
||
|
||
@dataclass
|
||
class RenderResult:
|
||
"""渲染结果。"""
|
||
|
||
output_path: Path
|
||
duration: float
|
||
file_size: int
|
||
width: int
|
||
height: int
|
||
|
||
|
||
# ── clip_type → layer role 映射 ──────────────────────────────────────────────
|
||
|
||
|
||
def _resolve_layer_role(clip_type: str, config: dict[str, Any]) -> str:
|
||
"""根据 clip_type 和 config.role 确定图层角色（向后兼容别名）。
|
||
|
||
实际实现移至 packages.domain.render_layer_utils.resolve_layer_role。
|
||
"""
|
||
return _resolve_layer_role_pure(clip_type, config)
|
||
|
||
|
||
# ── 图层默认 z_index ─────────────────────────────────────────────────────────
|
||
|
||
_LAYER_Z_INDEX: dict[str, int] = _IMPORTED_LAYER_Z_INDEX
|
||
|
||
# 图层默认 PiP 位置（相对输出画布的偏移）
|
||
_PIP_SCALE = 0.25 # PiP 占主画面的比例
|
||
|
||
|
||
# ── 统一渲染引擎 ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
class UnifiedRenderService:
|
||
"""统一渲染引擎。
|
||
|
||
输入 EditPlan + EditPlanClips + 素材路径映射，按时间线+图层执行渲染。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
plan: Any, # EditPlan
|
||
clips: list[Any], # list[EditPlanClip]
|
||
asset_path_map: dict[str, Path], # asset_id → local_path
|
||
work_dir: Path,
|
||
*,
|
||
output_width: int = DEFAULT_OUTPUT_WIDTH,
|
||
output_height: int = DEFAULT_OUTPUT_HEIGHT,
|
||
output_fps: int = DEFAULT_FPS,
|
||
transition_duration: float = DEFAULT_TRANSITION_DURATION,
|
||
asr_service: Any = None, # ASRService 实例，用于自动生成字幕
|
||
bgm_path: str | None = None, # BGM 本地文件路径
|
||
voiceover_audio_path: str | None = None, # 配音素材库音频本地路径
|
||
):
|
||
self.plan = plan
|
||
self.clips = clips
|
||
self.asset_path_map = asset_path_map
|
||
self.work_dir = work_dir
|
||
self.output_width = output_width
|
||
self.output_height = output_height
|
||
self.output_fps = output_fps
|
||
self.transition_duration = transition_duration
|
||
self.asr_service = asr_service
|
||
self.bgm_path = bgm_path
|
||
self.voiceover_audio_path = voiceover_audio_path
|
||
self._transition_engine = TransitionEngine(default_duration=transition_duration)
|
||
self._speed_engine = SpeedEngine()
|
||
self._asr_timeline_cache: Any = None # ASR 字幕结果缓存，避免重复调用
|
||
self._asr_timeline_cached = False
|
||
|
||
def render(self) -> RenderResult:
|
||
"""执行渲染，返回 RenderResult.
|
||
|
||
优化路径：
|
||
- 单图层单 clip → 直通模式（-vf），性能最优
|
||
- 其他情况 → 完整 filter_complex 渲染
|
||
|
||
字幕渲染流程：
|
||
1. 视频主渲染（直通或完整链路）
|
||
2. 如有 title/subtitle，叠加 ASS 字幕
|
||
|
||
音频后处理：
|
||
1. 主图层音频 concat 拼接
|
||
2. 独立音频轨 amix 混入
|
||
3. 合并到输出视频
|
||
|
||
Raises:
|
||
ValueError: 没有可渲染的片段时抛出
|
||
"""
|
||
t_start = time.time()
|
||
|
||
# 1. 解析 clips → ResolvedClips（跳过无素材的 clip）
|
||
resolved = self._resolve_clips()
|
||
if not resolved:
|
||
raise ValueError("没有可渲染的片段（所有片段素材缺失或下载失败）")
|
||
|
||
# 2. 分组为 RenderLayers
|
||
layers = self._group_clips_into_layers(resolved)
|
||
|
||
# 3. 计算视频总时长（用于字幕显示时长）
|
||
video_duration = self._estimate_total_duration(layers)
|
||
|
||
# 3.5 TTS 配音生成（如果配置了）
|
||
self._maybe_add_voiceover_layer(layers, video_duration=video_duration)
|
||
|
||
# 3.6 配音素材库音频（如果传入了本地路径）
|
||
self._maybe_add_voice_library_layer(layers, video_duration=video_duration)
|
||
|
||
# 4. 生成 ASS 字幕文件（如果有 title/subtitle 配置）
|
||
ass_path = self._maybe_generate_ass(video_duration)
|
||
|
||
# 4.5 解析画中画配置
|
||
pip_config = PiPConfig.from_dict((self.plan.config or {}).get("pip_config"))
|
||
pip_sources = self._resolve_pip_sources(pip_config) if pip_config.enabled else []
|
||
has_pip = len(pip_sources) > 0
|
||
|
||
# 灰度埋点：开始渲染
|
||
layer_roles = [layer.role for layer in layers]
|
||
clip_counts = {layer.role: len(layer.clips) for layer in layers}
|
||
logger.info(
|
||
"[unified-render] start render: plan_id=%s clip_count=%d layers=%s clip_counts=%s pip_layers=%d",
|
||
self.plan.id,
|
||
len(resolved),
|
||
layer_roles,
|
||
clip_counts,
|
||
len(pip_sources),
|
||
)
|
||
|
||
# 5. 视频主渲染
|
||
t_video_start = time.time()
|
||
video_only_path = self.work_dir / f"rendered_{self.plan.id}_video.mp4"
|
||
output_path = self.work_dir / f"rendered_{self.plan.id}.mp4"
|
||
|
||
# 有画中画时不走直通（需要额外图层叠加）
|
||
is_pass_through = self._can_use_pass_through(layers) and not has_pip
|
||
pass_through_has_audio = False
|
||
used_stream_copy = False
|
||
|
||
if is_pass_through:
|
||
# 先尝试 stream copy 优化（无重编码，性能提升 10 倍+）
|
||
# 条件不满足或失败时回退到带滤镜的直通渲染
|
||
stream_copy_ok = self._try_render_stream_copy(
|
||
layers, output_path, ass_path=ass_path, video_duration=video_duration
|
||
)
|
||
if stream_copy_ok:
|
||
used_stream_copy = True
|
||
# stream copy 模式下，直接探测输出是否有音频
|
||
clip = layers[0].clips[0]
|
||
info = probe_video_info(str(clip.local_path))
|
||
pass_through_has_audio = info.get("has_audio", True)
|
||
else:
|
||
# 回退到带滤镜的直通渲染
|
||
pass_through_has_audio = self._render_pass_through(
|
||
layers,
|
||
output_path,
|
||
ass_path=ass_path,
|
||
video_duration=video_duration,
|
||
)
|
||
else:
|
||
filter_complex, input_args = self._build_filter_complex(layers, ass_path=ass_path)
|
||
|
||
# 追加画中画滤镜
|
||
if has_pip:
|
||
filter_complex, input_args = self._append_pip_filters(filter_complex, input_args, pip_sources)
|
||
|
||
self._execute_ffmpeg(filter_complex, input_args, video_only_path)
|
||
|
||
t_video_end = time.time()
|
||
video_render_ms = int((t_video_end - t_video_start) * 1000)
|
||
logger.info(
|
||
"[unified-render] video render done: plan_id=%s duration_ms=%d pass_through=%s stream_copy=%s",
|
||
self.plan.id,
|
||
video_render_ms,
|
||
is_pass_through,
|
||
used_stream_copy,
|
||
)
|
||
|
||
# 6. 音频后处理混音（直通场景已合并处理，跳过）
|
||
t_audio_start = time.time()
|
||
audio_mix_ms = 0
|
||
has_audio = False
|
||
|
||
if is_pass_through:
|
||
# 直通场景已在一次调用中完成视频+音频
|
||
has_audio = pass_through_has_audio
|
||
# 直通模式下也支持 BGM 混音：提取音频 → 混 BGM → 合并回视频
|
||
if self.bgm_path and pass_through_has_audio:
|
||
config = self.plan.config or {}
|
||
bgm_config = config.get("bgm", {}) or {}
|
||
if bgm_config.get("enabled", False):
|
||
ctx = RenderContext(work_dir=self.work_dir, plan_id=self.plan.id)
|
||
from video_processing.bgm_mixer import BGMConfig, mix_bgm_with_main
|
||
|
||
bgm_cfg = BGMConfig.from_config_dict(self.bgm_path, bgm_config)
|
||
# 从直通输出中提取音频
|
||
main_audio_path = self.work_dir / f"pass_through_audio_{self.plan.id}.aac"
|
||
extract_cmd = [
|
||
FFMPEG_BIN,
|
||
"-y",
|
||
"-i",
|
||
str(output_path),
|
||
"-vn",
|
||
"-acodec",
|
||
"aac",
|
||
"-b:a",
|
||
"128k",
|
||
str(main_audio_path),
|
||
]
|
||
try:
|
||
from video_processing.ffmpeg_utils import run_ffmpeg
|
||
|
||
run_ffmpeg(extract_cmd)
|
||
final_audio = mix_bgm_with_main(ctx, main_audio_path, bgm_cfg, video_duration)
|
||
# 合并回视频
|
||
|
||
bgm_output = self.work_dir / f"rendered_{self.plan.id}_bgm.mp4"
|
||
merge_audio_video(ctx, output_path, final_audio, bgm_output)
|
||
output_path = bgm_output
|
||
logger.info("[unified-render] pass-through BGM mix done: plan_id=%s", self.plan.id)
|
||
except Exception:
|
||
logger.exception(
|
||
"[unified-render] pass-through BGM mix failed, skipping: plan_id=%s", self.plan.id
|
||
)
|
||
else:
|
||
config = self.plan.config or {}
|
||
bgm_config = config.get("bgm", {}) or {}
|
||
audio_tracks_config = config.get("audio_tracks") or {}
|
||
noise_reduction_config = config.get("audio_noise_reduction")
|
||
ctx = RenderContext(
|
||
work_dir=self.work_dir,
|
||
plan_id=self.plan.id,
|
||
noise_reduction_config=noise_reduction_config,
|
||
)
|
||
audio_path = mix_audio(
|
||
ctx,
|
||
layers,
|
||
video_duration,
|
||
bgm_path=self.bgm_path,
|
||
bgm_config=bgm_config,
|
||
audio_tracks_config=audio_tracks_config,
|
||
)
|
||
t_audio_end = time.time()
|
||
audio_mix_ms = int((t_audio_end - t_audio_start) * 1000)
|
||
has_audio = audio_path is not None
|
||
if has_audio:
|
||
logger.info(
|
||
"[unified-render] audio mix done: plan_id=%s duration_ms=%d",
|
||
self.plan.id,
|
||
audio_mix_ms,
|
||
)
|
||
# 7. 合并音视频
|
||
merge_audio_video(ctx, video_only_path, audio_path, output_path)
|
||
else:
|
||
# 无音频，直接用无声视频
|
||
import shutil
|
||
|
||
shutil.copy2(video_only_path, output_path)
|
||
|
||
# 8. 探测输出
|
||
duration, file_size, width, height = self._probe_output(output_path)
|
||
|
||
# 9. 片头片尾拼接（后处理）
|
||
intro_outro_config = IntroOutroConfig.from_dict((self.plan.config or {}).get("intro_outro"))
|
||
if intro_outro_config.has_intro or intro_outro_config.has_outro:
|
||
io_valid, io_err = intro_outro_config.validate()
|
||
if io_valid:
|
||
final_with_io = self.work_dir / f"rendered_{self.plan.id}_with_io.mp4"
|
||
intro_path = None
|
||
outro_path = None
|
||
|
||
# 生成片头
|
||
if intro_outro_config.has_intro:
|
||
intro_path = self.work_dir / f"intro_{self.plan.id}.mp4"
|
||
intro_ok = False
|
||
if intro_outro_config.intro_type == "video":
|
||
import shutil
|
||
|
||
src = Path(intro_outro_config.intro_video_path)
|
||
if src.exists():
|
||
shutil.copy2(src, intro_path)
|
||
intro_ok = True
|
||
else:
|
||
logger.warning("片头视频不存在，跳过片头: %s", src)
|
||
elif intro_outro_config.intro_type == "text":
|
||
intro_ok = IntroOutroEngine.generate_text_intro(
|
||
intro_path,
|
||
intro_outro_config,
|
||
self.output_width,
|
||
self.output_height,
|
||
self.output_fps,
|
||
)
|
||
|
||
if not intro_ok:
|
||
intro_path = None
|
||
|
||
# 生成片尾
|
||
if intro_outro_config.has_outro:
|
||
outro_path = self.work_dir / f"outro_{self.plan.id}.mp4"
|
||
outro_ok = False
|
||
if intro_outro_config.outro_type == "video":
|
||
import shutil
|
||
|
||
src = Path(intro_outro_config.outro_video_path)
|
||
if src.exists():
|
||
shutil.copy2(src, outro_path)
|
||
outro_ok = True
|
||
else:
|
||
logger.warning("片尾视频不存在，跳过片尾: %s", src)
|
||
elif intro_outro_config.outro_type in ("text", "follow"):
|
||
outro_ok = IntroOutroEngine.generate_text_outro(
|
||
outro_path,
|
||
intro_outro_config,
|
||
self.output_width,
|
||
self.output_height,
|
||
self.output_fps,
|
||
)
|
||
|
||
if not outro_ok:
|
||
outro_path = None
|
||
|
||
# 拼接
|
||
if intro_path or outro_path:
|
||
concat_ok = IntroOutroEngine.concat_with_intro_outro(
|
||
output_path,
|
||
intro_path,
|
||
outro_path,
|
||
final_with_io,
|
||
transition_duration=intro_outro_config.transition_duration,
|
||
transition_effect=intro_outro_config.transition_effect,
|
||
)
|
||
if concat_ok and final_with_io.exists():
|
||
output_path = final_with_io
|
||
# 重新探测
|
||
duration, file_size, width, height = self._probe_output(output_path)
|
||
logger.info("[unified-render] 片头片尾拼接完成: plan_id=%s", self.plan.id)
|
||
else:
|
||
logger.warning("[unified-render] 片头片尾拼接失败，使用原视频: plan_id=%s", self.plan.id)
|
||
else:
|
||
logger.warning("[unified-render] 片头片尾配置无效，跳过: %s", io_err)
|
||
|
||
t_total = int((time.time() - t_start) * 1000)
|
||
logger.info(
|
||
"[unified-render] render done: plan_id=%s total_ms=%d video_ms=%d audio_ms=%d "
|
||
"output_duration=%.2fs output_size=%d resolution=%dx%d has_audio=%s",
|
||
self.plan.id,
|
||
t_total,
|
||
video_render_ms,
|
||
audio_mix_ms if has_audio else 0,
|
||
duration,
|
||
file_size,
|
||
width,
|
||
height,
|
||
has_audio,
|
||
)
|
||
|
||
return RenderResult(
|
||
output_path=output_path,
|
||
duration=duration,
|
||
file_size=file_size,
|
||
width=width,
|
||
height=height,
|
||
)
|
||
|
||
def _estimate_total_duration(self, layers: list[RenderLayer]) -> float:
|
||
"""估算视频总时长（用于字幕等需要）。
|
||
|
||
实际实现移至 packages.domain.render_layer_utils.estimate_total_duration。
|
||
"""
|
||
return _estimate_total_duration_pure(layers, self.transition_duration)
|
||
|
||
def _maybe_generate_ass(self, video_duration: float) -> Path | None:
|
||
"""根据 plan.config 生成 ASS 字幕文件。
|
||
|
||
支持两种字幕模式：
|
||
1. 静态字幕 — title/subtitle 配置了 text 时，生成整段静态字幕
|
||
2. ASR 自动字幕 — subtitle.auto_generated=true 时，从音频自动识别生成时间轴字幕
|
||
|
||
Returns:
|
||
ASS 文件路径，没有字幕时返回 None
|
||
"""
|
||
config = self.plan.config or {}
|
||
title_cfg = config.get("title", {}) or {}
|
||
subtitle_cfg = config.get("subtitle", {}) or {}
|
||
|
||
title_enabled = title_cfg.get("enabled", True)
|
||
subtitle_enabled = subtitle_cfg.get("enabled", True)
|
||
title_text = title_cfg.get("text", "") or ""
|
||
subtitle_text = subtitle_cfg.get("text", "") or ""
|
||
auto_generated = subtitle_cfg.get("auto_generated", False)
|
||
|
||
has_title = title_enabled and bool(title_text.strip())
|
||
has_static_subtitle = subtitle_enabled and bool(subtitle_text.strip())
|
||
has_auto_subtitle = subtitle_enabled and auto_generated and self.asr_service is not None
|
||
|
||
if not has_title and not has_static_subtitle and not has_auto_subtitle:
|
||
return None
|
||
|
||
ass_path = self.work_dir / f"subtitles_{self.plan.id}.ass"
|
||
|
||
# ASR 自动字幕模式
|
||
if has_auto_subtitle:
|
||
try:
|
||
timeline = self._generate_asr_subtitles(video_duration, subtitle_cfg)
|
||
if timeline and timeline.segments:
|
||
generate_ass_from_timeline(
|
||
ass_path,
|
||
timeline,
|
||
video_width=self.output_width,
|
||
video_height=self.output_height,
|
||
subtitle_config=subtitle_cfg,
|
||
)
|
||
logger.info(
|
||
"ASR自动字幕生成完成: plan_id=%s segments=%d duration=%.1fs",
|
||
self.plan.id,
|
||
timeline.segment_count,
|
||
video_duration,
|
||
)
|
||
return ass_path
|
||
else:
|
||
# ASR 无结果，不生成字幕
|
||
logger.info("ASR自动字幕无识别结果，跳过字幕: plan_id=%s", self.plan.id)
|
||
return None
|
||
except Exception:
|
||
# ASR 失败降级：不生成字幕，不阻断主流程
|
||
logger.warning("ASR自动字幕生成失败，跳过字幕", exc_info=True)
|
||
return None
|
||
|
||
# 静态字幕模式（原有逻辑）
|
||
generate_ass_subtitles(
|
||
ass_path,
|
||
video_width=self.output_width,
|
||
video_height=self.output_height,
|
||
video_duration=video_duration,
|
||
title_text=title_text,
|
||
title_config=title_cfg,
|
||
subtitle_text=subtitle_text,
|
||
subtitle_config=subtitle_cfg,
|
||
)
|
||
|
||
logger.info(
|
||
"生成字幕: plan_id=%s title=%s subtitle=%s ass=%s",
|
||
self.plan.id,
|
||
has_title,
|
||
has_static_subtitle,
|
||
ass_path,
|
||
)
|
||
return ass_path
|
||
|
||
def _generate_asr_subtitles(self, video_duration: float, subtitle_cfg: dict) -> Any: # SubtitleTimeline
|
||
"""从视频素材音频中自动识别生成字幕时间轴。
|
||
|
||
MVP 版本：使用第一个有音频的素材做ASR，然后按比例映射到整个视频时长。
|
||
后续优化：支持多片段拼接后的完整音频ASR。
|
||
|
||
带缓存：同一 plan 只做一次 ASR，TTS 配音和字幕共用结果。
|
||
"""
|
||
# 检查缓存
|
||
if self._asr_timeline_cached:
|
||
return self._asr_timeline_cache
|
||
|
||
from packages.domain.subtitle import SubtitleTimeline
|
||
|
||
# 找第一个有本地路径的素材
|
||
first_asset_path = None
|
||
for clip in self.clips:
|
||
asset_id = getattr(clip, "asset_id", None)
|
||
if asset_id and asset_id in self.asset_path_map:
|
||
first_asset_path = self.asset_path_map[asset_id]
|
||
break
|
||
|
||
if first_asset_path is None:
|
||
logger.warning("ASR字幕生成失败：找不到可用素材音频")
|
||
result = SubtitleTimeline(segments=[], total_duration=video_duration)
|
||
self._asr_timeline_cache = result
|
||
self._asr_timeline_cached = True
|
||
return result
|
||
|
||
# 提取素材音频为 wav（16kHz单声道，ASR友好格式）
|
||
audio_path = self.work_dir / f"asr_audio_{self.plan.id}.wav"
|
||
try:
|
||
self._extract_audio(first_asset_path, audio_path)
|
||
except Exception:
|
||
logger.warning("ASR音频提取失败", exc_info=True)
|
||
return SubtitleTimeline(segments=[], total_duration=video_duration)
|
||
|
||
if not audio_path.exists():
|
||
return SubtitleTimeline(segments=[], total_duration=video_duration)
|
||
|
||
# 调用 ASR 服务
|
||
language = subtitle_cfg.get("language", "") or None
|
||
timeline = self.asr_service.transcribe(
|
||
audio_path,
|
||
language=language,
|
||
with_word_timestamps=True,
|
||
)
|
||
|
||
# 字幕后处理：合并短片段 + 拆分长片段
|
||
min_chars = int(subtitle_cfg.get("min_chars_per_segment", 8))
|
||
max_chars = int(subtitle_cfg.get("max_chars_per_line", 20))
|
||
|
||
if timeline.segments:
|
||
timeline = timeline.merge_short_segments(min_chars=min_chars)
|
||
timeline = timeline.split_long_segments(max_chars=max_chars)
|
||
|
||
# 清理临时音频文件
|
||
try:
|
||
audio_path.unlink(missing_ok=True)
|
||
except Exception:
|
||
pass
|
||
|
||
# 存入缓存
|
||
self._asr_timeline_cache = timeline
|
||
self._asr_timeline_cached = True
|
||
return timeline
|
||
|
||
def _extract_audio(self, video_path: Path, output_path: Path) -> None:
|
||
"""从视频中提取音频为16kHz单声道wav（ASR友好格式）。"""
|
||
cmd = [
|
||
FFMPEG_BIN,
|
||
"-y",
|
||
"-i",
|
||
str(video_path),
|
||
"-vn",
|
||
"-acodec",
|
||
"pcm_s16le",
|
||
"-ar",
|
||
"16000",
|
||
"-ac",
|
||
"1",
|
||
str(output_path),
|
||
]
|
||
|
||
try:
|
||
run_ffmpeg(cmd, timeout=120)
|
||
except Exception as e:
|
||
raise RuntimeError(f"音频提取失败: {str(e)[:200]}") from e
|
||
|
||
def _maybe_add_voiceover_layer(
|
||
self,
|
||
layers: list[RenderLayer],
|
||
*,
|
||
video_duration: float,
|
||
) -> bool:
|
||
"""根据 plan.config 生成 TTS 配音，加到 audio 图层.
|
||
|
||
支持三种触发方式：
|
||
1. config.tts.enabled = true → 标准 TTS 配置
|
||
2. 顶层 voice_id + custom_text → 桥接模式（自定义文案配音）
|
||
3. 顶层 voice_id + subtitle.auto_generated=true → ASR 字幕对齐配音（预设配音）
|
||
|
||
Returns:
|
||
是否成功添加了配音音轨
|
||
"""
|
||
config = self.plan.config or {}
|
||
tts_cfg = config.get("tts", {}) or {}
|
||
subtitle_cfg = config.get("subtitle", {}) or {}
|
||
use_subtitle_align = False # 是否使用字幕对齐模式
|
||
|
||
# 兼容前端顶层字段：voice_id / custom_text / voice_clone_profile_id
|
||
if not tts_cfg.get("enabled"):
|
||
top_voice_id = config.get("voice_id", "") or ""
|
||
top_text = config.get("custom_text", "") or ""
|
||
|
||
# 方式A：voice_id + custom_text → 整段配音
|
||
if top_voice_id and top_text:
|
||
tts_cfg = {
|
||
"enabled": True,
|
||
"voice_id": top_voice_id,
|
||
"text": top_text,
|
||
"align_mode": "full",
|
||
"overlap_mode": "replace",
|
||
}
|
||
logger.info(
|
||
"[unified-render] 检测到顶层 voice_id+custom_text，桥接到 tts 配置（整段）: plan_id=%s voice_id=%s text_len=%d",
|
||
self.plan.id,
|
||
top_voice_id,
|
||
len(top_text),
|
||
)
|
||
# 方式B：voice_id + 自动字幕 → 字幕对齐配音（预设配音模式）
|
||
elif top_voice_id and subtitle_cfg.get("auto_generated", False) and self.asr_service is not None:
|
||
tts_cfg = {
|
||
"enabled": True,
|
||
"voice_id": top_voice_id,
|
||
"text": "",
|
||
"align_mode": "subtitle",
|
||
"overlap_mode": "replace",
|
||
}
|
||
use_subtitle_align = True
|
||
logger.info(
|
||
"[unified-render] 检测到预设配音+自动字幕，使用字幕对齐模式: plan_id=%s voice_id=%s",
|
||
self.plan.id,
|
||
top_voice_id,
|
||
)
|
||
|
||
# 兼容前端顶层字段：voice_id / custom_text / voice_clone_profile_id
|
||
# 前端一键生成页面传 config.voice_id + config.custom_text，
|
||
# 统一渲染引擎从 config.tts 读，这里做桥接映射。
|
||
if not tts_cfg.get("enabled"):
|
||
top_voice_id = config.get("voice_id", "") or ""
|
||
top_text = config.get("custom_text", "") or ""
|
||
if top_voice_id and top_text:
|
||
tts_cfg = {
|
||
"enabled": True,
|
||
"voice_id": top_voice_id,
|
||
"text": top_text,
|
||
"align_mode": "full",
|
||
"overlap_mode": "replace",
|
||
}
|
||
logger.info(
|
||
"[unified-render] 检测到顶层 voice_id+custom_text，桥接到 tts 配置: plan_id=%s voice_id=%s text_len=%d",
|
||
self.plan.id,
|
||
top_voice_id,
|
||
len(top_text),
|
||
)
|
||
|
||
tts_config = TtsConfig.parse(tts_cfg)
|
||
if not tts_config.enabled:
|
||
return False
|
||
|
||
try:
|
||
from apps.worker.services.tts_service_factory import get_tts_service
|
||
|
||
tts_service = get_tts_service()
|
||
tts_engine = TtsEngine(tts_service, self.work_dir / "tts")
|
||
|
||
# 根据对齐模式选择生成方式
|
||
if use_subtitle_align or tts_config.align_mode == "subtitle":
|
||
# 字幕对齐模式：先做 ASR，再按字幕生成配音
|
||
if not self._asr_timeline_cached:
|
||
self._generate_asr_subtitles(video_duration, subtitle_cfg)
|
||
timeline = self._asr_timeline_cache
|
||
if timeline is None or not timeline.segments:
|
||
logger.warning("TTS 字幕对齐配音：ASR 无识别结果，跳过配音")
|
||
return False
|
||
|
||
# 转换为 TtsEngine 需要的字幕格式
|
||
subtitles = [
|
||
{
|
||
"text": seg.text,
|
||
"start_time": seg.start,
|
||
"end_time": seg.end,
|
||
}
|
||
for seg in timeline.segments
|
||
if getattr(seg, "text", "").strip()
|
||
]
|
||
if not subtitles:
|
||
logger.warning("TTS 字幕对齐配音：字幕文本为空，跳过配音")
|
||
return False
|
||
|
||
result = tts_engine.generate_subtitle_voiceover(tts_config, subtitles)
|
||
else:
|
||
# 整段配音模式
|
||
result = tts_engine.generate_full_voiceover(tts_config, total_duration=video_duration)
|
||
|
||
if not result.success or not result.segments:
|
||
logger.warning("TTS 配音生成失败，跳过: %s", result.error_message)
|
||
return False
|
||
|
||
# 获取主音轨图层（用于判断 replace 模式下是否静音原音）
|
||
# 这里只处理混音添加，replace 模式在外部处理
|
||
|
||
# 找到或创建 audio 图层
|
||
audio_layer = None
|
||
for layer in layers:
|
||
if layer.role == "audio":
|
||
audio_layer = layer
|
||
break
|
||
|
||
if audio_layer is None:
|
||
from video_processing.unified_render_service import _LAYER_Z_INDEX # type: ignore
|
||
|
||
z_index = _LAYER_Z_INDEX.get("audio", 2)
|
||
audio_layer = RenderLayer(role="audio", z_index=z_index)
|
||
layers.append(audio_layer)
|
||
|
||
# 把配音片段作为 audio clip 加入
|
||
for seg in result.segments:
|
||
if seg.audio_path is None:
|
||
continue
|
||
vo_clip = ResolvedClip(
|
||
clip_id=f"tts_{seg.start_time:.3f}",
|
||
asset_id="tts_voiceover",
|
||
local_path=seg.audio_path,
|
||
clip_type="audio",
|
||
order=len(audio_layer.clips),
|
||
start_time=seg.start_time,
|
||
duration=seg.duration,
|
||
config={"volume": tts_config.volume, "tts": True},
|
||
actual_duration=seg.duration,
|
||
)
|
||
audio_layer.clips.append(vo_clip)
|
||
|
||
logger.info(
|
||
"TTS 配音已添加: plan_id=%s voice_id=%s segments=%d total_%.2fs",
|
||
self.plan.id,
|
||
tts_config.voice_id,
|
||
len(result.segments),
|
||
result.total_duration,
|
||
)
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.warning("TTS 配音异常，跳过: %s", e)
|
||
return False
|
||
|
||
def _maybe_add_voice_library_layer(
|
||
self,
|
||
layers: list[RenderLayer],
|
||
*,
|
||
video_duration: float,
|
||
) -> bool:
|
||
"""将配音素材库音频作为整段配音加到 audio 图层.
|
||
|
||
与 TTS 配音共享同一套 audio 图层混音架构，
|
||
支持与 BGM、TTS 的音量平衡，不再走独立的后处理 mux 链路。
|
||
|
||
Returns:
|
||
是否成功添加了配音音轨
|
||
"""
|
||
if not self.voiceover_audio_path:
|
||
return False
|
||
|
||
audio_path = Path(self.voiceover_audio_path)
|
||
if not audio_path.exists() or audio_path.stat().st_size == 0:
|
||
logger.warning("配音素材库音频文件不存在或为空，跳过: %s", self.voiceover_audio_path)
|
||
return False
|
||
|
||
try:
|
||
# 找到或创建 audio 图层
|
||
audio_layer = None
|
||
for layer in layers:
|
||
if layer.role == "audio":
|
||
audio_layer = layer
|
||
break
|
||
|
||
if audio_layer is None:
|
||
from video_processing.unified_render_service import _LAYER_Z_INDEX # type: ignore
|
||
|
||
z_index = _LAYER_Z_INDEX.get("audio", 2)
|
||
audio_layer = RenderLayer(role="audio", z_index=z_index)
|
||
layers.append(audio_layer)
|
||
|
||
# 配音素材作为整段配音：从 0 开始，覆盖整个视频时长
|
||
# 音频不足视频时长时，混音层会按实际长度处理（amix 不自动循环）
|
||
vo_clip = ResolvedClip(
|
||
clip_id="voice_library_main",
|
||
asset_id="voice_library",
|
||
local_path=audio_path,
|
||
clip_type="audio",
|
||
order=len(audio_layer.clips),
|
||
start_time=0.0,
|
||
duration=video_duration,
|
||
config={"volume": 1.0, "voice_library": True},
|
||
actual_duration=video_duration,
|
||
)
|
||
audio_layer.clips.append(vo_clip)
|
||
|
||
logger.info(
|
||
"配音素材库音频已添加到 audio 图层: plan_id=%s duration=%.2fs",
|
||
self.plan.id,
|
||
video_duration,
|
||
)
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.warning("配音素材库音频添加失败，跳过: %s", e)
|
||
return False
|
||
|
||
@staticmethod
|
||
def _resolve_watermark_config(plan_config: dict[str, Any] | None) -> WatermarkConfig | None:
|
||
"""从 plan config 中解析水印配置，兼容两种存储格式.
|
||
|
||
支持格式：
|
||
1. 嵌套格式：config.watermark = {enabled, mode, text, image_path, ...}
|
||
2. 扁平格式（导出配置）：config.export.watermark_enabled + config.export.watermark_text
|
||
|
||
Returns:
|
||
WatermarkConfig 或 None（未启用水印时）
|
||
"""
|
||
if not plan_config or not isinstance(plan_config, dict):
|
||
return None
|
||
|
||
# 格式1: 嵌套 watermark 对象（优先）
|
||
wm_data = plan_config.get("watermark")
|
||
if isinstance(wm_data, dict) and wm_data:
|
||
config = WatermarkConfig.from_dict(wm_data)
|
||
if config is not None:
|
||
return config
|
||
|
||
# 格式2: 扁平 export.watermark_enabled + export.watermark_text
|
||
export_cfg = plan_config.get("export")
|
||
if isinstance(export_cfg, dict) and export_cfg:
|
||
enabled = export_cfg.get("watermark_enabled", False)
|
||
text = export_cfg.get("watermark_text", "") or ""
|
||
if enabled and text:
|
||
return WatermarkConfig(
|
||
mode="text",
|
||
text=str(text),
|
||
position=export_cfg.get("watermark_position", "bottom_right"),
|
||
opacity=float(export_cfg.get("watermark_opacity", 0.6)),
|
||
font_size=int(export_cfg.get("watermark_font_size", 24)),
|
||
font_color=str(export_cfg.get("watermark_font_color", "white")),
|
||
)
|
||
|
||
return None
|
||
|
||
def _can_use_pass_through(self, layers: list[RenderLayer]) -> bool:
|
||
"""判断是否可以走直通优化路径。
|
||
|
||
条件：
|
||
1. 只有 1 个图层
|
||
2. 该图层是视频图层（main/broll/background），不是 overlay/corner_voice/audio
|
||
3. 该图层只有 1 个 clip（无转场需求）
|
||
4. 没有贴纸（贴纸需要 filter_complex 或额外输入）
|
||
"""
|
||
if len(layers) != 1:
|
||
return False
|
||
layer = layers[0]
|
||
if layer.role not in ("main", "broll", "background"):
|
||
return False
|
||
if len(layer.clips) != 1:
|
||
return False
|
||
# 有贴纸时禁用直通（图片贴纸需要额外输入，统一走 filter_complex）
|
||
plan_config = getattr(self.plan, "config", None) or {}
|
||
if isinstance(plan_config, dict) and plan_config.get("stickers"):
|
||
return False
|
||
|
||
# 有水印时禁用直通（图片水印需要额外输入，统一走 filter_complex；
|
||
# 文字水印虽然可以 -vf 叠加，但为了保持路径统一也走 filter_complex）
|
||
wm_config = UnifiedRenderService._resolve_watermark_config(plan_config)
|
||
if wm_config is not None and wm_config.validate()[0]:
|
||
return False
|
||
|
||
# 有调速时仍然可以走直通（视频调速通过 setpts 实现，单输入即可）
|
||
|
||
return True
|
||
|
||
def _can_use_stream_copy(
|
||
self,
|
||
clip: ResolvedClip,
|
||
*,
|
||
ass_path: Path | None = None,
|
||
video_duration: float = 0.0,
|
||
) -> tuple[bool, str]:
|
||
"""判断是否可以走 stream copy（流拷贝，不重编码）。
|
||
|
||
性能提升：10 倍以上（典型场景从 20s → 1-2s）。
|
||
|
||
条件：
|
||
1. 视频编码为 h264（输出目标也是 h264）
|
||
2. 像素格式为 yuv420p
|
||
3. 分辨率与输出一致（不需要 scale/crop）
|
||
4. 帧率与输出一致（误差 < 0.1fps）
|
||
5. 无字幕叠加（字幕需要滤镜）
|
||
6. 无 trim 需求（或 trim 后恰好等于原时长）
|
||
7. 无转场、无特效（单 clip 直通已保证）
|
||
|
||
Returns:
|
||
(是否可以 copy, 原因说明)
|
||
"""
|
||
# 有字幕 → 需要滤镜 → 不能 copy
|
||
if ass_path is not None:
|
||
return False, "有字幕叠加"
|
||
|
||
# 有调速 → 需要重编码 → 不能 copy
|
||
speed = UnifiedRenderService._clip_speed(clip)
|
||
if abs(speed - 1.0) >= 1e-6:
|
||
return False, f"有调速: speed={speed:.2f}x"
|
||
|
||
# 有倒放 → 需要重编码 → 不能 copy
|
||
reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
|
||
if reverse_config.enabled and (reverse_config.reverse_video or reverse_config.reverse_audio):
|
||
return False, "有倒放效果"
|
||
|
||
# 探测输入视频参数
|
||
info = probe_video_info(str(clip.local_path))
|
||
|
||
# 编码必须是 h264
|
||
if info.get("video_codec", "") != "h264":
|
||
return False, f"视频编码不是h264: {info.get('video_codec', 'unknown')}"
|
||
|
||
# 像素格式必须是 yuv420p
|
||
if info.get("pix_fmt", "") != "yuv420p":
|
||
return False, f"像素格式不是yuv420p: {info.get('pix_fmt', 'unknown')}"
|
||
|
||
# 分辨率必须一致
|
||
if info.get("width", 0) != self.output_width or info.get("height", 0) != self.output_height:
|
||
return False, (
|
||
f"分辨率不匹配: "
|
||
f"{info.get('width', 0)}x{info.get('height', 0)} "
|
||
f"vs {self.output_width}x{self.output_height}"
|
||
)
|
||
|
||
# 帧率必须一致（误差 < 0.1fps）
|
||
fps_diff = abs(info.get("fps", 0) - self.output_fps)
|
||
if fps_diff > 0.1:
|
||
return False, f"帧率不匹配: {info.get('fps', 0)} vs {self.output_fps}"
|
||
|
||
# 检查是否需要 trim
|
||
effective_duration = UnifiedRenderService._clip_effective_duration(clip)
|
||
if effective_duration > 0:
|
||
# 有 trim 需求但视频时长足够，可用 -ss/-t 实现 copy trim
|
||
input_duration = info.get("duration", 0)
|
||
if input_duration <= 0:
|
||
return False, "无法探测输入时长"
|
||
# trim 起始点 + 目标时长 <= 输入时长
|
||
start_time = getattr(clip, "start_time", 0) or 0
|
||
if start_time + effective_duration > input_duration + 0.1:
|
||
return False, "trim 超出输入时长"
|
||
|
||
# video_duration 截断
|
||
if video_duration > 0 and effective_duration > 0:
|
||
final_duration = min(effective_duration, video_duration)
|
||
if final_duration != effective_duration:
|
||
# 也需要截断，但 -t 可以 copy 模式下用
|
||
pass
|
||
|
||
return True, "所有条件满足"
|
||
|
||
def _try_render_stream_copy(
|
||
self,
|
||
layers: list[RenderLayer],
|
||
output_path: Path,
|
||
*,
|
||
ass_path: Path | None = None,
|
||
video_duration: float = 0.0,
|
||
) -> bool:
|
||
"""尝试 stream copy 渲染，成功返回 True，失败返回 False（调用方回退到重编码）。
|
||
|
||
stream copy 模式：不重编码，直接拷贝视频/音频流，性能提升 10 倍+。
|
||
仅用于单 clip 直通场景且满足 copy 条件。
|
||
"""
|
||
clip = layers[0].clips[0]
|
||
role = layers[0].role
|
||
|
||
# 判断是否满足 copy 条件
|
||
can_copy, reason = self._can_use_stream_copy(clip, ass_path=ass_path, video_duration=video_duration)
|
||
if not can_copy:
|
||
logger.info(
|
||
"[unified-render] stream_copy 跳过: plan_id=%s reason=%s",
|
||
self.plan.id,
|
||
reason,
|
||
)
|
||
return False
|
||
|
||
# 构建 copy 命令
|
||
command = [
|
||
FFMPEG_BIN,
|
||
"-y",
|
||
]
|
||
|
||
# trim 支持（-ss 放在 -i 前 = input seeking，速度更快但精度稍差；
|
||
# 放在 -i 后 = output seeking，精度高但慢）
|
||
# 这里用 output seeking 保证精度，反正 copy 模式已经很快了
|
||
start_time = getattr(clip, "start_time", 0) or 0
|
||
effective_duration = UnifiedRenderService._clip_effective_duration(clip)
|
||
|
||
command.extend(["-i", str(clip.local_path)])
|
||
|
||
if start_time > 0:
|
||
command.extend(["-ss", f"{start_time:.3f}"])
|
||
|
||
# 计算最终时长
|
||
final_duration = effective_duration
|
||
if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration):
|
||
final_duration = video_duration
|
||
if final_duration > 0:
|
||
command.extend(["-t", f"{final_duration:.3f}"])
|
||
|
||
# 流拷贝
|
||
command.extend(
|
||
[
|
||
"-c:v",
|
||
"copy",
|
||
"-c:a",
|
||
"copy",
|
||
"-movflags",
|
||
"+faststart",
|
||
str(output_path),
|
||
]
|
||
)
|
||
|
||
logger.info(
|
||
"[unified-render] stream_copy 渲染: plan_id=%s clip=%s role=%s duration=%.2fs",
|
||
self.plan.id,
|
||
clip.clip_id,
|
||
role,
|
||
final_duration,
|
||
)
|
||
|
||
try:
|
||
run_ffmpeg(command)
|
||
# 验证输出文件存在且有大小
|
||
if output_path.exists() and output_path.stat().st_size > 0:
|
||
logger.info(
|
||
"[unified-render] stream_copy 成功: plan_id=%s size=%d",
|
||
self.plan.id,
|
||
output_path.stat().st_size,
|
||
)
|
||
return True
|
||
else:
|
||
logger.warning("[unified-render] stream_copy 输出为空: plan_id=%s", self.plan.id)
|
||
return False
|
||
except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
|
||
logger.warning(
|
||
"[unified-render] stream_copy 失败，回退到重编码: plan_id=%s error=%s",
|
||
self.plan.id,
|
||
str(e)[:200],
|
||
)
|
||
# 清理可能的损坏输出文件
|
||
if output_path.exists():
|
||
try:
|
||
output_path.unlink()
|
||
except OSError as unlink_err:
|
||
logger.warning(
|
||
"[unified-render] 损坏输出文件清理失败: path=%s error=%s",
|
||
output_path,
|
||
unlink_err,
|
||
)
|
||
return False
|
||
|
||
def _render_pass_through(
|
||
self,
|
||
layers: list[RenderLayer],
|
||
output_path: Path,
|
||
*,
|
||
ass_path: Path | None = None,
|
||
video_duration: float = 0.0,
|
||
) -> bool:
|
||
"""单图层单 clip 直通渲染（使用 -vf 而非 -filter_complex），一次性输出带音频的最终视频。
|
||
|
||
性能优化：
|
||
- 避免 filter_complex 的解析和调度开销，单clip场景性能提升 ~30%
|
||
- 视频+音频一次FFmpeg调用完成，省去后续音频提取+音视频合并两次调用
|
||
|
||
Args:
|
||
layers: 图层列表（只有1个图层1个clip）
|
||
output_path: 输出文件路径
|
||
ass_path: ASS 字幕文件路径，有则叠加字幕
|
||
video_duration: 视频总时长（用于截断音频，0表示不额外截断）
|
||
|
||
Returns:
|
||
True 表示输出包含音频（近似判断，实际以输出文件为准）
|
||
"""
|
||
clip = layers[0].clips[0]
|
||
role = layers[0].role
|
||
|
||
# 构建视频滤镜链（与 _build_filter_complex 中预处理逻辑一致）
|
||
filters: list[str] = []
|
||
|
||
# trim
|
||
effective_duration = UnifiedRenderService._clip_effective_duration(clip)
|
||
|
||
if effective_duration > 0:
|
||
filters.append(f"trim=duration={effective_duration}")
|
||
filters.append("setpts=PTS-STARTPTS")
|
||
|
||
# 调速 — 与 filter_complex 路径一致
|
||
speed = UnifiedRenderService._clip_speed(clip)
|
||
if abs(speed - 1.0) >= 1e-6:
|
||
filters.append(f"setpts=PTS/{speed:.4f}")
|
||
|
||
# 倒放滤镜
|
||
reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
|
||
if reverse_config.enabled and reverse_config.reverse_video:
|
||
reverse_filter = ReverseEngine.build_video_filter(reverse_config, duration=effective_duration)
|
||
if reverse_filter:
|
||
filters.append(reverse_filter)
|
||
|
||
# scale + pad（等比缩放+留黑边）
|
||
if role in ("overlay", "corner_voice"):
|
||
pip_w = int(self.output_width * _PIP_SCALE)
|
||
pip_h = int(self.output_height * _PIP_SCALE)
|
||
filters.append(f"scale={pip_w}:{pip_h}")
|
||
elif role == "background":
|
||
# background: 铺满裁剪（作为底图，覆盖全屏）
|
||
filters.append(f"scale={self.output_width}:{self.output_height}:force_original_aspect_ratio=increase")
|
||
filters.append(f"crop={self.output_width}:{self.output_height}")
|
||
else:
|
||
# main / broll: 等比缩放 + 居中留黑边（保持原始比例，不裁剪内容）
|
||
filters.append(f"scale={self.output_width}:{self.output_height}:force_original_aspect_ratio=decrease")
|
||
filters.append(f"pad={self.output_width}:{self.output_height}:trunc((ow-iw)/2):trunc((oh-ih)/2):black")
|
||
|
||
# 调色滤镜
|
||
color_grade = ColorGradeConfig.from_dict(clip.config.get("color_grade"))
|
||
if color_grade.enabled and color_grade.has_effect():
|
||
grade_filter = ColorGradeEngine.build_filter(color_grade)
|
||
if grade_filter:
|
||
filters.append(grade_filter)
|
||
# chroma key 绿幕抠像
|
||
try:
|
||
from video_processing.chroma_key_engine import ChromaKeyConfig, ChromaKeyEngine
|
||
|
||
ck_config = ChromaKeyConfig.from_dict(clip.config.get("chroma_key"))
|
||
if ck_config.has_effect():
|
||
ck_engine = ChromaKeyEngine(ck_config)
|
||
ck_full = ck_engine.build_filter("[in]", "[out]")
|
||
ck_filter_part = ck_full[len("[in]") : -len("[out]")]
|
||
filters.append(ck_filter_part)
|
||
except Exception as e:
|
||
logger.warning("[unified-render] chroma key 直通模式应用失败，跳过: %s", e)
|
||
|
||
filters.append("setpts=PTS-STARTPTS")
|
||
filters.append(f"fps={self.output_fps}")
|
||
filters.append("format=yuv420p")
|
||
|
||
# 字幕叠加
|
||
if ass_path is not None:
|
||
ass_filter_path = str(ass_path).replace("\\", "/").replace(":", "\\:")
|
||
filters.append(f"subtitles='{ass_filter_path}'")
|
||
|
||
vf_str = ",".join(filters)
|
||
|
||
# 最终输出时长：取 clip 调速后有效时长和 video_duration 的较小值
|
||
# 注意：必须用调速后的时长，否则减速场景（speed<1）会被 -t 截断
|
||
adjusted_duration = UnifiedRenderService._clip_adjusted_duration(clip)
|
||
final_duration = adjusted_duration
|
||
if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration):
|
||
final_duration = video_duration
|
||
|
||
command = [
|
||
FFMPEG_BIN,
|
||
"-y",
|
||
"-i",
|
||
str(clip.local_path),
|
||
"-vf",
|
||
vf_str,
|
||
"-c:v",
|
||
"libx264",
|
||
"-crf",
|
||
"23",
|
||
"-preset",
|
||
"medium",
|
||
"-pix_fmt",
|
||
"yuv420p",
|
||
"-movflags",
|
||
"+faststart",
|
||
]
|
||
|
||
# 音频处理：background 通常是图片无音频，跳过；其他编码为 aac
|
||
# background 以外的视频素材，默认带音频
|
||
has_audio = role != "background"
|
||
if has_audio:
|
||
af_parts: list[str] = []
|
||
|
||
# 音频降噪
|
||
try:
|
||
from video_processing.noise_reduction_engine import NoiseReductionConfig, NoiseReductionEngine
|
||
|
||
plan_config = getattr(self.plan, "config", {}) or {}
|
||
nr_config = NoiseReductionConfig.from_dict(plan_config.get("audio_noise_reduction"))
|
||
if nr_config.has_effect():
|
||
nr_engine = NoiseReductionEngine(nr_config)
|
||
nr_full = nr_engine.build_filter("[in]", "[out]")
|
||
nr_filter_part = nr_full[len("[in]") : -len("[out]")]
|
||
af_parts.append(nr_filter_part)
|
||
except Exception as e:
|
||
logger.warning("[unified-render] 直通模式音频降噪应用失败，跳过: %s", e)
|
||
|
||
# 音频调速（与视频setpts对应，保持音画同步）
|
||
speed = UnifiedRenderService._clip_speed(clip)
|
||
if abs(speed - 1.0) >= 1e-6:
|
||
try:
|
||
from video_processing.speed_engine import SpeedConfig, SpeedEngine
|
||
|
||
speed_cfg = SpeedConfig(speed=speed)
|
||
speed_engine = SpeedEngine()
|
||
af_parts.append(speed_engine.build_audio_filter(speed_cfg))
|
||
except Exception as e:
|
||
logger.warning("[unified-render] 直通模式音频调速应用失败，跳过: %s", e)
|
||
|
||
# 音频倒放
|
||
reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
|
||
if reverse_config.enabled and reverse_config.reverse_audio:
|
||
af_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration)
|
||
if af_filter:
|
||
af_parts.append(af_filter)
|
||
|
||
if af_parts:
|
||
command.extend(["-af", ",".join(af_parts)])
|
||
|
||
command.extend(["-c:a", "aac", "-b:a", "128k"])
|
||
|
||
# 统一截断时长（同时作用于视频和音频）
|
||
if final_duration > 0:
|
||
command.extend(["-t", f"{final_duration:.3f}"])
|
||
|
||
command.append(str(output_path))
|
||
|
||
logger.info(
|
||
"直通渲染: plan_id=%s clip=%s role=%s duration=%.2fs has_audio=%s",
|
||
self.plan.id,
|
||
clip.clip_id,
|
||
role,
|
||
effective_duration,
|
||
has_audio,
|
||
)
|
||
try:
|
||
run_ffmpeg(command)
|
||
except subprocess.CalledProcessError as e:
|
||
stderr_text = (e.stderr or "").strip()
|
||
stderr_tail = stderr_text[-1500:] if len(stderr_text) > 1500 else stderr_text
|
||
logger.error(
|
||
"直通渲染失败: plan_id=%s clip=%s exit_code=%d\nvf=%s\nstderr(last 1500):\n%s",
|
||
self.plan.id,
|
||
clip.clip_id,
|
||
e.returncode,
|
||
vf_str[:2000],
|
||
stderr_tail,
|
||
)
|
||
raise
|
||
|
||
return has_audio
|
||
|
||
# ── 内部方法 ──────────────────────────────────────────────────────────────
|
||
|
||
def _resolve_clips(self) -> list[ResolvedClip]:
|
||
"""将 EditPlanClip 列表解析为 ResolvedClip 列表。
|
||
|
||
跳过 asset_id 为空或在 asset_path_map 中找不到的片段。
|
||
支持多段裁剪：一个 clip 配置了 trim_segments 时会展开为多个 ResolvedClip。
|
||
"""
|
||
resolved: list[ResolvedClip] = []
|
||
for clip in self.clips:
|
||
asset_id = clip.asset_id
|
||
if not asset_id:
|
||
logger.warning("片段无素材: clip_id=%s", clip.id)
|
||
continue
|
||
|
||
local_path = self.asset_path_map.get(asset_id)
|
||
if local_path is None or not local_path.exists():
|
||
logger.warning("素材不存在: clip_id=%s asset_id=%s", clip.id, asset_id)
|
||
continue
|
||
|
||
# 探测实际时长
|
||
try:
|
||
actual_duration = probe_duration(local_path)
|
||
except Exception:
|
||
actual_duration = clip.duration or 5.0
|
||
|
||
# 检查是否有多段裁剪配置
|
||
clip_config = clip.config or {}
|
||
trim_segments = TrimEngine.parse_segments_from_config(clip_config)
|
||
|
||
if trim_segments and len(trim_segments) > 1:
|
||
# 多段裁剪：展开为多个 clip
|
||
resolved_segments = TrimEngine.resolve_segments(trim_segments, actual_duration)
|
||
for i, seg in enumerate(resolved_segments):
|
||
# 每个段生成一个独立的 ResolvedClip
|
||
seg_clip_id = f"{clip.id}_seg_{seg.segment_id}"
|
||
seg_order = clip.order + seg.order * 0.001 + i * 0.0001 # 保持排序
|
||
seg_start = seg.trim.start_time
|
||
seg_duration = seg.trim.duration
|
||
|
||
rc = ResolvedClip(
|
||
clip_id=seg_clip_id,
|
||
asset_id=asset_id,
|
||
local_path=local_path,
|
||
clip_type=clip.clip_type,
|
||
order=seg_order,
|
||
start_time=seg_start,
|
||
duration=seg_duration,
|
||
transition_effect=clip.transition_effect or "cut",
|
||
transition_duration=getattr(clip, "transition_duration", 0.0) or 0.0,
|
||
playback_speed=getattr(clip, "playback_speed", 1.0) or 1.0,
|
||
config={**clip_config, "_segment_id": seg.segment_id},
|
||
actual_duration=actual_duration,
|
||
trim_config=seg.trim,
|
||
)
|
||
resolved.append(rc)
|
||
continue
|
||
|
||
# 单段裁剪（或无裁剪）
|
||
# 解析裁剪配置：config 优先，否则用 clip.start_time + clip.duration
|
||
trim_config = extract_trim_from_clip_config(clip_config)
|
||
if trim_config is None and (clip.start_time > 0 or clip.duration > 0):
|
||
# 用旧字段构造
|
||
trim_config = TrimConfig(
|
||
start_time=clip.start_time,
|
||
duration=clip.duration,
|
||
)
|
||
|
||
# 钳制到实际素材时长
|
||
effective_trim: TrimConfig | None = None
|
||
final_start = clip.start_time
|
||
final_duration = clip.duration
|
||
|
||
if trim_config is not None and actual_duration > 0:
|
||
effective_trim = trim_config.validate_and_resolve(actual_duration)
|
||
if effective_trim.is_valid:
|
||
final_start = effective_trim.start_time
|
||
final_duration = effective_trim.duration
|
||
else:
|
||
# 裁剪无效 → 使用完整素材
|
||
logger.warning("裁剪配置无效，使用完整素材: clip_id=%s", clip.id)
|
||
effective_trim = None
|
||
final_start = 0.0
|
||
final_duration = actual_duration
|
||
|
||
rc = ResolvedClip(
|
||
clip_id=clip.id,
|
||
asset_id=asset_id,
|
||
local_path=local_path,
|
||
clip_type=clip.clip_type,
|
||
order=clip.order,
|
||
start_time=final_start,
|
||
duration=final_duration,
|
||
transition_effect=clip.transition_effect or "cut",
|
||
transition_duration=getattr(clip, "transition_duration", 0.0) or 0.0,
|
||
playback_speed=getattr(clip, "playback_speed", 1.0) or 1.0,
|
||
config=clip_config,
|
||
actual_duration=actual_duration,
|
||
trim_config=effective_trim,
|
||
)
|
||
resolved.append(rc)
|
||
|
||
# 按 order 排序
|
||
resolved.sort(key=lambda c: c.order)
|
||
return resolved
|
||
|
||
def _group_clips_into_layers(self, resolved_clips: list[ResolvedClip]) -> list[RenderLayer]:
|
||
"""将 ResolvedClips 分组为 RenderLayers。
|
||
|
||
分组规则见 _resolve_layer_role 函数文档。
|
||
"""
|
||
layer_map: dict[str, RenderLayer] = {}
|
||
|
||
for clip in resolved_clips:
|
||
role = _resolve_layer_role(clip.clip_type, clip.config)
|
||
if role not in layer_map:
|
||
z = _LAYER_Z_INDEX.get(role, 0)
|
||
layer_map[role] = RenderLayer(role=role, z_index=z)
|
||
layer_map[role].clips.append(clip)
|
||
|
||
# 每个 layer 内的 clips 按 order 排序
|
||
for layer in layer_map.values():
|
||
layer.clips.sort(key=lambda c: c.order)
|
||
|
||
# 计算 PiP 位置
|
||
pip_width = int(self.output_width * _PIP_SCALE)
|
||
margin = 20 # 边距
|
||
|
||
if "overlay" in layer_map:
|
||
layer_map["overlay"].position = (
|
||
self.output_width - pip_width - margin,
|
||
margin,
|
||
)
|
||
if "corner_voice" in layer_map:
|
||
layer_map["corner_voice"].position = (
|
||
self.output_width - pip_width - margin,
|
||
margin,
|
||
)
|
||
|
||
# 按 z_index 排序返回
|
||
layers = sorted(layer_map.values(), key=lambda lyr: lyr.z_index)
|
||
return layers
|
||
|
||
def _build_filter_complex(
|
||
self, layers: list[RenderLayer], *, ass_path: Path | None = None
|
||
) -> tuple[str, list[str]]:
|
||
"""构建 FFmpeg filter_complex 字符串和输入参数列表。
|
||
|
||
Args:
|
||
layers: 图层列表
|
||
ass_path: ASS 字幕文件路径，有则在最后叠加字幕
|
||
|
||
Returns:
|
||
(filter_complex_str, input_args_list)
|
||
input_args_list 是 ["-i", path1, "-i", path2, ...] 格式
|
||
"""
|
||
if not layers:
|
||
raise ValueError("没有可渲染的图层")
|
||
|
||
# 收集所有 clips（按图层顺序，同层按 order）
|
||
all_clips: list[ResolvedClip] = []
|
||
for layer in layers:
|
||
all_clips.extend(layer.clips)
|
||
|
||
# 构建输入参数
|
||
input_args: list[str] = []
|
||
clip_to_input_idx: dict[str, int] = {}
|
||
for i, clip in enumerate(all_clips):
|
||
input_args.extend(["-i", str(clip.local_path)])
|
||
clip_to_input_idx[clip.clip_id] = i
|
||
|
||
filter_parts: list[str] = []
|
||
|
||
# Step 1: 预处理每个 clip — trim + scale + setpts
|
||
# 为每个 clip 生成预处理后的标签 [v0], [v1], ...
|
||
preprocessed_labels: list[str] = []
|
||
for i, clip in enumerate(all_clips):
|
||
label = f"v{i}"
|
||
role = _resolve_layer_role(clip.clip_type, clip.config)
|
||
|
||
filters: list[str] = []
|
||
|
||
# trim — 裁剪到指定区间，精确到帧
|
||
effective_duration = UnifiedRenderService._clip_effective_duration(clip)
|
||
trim_start = getattr(clip, "start_time", 0) or 0
|
||
|
||
if effective_duration > 0:
|
||
if trim_start > 0:
|
||
filters.append(f"trim=start={trim_start:.3f}:duration={effective_duration:.3f}")
|
||
else:
|
||
filters.append(f"trim=duration={effective_duration:.3f}")
|
||
filters.append("setpts=PTS-STARTPTS")
|
||
|
||
# 调速 — 基于 setpts 改变播放速度
|
||
speed = UnifiedRenderService._clip_speed(clip)
|
||
if abs(speed - 1.0) >= 1e-6:
|
||
filters.append(f"setpts=PTS/{speed:.4f}")
|
||
|
||
# 倒放滤镜（在 trim 之后、scale 之前应用）
|
||
reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
|
||
if reverse_config.enabled and reverse_config.reverse_video:
|
||
reverse_filter = ReverseEngine.build_video_filter(reverse_config, duration=effective_duration)
|
||
if reverse_filter:
|
||
filters.append(reverse_filter)
|
||
|
||
# scale
|
||
if role in ("overlay", "corner_voice"):
|
||
pip_w = int(self.output_width * _PIP_SCALE)
|
||
pip_h = int(self.output_height * _PIP_SCALE)
|
||
filters.append(f"scale={pip_w}:{pip_h}")
|
||
elif role == "background":
|
||
filters.append(f"scale={self.output_width}:{self.output_height}:force_original_aspect_ratio=increase")
|
||
filters.append(f"crop={self.output_width}:{self.output_height}")
|
||
else:
|
||
# main / broll: 等比缩放 + 居中留黑边（保持原始比例，不裁剪内容）
|
||
# concat 要求所有输入分辨率完全一致，pad 模式确保不同宽高比的素材都能正常拼接
|
||
filters.append(f"scale={self.output_width}:{self.output_height}:force_original_aspect_ratio=decrease")
|
||
filters.append(f"pad={self.output_width}:{self.output_height}:trunc((ow-iw)/2):trunc((oh-ih)/2):black")
|
||
|
||
# 调色滤镜（每个 clip 独立的 color grade 配置）
|
||
color_grade = ColorGradeConfig.from_dict(clip.config.get("color_grade"))
|
||
if color_grade.enabled and color_grade.has_effect():
|
||
grade_filter = ColorGradeEngine.build_filter(color_grade)
|
||
if grade_filter:
|
||
filters.append(grade_filter)
|
||
# chroma key 绿幕抠像（在 scale 之后，fps 之前）
|
||
try:
|
||
from video_processing.chroma_key_engine import ChromaKeyConfig, ChromaKeyEngine
|
||
|
||
ck_config = ChromaKeyConfig.from_dict(clip.config.get("chroma_key"))
|
||
if ck_config.has_effect():
|
||
ck_engine = ChromaKeyEngine(ck_config)
|
||
# 提取滤镜部分（不带输入输出标签）
|
||
ck_full = ck_engine.build_filter("[in]", "[out]")
|
||
ck_filter_part = ck_full[len("[in]") : -len("[out]")]
|
||
filters.append(ck_filter_part)
|
||
except Exception as e:
|
||
logger.warning("[unified-render] chroma key 应用失败，跳过 clip=%s: %s", clip.clip_id, e)
|
||
|
||
filters.append("setpts=PTS-STARTPTS")
|
||
filters.append(f"fps={self.output_fps}")
|
||
|
||
filter_str = f"[{i}:v]{','.join(filters)}[{label}]"
|
||
filter_parts.append(filter_str)
|
||
preprocessed_labels.append(label)
|
||
|
||
# Step 2: 同层 clips 用 xfade 串联
|
||
layer_output_labels: dict[str, str] = {}
|
||
for layer in layers:
|
||
layer_clip_indices = [all_clips.index(c) for c in layer.clips]
|
||
layer_labels = [preprocessed_labels[i] for i in layer_clip_indices]
|
||
# 使用调速后的实际时长，与 Step 1 的调速处理保持一致
|
||
layer_durations = [UnifiedRenderService._clip_adjusted_duration(all_clips[i]) for i in layer_clip_indices]
|
||
layer_transitions = [all_clips[i].transition_effect for i in layer_clip_indices]
|
||
layer_transition_durations = [all_clips[i].transition_duration for i in layer_clip_indices]
|
||
|
||
if len(layer_labels) == 1:
|
||
# 单 clip 层，直接使用预处理标签
|
||
layer_output_labels[layer.role] = layer_labels[0]
|
||
else:
|
||
out_label = f"{layer.role}_merged"
|
||
# 判断是否全部为硬切：是则用 concat filter，否则用 xfade 转场链
|
||
all_cut = all(
|
||
t is None or t == "" or str(t).lower() == "cut"
|
||
for t in layer_transitions[1:] # 第一个 clip 的转场忽略
|
||
)
|
||
if all_cut:
|
||
# 全硬切：用 concat filter，性能远优于 xfade
|
||
concat_inputs = "".join(f"[{label}]" for label in layer_labels)
|
||
filter_parts.append(f"{concat_inputs}concat=n={len(layer_labels)}:v=1:a=0[{out_label}]")
|
||
logger.info(
|
||
"[unified-render] layer=%s clips=%d using concat (all hard-cut)",
|
||
layer.role,
|
||
len(layer_labels),
|
||
)
|
||
else:
|
||
# 有转场效果：用 TransitionEngine 构建 xfade 链
|
||
layer_dur = 0.0
|
||
for d in layer_transition_durations:
|
||
if d > 0:
|
||
layer_dur = d
|
||
break
|
||
xfade_filter, _ = self._transition_engine.build_xfade_chain(
|
||
clip_durations=layer_durations,
|
||
clip_video_labels=layer_labels,
|
||
transitions=layer_transitions,
|
||
transition_duration=layer_dur if layer_dur > 0 else None,
|
||
output_label=out_label,
|
||
)
|
||
if xfade_filter:
|
||
filter_parts.append(xfade_filter)
|
||
layer_output_labels[layer.role] = out_label
|
||
|
||
# Step 3: 合成各层
|
||
# 找到主层 — background 优先作为底图，其次 broll / main
|
||
final_video_label = None
|
||
|
||
if "background" in layer_output_labels:
|
||
final_video_label = layer_output_labels["background"]
|
||
# b_roll / main 叠加到 background 上
|
||
for role in ("broll", "main"):
|
||
if role in layer_output_labels:
|
||
base_label = layer_output_labels[role]
|
||
combined_label = f"combined_{role}"
|
||
filter_parts.append(f"[{final_video_label}][{base_label}]overlay=(W-w)/2:(H-h)/2[{combined_label}]")
|
||
final_video_label = combined_label
|
||
else:
|
||
# 无 background 时，取 broll 或 main 作为基础
|
||
for role in ("broll", "main"):
|
||
if role in layer_output_labels:
|
||
final_video_label = layer_output_labels[role]
|
||
break
|
||
|
||
if final_video_label is None:
|
||
# 没有任何主层，使用第一个层
|
||
final_video_label = layer_output_labels[layers[0].role]
|
||
|
||
# 叠加 overlay 层
|
||
for layer in layers:
|
||
if layer.role in ("overlay", "corner_voice"):
|
||
if layer.role not in layer_output_labels:
|
||
continue
|
||
overlay_label = layer_output_labels[layer.role]
|
||
x, y = layer.position or (
|
||
self.output_width - int(self.output_width * _PIP_SCALE) - 20,
|
||
20,
|
||
)
|
||
combined_label = f"combined_{layer.role}"
|
||
filter_parts.append(f"[{final_video_label}][{overlay_label}]overlay={x}:{y}[{combined_label}]")
|
||
final_video_label = combined_label
|
||
|
||
# 叠加水印（在字幕之前）
|
||
watermark_config = UnifiedRenderService._resolve_watermark_config(self.plan.config)
|
||
if watermark_config is not None:
|
||
wm_valid, wm_err = watermark_config.validate()
|
||
if wm_valid:
|
||
wm_label = "watermarked"
|
||
if watermark_config.mode == "image":
|
||
# 图片水印：检查图片是否存在
|
||
wm_path = Path(watermark_config.image_path)
|
||
if wm_path.exists():
|
||
# 图片水印需要额外输入，放在 filter 开头
|
||
wm_idx = len(all_clips) # 水印图是最后一个输入
|
||
wm_scale = int(self.output_width * watermark_config.scale)
|
||
|
||
# 透明度
|
||
wm_filters = f"scale={wm_scale}:-1"
|
||
if watermark_config.opacity < 1.0:
|
||
wm_filters += f",format=rgba,colorchannelmixer=aa={watermark_config.opacity}"
|
||
|
||
filter_parts.insert(0, f"[{wm_idx}:v]{wm_filters}[wm_scaled]")
|
||
input_args.extend(["-i", str(wm_path)])
|
||
|
||
# 位置计算（水印高度用 scale 后的宽度近似）
|
||
wm_h = wm_scale # 近似（正方形假设）
|
||
x, y = WatermarkEngine.calc_position(
|
||
watermark_config.position,
|
||
self.output_width,
|
||
self.output_height,
|
||
wm_scale,
|
||
wm_h,
|
||
watermark_config.margin_x,
|
||
watermark_config.margin_y,
|
||
)
|
||
|
||
# 滚动水印
|
||
if watermark_config.scroll:
|
||
x_expr = f"W-mod({watermark_config.scroll_speed}*t\\,W+w)"
|
||
overlay = f"[{final_video_label}][wm_scaled]overlay=x={x_expr}:y={y}[{wm_label}]"
|
||
else:
|
||
overlay = f"[{final_video_label}][wm_scaled]overlay=x={x}:y={y}[{wm_label}]"
|
||
|
||
filter_parts.append(overlay)
|
||
final_video_label = wm_label
|
||
else:
|
||
logger.warning("水印图片不存在，跳过水印: %s", wm_path)
|
||
elif watermark_config.mode == "text":
|
||
# 文字水印
|
||
try:
|
||
text_wm = WatermarkEngine.build_text_watermark_filter(
|
||
f"[{final_video_label}]",
|
||
f"[{wm_label}]",
|
||
watermark_config,
|
||
self.output_width,
|
||
self.output_height,
|
||
)
|
||
filter_parts.append(text_wm)
|
||
final_video_label = wm_label
|
||
except Exception as e:
|
||
logger.warning("文字水印构建失败，跳过: %s", e)
|
||
# 贴纸叠加（图片贴纸 + 文字贴纸）
|
||
sticker_filter, sticker_extra_inputs = self._build_sticker_filters(final_video_label, "after_stickers")
|
||
if sticker_filter:
|
||
filter_parts.append(sticker_filter)
|
||
# 图片贴纸需要额外输入
|
||
for img_path in sticker_extra_inputs:
|
||
input_args.extend(["-i", img_path])
|
||
final_video_label = "after_stickers"
|
||
|
||
# 叠加字幕（如有）+ 最终像素格式
|
||
if ass_path is not None:
|
||
ass_filter_path = str(ass_path).replace("\\", "/").replace(":", "\\:")
|
||
filter_parts.append(f"[{final_video_label}]subtitles='{ass_filter_path}',format=yuv420p[final_video]")
|
||
else:
|
||
filter_parts.append(f"[{final_video_label}]format=yuv420p[final_video]")
|
||
|
||
filter_complex = ";".join(filter_parts)
|
||
return filter_complex, input_args
|
||
|
||
def _execute_ffmpeg(
|
||
self,
|
||
filter_complex: str,
|
||
input_args: list[str],
|
||
output_path: Path,
|
||
) -> None:
|
||
"""执行 FFmpeg 渲染命令。
|
||
|
||
失败时记录完整 filter_complex 以便排查（如 exit code 183）。
|
||
"""
|
||
command = [
|
||
FFMPEG_BIN,
|
||
"-y",
|
||
*input_args,
|
||
"-filter_complex",
|
||
filter_complex,
|
||
"-map",
|
||
"[final_video]",
|
||
"-c:v",
|
||
"libx264",
|
||
"-crf",
|
||
"23",
|
||
"-preset",
|
||
"medium",
|
||
"-pix_fmt",
|
||
"yuv420p",
|
||
"-movflags",
|
||
"+faststart",
|
||
str(output_path),
|
||
]
|
||
|
||
logger.info(
|
||
"执行渲染: plan_id=%s inputs=%d output=%s",
|
||
self.plan.id,
|
||
input_args.count("-i"),
|
||
output_path,
|
||
)
|
||
try:
|
||
run_ffmpeg(command)
|
||
except subprocess.CalledProcessError as e:
|
||
# 额外记录 filter_complex + stderr，方便排查滤镜链构建问题
|
||
stderr_text = (e.stderr or "").strip()
|
||
stderr_tail = stderr_text[-1500:] if len(stderr_text) > 1500 else stderr_text
|
||
logger.error(
|
||
"渲染失败: plan_id=%s exit_code=%d\nfilter_complex:\n%s\nstderr(last 1500):\n%s",
|
||
self.plan.id,
|
||
e.returncode,
|
||
filter_complex[:5000],
|
||
stderr_tail,
|
||
)
|
||
raise
|
||
|
||
def _build_sticker_filters(self, input_label: str, output_label: str) -> tuple[str, list[str]]:
|
||
"""构建贴纸叠加滤镜链.
|
||
|
||
Args:
|
||
input_label: 输入视频标签
|
||
output_label: 输出视频标签
|
||
|
||
Returns:
|
||
(filter_str, extra_input_paths)
|
||
filter_str: 贴纸滤镜字符串（空表示无贴纸）
|
||
extra_input_paths: 额外需要的输入文件路径（图片贴纸）
|
||
"""
|
||
plan_config = getattr(self.plan, "config", None) or {}
|
||
if isinstance(plan_config, dict):
|
||
stickers_data = plan_config.get("stickers", [])
|
||
else:
|
||
stickers_data = []
|
||
|
||
if not stickers_data:
|
||
return "", []
|
||
|
||
try:
|
||
result = StickerEngine.build_sticker_chain(
|
||
stickers=stickers_data,
|
||
input_label=f"[{input_label}]",
|
||
output_label=f"[{output_label}]",
|
||
canvas_w=self.output_width,
|
||
canvas_h=self.output_height,
|
||
)
|
||
return result.filter_str, result.extra_inputs
|
||
except Exception as e:
|
||
logger.warning("贴纸滤镜构建失败，跳过贴纸: %s", e)
|
||
return "", []
|
||
|
||
def _probe_output(self, output_path: Path) -> tuple[float, int, int, int]:
|
||
"""探测输出文件的时长、大小、宽高.
|
||
|
||
Returns:
|
||
(duration, file_size, width, height)
|
||
"""
|
||
info = probe_video_info(str(output_path))
|
||
file_size = output_path.stat().st_size if output_path.exists() else 0
|
||
return (
|
||
info["duration"],
|
||
file_size,
|
||
info["width"],
|
||
info["height"],
|
||
)
|
||
|
||
@staticmethod
|
||
def _clip_effective_duration(clip: ResolvedClip) -> float:
|
||
"""计算 clip 的有效时长（原速 trim 后时长）。
|
||
|
||
实际实现移至 packages.domain.render_layer_utils.clip_effective_duration。
|
||
"""
|
||
return _clip_effective_duration_pure(clip.duration, clip.actual_duration)
|
||
|
||
# ── 画中画（PiP）相关方法 ──────────────────────────────────────────────────
|
||
|
||
def _resolve_pip_sources(self, pip_config: PiPConfig) -> list[tuple[str, PiPLayerConfig, Path]]:
|
||
"""解析画中画图层的素材源，返回可用的图层列表.
|
||
|
||
降级策略：素材不存在或无效的图层自动跳过，不阻断渲染。
|
||
|
||
Returns:
|
||
[(input_label_placeholder, layer_config, local_path), ...]
|
||
input_label 在 build_pip_filters 中会用实际的输入索引替换
|
||
"""
|
||
if not pip_config.enabled:
|
||
return []
|
||
|
||
engine = PiPEngine(
|
||
output_width=self.output_width,
|
||
output_height=self.output_height,
|
||
output_fps=self.output_fps,
|
||
)
|
||
|
||
result = []
|
||
for i, layer in enumerate(pip_config.layers):
|
||
path = engine.validate_layer_source(layer, self.asset_path_map)
|
||
if path is None:
|
||
logger.warning("PiP图层素材不可用，跳过: layer_index=%d source=%s", i, layer.source)
|
||
continue
|
||
# 标签占位，实际输入索引由 build_pip_filters 内部管理
|
||
result.append((f"pip_src_{i}", layer, path))
|
||
|
||
return result
|
||
|
||
def _append_pip_filters(
|
||
self,
|
||
filter_complex: str,
|
||
input_args: list[str],
|
||
pip_sources: list[tuple[str, Any, Path]],
|
||
) -> tuple[str, list[str]]:
|
||
"""将画中画滤镜追加到 filter_complex 末尾.
|
||
|
||
处理逻辑：
|
||
1. 将原 final_video 标签重命名为 pip_base（作为PiP的底层视频）
|
||
2. 追加 PiP 预处理和 overlay 滤镜
|
||
3. PiP 最终输出命名为 final_video
|
||
|
||
Args:
|
||
filter_complex: 原 filter_complex 字符串
|
||
input_args: 原输入参数列表
|
||
pip_sources: PiP 素材列表 [(label, layer_config, path), ...]
|
||
|
||
Returns:
|
||
(new_filter_complex, new_input_args)
|
||
"""
|
||
if not pip_sources:
|
||
return filter_complex, input_args
|
||
|
||
pip_engine = PiPEngine(
|
||
output_width=self.output_width,
|
||
output_height=self.output_height,
|
||
output_fps=self.output_fps,
|
||
)
|
||
|
||
# 1. 将原 final_video 改为 pip_base
|
||
new_filter = filter_complex.replace("[final_video]", "[pip_base]")
|
||
|
||
# 2. 构建 PiP 滤镜链
|
||
# 主输入数量 = len(input_args) // 2（每个输入占 "-i path" 两个参数）
|
||
base_input_idx = len(input_args) // 2
|
||
pip_filter_parts, pip_input_args, final_label = pip_engine.build_pip_filters(
|
||
base_label="pip_base",
|
||
pip_sources=pip_sources,
|
||
base_input_idx=base_input_idx,
|
||
)
|
||
|
||
if not pip_filter_parts:
|
||
# 没有有效PiP滤镜，恢复原标签
|
||
return filter_complex, input_args
|
||
|
||
# 3. 追加 PiP 滤镜 + 最终格式转换（输出为 final_video）
|
||
pip_filter_str = ";".join(pip_filter_parts)
|
||
final_format = f"[{final_label}]format=yuv420p[final_video]"
|
||
new_filter = f"{new_filter};{pip_filter_str};{final_format}"
|
||
|
||
# 4. 追加输入参数
|
||
new_input_args = list(input_args) + pip_input_args
|
||
|
||
logger.info(
|
||
"[unified-render] appended PiP filters: layers=%d new_inputs=%d",
|
||
len(pip_sources),
|
||
len(pip_input_args) // 2,
|
||
)
|
||
|
||
return new_filter, new_input_args
|
||
|
||
@staticmethod
|
||
def _clip_speed(clip: ResolvedClip) -> float:
|
||
"""获取 clip 的播放速度，无效值回退到 1.0。
|
||
|
||
实际实现移至 packages.domain.render_layer_utils.clip_playback_speed。
|
||
"""
|
||
return _clip_playback_speed_pure(getattr(clip, "playback_speed", 1.0))
|
||
|
||
@staticmethod
|
||
def _clip_adjusted_duration(clip: ResolvedClip) -> float:
|
||
"""计算调速后的 clip 实际时长（用于拼接计算）。
|
||
|
||
实际实现移至 packages.domain.render_layer_utils.clip_adjusted_duration。
|
||
"""
|
||
return _clip_adjusted_duration_pure(
|
||
clip.duration,
|
||
clip.actual_duration,
|
||
getattr(clip, "playback_speed", 1.0),
|
||
)
|