"""字幕渲染引擎 — 统一管理字幕样式配置与视频烧录. 与现有模块的关系: - 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.path_security import PathSecurityError, is_in_allowed_dirs, safe_resolve_path from video_processing.render_subtitles import generate_ass_subtitles from video_processing.subtitle_generator import generate_ass_from_timeline logger = logging.getLogger(__name__) # ── 常量 ────────────────────────────────────────────────────────────────────── ALLOWED_SUBTITLE_EXTENSIONS = {".srt", ".ass", ".vtt", ".sub"} # 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", 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