diff --git a/apps/api/app/api/routes/generation_preview.py b/apps/api/app/api/routes/generation_preview.py index 746eac25f..2246df9a1 100755 --- a/apps/api/app/api/routes/generation_preview.py +++ b/apps/api/app/api/routes/generation_preview.py @@ -125,6 +125,10 @@ def _resolve_strategy_id_from_template( mode, template_id, ) + # 画中画已下线，pip/voice_pip 统一映射为 one_take + if mode in ("pip", "voice_pip"): + logger.info("[预览生成] %s → one_take (画中画已下线)", mode) + mode = "one_take" return mode except Exception: logger.debug( @@ -146,6 +150,10 @@ def _resolve_strategy_id_from_template( mode, template_id, ) + # 画中画已下线，pip/voice_pip 统一映射为 one_take + if mode in ("pip", "voice_pip"): + logger.info("[预览生成] %s → one_take (画中画已下线)", mode) + mode = "one_take" return mode except Exception: logger.warning( diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py index 350d20498..795514544 100755 --- a/apps/api/app/api/routes/generation_tasks.py +++ b/apps/api/app/api/routes/generation_tasks.py @@ -271,13 +271,19 @@ def create_generation_task( detail="系统繁忙，请稍后再试", ) from e + # 画中画已下线：strategy_id 中的 pip/voice_pip 统一映射为 one_take + effective_strategy_id = request.strategy_id + if effective_strategy_id in ("pip", "voice_pip"): + logger.info("画中画已下线，strategy_id %s → one_take", effective_strategy_id) + effective_strategy_id = "one_take" + try: for _ in range(count): task = use_case.execute( CreateGenerationTaskCommand( project_id=project_id, asset_library_id=asset_library_id, - strategy_id=request.strategy_id, + strategy_id=effective_strategy_id, voice_library_id=request.voice_library_id, template_id=request.template_id, asset_ids=resolved_asset_ids, diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index fb917d7d1..d2c9ffb83 100644 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -235,10 +235,8 @@ def _build_plan_and_clips_from_task( """根据模式和下载的素材路径，构建虚拟 plan + clips + asset_path_map。 模式 → clip_type 映射： - ONE_TAKE: N 个 main clips - PIP: 1 main + N-1 overlay + ONE_TAKE: N 个 main clips（默认，pip/voice_pip 已统一映射为此模式） VOICE_OVER: N 个 main(config.role=b_roll) - VOICE_PIP: 1 background + 1 corner_voice + N-2 b_roll Returns: (virtual_plan, virtual_clips, asset_path_map) @@ -255,23 +253,14 @@ def _build_plan_and_clips_from_task( path_to_asset_id[p] = asset_id path_duration[p] = probe_duration(p) + # 产品已确认全面下线画中画，pip/voice_pip统一走one_take（顺序拼接） + if mode in ("pip", "voice_pip"): + logger.info("画中画模式已下线，%s 强制映射为 one_take", mode) + mode = "one_take" + clips: list[_VirtualClip] = [] - if mode == "pip": - # 1 main + N-1 overlay - for i, p in enumerate(downloaded_paths): - clip_type = "main" if i == 0 else "overlay" - clips.append( - _VirtualClip( - id=f"vc_{i:03d}", - plan_id=task_id, - clip_type=clip_type, - order=i, - asset_id=path_to_asset_id[p], - duration=path_duration[p], - ) - ) - elif mode == "voice_over": + if mode == "voice_over": # N 个 main(config.role=b_roll) for i, p in enumerate(downloaded_paths): clips.append( @@ -285,25 +274,6 @@ def _build_plan_and_clips_from_task( config={"role": "b_roll"}, ) ) - elif mode == "voice_pip": - # 1 background + 1 corner_voice + N-2 b_roll - for i, p in enumerate(downloaded_paths): - if i == 0: - clip_type = "background" - elif i == 1: - clip_type = "corner_voice" - else: - clip_type = "b_roll" - clips.append( - _VirtualClip( - id=f"vc_{i:03d}", - plan_id=task_id, - clip_type=clip_type, - order=i, - asset_id=path_to_asset_id[p], - duration=path_duration[p], - ) - ) else: # ONE_TAKE (default): N 个 main clips for i, p in enumerate(downloaded_paths): diff --git a/packages/domain/ass_subtitle_builder.py b/packages/domain/ass_subtitle_builder.py index 143d56f80..85fa85a9d 100755 --- a/packages/domain/ass_subtitle_builder.py +++ b/packages/domain/ass_subtitle_builder.py @@ -175,6 +175,48 @@ def format_ass_time(seconds: float) -> str: # ── 完整 ASS 内容生成 ───────────────────────────────────────────────────────── + +def _wrap_title_text( + text: str, + video_width: int, + font_size: int, + margin_l: int = TITLE_MARGIN_SIDE, + margin_r: int = TITLE_MARGIN_SIDE, +) -> str: + """根据视频宽度和字号自动换行标题文本。 + + 中文字符按 font_size 像素宽度估算，英文/数字按半角估算。 + 超过可用宽度时插入 \\N (ASS 硬换行)。 + """ + if not text or video_width <= 0 or font_size <= 0: + return text + + available_width = video_width - margin_l - margin_r + if available_width <= 0: + return text + + lines: list[str] = [] + current_line = "" + current_width = 0.0 + + for ch in text: + # CJK 字符按全角估算，其他按半角 + char_width = float(font_size) if ord(ch) > 0x2E80 else font_size * 0.55 + + if current_width + char_width > available_width and current_line: + lines.append(current_line) + current_line = ch + current_width = char_width + else: + current_line += ch + current_width += char_width + + if current_line: + lines.append(current_line) + + return "\\N".join(lines) + + def build_ass_content( *, video_width: int, @@ -248,7 +290,11 @@ def build_ass_content( ) ) - safe_title_text = escape_ass_text(title_text) + # 根据视频宽度和字号自动换行标题，防止超出画面 + # 先 escape 特殊字符，再插入换行符 \N，避免顺序颠倒导致 \N 被转义 + title_font_size = int(title_config.get("size", 48)) + safe_title_text_raw = escape_ass_text(title_text) + safe_title_text = _wrap_title_text(safe_title_text_raw, video_width, title_font_size) events.append( "Dialogue: 0,0:00:00.00," f"{format_ass_time(video_duration)}," "TitleStyle,,0,0,0,," f"{safe_title_text}" diff --git a/tests/integration/test_four_mode_rendering.py b/tests/integration/test_four_mode_rendering.py index eaa93506d..581aa0729 100644 --- a/tests/integration/test_four_mode_rendering.py +++ b/tests/integration/test_four_mode_rendering.py @@ -1,7 +1,8 @@ -"""四模式渲染集成测试. +"""剪辑模式渲染集成测试. -验证 4 种剪辑模式（ONE_TAKE / PIP / VOICE_OVER / VOICE_PIP）通过 +验证剪辑模式（ONE_TAKE / VOICE_OVER）通过 _build_plan_and_clips_from_task + UnifiedRenderService 的完整渲染流程。 +注：PIP / VOICE_PIP 已下线，统一映射为 ONE_TAKE。 需要 ffmpeg 可用；CI 无 ffmpeg 时自动跳过。 """ @@ -109,14 +110,13 @@ class TestBuildPlanAndClips: assert all(c.clip_type == "main" for c in clips) assert len(asset_map) == 3 - def test_pip_mode(self): + def test_pip_mode_maps_to_one_take(self): + """PIP 已下线，映射为 one_take → 全部 main clips。""" paths = self._make_paths(3) plan, clips, asset_map = _build_plan_and_clips_from_task("t2", paths, "pip") assert len(clips) == 3 - assert clips[0].clip_type == "main" - assert clips[1].clip_type == "overlay" - assert clips[2].clip_type == "overlay" + assert all(c.clip_type == "main" for c in clips) def test_voice_over_mode(self): paths = self._make_paths(3) @@ -126,15 +126,13 @@ class TestBuildPlanAndClips: assert all(c.clip_type == "main" for c in clips) assert all(c.config.get("role") == "b_roll" for c in clips) - def test_voice_pip_mode(self): + def test_voice_pip_mode_maps_to_one_take(self): + """VOICE_PIP 已下线，映射为 one_take → 全部 main clips。""" paths = self._make_paths(4) plan, clips, asset_map = _build_plan_and_clips_from_task("t4", paths, "voice_pip") assert len(clips) == 4 - assert clips[0].clip_type == "background" - assert clips[1].clip_type == "corner_voice" - assert clips[2].clip_type == "b_roll" - assert clips[3].clip_type == "b_roll" + assert all(c.clip_type == "main" for c in clips) def test_unknown_mode_defaults_to_one_take(self): paths = self._make_paths(2) @@ -166,13 +164,13 @@ class TestFourModeLayerGrouping: roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips} assert roles == {"main"} - def test_pip_layers(self): - """PIP: 1 main + 2 overlay → main + overlay。""" + def test_pip_layers_now_one_take(self): + """PIP 已下线 → one_take: 3 main → 1 main layer。""" paths = [Path(f"/tmp/pip_{i}.mp4") for i in range(3)] _, clips, _ = _build_plan_and_clips_from_task("pip", paths, "pip") roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips} - assert roles == {"main", "overlay"} + assert roles == {"main"} def test_voice_over_layers(self): """VOICE_OVER: 3 main(b_roll) → broll。""" @@ -182,13 +180,13 @@ class TestFourModeLayerGrouping: roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips} assert roles == {"broll"} - def test_voice_pip_layers(self): - """VOICE_PIP: 1 bg + 1 corner_voice + 2 b_roll → 3 个图层。""" + def test_voice_pip_layers_now_one_take(self): + """VOICE_PIP 已下线 → one_take: 4 main → main layer。""" paths = [Path(f"/tmp/vpip_{i}.mp4") for i in range(4)] _, clips, _ = _build_plan_and_clips_from_task("vpip", paths, "voice_pip") roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips} - assert roles == {"background", "corner_voice", "broll"} + assert roles == {"main"} # ── 端到端渲染测试（需要 ffmpeg） ───────────────────────────────────────────── diff --git a/tests/unit/test_ass_subtitle_builder.py b/tests/unit/test_ass_subtitle_builder.py index 971255014..2772e7948 100755 --- a/tests/unit/test_ass_subtitle_builder.py +++ b/tests/unit/test_ass_subtitle_builder.py @@ -8,6 +8,7 @@ from packages.domain.ass_subtitle_builder import ( TITLE_MARGIN_BOTTOM, TITLE_MARGIN_SIDE, TITLE_MARGIN_TOP, + _wrap_title_text, build_ass_content, build_ass_style, escape_ass_text, @@ -446,3 +447,64 @@ class TestConstants: def test_title_margin_side(self): assert TITLE_MARGIN_SIDE == 40 + +# ── 标题自动换行 ────────────────────────────────────────────────────────────── + + +class TestWrapTitleText: + """测试标题自动换行逻辑。""" + + def test_short_title_no_wrap(self): + """短标题不需要换行。""" + result = _wrap_title_text("你好世界", video_width=1080, font_size=48) + assert "\\N" not in result + assert result == "你好世界" + + def test_long_title_wraps(self): + """长标题应该被换行。""" + # 1080p竖屏 480px宽，字号48，边距40*2 + # 可用宽度 = 480 - 40 - 40 = 400 + # 每个中文字符 = 48px，最多约 8 个字符一行 + long_title = "这是一个非常非常长的标题需要换行处理" + result = _wrap_title_text(long_title, video_width=480, font_size=48) + assert "\\N" in result + lines = result.split("\\N") + assert len(lines) >= 2 + + def test_empty_text(self): + """空文本直接返回。""" + assert _wrap_title_text("", 1080, 48) == "" + + def test_english_half_width(self): + """英文字符按半角计算。""" + # 英文宽度 = 48 * 0.55 = 26.4px + # 可用宽度 = 480 - 80 = 400, 约15个字符 + result = _wrap_title_text("a" * 20, video_width=480, font_size=48) + # 20个英文字符 * 26.4 = 528 > 400，应该换行 + assert "\\N" in result + + def test_zero_width(self): + """视频宽度为0时直接返回原文。""" + assert _wrap_title_text("测试", 0, 48) == "测试" + + def test_zero_font_size(self): + """字号为0时直接返回原文。""" + assert _wrap_title_text("测试", 1080, 0) == "测试" + + def test_build_ass_content_integration(self): + """集成测试：build_ass_content 中的标题应该自动换行。""" + long_title = "这是一段非常长的标题文字用于测试自动换行功能是否正常工作" + content = build_ass_content( + video_width=480, + video_height=854, + video_duration=10.0, + title_text=long_title, + title_config={"size": 48, "position": "top"}, + ) + # 检查 Dialogue 行中包含 \N 换行 + for line in content.split("\n"): + if "Dialogue" in line and "TitleStyle" in line: + assert "\\N" in line, f"标题应该包含换行符: {line}" + break + else: + pytest.fail("未找到 TitleStyle Dialogue 行") diff --git a/tests/unit/test_generation_preview.py b/tests/unit/test_generation_preview.py old mode 100755 new mode 100644 index 0f02fb8e3..ac71c1f1d --- a/tests/unit/test_generation_preview.py +++ b/tests/unit/test_generation_preview.py @@ -1300,14 +1300,14 @@ class TestResolveStrategyIdFromTemplate: from app.api.routes.generation_preview import _resolve_strategy_id_from_template mock_new_template = MagicMock() - mock_new_template.editing_mode = "pip" + mock_new_template.editing_mode = "standard" mock_db = MagicMock() with patch("app.api.routes.generation_preview.SQLAlchemyEditTemplateRepository") as MockNewRepo: MockNewRepo.return_value.get.return_value = mock_new_template result = _resolve_strategy_id_from_template("tpl_001", mock_db, "user_1") - assert result == "pip" + assert result == "standard" def test_new_template_not_found_fallback_to_old(self): """新模板系统未找到 → fallback 旧模板系统返回 mode。""" @@ -1396,7 +1396,7 @@ class TestResolveStrategyIdFromTemplate: assert result == "one_take" def test_voice_pip_mode(self): - """模板 editing_mode=voice_pip → 返回 'voice_pip'。""" + """模板 editing_mode=voice_pip → 返回 one_take（画中画已下线，统一映射为one_take）。""" from app.api.routes.generation_preview import _resolve_strategy_id_from_template mock_new_template = MagicMock() @@ -1407,14 +1407,14 @@ class TestResolveStrategyIdFromTemplate: MockNewRepo.return_value.get.return_value = mock_new_template result = _resolve_strategy_id_from_template("tpl_voice", mock_db, "user_1") - assert result == "voice_pip" + assert result == "one_take" class TestPreviewRoutePassesStrategyId: """验证预览路由正确传递 strategy_id 到 CreateGenerationTaskCommand。""" def test_strategy_id_from_new_template(self): - """预览路由从新模板读取 strategy_id=pip。""" + """预览路由从新模板读取 strategy_id=standard。""" from app.api.routes.generation_preview import create_preview_generation_task from app.schemas.generation_task import CreatePreviewGenerationTaskRequest @@ -1424,7 +1424,7 @@ class TestPreviewRoutePassesStrategyId: mock_db = MagicMock() mock_new_template = MagicMock() - mock_new_template.editing_mode = "pip" + mock_new_template.editing_mode = "standard" task = _make_task() @@ -1456,7 +1456,7 @@ class TestPreviewRoutePassesStrategyId: call_args = MockUC.return_value.execute.call_args cmd = call_args[0][0] - assert cmd.strategy_id == "pip", f"Expected strategy_id='pip', got '{cmd.strategy_id}'" + assert cmd.strategy_id == "standard", f"Expected strategy_id='standard', got '{cmd.strategy_id}'" def test_strategy_id_fallback_to_old_template(self): """新模板无数据时，从旧模板读取 strategy_id=standard。"""