Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 25873c0b89 | |||
| 3d9e084a32 | |||
| 486fa7b20d | |||
| 17761124c2 |
@@ -52,6 +52,7 @@ from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_fr
|
||||
from video_processing.tts_engine import TtsEngine
|
||||
from video_processing.watermark_engine import WatermarkConfig, WatermarkEngine
|
||||
|
||||
from packages.domain.ass_subtitle_builder import build_ass_content
|
||||
from packages.domain.render_layer_utils import LAYER_Z_INDEX as _IMPORTED_LAYER_Z_INDEX
|
||||
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
|
||||
@@ -131,6 +132,83 @@ _PIP_SCALE = 0.25 # PiP 占主画面的比例
|
||||
# ── 统一渲染引擎 ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _overlay_title_on_ass(
|
||||
ass_path: Path,
|
||||
*,
|
||||
title_text: str,
|
||||
title_config: dict,
|
||||
video_width: int,
|
||||
video_height: int,
|
||||
video_duration: float,
|
||||
) -> None:
|
||||
"""在已有的 ASS 文件上叠加标题事件。
|
||||
|
||||
用于 ASR 字幕路径:ASR 生成的 ASS 只含字幕事件,此函数将标题
|
||||
作为独立的 TitleStyle + Dialogue 事件追加进去,使标题显示在
|
||||
ASR 字幕之上(封面抽帧时也能看到标题)。
|
||||
|
||||
Args:
|
||||
ass_path: 已有的 ASS 文件路径(由 generate_ass_from_timeline 生成)
|
||||
title_text: 标题文本
|
||||
title_config: 标题样式配置
|
||||
video_width: 视频宽度
|
||||
video_height: 视频高度
|
||||
video_duration: 视频时长
|
||||
"""
|
||||
if not title_text or not title_text.strip():
|
||||
return
|
||||
|
||||
# 生成仅包含标题的 ASS 内容
|
||||
title_only_content = build_ass_content(
|
||||
video_width=video_width,
|
||||
video_height=video_height,
|
||||
video_duration=video_duration,
|
||||
title_text=title_text,
|
||||
title_config=title_config,
|
||||
)
|
||||
if not title_only_content:
|
||||
return
|
||||
|
||||
# 从 title_only_content 中提取 TitleStyle 行和标题 Dialogue 行
|
||||
title_style_line = None
|
||||
title_dialogue_line = None
|
||||
for line in title_only_content.splitlines():
|
||||
if line.startswith("Style: TitleStyle"):
|
||||
title_style_line = line
|
||||
elif "TitleStyle" in line and line.startswith("Dialogue:"):
|
||||
title_dialogue_line = line
|
||||
|
||||
if not title_style_line or not title_dialogue_line:
|
||||
logger.warning("标题 ASS 内容解析失败,跳过叠加")
|
||||
return
|
||||
|
||||
# 读取现有 ASS 文件
|
||||
existing_content = ass_path.read_text(encoding="utf-8")
|
||||
|
||||
# 插入 TitleStyle 到 [V4+ Styles] 段(最后一个 Style: 行之后)
|
||||
# 插入标题 Dialogue 到 [Events] 段(Format 行之后)
|
||||
lines = existing_content.splitlines()
|
||||
last_style_idx = -1
|
||||
events_format_idx = -1
|
||||
for i, line in enumerate(lines):
|
||||
if line.startswith("Style:"):
|
||||
last_style_idx = i
|
||||
if line.startswith("Format: Layer,"):
|
||||
events_format_idx = i
|
||||
|
||||
if last_style_idx >= 0:
|
||||
lines.insert(last_style_idx + 1, title_style_line)
|
||||
# events_format_idx 需要 +1 因为插入了一行
|
||||
events_format_idx += 1
|
||||
|
||||
# 2. 在 Events Format 行之后、第一个 Dialogue 之前插入标题 Dialogue
|
||||
# 标题应该显示在整个视频时长,放在最前面(最先渲染,在底层)
|
||||
if events_format_idx >= 0:
|
||||
lines.insert(events_format_idx + 1, title_dialogue_line)
|
||||
|
||||
ass_path.write_text("\n".join(lines), encoding="utf-8")
|
||||
|
||||
|
||||
class UnifiedRenderService:
|
||||
"""统一渲染引擎。
|
||||
|
||||
@@ -515,14 +593,71 @@ class UnifiedRenderService:
|
||||
timeline.segment_count,
|
||||
video_duration,
|
||||
)
|
||||
# ASR 路径也需要叠加标题(标题作为独立 ASS Event 追加到 ASR 字幕之上)
|
||||
# 用独立 try-except 包裹,避免叠加失败时覆盖已生成的 ASR 数据
|
||||
if has_title:
|
||||
try:
|
||||
_overlay_title_on_ass(
|
||||
ass_path,
|
||||
title_text=title_text,
|
||||
title_config=title_cfg,
|
||||
video_width=self.output_width,
|
||||
video_height=self.output_height,
|
||||
video_duration=video_duration,
|
||||
)
|
||||
logger.info(
|
||||
"ASR字幕叠加标题: plan_id=%s title=%s",
|
||||
self.plan.id,
|
||||
title_text[:30],
|
||||
)
|
||||
except Exception:
|
||||
logger.warning(
|
||||
"ASR字幕叠加标题失败,保留纯ASR字幕: plan_id=%s",
|
||||
self.plan.id,
|
||||
exc_info=True,
|
||||
)
|
||||
return ass_path
|
||||
else:
|
||||
# ASR 无结果,不生成字幕
|
||||
# ASR 无结果:如果有标题,仍然生成标题 ASS
|
||||
if has_title:
|
||||
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,
|
||||
)
|
||||
logger.info(
|
||||
"ASR无结果但生成标题: plan_id=%s title=%s",
|
||||
self.plan.id,
|
||||
title_text[:30],
|
||||
)
|
||||
return ass_path
|
||||
logger.info("ASR自动字幕无识别结果,跳过字幕: plan_id=%s", self.plan.id)
|
||||
return None
|
||||
except Exception:
|
||||
# ASR 失败降级:不生成字幕,不阻断主流程
|
||||
logger.warning("ASR自动字幕生成失败,跳过字幕", exc_info=True)
|
||||
# ASR 失败降级:如果有标题,仍然生成标题 ASS
|
||||
if has_title:
|
||||
try:
|
||||
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,
|
||||
)
|
||||
logger.info(
|
||||
"ASR失败但生成标题: plan_id=%s title=%s",
|
||||
self.plan.id,
|
||||
title_text[:30],
|
||||
)
|
||||
return ass_path
|
||||
except Exception:
|
||||
logger.warning("ASR失败后标题生成也失败", exc_info=True)
|
||||
else:
|
||||
logger.warning("ASR自动字幕生成失败,跳过字幕", exc_info=True)
|
||||
return None
|
||||
|
||||
# 静态字幕模式(原有逻辑)
|
||||
|
||||
@@ -1205,6 +1205,7 @@ def _render_video(
|
||||
plan_cfg["voice_id"] = voice_ids[0]
|
||||
subtitle_cfg = plan_cfg.get("subtitle", {}) or {}
|
||||
subtitle_cfg["auto_generated"] = True
|
||||
subtitle_cfg["enabled"] = True # 确保 ASR 字幕路径被触发,标题叠加也依赖此路径
|
||||
plan_cfg["subtitle"] = subtitle_cfg
|
||||
virtual_plan.config = plan_cfg
|
||||
logger.info(
|
||||
|
||||
@@ -0,0 +1,242 @@
|
||||
"""测试 ASR 字幕路径的标题叠加功能。
|
||||
|
||||
验证:
|
||||
1. _overlay_title_on_ass 函数正确地将标题事件追加到 ASR 生成的 ASS 文件中
|
||||
2. _maybe_generate_ass 在 ASR 路径中正确叠加标题
|
||||
3. ASR 无结果但有标题时,仍然生成标题 ASS
|
||||
4. ASR 失败但有标题时,降级生成标题 ASS
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from apps.worker.video_processing.unified_render_service import _overlay_title_on_ass
|
||||
|
||||
|
||||
class TestOverlayTitleOnAss:
|
||||
"""_overlay_title_on_ass 函数测试"""
|
||||
|
||||
def test_overlay_title_adds_style_and_dialogue(self, tmp_path):
|
||||
"""标题 Style 和 Dialogue 正确插入 ASS 文件"""
|
||||
# 准备一个模拟 ASR 生成的 ASS 文件
|
||||
ass_content = """[Script Info]
|
||||
ScriptType: v4.00+
|
||||
PlayResX: 1280
|
||||
PlayResY: 720
|
||||
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
|
||||
Style: Default,思源黑体,24,&H00FFFFFF,&H000000FF,&H00000000,&H00000000,-1,0,0,0,100,100,0,0,1,1.5,0,2,40,40,60,1
|
||||
|
||||
[Events]
|
||||
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
|
||||
Dialogue: 0,0:00:00.00,0:00:10.00,Default,,0,0,0,,这是ASR字幕
|
||||
"""
|
||||
ass_path = tmp_path / "test.ass"
|
||||
ass_path.write_text(ass_content, encoding="utf-8")
|
||||
|
||||
# 叠加标题
|
||||
_overlay_title_on_ass(
|
||||
ass_path,
|
||||
title_text="测试标题",
|
||||
title_config={"position": "top", "font": "思源黑体", "size": 48, "color": "#ffffff"},
|
||||
video_width=1280,
|
||||
video_height=720,
|
||||
video_duration=10.0,
|
||||
)
|
||||
|
||||
# 验证结果
|
||||
result = ass_path.read_text(encoding="utf-8")
|
||||
assert "Style: TitleStyle" in result, "TitleStyle 应被插入"
|
||||
assert "测试标题" in result, "标题文本应出现在 Dialogue 中"
|
||||
# 原有的 ASR 字幕应该保留
|
||||
assert "这是ASR字幕" in result, "原有 ASR 字幕应保留"
|
||||
# TitleStyle 应该在 Default Style 之后
|
||||
lines = result.splitlines()
|
||||
style_lines = [i for i, ln in enumerate(lines) if ln.startswith("Style:")]
|
||||
assert len(style_lines) >= 2, "应有至少两个 Style 行"
|
||||
|
||||
def test_overlay_title_empty_text_noop(self, tmp_path):
|
||||
"""空标题文本时不修改 ASS 文件"""
|
||||
ass_content = "[Script Info]\n\n[V4+ Styles]\nStyle: Default,test\n\n[Events]\nFormat: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text\nDialogue: 0,0:00:00.00,0:00:10.00,Default,,0,0,0,,test\n"
|
||||
ass_path = tmp_path / "test.ass"
|
||||
ass_path.write_text(ass_content, encoding="utf-8")
|
||||
|
||||
_overlay_title_on_ass(
|
||||
ass_path,
|
||||
title_text="",
|
||||
title_config={},
|
||||
video_width=1280,
|
||||
video_height=720,
|
||||
video_duration=10.0,
|
||||
)
|
||||
|
||||
result = ass_path.read_text(encoding="utf-8")
|
||||
assert "TitleStyle" not in result, "空标题不应添加 TitleStyle"
|
||||
|
||||
def test_overlay_title_preserves_asr_events(self, tmp_path):
|
||||
"""叠加标题后 ASR 字幕事件保持不变"""
|
||||
ass_content = """[Script Info]
|
||||
ScriptType: v4.00+
|
||||
PlayResX: 1920
|
||||
PlayResY: 1080
|
||||
|
||||
[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
|
||||
Style: Default,思源黑体,24,&H00FFFFFF,&H000000FF,&H00000000,&H00000000,-1,0,0,0,100,100,0,0,1,1.5,0,2,40,40,60,1
|
||||
|
||||
[Events]
|
||||
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
|
||||
Dialogue: 0,0:00:00.50,0:00:03.00,Default,,0,0,0,,第一段字幕
|
||||
Dialogue: 0,0:00:03.50,0:00:06.00,Default,,0,0,0,,第二段字幕
|
||||
Dialogue: 0,0:00:06.50,0:00:10.00,Default,,0,0,0,,第三段字幕
|
||||
"""
|
||||
ass_path = tmp_path / "test.ass"
|
||||
ass_path.write_text(ass_content, encoding="utf-8")
|
||||
|
||||
_overlay_title_on_ass(
|
||||
ass_path,
|
||||
title_text="我的标题",
|
||||
title_config={"position": "top", "size": 48},
|
||||
video_width=1920,
|
||||
video_height=1080,
|
||||
video_duration=10.0,
|
||||
)
|
||||
|
||||
result = ass_path.read_text(encoding="utf-8")
|
||||
# 所有 ASR 字幕段都应保留
|
||||
assert "第一段字幕" in result
|
||||
assert "第二段字幕" in result
|
||||
assert "第三段字幕" in result
|
||||
# 标题也应存在
|
||||
assert "我的标题" in result
|
||||
|
||||
|
||||
class TestMaybeGenerateAssWithTitle:
|
||||
"""_maybe_generate_ass 方法在 ASR 路径中标题叠加的集成测试"""
|
||||
|
||||
def _make_service(self, tmp_path, plan_config, asr_service=None):
|
||||
"""创建简化的 UnifiedRenderService 实例用于测试"""
|
||||
from apps.worker.video_processing.unified_render_service import UnifiedRenderService
|
||||
|
||||
service = object.__new__(UnifiedRenderService)
|
||||
service.plan = MagicMock()
|
||||
service.plan.id = "test_plan_001"
|
||||
service.plan.config = plan_config
|
||||
service.work_dir = tmp_path
|
||||
service.output_width = 1280
|
||||
service.output_height = 720
|
||||
service.asr_service = asr_service
|
||||
service._asr_timeline_cached = False
|
||||
service._asr_timeline_cache = None
|
||||
return service
|
||||
|
||||
def test_asr_path_with_title_overlays_title(self, tmp_path):
|
||||
"""ASR 路径 + 有标题 → 标题叠加到 ASS 文件"""
|
||||
plan_config = {
|
||||
"title": {
|
||||
"text": "测试标题",
|
||||
"enabled": True,
|
||||
"position": "top",
|
||||
"size": 48,
|
||||
"font": "思源黑体",
|
||||
"color": "#ffffff",
|
||||
},
|
||||
"subtitle": {"enabled": True, "auto_generated": True},
|
||||
}
|
||||
|
||||
# Mock ASR service
|
||||
mock_asr = MagicMock()
|
||||
service = self._make_service(tmp_path, plan_config, asr_service=mock_asr)
|
||||
|
||||
# Mock _generate_asr_subtitles to return a timeline with segments
|
||||
mock_timeline = MagicMock()
|
||||
mock_segment = MagicMock()
|
||||
mock_segment.start = 0.0
|
||||
mock_segment.end = 3.0
|
||||
mock_segment.text = "ASR识别的文字"
|
||||
mock_timeline.segments = [mock_segment]
|
||||
mock_timeline.segment_count = 1
|
||||
|
||||
with patch.object(service, "_generate_asr_subtitles", return_value=mock_timeline):
|
||||
result = service._maybe_generate_ass(video_duration=10.0)
|
||||
|
||||
assert result is not None, "应生成 ASS 文件"
|
||||
content = result.read_text(encoding="utf-8")
|
||||
assert "测试标题" in content, "标题应出现在 ASS 文件中"
|
||||
assert "ASR识别的文字" in content, "ASR 字幕也应保留"
|
||||
|
||||
def test_asr_no_result_with_title_generates_title_ass(self, tmp_path):
|
||||
"""ASR 无结果 + 有标题 → 仍然生成标题 ASS"""
|
||||
plan_config = {
|
||||
"title": {"text": "仅标题", "enabled": True, "position": "top", "size": 48},
|
||||
"subtitle": {"enabled": True, "auto_generated": True},
|
||||
}
|
||||
|
||||
mock_asr = MagicMock()
|
||||
service = self._make_service(tmp_path, plan_config, asr_service=mock_asr)
|
||||
|
||||
# Mock ASR returns empty timeline
|
||||
mock_timeline = MagicMock()
|
||||
mock_timeline.segments = []
|
||||
|
||||
with patch.object(service, "_generate_asr_subtitles", return_value=mock_timeline):
|
||||
result = service._maybe_generate_ass(video_duration=10.0)
|
||||
|
||||
assert result is not None, "有标题时应生成 ASS 文件"
|
||||
content = result.read_text(encoding="utf-8")
|
||||
assert "仅标题" in content, "标题应出现在 ASS 文件中"
|
||||
|
||||
def test_asr_failure_with_title_generates_title_ass(self, tmp_path):
|
||||
"""ASR 失败 + 有标题 → 降级生成标题 ASS"""
|
||||
plan_config = {
|
||||
"title": {"text": "降级标题", "enabled": True, "position": "top", "size": 48},
|
||||
"subtitle": {"enabled": True, "auto_generated": True},
|
||||
}
|
||||
|
||||
mock_asr = MagicMock()
|
||||
service = self._make_service(tmp_path, plan_config, asr_service=mock_asr)
|
||||
|
||||
# Mock ASR raises exception
|
||||
with patch.object(service, "_generate_asr_subtitles", side_effect=RuntimeError("ASR error")):
|
||||
result = service._maybe_generate_ass(video_duration=10.0)
|
||||
|
||||
assert result is not None, "ASR 失败但有标题时应生成 ASS 文件"
|
||||
content = result.read_text(encoding="utf-8")
|
||||
assert "降级标题" in content, "标题应出现在降级 ASS 文件中"
|
||||
|
||||
def test_overlay_failure_preserves_asr_data(self, tmp_path):
|
||||
"""_overlay_title_on_ass 抛异常时,ASR 生成的 ASS 文件应保留并返回"""
|
||||
plan_config = {
|
||||
"title": {"text": "测试标题", "enabled": True, "position": "top", "size": 48},
|
||||
"subtitle": {"enabled": True, "auto_generated": True},
|
||||
}
|
||||
|
||||
mock_asr = MagicMock()
|
||||
service = self._make_service(tmp_path, plan_config, asr_service=mock_asr)
|
||||
|
||||
mock_timeline = MagicMock()
|
||||
mock_segment = MagicMock()
|
||||
mock_segment.start = 0.0
|
||||
mock_segment.end = 3.0
|
||||
mock_segment.text = "ASR识别的文字"
|
||||
mock_timeline.segments = [mock_segment]
|
||||
mock_timeline.segment_count = 1
|
||||
|
||||
with patch.object(service, "_generate_asr_subtitles", return_value=mock_timeline):
|
||||
with patch(
|
||||
"apps.worker.video_processing.unified_render_service._overlay_title_on_ass",
|
||||
side_effect=RuntimeError("模拟叠加标题失败"),
|
||||
):
|
||||
result = service._maybe_generate_ass(video_duration=10.0)
|
||||
|
||||
# 即使 _overlay_title_on_ass 失败,仍返回 ASS 文件
|
||||
assert result is not None, "应返回 ASS 文件路径"
|
||||
assert result.exists(), "ASS 文件应存在"
|
||||
content = result.read_text(encoding="utf-8")
|
||||
assert "ASR识别的文字" in content, "ASR 字幕数据应保留"
|
||||
Reference in New Issue
Block a user