apps/worker/video_processing/dedup_helpers.py

@@ -29,6 +29,7 @@ def create_video_record_and_dedup(
     height: int = 720,
     fps: float = 25.0,
     name: str = "",
+    thumbnail_url: str = "",
 ) -> int:
     """创建 GeneratedVideo 记录，计算指纹并执行查重（历史 + 批次）。
 
@@ -79,17 +80,22 @@ def create_video_record_and_dedup(
         video_repo.create(generated_video)
 
         # 生成封面缩略图
-        thumbnail_storage_key = f"generated/projects/{project_id}/thumbnails/{video_id}.jpg"
-        try:
-            from video_processing.thumbnail_generator import generate_and_upload_thumbnail
+        if thumbnail_url:
+            generated_video.thumbnail_url = thumbnail_url
+            video_repo.update_thumbnail(video_id, thumbnail_url)
+            logger.info("Thumbnail reused (pre-generated) for video %s", video_id)
+        else:
+            thumbnail_storage_key = f"generated/projects/{project_id}/thumbnails/{video_id}.jpg"
+            try:
+                from video_processing.thumbnail_generator import generate_and_upload_thumbnail
 
-            thumbnail_url = generate_and_upload_thumbnail(video_path, thumbnail_storage_key)
-            if thumbnail_url:
-                generated_video.thumbnail_url = thumbnail_url
-                video_repo.update_thumbnail(video_id, thumbnail_url)
-                logger.info("Thumbnail generated for video %s: %s", video_id, thumbnail_url)
-        except Exception as thumb_err:
-            logger.warning("Thumbnail generation failed for %s: %s", video_id, thumb_err)
+                _thumbnail_url = generate_and_upload_thumbnail(video_path, thumbnail_storage_key)
+                if _thumbnail_url:
+                    generated_video.thumbnail_url = _thumbnail_url
+                    video_repo.update_thumbnail(video_id, _thumbnail_url)
+                    logger.info("Thumbnail generated for video %s: %s", video_id, _thumbnail_url)
+            except Exception as thumb_err:
+                logger.warning("Thumbnail generation failed for %s: %s", video_id, thumb_err)
 
         # 计算视频指纹
         deduplicator = VideoDeduplicator()

apps/worker/video_processing/render_adapter.py

@@ -63,6 +63,7 @@ class RenderAdapterResult:
     success: bool
     output_url: str = ""
     output_path: Path | None = None
+    thumbnail_url: str = ""
     duration: float = 0.0
     file_size: int = 0
     width: int = 0
@@ -224,6 +225,22 @@ class RenderAdapter:
             storage_key = f"rendered/{plan_id}/{job_id or plan_id}.mp4"
             output_url = upload_to_oss(result.output_path, storage_key)
 
+            self._report_progress(progress_cb, 90.0, "生成封面缩略图")
+
+            # 5. 生成缩略图（在清理临时目录前）
+            thumbnail_url = ""
+            try:
+                from video_processing.thumbnail_generator import generate_and_upload_thumbnail
+
+                thumb_storage_key = f"rendered/{plan_id}/thumbnail.jpg"
+                thumbnail_url = generate_and_upload_thumbnail(str(result.output_path), thumb_storage_key)
+            except Exception as thumb_err:
+                logger.warning(
+                    "[render-adapter] 缩略图生成失败（不影响主流程）: plan_id=%s error=%s",
+                    plan_id,
+                    thumb_err,
+                )
+
             self._report_progress(progress_cb, 100.0, "渲染完成")
 
             logger.info(
@@ -242,6 +259,7 @@ class RenderAdapter:
                 success=True,
                 output_url=output_url or "",
                 output_path=result.output_path,
+                thumbnail_url=thumbnail_url,
                 duration=result.duration,
                 file_size=result.file_size,
                 width=result.width,

apps/worker/worker_app/tasks/edit_plan_generation.py

@@ -135,6 +135,7 @@ def _finalize_render_success(
     generation_task_id: str,
     output_path: Path,
     engine: str,
+    thumbnail_url: str = "",
 ) -> dict:
     """渲染成功后的统一收尾：查重 + 更新状态 + 返回结果。"""
     # 创建 GeneratedVideo 记录 + 查重
@@ -161,6 +162,7 @@ def _finalize_render_success(
                 height=height,
                 fps=OUTPUT_FPS,
                 name=video_name,
+                thumbnail_url=thumbnail_url,
             )
         except Exception as dedup_err:
             logger.warning("查重失败（不影响渲染结果）: %s", dedup_err)
@@ -279,6 +281,7 @@ def _render_with_unified(
 
     output_path = result.output_path or Path("")
     output_url = result.output_url
+    thumbnail_url = result.thumbnail_url or ""
     storage_key = f"rendered/{plan_id}/output.mp4"
 
     # 用 adapter 返回的 clip 明细（以 adapter 的结果为准）
@@ -303,6 +306,7 @@ def _render_with_unified(
         generation_task_id=generation_task_id,
         output_path=output_path,
         engine="unified",
+        thumbnail_url=thumbnail_url,
     )
 
 
@@ -367,6 +371,23 @@ def _render_with_legacy(
     )
 
     logger.info("执行 FFmpeg (legacy): plan_id=%s cmd=%s", plan_id, " ".join(compose_cmd.command)[:500])
+
+    # 开始渲染，更新进度
+    if generation_task_id:
+        try:
+            gen_task = gen_task_repo.get(generation_task_id)
+            if gen_task and gen_task.progress < 40.0:
+                gen_task.progress = 40.0
+                gen_task.append_log(
+                    stage="render_start",
+                    message="开始FFmpeg渲染（legacy）",
+                    level="INFO",
+                    progress=40.0,
+                )
+                gen_task_repo.update(gen_task)
+        except Exception:
+            pass
+
     try:
         from video_processing.ffmpeg_utils import run_ffmpeg
 
@@ -550,10 +571,42 @@ def _render_with_legacy(
         except Exception as tts_err:
             logger.warning("legacy TTS 配音混音失败（不影响主流程）: plan_id=%s err=%s", plan_id, tts_err)
 
+    # 渲染完成，更新进度
+    if generation_task_id:
+        try:
+            gen_task = gen_task_repo.get(generation_task_id)
+            if gen_task and gen_task.progress < 80.0:
+                gen_task.progress = 80.0
+                gen_task.append_log(
+                    stage="render_done",
+                    message="FFmpeg渲染完成（legacy）",
+                    level="INFO",
+                    progress=80.0,
+                )
+                gen_task_repo.update(gen_task)
+        except Exception:
+            pass
+
     # 上传到 OSS
     storage_key = f"rendered/{plan_id}/output.mp4"
     output_url = upload_to_oss(output_path, storage_key)
 
+    # 上传完成，更新进度
+    if generation_task_id:
+        try:
+            gen_task = gen_task_repo.get(generation_task_id)
+            if gen_task and gen_task.progress < 95.0:
+                gen_task.progress = 95.0
+                gen_task.append_log(
+                    stage="upload_done",
+                    message="OSS上传完成（legacy）",
+                    level="INFO",
+                    progress=95.0,
+                )
+                gen_task_repo.update(gen_task)
+        except Exception:
+            pass
+
     return _finalize_render_success(
         plan=plan,
         plan_repo=plan_repo,

tests/unit/test_generated_video_creation_logic.py

@@ -157,3 +157,183 @@ class TestVideoNameParameter:
         title_text = "我的旅行vlog"
         video_name_3 = title_text.strip() or f"generated-{generation_task_id[:8]}.mp4"
         assert video_name_3 == "我的旅行vlog"
+
+
+class TestThumbnailInDedupHelpers:
+    """验证 dedup_helpers 中缩略图相关逻辑（全 mock，不依赖 cv2）。"""
+
+    @classmethod
+    def setup_class(cls):
+        """用 mock 模块替代需要 cv2 的 dedup 模块，避免导入失败。
+
+        注意：光往 sys.modules 塞不够，patch() 走属性访问链，
+        必须给 video_processing 包设置对应子模块属性。
+        """
+        import sys
+        from unittest.mock import MagicMock
+
+        # 先 mock 掉 cv2
+        if "cv2" not in sys.modules:
+            sys.modules["cv2"] = MagicMock()
+
+        # mock video_processing.dedup
+        mock_dedup = MagicMock()
+        mock_dedup.VideoDeduplicator = MagicMock()
+        sys.modules["video_processing.dedup"] = mock_dedup
+
+        # mock video_processing.thumbnail_generator
+        mock_thumb = MagicMock()
+        mock_thumb.generate_and_upload_thumbnail = MagicMock()
+        sys.modules["video_processing.thumbnail_generator"] = mock_thumb
+
+        # 关键：给 video_processing 包设置子模块属性，让 patch() 能通过属性访问找到
+        import video_processing
+
+        video_processing.dedup = mock_dedup
+        video_processing.thumbnail_generator = mock_thumb
+
+    def test_pre_generated_thumbnail_url_is_reused(self):
+        """传入 thumbnail_url 时直接复用，不调用 generate_and_upload_thumbnail。"""
+        from video_processing.dedup_helpers import create_video_record_and_dedup
+
+        engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(engine)
+        Session = sessionmaker(bind=engine)
+        session = Session()
+
+        pre_thumb_url = "https://oss.example.com/pre-thumb.jpg"
+
+        try:
+            with patch("video_processing.dedup.VideoDeduplicator") as mock_dedup_cls:
+                mock_dedup = mock_dedup_cls.return_value
+                mock_dedup.compute_fingerprint.return_value = MagicMock(to_dict=lambda: {})
+                mock_dedup.check_duplicate.return_value = None
+                mock_dedup.check_batch_duplicate.return_value = None
+
+                with patch("video_processing.thumbnail_generator.generate_and_upload_thumbnail") as mock_gen:
+                    result = create_video_record_and_dedup(
+                        generation_task_id="task-thumb-reuse",
+                        project_id="proj-1",
+                        batch_id="",
+                        file_url="https://oss.example.com/video.mp4",
+                        file_size=1024,
+                        duration=10.0,
+                        video_path="/tmp/fake.mp4",
+                        mode="edit_plan",
+                        session=session,
+                        width=1280,
+                        height=720,
+                        fps=25.0,
+                        thumbnail_url=pre_thumb_url,
+                    )
+
+            assert result == 1
+            # 预生成缩略图时不应调用 generate_and_upload_thumbnail
+            mock_gen.assert_not_called()
+
+            from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel
+
+            video = session.query(GeneratedVideoModel).filter_by(generation_task_id="task-thumb-reuse").first()
+            assert video is not None
+            assert video.thumbnail_url == pre_thumb_url
+        finally:
+            session.close()
+
+    def test_thumbnail_generated_when_not_provided(self):
+        """未传 thumbnail_url 时调用 generate_and_upload_thumbnail 生成。"""
+        from video_processing.dedup_helpers import create_video_record_and_dedup
+
+        engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(engine)
+        Session = sessionmaker(bind=engine)
+        session = Session()
+
+        generated_thumb_url = "https://oss.example.com/generated-thumb.jpg"
+
+        try:
+            with patch("video_processing.dedup.VideoDeduplicator") as mock_dedup_cls:
+                mock_dedup = mock_dedup_cls.return_value
+                mock_dedup.compute_fingerprint.return_value = MagicMock(to_dict=lambda: {})
+                mock_dedup.check_duplicate.return_value = None
+                mock_dedup.check_batch_duplicate.return_value = None
+
+                with patch(
+                    "video_processing.thumbnail_generator.generate_and_upload_thumbnail",
+                    return_value=generated_thumb_url,
+                ) as mock_gen:
+                    result = create_video_record_and_dedup(
+                        generation_task_id="task-thumb-gen",
+                        project_id="proj-1",
+                        batch_id="",
+                        file_url="https://oss.example.com/video.mp4",
+                        file_size=1024,
+                        duration=10.0,
+                        video_path="/tmp/fake.mp4",
+                        mode="edit_plan",
+                        session=session,
+                        width=1280,
+                        height=720,
+                        fps=25.0,
+                    )
+
+            assert result == 1
+            # 应调用一次缩略图生成
+            mock_gen.assert_called_once()
+            # 验证参数：video_path 和 storage_key
+            call_args = mock_gen.call_args
+            assert call_args[0][0] == "/tmp/fake.mp4"
+            assert "thumbnails" in call_args[0][1]
+
+            from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel
+
+            video = session.query(GeneratedVideoModel).filter_by(generation_task_id="task-thumb-gen").first()
+            assert video is not None
+            assert video.thumbnail_url == generated_thumb_url
+        finally:
+            session.close()
+
+    def test_thumbnail_generation_failure_does_not_block(self):
+        """缩略图生成失败不影响主流程。"""
+        from video_processing.dedup_helpers import create_video_record_and_dedup
+
+        engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(engine)
+        Session = sessionmaker(bind=engine)
+        session = Session()
+
+        try:
+            with patch("video_processing.dedup.VideoDeduplicator") as mock_dedup_cls:
+                mock_dedup = mock_dedup_cls.return_value
+                mock_dedup.compute_fingerprint.return_value = MagicMock(to_dict=lambda: {})
+                mock_dedup.check_duplicate.return_value = None
+                mock_dedup.check_batch_duplicate.return_value = None
+
+                with patch(
+                    "video_processing.thumbnail_generator.generate_and_upload_thumbnail",
+                    side_effect=RuntimeError("cv2 not found"),
+                ):
+                    result = create_video_record_and_dedup(
+                        generation_task_id="task-thumb-fail",
+                        project_id="proj-1",
+                        batch_id="",
+                        file_url="https://oss.example.com/video.mp4",
+                        file_size=1024,
+                        duration=10.0,
+                        video_path="/tmp/fake.mp4",
+                        mode="edit_plan",
+                        session=session,
+                        width=1280,
+                        height=720,
+                        fps=25.0,
+                    )
+
+            assert result == 1  # 不阻断
+
+            from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel
+
+            video = session.query(GeneratedVideoModel).filter_by(generation_task_id="task-thumb-fail").first()
+            assert video is not None
+            # 缩略图生成失败时 thumbnail_url 为 None 或空串
+            assert not video.thumbnail_url
+        finally:
+            session.close()

tests/unit/test_render_adapter.py

@@ -308,13 +308,99 @@ class TestRenderPlan:
         assert result.width == 1280
         assert result.height == 720
         assert result.clip_count == 2
+        # 缩略图URL（即使生成失败也应该是空串，不为None）
+        assert hasattr(result, "thumbnail_url")
+
+    @patch("video_processing.render_adapter.upload_to_oss")
+    @patch("video_processing.render_adapter.UnifiedRenderService")
+    @patch("video_processing.render_adapter.download_asset")
+    def test_thumbnail_generated_on_success(self, mock_download, mock_render_cls, mock_upload, tmp_path):
+        """渲染成功后生成缩略图，thumbnail_url 正确返回。"""
+
+        def _fake_download(storage_key, local_path):
+            local_path.parent.mkdir(parents=True, exist_ok=True)
+            local_path.write_bytes(b"fake video")
+            return True
+
+        mock_download.side_effect = _fake_download
+
+        mock_render = MagicMock()
+        mock_render.render.return_value = MagicMock(
+            output_path=tmp_path / "out.mp4",
+            duration=5.0,
+            file_size=1024,
+            width=1280,
+            height=720,
+        )
+        mock_render_cls.return_value = mock_render
+        mock_upload.return_value = "https://oss.example.com/out.mp4"
+
+        fake_thumb = "https://oss.example.com/rendered/plan_thumb/thumbnail.jpg"
+
+        plan = FakePlan(id="plan_thumb")
+        clips = [_make_clip("c1", order=0, duration=5.0)]
+        asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
+
+        with patch(
+            "video_processing.thumbnail_generator.generate_and_upload_thumbnail",
+            return_value=fake_thumb,
+        ):
+            result = adapter.render_plan(
+                "plan_thumb",
+                work_dir=tmp_path / "work",
+            )
+
+        assert result.success
+        assert result.thumbnail_url == fake_thumb
+
+    @patch("video_processing.render_adapter.upload_to_oss")
+    @patch("video_processing.render_adapter.UnifiedRenderService")
+    @patch("video_processing.render_adapter.download_asset")
+    def test_thumbnail_failure_does_not_block(self, mock_download, mock_render_cls, mock_upload, tmp_path):
+        """缩略图生成失败不影响主流程，thumbnail_url 为空串。"""
+
+        def _fake_download(storage_key, local_path):
+            local_path.parent.mkdir(parents=True, exist_ok=True)
+            local_path.write_bytes(b"fake video")
+            return True
+
+        mock_download.side_effect = _fake_download
+
+        mock_render = MagicMock()
+        mock_render.render.return_value = MagicMock(
+            output_path=tmp_path / "out.mp4",
+            duration=5.0,
+            file_size=1024,
+            width=1280,
+            height=720,
+        )
+        mock_render_cls.return_value = mock_render
+        mock_upload.return_value = "https://oss.example.com/out.mp4"
+
+        plan = FakePlan(id="plan_thumb_fail")
+        clips = [_make_clip("c1", order=0, duration=5.0)]
+        asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
+
+        with patch(
+            "video_processing.thumbnail_generator.generate_and_upload_thumbnail",
+            side_effect=RuntimeError("cv2 not available"),
+        ):
+            result = adapter.render_plan(
+                "plan_thumb_fail",
+                work_dir=tmp_path / "work",
+            )
+
+        assert result.success  # 不阻断
+        assert result.thumbnail_url == ""
 
         # 验证 UnifiedRenderService 被正确调用
         mock_render_cls.assert_called_once()
         call_kwargs = mock_render_cls.call_args
         assert call_kwargs.kwargs["plan"] is plan
-        assert len(call_kwargs.kwargs["clips"]) == 2
-        assert len(call_kwargs.kwargs["asset_path_map"]) == 2
+        assert len(call_kwargs.kwargs["clips"]) == 1
+        assert len(call_kwargs.kwargs["asset_path_map"]) == 1
 
     @patch("video_processing.render_adapter.upload_to_oss")
     @patch("video_processing.render_adapter.UnifiedRenderService")