"""字幕渲染引擎 — 统一管理字幕样式配置与视频烧录. 与现有模块的关系: - 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 pathlib import Path from typing import Any from video_processing.path_security import PathSecurityError, is_in_allowed_dirs, safe_resolve_path from packages.domain.subtitle_style import ( ALLOWED_SUBTITLE_EXTENSIONS, DEFAULT_FONT_SIZE, DEFAULT_POSITION, SubtitleSegment, SubtitleStyle, ) from packages.domain.subtitle_style import escape_ass_text as _escape_ass_text # noqa: F401 向后兼容导出 from packages.domain.subtitle_style import format_ass_time as _format_ass_time from packages.domain.subtitle_style import hex_to_ass_bgr as _hex_to_ass_bgr # noqa: F401 from packages.domain.subtitle_style import hex_to_ass_color as _hex_to_ass_color # noqa: F401 from packages.domain.subtitle_style import opacity_to_ass_alpha as _opacity_to_ass_alpha # noqa: F401 from packages.domain.subtitle_style import wrap_text as _wrap_text logger = logging.getLogger(__name__) # ── 字幕渲染引擎 ────────────────────────────────────────────────────────────── 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", work_dir: Path | str | None = None, ) -> str: """生成 FFmpeg subtitles 滤镜字符串. Args: ass_path: ASS 字幕文件路径 video_input_label: 视频输入标签(如 "0:v" 或 "[v_out]") output_label: 输出标签 work_dir: 工作目录(必填,用于路径安全校验,防止路径遍历绕过) Returns: filter_complex 片段,如 "[0:v]subtitles=xxx.ass[subtitled]" Raises: PathSecurityError: 字幕路径不安全或 work_dir 未提供 """ # ── 安全校验:字幕文件路径白名单 ── ass_path_str = str(ass_path) if work_dir is None or not str(work_dir).strip(): raise PathSecurityError("work_dir 必须提供,不能为 None 或空") _validate_subtitle_path(ass_path_str, Path(work_dir)) # FFmpeg subtitles filter 的路径需要转义: # - Windows 路径的 \ → / # - 冒号 : → \: # - 单引号 ' → '\'' safe_path = ass_path_str.replace("\\", "/").replace(":", "\\:").replace("'", "'\\''") return f"{video_input_label}subtitles='{safe_path}'[{output_label}]" def _validate_subtitle_path(subtitle_path: str, work_dir: Path) -> None: """校验字幕文件路径安全性. 规则: - 必须是本地路径(不支持远程URL字幕) - local:// schema → 必须在 work_dir 内 - 相对路径 → 必须在 work_dir 内 - 绝对路径 → 必须在允许目录白名单内 - 扩展名必须是字幕格式 Raises: PathSecurityError: 路径不安全 """ if not subtitle_path or not isinstance(subtitle_path, str): raise PathSecurityError("字幕路径不能为空") # 不允许远程URL字幕(subtitles滤镜不支持远程加载,且有SSRF风险) if subtitle_path.startswith(("http://", "https://", "oss://")): raise PathSecurityError("不允许使用远程URL字幕文件") is_abs = subtitle_path.startswith("/") and not subtitle_path.startswith("local://") resolved_path = safe_resolve_path( subtitle_path, work_dir, allow_outside=is_abs, allowed_extensions=ALLOWED_SUBTITLE_EXTENSIONS, ) # 绝对路径额外检查白名单目录(用realpath规范化后的真实路径比较,防止 ../ 遍历绕过) if is_abs: resolved_work_dir = work_dir.resolve() try: resolved_path.relative_to(resolved_work_dir) except ValueError as _e: if not is_in_allowed_dirs(resolved_path): raise PathSecurityError(f"字幕路径不在允许目录内: {subtitle_path[:80]}") from _e