apps/worker/video_processing/dedup_helpers.py

@@ -28,6 +28,7 @@ def create_video_record_and_dedup(
     width: int = 1280,
     height: int = 720,
     fps: float = 25.0,
+    name: str = "",
 ) -> int:
     """创建 GeneratedVideo 记录，计算指纹并执行查重（历史 + 批次）。
 
@@ -57,11 +58,13 @@ def create_video_record_and_dedup(
 
     try:
         video_id = uuid4().hex
+        # 使用传入的名称，没有则 fallback 到默认命名
+        video_name = name.strip() if name else f"generated-{generation_task_id[:8]}.mp4"
         generated_video = GeneratedVideo(
             id=video_id,
             project_id=project_id,
             generation_task_id=generation_task_id,
-            name=f"generated-{generation_task_id[:8]}.mp4",
+            name=video_name,
             file_url=file_url,
             file_size=file_size,
             duration=duration,

apps/worker/worker_app/tasks/edit_plan_generation.py

@@ -141,6 +141,10 @@ def _finalize_render_success(
     project_id = plan.project_id or ""
     batch_id = plan.config.get("batch_id", "")
     mode = plan.config.get("mode", "edit_plan")
+    # 从 plan.config.title.text 读取视频名称
+    plan_config = plan.config or {}
+    title_cfg = plan_config.get("title", {}) or {}
+    video_name = (title_cfg.get("text") or "").strip() or f"generated-{generation_task_id[:8]}.mp4"
     if generation_task_id:
         try:
             create_video_record_and_dedup(
@@ -156,6 +160,7 @@ def _finalize_render_success(
                 width=width,
                 height=height,
                 fps=OUTPUT_FPS,
+                name=video_name,
             )
         except Exception as dedup_err:
             logger.warning("查重失败（不影响渲染结果）: %s", dedup_err)
@@ -395,9 +400,74 @@ def _render_with_legacy(
         _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, error_msg)
         return {"status": "error", "message": error_msg}
 
-    # 获取文件大小
+    # 获取文件大小 + 实际时长
     file_size = output_path.stat().st_size if output_path.exists() else 0
     duration = compose_cmd.estimated_duration or 0.0
+    try:
+        from video_processing.ffmpeg_utils import probe_duration
+
+        actual_duration = probe_duration(str(output_path))
+        if actual_duration > 0:
+            duration = actual_duration
+    except Exception:
+        pass
+
+    # ── 标题/字幕叠加（legacy 引擎补齐） ────────────────────────────────
+    plan_config = plan.config or {}
+    title_cfg = plan_config.get("title", {}) or {}
+    subtitle_cfg = plan_config.get("subtitle", {}) or {}
+    title_text = title_cfg.get("text", "") or ""
+    subtitle_text = subtitle_cfg.get("text", "") or ""
+    title_enabled = title_cfg.get("enabled", True) and bool(title_text.strip())
+    subtitle_enabled = subtitle_cfg.get("enabled", True) and bool(subtitle_text.strip())
+    # ASR 自动字幕 legacy 暂不支持（需要额外 ASR 服务，统一用 unified 引擎）
+    has_subtitle_overlay = title_enabled or subtitle_enabled
+
+    if has_subtitle_overlay and output_path.exists() and duration > 0:
+        try:
+            from video_processing.ffmpeg_utils import run_ffmpeg
+            from video_processing.render_subtitles import generate_ass_subtitles
+
+            ass_path = tmpdir_path / f"subtitles_{plan_id}.ass"
+            generate_ass_subtitles(
+                ass_path,
+                video_width=output_width,
+                video_height=output_height,
+                video_duration=duration,
+                title_text=title_text,
+                title_config=title_cfg,
+                subtitle_text=subtitle_text,
+                subtitle_config=subtitle_cfg,
+            )
+            # 用 subtitles 滤镜叠加 ASS 字幕，音频直接 copy
+            subtitled_path = tmpdir_path / f"{plan_id}_subtitled.mp4"
+            # 处理 Windows 路径下的 ass 滤镜转义问题
+            ass_filter_path = str(ass_path).replace("\\", "/").replace(":", r"\:")
+            run_ffmpeg(
+                [
+                    "ffmpeg",
+                    "-y",
+                    "-i",
+                    str(output_path),
+                    "-vf",
+                    f"subtitles='{ass_filter_path}'",
+                    "-c:a",
+                    "copy",
+                    str(subtitled_path),
+                ],
+                timeout=1800,
+            )
+            if subtitled_path.exists() and subtitled_path.stat().st_size > 0:
+                output_path = subtitled_path
+                file_size = subtitled_path.stat().st_size
+                logger.info(
+                    "legacy 标题/字幕叠加完成: plan_id=%s title=%s subtitle=%s",
+                    plan_id,
+                    title_enabled,
+                    subtitle_enabled,
+                )
+        except Exception as sub_err:
+            logger.warning("legacy 标题/字幕叠加失败（不影响主流程）: plan_id=%s err=%s", plan_id, sub_err)
 
     # 上传到 OSS
     storage_key = f"rendered/{plan_id}/output.mp4"

tests/unit/test_generated_video_creation_logic.py

@@ -111,3 +111,49 @@ class TestVideoCreationLogic:
         generation_task_id = ""
         project_id = "proj-456"
         assert bool(generation_task_id) is False
+
+
+class TestVideoNameParameter:
+    """验证成片库视频名称逻辑：用户设置标题时用标题，没设置时用默认命名。"""
+
+    def test_name_from_user_title(self):
+        """用户设置了标题 → 用标题作为视频名称。"""
+        # 验证函数签名包含 name 参数
+        import inspect
+
+        from video_processing.dedup_helpers import create_video_record_and_dedup
+
+        sig = inspect.signature(create_video_record_and_dedup)
+        assert "name" in sig.parameters, "create_video_record_and_dedup 应支持 name 参数"
+
+    def test_generated_video_requires_name(self):
+        """GeneratedVideo.create 要求 name 非空。"""
+        from packages.domain import GeneratedVideo
+
+        with pytest.raises(ValueError, match="name cannot be empty"):
+            GeneratedVideo.create(
+                project_id="proj-1",
+                generation_task_id="task-1",
+                name="",
+                file_url="https://example.com/test.mp4",
+            )
+
+    def test_name_fallback_when_empty(self):
+        """name 为空或纯空格时，调用方应 fallback 到默认命名。"""
+        # 模拟 _finalize_render_success 中的逻辑
+        title_text_empty = ""
+        title_text_spaces = "   "
+        generation_task_id = "task-abcdef123456"
+
+        # 空标题 → fallback
+        video_name_1 = title_text_empty.strip() or f"generated-{generation_task_id[:8]}.mp4"
+        assert video_name_1 == f"generated-{generation_task_id[:8]}.mp4"
+
+        # 纯空格 → fallback
+        video_name_2 = title_text_spaces.strip() or f"generated-{generation_task_id[:8]}.mp4"
+        assert video_name_2 == f"generated-{generation_task_id[:8]}.mp4"
+
+        # 有标题 → 用标题
+        title_text = "我的旅行vlog"
+        video_name_3 = title_text.strip() or f"generated-{generation_task_id[:8]}.mp4"
+        assert video_name_3 == "我的旅行vlog"