450 lines
17 KiB
Python
Executable File
450 lines
17 KiB
Python
Executable File
"""字幕渲染引擎 — 统一管理字幕样式配置与视频烧录.
|
||
|
||
与现有模块的关系：
|
||
- 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
|