feat: 多轨道混音 + 字幕渲染引擎 + 视频拼接（后端） (#312)

2026-07-14 15:20:47 +08:00
parent b05966ff48
commit f268e208de
6 changed files with 2676 additions and 2 deletions
@@ -0,0 +1,633 @@
+"""字幕渲染引擎 — 统一管理字幕样式配置与视频烧录.
+
+与现有模块的关系：
+- render_subtitles.py：生成静态整段标题/字幕的 ASS 文件
+- subtitle_generator.py：从 ASR 时间轴生成 ASS 文件
+- 本模块：统一的字幕样式配置 + 烧录滤镜生成 + 多源字幕合并
+
+支持的字幕来源：
+1. 静态标题/字幕（title_config / subtitle_config）
+2. ASR 自动字幕（asr_subtitle_timeline）
+3. 手动字幕（manual_subtitles 时间轴）
+
+支持的样式配置：
+- 字体、字号、颜色
+- 描边（颜色、宽度）
+- 阴影（偏移、模糊、颜色）
+- 背景框（颜色、透明度、圆角、边距）
+- 位置（9宫格 + 自定义坐标）
+- 对齐方式
+- 动画（淡入淡出、滑入滑出、打字机）
+- 多行/换行规则
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from video_processing.render_subtitles import generate_ass_subtitles
+from video_processing.subtitle_generator import generate_ass_from_timeline
+
+logger = logging.getLogger(__name__)
+
+
+# ── 常量 ──────────────────────────────────────────────────────────────────────
+
+# 9宫格位置映射（ASS alignment 编号）
+POSITION_ALIGNMENT = {
+    "top_left": 7,
+    "top_center": 8,
+    "top_right": 9,
+    "middle_left": 4,
+    "center": 5,
+    "middle_right": 6,
+    "bottom_left": 1,
+    "bottom_center": 2,
+    "bottom_right": 3,
+}
+
+# 位置简称兼容
+POSITION_ALIASES = {
+    "top": "top_center",
+    "bottom": "bottom_center",
+    "middle": "center",
+    "left": "middle_left",
+    "right": "middle_right",
+}
+
+DEFAULT_FONT = "思源黑体"
+DEFAULT_FONT_SIZE = 24
+DEFAULT_COLOR = "#FFFFFF"
+DEFAULT_STROKE_COLOR = "#000000"
+DEFAULT_STROKE_WIDTH = 1.5
+DEFAULT_POSITION = "bottom_center"
+DEFAULT_MAX_CHARS_PER_LINE = 20
+
+
+# ── 字幕样式配置 ────────────────────────────────────────────────────────────
+
+
+@dataclass
+class SubtitleStyle:
+    """字幕样式配置."""
+
+    font_name: str = DEFAULT_FONT
+    font_size: int = DEFAULT_FONT_SIZE
+    font_color: str = DEFAULT_COLOR
+    bold: bool = False
+    italic: bool = False
+
+    # 描边
+    stroke_enabled: bool = True
+    stroke_color: str = DEFAULT_STROKE_COLOR
+    stroke_width: float = DEFAULT_STROKE_WIDTH
+
+    # 阴影
+    shadow_enabled: bool = False
+    shadow_color: str = "#000000"
+    shadow_offset_x: int = 2
+    shadow_offset_y: int = 2
+    shadow_blur: float = 0.0
+
+    # 背景框
+    background_enabled: bool = False
+    background_color: str = "#000000"
+    background_opacity: float = 0.5  # 0.0 ~ 1.0
+    background_padding: int = 8
+    background_radius: int = 4
+
+    # 位置
+    position: str = DEFAULT_POSITION  # 9宫格位置名
+    margin_v: int = 60  # 垂直边距
+    margin_l: int = 40  # 左边距
+    margin_r: int = 40  # 右边距
+
+    # 多行
+    max_chars_per_line: int = DEFAULT_MAX_CHARS_PER_LINE
+    line_spacing: int = 0  # 行间距
+
+    # 动画
+    fade_in: float = 0.0  # 淡入时长（秒）
+    fade_out: float = 0.0  # 淡出时长（秒）
+    animation_type: str = "none"  # none/fade/slide/typewriter
+
+    @classmethod
+    def from_dict(cls, config: dict[str, Any] | None) -> "SubtitleStyle":
+        """从字典创建样式配置，带安全类型转换."""
+        if not config or not isinstance(config, dict):
+            return cls()
+
+        def safe_str(key: str, default: str) -> str:
+            val = config.get(key, default)
+            return str(val) if val is not None else default
+
+        def safe_int(key: str, default: int) -> int:
+            try:
+                return int(config.get(key, default))
+            except (TypeError, ValueError):
+                return default
+
+        def safe_float(key: str, default: float) -> float:
+            try:
+                return float(config.get(key, default))
+            except (TypeError, ValueError):
+                return default
+
+        def safe_bool(key: str, default: bool) -> bool:
+            return bool(config.get(key, default))
+
+        position = safe_str("position", DEFAULT_POSITION)
+        position = POSITION_ALIASES.get(position, position)
+        if position not in POSITION_ALIGNMENT:
+            position = DEFAULT_POSITION
+
+        return cls(
+            font_name=safe_str("font", DEFAULT_FONT),
+            font_size=safe_int("size", DEFAULT_FONT_SIZE),
+            font_color=safe_str("color", DEFAULT_COLOR),
+            bold=safe_bool("bold", False),
+            italic=safe_bool("italic", False),
+            stroke_enabled=safe_bool("stroke_enabled", True),
+            stroke_color=safe_str("stroke_color", DEFAULT_STROKE_COLOR),
+            stroke_width=safe_float("stroke_width", DEFAULT_STROKE_WIDTH),
+            shadow_enabled=safe_bool("shadow_enabled", False),
+            shadow_color=safe_str("shadow_color", "#000000"),
+            shadow_offset_x=safe_int("shadow_offset_x", 2),
+            shadow_offset_y=safe_int("shadow_offset_y", 2),
+            shadow_blur=safe_float("shadow_blur", 0.0),
+            background_enabled=safe_bool("background_enabled", False),
+            background_color=safe_str("background_color", "#000000"),
+            background_opacity=max(0.0, min(1.0, safe_float("background_opacity", 0.5))),
+            background_padding=safe_int("background_padding", 8),
+            background_radius=safe_int("background_radius", 4),
+            position=position,
+            margin_v=safe_int("margin_v", 60),
+            margin_l=safe_int("margin_l", 40),
+            margin_r=safe_int("margin_r", 40),
+            max_chars_per_line=safe_int("max_chars_per_line", DEFAULT_MAX_CHARS_PER_LINE),
+            line_spacing=safe_int("line_spacing", 0),
+            fade_in=max(0.0, safe_float("fade_in", 0.0)),
+            fade_out=max(0.0, safe_float("fade_out", 0.0)),
+            animation_type=safe_str("animation_type", "none"),
+        )
+
+    @property
+    def alignment(self) -> int:
+        """获取 ASS alignment 编号."""
+        return POSITION_ALIGNMENT.get(self.position, 2)
+
+    @property
+    def ass_font_color(self) -> str:
+        """ASS 格式颜色 &HAABBGGRR."""
+        return _hex_to_ass_color(self.font_color)
+
+    @property
+    def ass_stroke_color(self) -> str:
+        return _hex_to_ass_color(self.stroke_color)
+
+    @property
+    def ass_shadow_color(self) -> str:
+        return _hex_to_ass_color(self.shadow_color)
+
+    @property
+    def ass_background_color(self) -> str:
+        """背景框颜色（ASS BackColour），带透明度."""
+        alpha_hex = _opacity_to_ass_alpha(self.background_opacity)
+        color_bgr = _hex_to_ass_bgr(self.background_color)
+        return f"&H{alpha_hex}{color_bgr}"
+
+
+# ── 工具函数 ──────────────────────────────────────────────────────────────────
+
+
+def _hex_to_ass_color(hex_color: str) -> str:
+    """HEX → ASS 颜色 &HAABBGGRR（默认不透明）."""
+    hex_color = hex_color.lstrip("#")
+    if len(hex_color) != 6:
+        return "&H00FFFFFF"
+    r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6]
+    return f"&H00{b.upper()}{g.upper()}{r.upper()}"
+
+
+def _hex_to_ass_bgr(hex_color: str) -> str:
+    """HEX → ASS BGR 部分（不含 alpha）."""
+    hex_color = hex_color.lstrip("#")
+    if len(hex_color) != 6:
+        return "FFFFFF"
+    r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6]
+    return f"{b.upper()}{g.upper()}{r.upper()}"
+
+
+def _opacity_to_ass_alpha(opacity: float) -> str:
+    """不透明度 → ASS alpha（00=不透明，FF=完全透明）."""
+    alpha = 255 - int(opacity * 255)
+    return f"{alpha:02X}"
+
+
+def _escape_ass_text(text: str) -> str:
+    """转义 ASS 文本特殊字符."""
+    text = text.replace("\r\n", "\\N").replace("\n", "\\N").replace("\r", "\\N")
+    text = text.replace("{", "(").replace("}", ")")
+    return text
+
+
+def _format_ass_time(seconds: float) -> str:
+    """秒 → ASS 时间格式 H:MM:SS.cc."""
+    hours = int(seconds // 3600)
+    minutes = int((seconds % 3600) // 60)
+    secs = seconds % 60
+    return f"{hours}:{minutes:02d}:{secs:05.2f}"
+
+
+def _wrap_text(text: str, max_chars: int) -> list[str]:
+    """按字数换行，优先标点断开."""
+    if len(text) <= max_chars:
+        return [text]
+
+    lines: list[str] = []
+    remaining = text
+
+    while len(remaining) > max_chars:
+        break_point = max_chars
+        punctuations = "，。！？、；：,.;:!?"
+
+        for i in range(max_chars, max_chars // 2, -1):
+            if i < len(remaining) and remaining[i] in punctuations:
+                break_point = i + 1
+                break
+
+        lines.append(remaining[:break_point])
+        remaining = remaining[break_point:]
+
+    if remaining:
+        lines.append(remaining)
+
+    return lines
+
+
+# ── 字幕片段 ──────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class SubtitleSegment:
+    """单个字幕片段."""
+
+    start: float  # 开始时间（秒）
+    end: float  # 结束时间（秒）
+    text: str  # 字幕文本
+    style_name: str = "Default"  # 使用的样式名
+
+
+# ── 字幕渲染引擎 ──────────────────────────────────────────────────────────────
+
+
+class SubtitleRenderEngine:
+    """字幕渲染引擎 — 统一管理多源字幕的 ASS 文件生成.
+
+    支持合并多个字幕来源到同一个 ASS 文件：
+    - 标题（顶部，单独样式）
+    - 字幕（底部，单独样式）
+    - ASR 时间轴字幕
+    - 手动字幕
+
+    输出一个统一的 ASS 文件，供 FFmpeg subtitles filter 烧录。
+    """
+
+    def __init__(
+        self,
+        video_width: int = 1080,
+        video_height: int = 1920,
+        video_duration: float = 0.0,
+    ):
+        self.video_width = video_width
+        self.video_height = video_height
+        self.video_duration = video_duration
+        self._styles: dict[str, SubtitleStyle] = {}
+        self._segments: list[SubtitleSegment] = []
+        self._style_counter = 0
+
+    # ── 样式管理 ──────────────────────────────────────────────────────
+
+    def add_style(self, name: str, style: SubtitleStyle) -> str:
+        """注册一个样式，返回样式名."""
+        self._styles[name] = style
+        return name
+
+    def get_or_create_style(self, base_name: str, style: SubtitleStyle) -> str:
+        """获取或创建样式（避免重复）."""
+        if base_name in self._styles:
+            return base_name
+        self._styles[base_name] = style
+        return base_name
+
+    # ── 字幕源添加 ────────────────────────────────────────────────────
+
+    def add_title(self, text: str, style: SubtitleStyle | None = None) -> None:
+        """添加整段标题（显示整个视频时长）."""
+        if not text or not text.strip():
+            return
+
+        style = style or SubtitleStyle(
+            position="top_center",
+            font_size=48,
+            bold=True,
+            stroke_enabled=True,
+            stroke_width=2.0,
+        )
+        style_name = self.get_or_create_style("TitleStyle", style)
+
+        self._segments.append(
+            SubtitleSegment(
+                start=0.0,
+                end=self.video_duration if self.video_duration > 0 else 9999.0,
+                text=text.strip(),
+                style_name=style_name,
+            )
+        )
+
+    def add_subtitle_text(self, text: str, style: SubtitleStyle | None = None) -> None:
+        """添加整段字幕（显示整个视频时长）."""
+        if not text or not text.strip():
+            return
+
+        style = style or SubtitleStyle()
+        style_name = self.get_or_create_style("SubtitleStyle", style)
+
+        self._segments.append(
+            SubtitleSegment(
+                start=0.0,
+                end=self.video_duration if self.video_duration > 0 else 9999.0,
+                text=text.strip(),
+                style_name=style_name,
+            )
+        )
+
+    def add_timeline_segments(
+        self,
+        segments: list[dict] | list[SubtitleSegment],
+        style: SubtitleStyle | None = None,
+    ) -> None:
+        """添加时间轴字幕片段（ASR 或手动字幕）.
+
+        segments 可以是：
+        - SubtitleSegment 列表
+        - dict 列表，每个 dict 含 start/end/text 字段
+        """
+        if not segments:
+            return
+
+        style = style or SubtitleStyle()
+        style_name = self.get_or_create_style("Default", style)
+
+        for seg in segments:
+            if isinstance(seg, SubtitleSegment):
+                seg.style_name = style_name
+                self._segments.append(seg)
+            elif isinstance(seg, dict):
+                try:
+                    start = float(seg.get("start", 0))
+                    end = float(seg.get("end", 0))
+                    text = str(seg.get("text", ""))
+                    if end > start and text.strip():
+                        self._segments.append(
+                            SubtitleSegment(
+                                start=start,
+                                end=end,
+                                text=text.strip(),
+                                style_name=style_name,
+                            )
+                        )
+                except (TypeError, ValueError):
+                    continue
+
+    def add_asr_timeline(self, timeline: Any, style: SubtitleStyle | None = None) -> None:
+        """从 SubtitleTimeline 对象添加 ASR 字幕."""
+        if not timeline or not hasattr(timeline, "segments") or not timeline.segments:
+            return
+
+        style = style or SubtitleStyle()
+        style_name = self.get_or_create_style("ASRStyle", style)
+
+        for seg in timeline.segments:
+            if hasattr(seg, "start") and hasattr(seg, "end") and hasattr(seg, "text"):
+                if seg.end > seg.start and seg.text.strip():
+                    self._segments.append(
+                        SubtitleSegment(
+                            start=seg.start,
+                            end=seg.end,
+                            text=seg.text.strip(),
+                            style_name=style_name,
+                        )
+                    )
+
+    # ── ASS 文件生成 ──────────────────────────────────────────────────
+
+    def generate_ass(self, output_path: Path) -> Path:
+        """生成 ASS 字幕文件.
+
+        Returns:
+            生成的文件路径；如果没有字幕内容，返回空文件。
+        """
+        if not self._segments:
+            output_path.parent.mkdir(parents=True, exist_ok=True)
+            output_path.write_text("", encoding="utf-8")
+            return output_path
+
+        # 确保至少有 Default 样式
+        if "Default" not in self._styles:
+            self._styles["Default"] = SubtitleStyle()
+
+        # 生成样式行
+        style_lines = []
+        for name, style in self._styles.items():
+            style_lines.append(self._build_ass_style_line(name, style))
+
+        # 生成事件行（按时间排序）
+        self._segments.sort(key=lambda s: s.start)
+        event_lines = []
+        for seg in self._segments:
+            event_lines.append(self._build_ass_event_line(seg))
+
+        # 组装文件
+        ass_content = f"""[Script Info]
+ScriptType: v4.00+
+PlayResX: {self.video_width}
+PlayResY: {self.video_height}
+ScaledBorderAndShadow: yes
+WrapStyle: 2
+Encoding: UTF-8
+
+[V4+ Styles]
+Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
+{chr(10).join(style_lines)}
+
+[Events]
+Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
+{chr(10).join(event_lines)}
+"""
+
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        output_path.write_text(ass_content, encoding="utf-8")
+        return output_path
+
+    def _build_ass_style_line(self, name: str, style: SubtitleStyle) -> str:
+        """构建一条 ASS Style 行."""
+        bold_val = -1 if style.bold else 0
+        italic_val = -1 if style.italic else 0
+
+        # BorderStyle: 1=outline+shadow, 3=opaque box（背景框）
+        if style.background_enabled:
+            border_style = 3
+            back_color = style.ass_background_color
+        else:
+            border_style = 1
+            back_color = style.ass_shadow_color if style.shadow_enabled else style.ass_font_color
+
+        outline_val = style.stroke_width if style.stroke_enabled else 0.0
+        shadow_val = style.shadow_offset_y if style.shadow_enabled else 0
+
+        return (
+            f"Style: {name},{style.font_name},{style.font_size},{style.ass_font_color},"
+            f"&H000000FF,{style.ass_stroke_color},{back_color},"
+            f"{bold_val},{italic_val},0,0,100,100,0,0,"
+            f"{border_style},{outline_val},{shadow_val},{style.alignment},"
+            f"{style.margin_l},{style.margin_r},{style.margin_v},1"
+        )
+
+    def _build_ass_event_line(self, seg: SubtitleSegment) -> str:
+        """构建一条 ASS Dialogue 事件行."""
+        style = self._styles.get(seg.style_name, SubtitleStyle())
+        max_chars = style.max_chars_per_line
+
+        # 自动换行
+        lines = _wrap_text(seg.text, max_chars)
+        display_text = "\\N".join(lines)
+
+        # 动画效果（淡入淡出）
+        effect_tags = ""
+        if style.fade_in > 0 or style.fade_out > 0:
+            fade_in_ms = int(style.fade_in * 1000)
+            fade_out_ms = int(style.fade_out * 1000)
+            effect_tags = f"{{\\fad({fade_in_ms},{fade_out_ms})}}"
+
+        safe_text = _escape_ass_text(display_text)
+        start_time = _format_ass_time(max(0, seg.start))
+        end_time = _format_ass_time(max(seg.start + 0.1, seg.end))
+
+        return f"Dialogue: 0,{start_time},{end_time},{seg.style_name},,0,0,0,," f"{effect_tags}{safe_text}"
+
+    @property
+    def has_subtitles(self) -> bool:
+        """是否有字幕内容."""
+        return len(self._segments) > 0
+
+
+# ── 便捷函数：从 plan.config 快速生成 ASS ────────────────────────────────────
+
+
+def build_subtitles_from_plan(
+    output_path: Path,
+    plan_config: dict,
+    *,
+    video_width: int,
+    video_height: int,
+    video_duration: float,
+    asr_timeline: Any = None,
+) -> Path | None:
+    """从 plan.config 构建字幕 ASS 文件.
+
+    支持的配置项：
+    - title_config: 标题配置（含 text/style）
+    - subtitle_config: 字幕配置（含 text/style）
+    - asr_subtitles: ASR 字幕开关 + 样式
+    - manual_subtitles: 手动字幕片段列表
+
+    Returns:
+        生成的 ASS 文件路径；如果没有任何字幕，返回 None
+    """
+    engine = SubtitleRenderEngine(
+        video_width=video_width,
+        video_height=video_height,
+        video_duration=video_duration,
+    )
+
+    has_any = False
+
+    # 1. 标题
+    title_cfg = plan_config.get("title_config") or {}
+    if isinstance(title_cfg, dict):
+        title_text = str(title_cfg.get("text", ""))
+        title_enabled = title_cfg.get("enabled", True)
+        if title_enabled and title_text.strip():
+            style_dict = title_cfg.get("style") or {}
+            style = SubtitleStyle.from_dict(style_dict)
+            # 标题默认样式：顶部、大字号、粗体
+            if style.position == DEFAULT_POSITION and style.font_size == DEFAULT_FONT_SIZE:
+                style.position = "top_center"
+                style.font_size = 48
+                style.bold = True
+            engine.add_title(title_text, style)
+            has_any = True
+
+    # 2. 静态字幕
+    sub_cfg = plan_config.get("subtitle_config") or {}
+    if isinstance(sub_cfg, dict):
+        sub_text = str(sub_cfg.get("text", ""))
+        sub_enabled = sub_cfg.get("enabled", True)
+        if sub_enabled and sub_text.strip():
+            style_dict = sub_cfg.get("style") or {}
+            style = SubtitleStyle.from_dict(style_dict)
+            engine.add_subtitle_text(sub_text, style)
+            has_any = True
+
+    # 3. ASR 自动字幕
+    asr_cfg = plan_config.get("asr_subtitles") or {}
+    if isinstance(asr_cfg, dict) and asr_cfg.get("enabled", False):
+        if asr_timeline is not None:
+            style_dict = asr_cfg.get("style") or {}
+            style = SubtitleStyle.from_dict(style_dict)
+            engine.add_asr_timeline(asr_timeline, style)
+            has_any = has_any or engine.has_subtitles
+
+    # 4. 手动字幕
+    manual_segs = plan_config.get("manual_subtitles") or []
+    if isinstance(manual_segs, list) and manual_segs:
+        style_dict = (plan_config.get("manual_subtitle_style") or {}) or {}
+        style = SubtitleStyle.from_dict(style_dict)
+        engine.add_timeline_segments(manual_segs, style)
+        has_any = has_any or engine.has_subtitles
+
+    if not has_any:
+        return None
+
+    return engine.generate_ass(output_path)
+
+
+# ── FFmpeg 烧录滤镜生成 ───────────────────────────────────────────────────────
+
+
+def build_subtitle_filter(
+    ass_path: Path | str,
+    *,
+    video_input_label: str = "0:v",
+    output_label: str = "subtitled",
+) -> str:
+    """生成 FFmpeg subtitles 滤镜字符串.
+
+    Args:
+        ass_path: ASS 字幕文件路径
+        video_input_label: 视频输入标签（如 "0:v" 或 "[v_out]"）
+        output_label: 输出标签
+
+    Returns:
+        filter_complex 片段，如 "[0:v]subtitles=xxx.ass[subtitled]"
+    """
+    # FFmpeg subtitles filter 的路径需要转义：
+    # - Windows 路径的 \ → /
+    # - 冒号 : → \:
+    # - 单引号 ' → '\''
+    safe_path = str(ass_path).replace("\\", "/").replace(":", "\\:").replace("'", "'\\''")
+    return f"{video_input_label}subtitles='{safe_path}'[{output_label}]"