apps/worker/worker_app/tasks/edit_plan_generation.py

@@ -141,7 +141,7 @@ def _finalize_render_success(
     project_id = plan.project_id or ""
     batch_id = plan.config.get("batch_id", "")
     mode = plan.config.get("mode", "edit_plan")
-    if generation_task_id and project_id:
+    if generation_task_id:
         try:
             create_video_record_and_dedup(
                 generation_task_id=generation_task_id,

scripts/backfill_generated_videos.py

@@ -0,0 +1,145 @@
+"""补写 generated_videos 记录 — 修复一键生成场景下 project_id 为空导致未写表的问题.
+
+背景：#535 之前的一键生成场景，由于 edit_plan.project_id 为空，
+_finalize_render_success 中 `if generation_task_id and project_id:` 判断不通过，
+导致渲染成功的视频没有写入 generated_videos 表，成片库看不到。
+
+本脚本扫描所有 status=completed 的 generation_task，
+如果没有对应的 generated_videos 记录，则从 plan.config.rendered_url 补写。
+
+用法：
+    PYTHONPATH="apps/worker:apps/api:packages" python scripts/backfill_generated_videos.py
+
+安全说明：
+- 只读事务扫描，补写前检查是否已存在，不会重复写入
+- 失败不中断，继续处理下一个
+"""
+
+from __future__ import annotations
+
+import logging
+import sys
+from datetime import datetime, timezone
+from uuid import uuid4
+
+# 配置日志
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+
+def main() -> int:
+    from sqlalchemy import exists
+    from sqlalchemy.orm import Session
+
+    from packages.adapters.sqlalchemy_impl.models import (
+        EditPlanModel,
+        GeneratedVideoModel,
+        GenerationTaskModel,
+    )
+    from packages.adapters.sqlalchemy_impl.session import get_db_session
+
+    db: Session = next(get_db_session())
+
+    try:
+        # 1. 查询所有 status=completed 的 generation_task
+        completed_tasks = (
+            db.query(GenerationTaskModel)
+            .filter(
+                GenerationTaskModel.status == "completed",
+            )
+            .all()
+        )
+
+        logger.info("找到 %d 个已完成的生成任务", len(completed_tasks))
+
+        fixed_count = 0
+        skipped_count = 0
+        failed_count = 0
+
+        for task in completed_tasks:
+            task_id = task.id
+
+            # 2. 检查是否已有 generated_videos 记录
+            exists_video = db.query(exists().where(GeneratedVideoModel.generation_task_id == task_id)).scalar()
+
+            if exists_video:
+                skipped_count += 1
+                continue
+
+            # 3. 从 source_edit_plan_id 找到对应的 edit_plan，取 rendered_url
+            if not task.source_edit_plan_id:
+                logger.warning("任务 %s 无 source_edit_plan_id，跳过", task_id)
+                skipped_count += 1
+                continue
+
+            plan = (
+                db.query(EditPlanModel)
+                .filter(
+                    EditPlanModel.id == task.source_edit_plan_id,
+                )
+                .first()
+            )
+
+            if not plan:
+                logger.warning("任务 %s 对应的剪辑计划不存在，跳过", task_id)
+                skipped_count += 1
+                continue
+
+            # 从 config 中取 rendered_url
+            config = plan.config or {}
+            rendered_url = config.get("rendered_url", "")
+            rendered_storage_key = config.get("rendered_storage_key", "")
+
+            if not rendered_url:
+                logger.warning("任务 %s 的剪辑计划无 rendered_url，跳过", task_id)
+                skipped_count += 1
+                continue
+
+            # 4. 补写 generated_videos 记录
+            try:
+                video_id = uuid4().hex
+                video = GeneratedVideoModel(
+                    id=video_id,
+                    project_id=task.project_id or "",
+                    generation_task_id=task_id,
+                    name=f"generated-{task_id[:8]}.mp4",
+                    file_url=rendered_url,
+                    storage_key=rendered_storage_key,
+                    file_size=0,
+                    duration=0.0,
+                    width=0,
+                    height=0,
+                    fps=0.0,
+                    status="completed",
+                    generation_params={"mode": "backfill", "plan_id": plan.id},
+                    review_status="pending",
+                    created_at=task.completed_at or datetime.now(timezone.utc),
+                    updated_at=datetime.now(timezone.utc),
+                )
+                db.add(video)
+                db.commit()
+                fixed_count += 1
+                logger.info("已补写视频记录: task_id=%s video_id=%s url=%s", task_id, video_id, rendered_url[:80])
+            except Exception as e:
+                db.rollback()
+                failed_count += 1
+                logger.error("补写视频记录失败: task_id=%s error=%s", task_id, e)
+
+        logger.info(
+            "补写完成：共 %d 个已完成任务，补写 %d 条，跳过 %d 条，失败 %d 条",
+            len(completed_tasks),
+            fixed_count,
+            skipped_count,
+            failed_count,
+        )
+        return 0 if failed_count == 0 else 1
+
+    finally:
+        db.close()
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/unit/test_generated_video_creation_logic.py

@@ -0,0 +1,113 @@
+"""测试：渲染完成后 generated_videos 写表判断逻辑.
+
+验证：project_id 为空时，只要有 generation_task_id 就应该写入 generated_videos。
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from packages.adapters.sqlalchemy_impl.generation_task_repository import (
+    SQLAlchemyGenerationTaskRepository,
+)
+from packages.adapters.sqlalchemy_impl.models import Base
+from packages.domain import GenerationTask, GenerationTaskStatus
+
+
+def _make_task(**kwargs) -> GenerationTask:
+    defaults = dict(
+        project_id="proj-1",
+        asset_library_id="lib-1",
+        created_by_user_id="user-1",
+        status=GenerationTaskStatus.PENDING,
+        progress=0.0,
+        source_edit_plan_id="plan-1",
+    )
+    defaults.update(kwargs)
+    return GenerationTask(id="task-" + kwargs.get("project_id", "t1")[:8], **defaults)
+
+
+class TestCleanupStaleRunning:
+    """复用已有的 repo 测试基础设施，验证写表判断逻辑。"""
+
+    def _setup(self):
+        engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(engine)
+        session = sessionmaker(bind=engine)()
+        return SQLAlchemyGenerationTaskRepository(session), session
+
+    def test_generation_task_without_project_id_can_be_created(self):
+        """验证：project_id 为空的 generation_task 可以正常创建（模拟一键生成场景）。"""
+        repo, session = self._setup()
+        try:
+            task = _make_task(
+                project_id="",
+                status=GenerationTaskStatus.COMPLETED,
+                progress=100.0,
+                result_count=1,
+                source_edit_plan_id="plan-empty-proj",
+            )
+            task.id = "task-empty-proj"
+            created = repo.create(task)
+            assert created.project_id == ""
+
+            fetched = repo.get("task-empty-proj")
+            assert fetched is not None
+            assert fetched.project_id == ""
+            assert fetched.status.value == "completed"
+            assert fetched.progress == 100.0
+        finally:
+            session.close()
+
+    def test_generation_task_with_project_id_can_be_created(self):
+        """验证：project_id 有值的 generation_task 正常（回归测试）。"""
+        repo, session = self._setup()
+        try:
+            task = _make_task(
+                project_id="proj-123",
+                status=GenerationTaskStatus.COMPLETED,
+                progress=100.0,
+                result_count=1,
+            )
+            task.id = "task-with-proj"
+            created = repo.create(task)
+            assert created.project_id == "proj-123"
+
+            fetched = repo.get("task-with-proj")
+            assert fetched is not None
+            assert fetched.project_id == "proj-123"
+        finally:
+            session.close()
+
+
+class TestVideoCreationLogic:
+    """验证写表判断逻辑（纯逻辑测试，不依赖 DB）。"""
+
+    def test_should_create_video_with_generation_task_id_and_empty_project(self):
+        """有 generation_task_id、无 project_id → 应该创建视频（修复点）。"""
+        generation_task_id = "task-123"
+        project_id = ""
+        # 修复后：只要有 generation_task_id 就创建
+        assert bool(generation_task_id) is True
+        # 之前的错误逻辑：if generation_task_id and project_id → False
+        assert bool(generation_task_id and project_id) is False
+        # 修复后的逻辑：if generation_task_id → True
+        assert bool(generation_task_id) is True
+
+    def test_should_create_video_with_both_ids(self):
+        """有 generation_task_id、有 project_id → 应该创建视频（回归）。"""
+        generation_task_id = "task-123"
+        project_id = "proj-456"
+        assert bool(generation_task_id) is True
+        assert bool(generation_task_id and project_id) is True
+
+    def test_should_not_create_video_without_generation_task_id(self):
+        """无 generation_task_id → 不创建视频（边界）。"""
+        generation_task_id = ""
+        project_id = "proj-456"
+        assert bool(generation_task_id) is False