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(

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,

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):

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}"

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） ─────────────────────────────────────────────

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 行")

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。"""