apps/api/app/api/routes/templates_editor/_fallback.py

@@ -98,6 +98,15 @@ def _auto_fallback_assign_assets(
     clips_without_asset = [c for c in all_clips if not c.asset_id]
     config_asset_ids = (plan_check.config or {}).get("asset_ids", [])
 
+    logger.info(
+        "模板编辑器自动兜底3 诊断: plan=%s total_clips=%d "
+        "clips_without_asset=%d config_asset_ids=%r",
+        plan_id,
+        len(all_clips),
+        len(clips_without_asset),
+        config_asset_ids[:5] if config_asset_ids else [],
+    )
+
     if clips_without_asset and config_asset_ids:
         logger.info(
             "模板编辑器自动兜底3: plan=%s 为 %d 个无素材片段分配 %d 个指定素材",
@@ -105,11 +114,42 @@ def _auto_fallback_assign_assets(
             len(clips_without_asset),
             len(config_asset_ids),
         )
+        assigned = 0
         for i, clip in enumerate(clips_without_asset):
             asset_idx = i % len(config_asset_ids)
-            svc.assign_asset(clip.id, config_asset_ids[asset_idx])
-        logger.info("模板编辑器自动兜底3: plan=%s 素材分配完成", plan_id)
-        clips_without_asset = []
+            try:
+                svc.assign_asset(clip.id, config_asset_ids[asset_idx])
+                assigned += 1
+            except Exception as exc:
+                logger.error(
+                    "模板编辑器自动兜底3: plan=%s clip=%s 分配素材 %s 失败: %s",
+                    plan_id,
+                    clip.id,
+                    config_asset_ids[asset_idx],
+                    exc,
+                )
+        logger.info(
+            "模板编辑器自动兜底3: plan=%s 素材分配完成 assigned=%d/%d",
+            plan_id,
+            assigned,
+            len(clips_without_asset),
+        )
+        # 重新检查剩余无素材片段
+        all_clips_after = svc.list_clips(plan_id)
+        clips_without_asset = [c for c in all_clips_after if not c.asset_id]
+        if clips_without_asset:
+            logger.warning(
+                "模板编辑器自动兜底3: plan=%s 仍有 %d 个片段无素材",
+                plan_id,
+                len(clips_without_asset),
+            )
+    elif not clips_without_asset:
+        logger.info("模板编辑器自动兜底3: plan=%s 所有片段已有素材，跳过", plan_id)
+    elif not config_asset_ids:
+        logger.info(
+            "模板编辑器自动兜底3: plan=%s config.asset_ids 为空，跳过分配",
+            plan_id,
+        )
 
     return clips_without_asset
 

apps/api/app/api/routes/templates_editor/generation.py

@@ -74,7 +74,7 @@ def generate_editor_draft(
         plan_svc, plan_id, plan_check, clips_without_asset, asset_library_repo, asset_repo
     )
 
-    # 检查是否可生成
+    # 检查是否可生成（含最后防线自动修复 + 诊断日志）
     try:
         can_gen, reason = plan_svc.can_generate(plan_id)
     except ValueError as exc:

apps/api/app/services/edit_plan_service.py

@@ -571,6 +571,10 @@ class EditPlanService:
     def can_generate(self, plan_id: str) -> tuple[bool, str]:
         """检查是否可以触发渲染
 
+        包含最后一道防线的自动修复：
+        - 如果 clips 存在但都没有 asset_id，且 config.asset_ids 非空，
+          直接在内部执行素材分配，不再依赖前置 fallback 链路。
+
         Returns:
             tuple: (can_generate, reason)
         """
@@ -586,8 +590,59 @@ class EditPlanService:
             return False, "请先添加片段后再生成视频"
 
         # 检查是否至少有一个片段分配了素材
-        if not any(c.asset_id for c in clips):
-            return False, "没有可渲染的就绪片段，请确保已选择素材"
+        has_asset = any(c.asset_id for c in clips)
+        config_asset_ids_count = len((plan.config or {}).get("asset_ids", []))
+        clips_with_asset_count = sum(1 for c in clips if c.asset_id)
+        logger.info(
+            "can_generate 诊断: plan=%s status=%s total_clips=%d " "clips_with_asset=%d config_asset_ids_count=%d",
+            plan_id,
+            plan.status,
+            len(clips),
+            clips_with_asset_count,
+            config_asset_ids_count,
+        )
+        if not has_asset:
+            # ── 最后防线：自动从 config.asset_ids 分配素材 ──
+            config_asset_ids = (plan.config or {}).get("asset_ids", [])
+            if config_asset_ids:
+                logger.warning(
+                    "can_generate 最后防线触发: plan=%s clips=%d 均无素材，" "从 config.asset_ids(%d个) 自动分配",
+                    plan_id,
+                    len(clips),
+                    len(config_asset_ids),
+                )
+                clips_without_asset = [c for c in clips if not c.asset_id]
+                assigned_count = 0
+                for i, clip in enumerate(clips_without_asset):
+                    asset_idx = i % len(config_asset_ids)
+                    try:
+                        self.assign_asset(clip.id, config_asset_ids[asset_idx])
+                        assigned_count += 1
+                    except Exception as exc:
+                        logger.warning(
+                            "can_generate 最后防线: plan=%s clip=%s 分配素材 %s 失败: %s",
+                            plan_id,
+                            clip.id,
+                            config_asset_ids[asset_idx],
+                            exc,
+                        )
+                logger.info(
+                    "can_generate 最后防线: plan=%s 已为 %d/%d 个片段分配素材",
+                    plan_id,
+                    assigned_count,
+                    len(clips_without_asset),
+                )
+                # 重新加载 clips 验证分配结果
+                clips = self._clip_repo.list_by_plan(plan_id)
+                if not any(c.asset_id for c in clips):
+                    return False, "没有可渲染的就绪片段，自动修复后仍未分配素材"
+            else:
+                logger.warning(
+                    "can_generate 失败: plan=%s clips=%d 均无素材，" "且 config.asset_ids 为空，无法自动修复",
+                    plan_id,
+                    len(clips),
+                )
+                return False, "没有可渲染的就绪片段，请确保已选择素材"
 
         return True, ""
 

tests/unit/test_1286_can_generate_auto_repair.py

@@ -0,0 +1,331 @@
+"""
+#1286 can_generate 最后防线自动修复 — 单元测试
+
+覆盖场景：
+1. clips 无素材 + config.asset_ids 非空 → 自动分配成功 → can_generate 返回 True
+2. clips 无素材 + config.asset_ids 为空 → 无法修复 → can_generate 返回 False
+3. clips 已有素材 → 正常通过，不触发自动修复
+4. 自动修复后素材数量与 clips 数量一致（循环分配验证）
+5. 自动修复不影响预览生成流程（预览不依赖 EditPlan clips）
+6. config 为 None 时安全降级
+7. 无 clips 时仍返回 False
+8. 部分 clips 有素材时正常通过（不触发自动修复）
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+from typing import Any, List, Optional
+from unittest.mock import MagicMock
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+from packages.domain.edit_plan import EditPlan, EditPlanStatus
+from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
+
+# ---------------------------------------------------------------------------
+# Stub Repositories
+# ---------------------------------------------------------------------------
+
+
+class StubEditPlanRepository:
+    def __init__(self) -> None:
+        self._plans: dict[str, EditPlan] = {}
+        self._counter = 0
+
+    def _next_id(self) -> str:
+        self._counter += 1
+        return f"plan-auto-{self._counter:03d}"
+
+    def get(self, plan_id: str) -> Optional[EditPlan]:
+        return self._plans.get(plan_id)
+
+    def create(self, plan: EditPlan) -> EditPlan:
+        if not plan.id:
+            plan = EditPlan(
+                id=self._next_id(),
+                template_id=plan.template_id,
+                name=plan.name,
+                status=plan.status,
+                total_duration=plan.total_duration,
+                source_edit_plan_id=plan.source_edit_plan_id,
+                project_id=plan.project_id,
+                created_by_user_id=plan.created_by_user_id,
+                config=plan.config,
+            )
+        self._plans[plan.id] = plan
+        return plan
+
+    def update(self, plan: EditPlan) -> EditPlan:
+        self._plans[plan.id] = plan
+        return plan
+
+    def delete(self, plan_id: str) -> bool:
+        return self._plans.pop(plan_id, None) is not None
+
+    def list_all(self, **kwargs):
+        return list(self._plans.values())
+
+
+class StubEditPlanClipRepository:
+    def __init__(self) -> None:
+        self._clips: dict[str, EditPlanClip] = {}
+        self._counter = 0
+
+    def _next_id(self) -> str:
+        self._counter += 1
+        return f"clip-auto-{self._counter:03d}"
+
+    def get(self, clip_id: str) -> Optional[EditPlanClip]:
+        return self._clips.get(clip_id)
+
+    def create(self, clip: EditPlanClip) -> EditPlanClip:
+        if not clip.id:
+            clip = EditPlanClip(
+                id=self._next_id(),
+                plan_id=clip.plan_id,
+                clip_type=clip.clip_type,
+                order=clip.order,
+                duration=clip.duration,
+                status=clip.status,
+                asset_id=clip.asset_id,
+                text_content=clip.text_content,
+                config=clip.config,
+                template_clip_config_id=clip.template_clip_config_id,
+                transition_effect=clip.transition_effect,
+            )
+        self._clips[clip.id] = clip
+        return clip
+
+    def update(self, clip: EditPlanClip) -> EditPlanClip:
+        self._clips[clip.id] = clip
+        return clip
+
+    def delete(self, clip_id: str) -> bool:
+        return self._clips.pop(clip_id, None) is not None
+
+    def list_by_plan(
+        self,
+        plan_id: str,
+        *,
+        status: Optional[EditPlanClipStatus] = None,
+        skip: int = 0,
+        limit: int = 100,
+    ) -> List[EditPlanClip]:
+        clips = [c for c in self._clips.values() if c.plan_id == plan_id]
+        if status is not None:
+            clips = [c for c in clips if c.status == status]
+        return sorted(clips, key=lambda c: c.order)[skip : skip + limit]
+
+    def delete_by_plan(self, plan_id: str) -> int:
+        to_del = [cid for cid, c in self._clips.items() if c.plan_id == plan_id]
+        for cid in to_del:
+            del self._clips[cid]
+        return len(to_del)
+
+    def count_by_plan(self, plan_id: str) -> int:
+        return sum(1 for c in self._clips.values() if c.plan_id == plan_id)
+
+
+class StubGenerationTaskRepository:
+    def __init__(self):
+        pass
+
+
+def _make_service():
+    from app.services.edit_plan_service import EditPlanService
+
+    db = MagicMock()
+    svc = EditPlanService(db)
+    svc._plan_repo = StubEditPlanRepository()
+    svc._clip_repo = StubEditPlanClipRepository()
+    svc._generation_task_repo = StubGenerationTaskRepository()
+    return svc
+
+
+# ===========================================================================
+# 测试用例
+# ===========================================================================
+
+
+class TestCanGenerateAutoRepair:
+    """#1286 can_generate 最后防线自动修复"""
+
+    def test_auto_repair_with_config_asset_ids(self):
+        """clips 无素材 + config.asset_ids 非空 → 自动分配成功 → can_generate True"""
+        svc = _make_service()
+        p = svc.create_plan(
+            "tpl-001",
+            "测试",
+            config={"asset_ids": ["asset-001", "asset-002"]},
+        )
+        svc.transition_status(p.id, EditPlanStatus.EDITING)
+        svc.create_clip(p.id, "intro", 0)
+        svc.create_clip(p.id, "main", 1)
+
+        # clips 无素材
+        clips = svc.list_clips(p.id)
+        assert all(not c.asset_id for c in clips)
+
+        # can_generate 应触发自动修复
+        can, reason = svc.can_generate(p.id)
+        assert can is True
+        assert reason == ""
+
+        # 验证 clips 已被分配素材
+        clips_after = svc.list_clips(p.id)
+        assert all(c.asset_id is not None for c in clips_after)
+
+    def test_auto_repair_fails_without_config_asset_ids(self):
+        """clips 无素材 + config.asset_ids 为空 → 无法修复 → False"""
+        svc = _make_service()
+        p = svc.create_plan("tpl-001", "测试", config={})
+        svc.transition_status(p.id, EditPlanStatus.EDITING)
+        svc.create_clip(p.id, "intro", 0)
+
+        can, reason = svc.can_generate(p.id)
+        assert can is False
+        assert "没有可渲染" in reason or "素材" in reason
+
+    def test_no_repair_when_clips_have_assets(self):
+        """clips 已有素材 → 正常通过，不触发自动修复"""
+        svc = _make_service()
+        p = svc.create_plan(
+            "tpl-001",
+            "测试",
+            config={"asset_ids": ["asset-001", "asset-002"]},
+        )
+        svc.transition_status(p.id, EditPlanStatus.EDITING)
+        clip = svc.create_clip(p.id, "intro", 0)
+        svc.assign_asset(clip.id, "asset-001")
+
+        can, reason = svc.can_generate(p.id)
+        assert can is True
+        # 确认 clip 的 asset_id 没有被改变
+        clips = svc.list_clips(p.id)
+        assert clips[0].asset_id == "asset-001"
+
+    def test_auto_repair_circular_assignment(self):
+        """素材少于 clips 时循环分配（取模）"""
+        svc = _make_service()
+        p = svc.create_plan(
+            "tpl-001",
+            "测试",
+            config={"asset_ids": ["asset-A"]},
+        )
+        svc.transition_status(p.id, EditPlanStatus.EDITING)
+        svc.create_clip(p.id, "intro", 0)
+        svc.create_clip(p.id, "main", 1)
+        svc.create_clip(p.id, "outro", 2)
+
+        can, reason = svc.can_generate(p.id)
+        assert can is True
+
+        clips = svc.list_clips(p.id)
+        # 所有 3 个 clips 都应被分配了同一个 asset-A
+        assert all(c.asset_id == "asset-A" for c in clips)
+
+    def test_auto_repair_does_not_affect_preview_flow(self):
+        """预览生成走 generation_preview.py，不依赖 EditPlan clips 的 can_generate"""
+        svc = _make_service()
+        p = svc.create_plan(
+            "tpl-001",
+            "测试",
+            config={"asset_ids": ["asset-001"]},
+        )
+        # 不切到 editing 状态，模拟预览场景
+        svc.create_clip(p.id, "intro", 0)
+
+        can, reason = svc.can_generate(p.id)
+        assert can is False
+        assert "编辑" in reason or "模板" in reason
+
+    def test_auto_repair_empty_config(self):
+        """config 为 None 时也不报错"""
+        svc = _make_service()
+        p = svc.create_plan("tpl-001", "测试")
+        svc.transition_status(p.id, EditPlanStatus.EDITING)
+        svc.create_clip(p.id, "intro", 0)
+
+        can, reason = svc.can_generate(p.id)
+        assert can is False
+        assert "没有可渲染" in reason or "素材" in reason
+
+    def test_auto_repair_no_clips_still_fails(self):
+        """没有 clips 时仍然返回 False（不进入自动修复分支）"""
+        svc = _make_service()
+        p = svc.create_plan(
+            "tpl-001",
+            "测试",
+            config={"asset_ids": ["asset-001"]},
+        )
+        svc.transition_status(p.id, EditPlanStatus.EDITING)
+
+        can, reason = svc.can_generate(p.id)
+        assert can is False
+        assert "请先添加片段" in reason
+
+    def test_auto_repair_partial_assets_still_passes(self):
+        """部分 clips 有素材、部分没有 → 至少有一个有素材 → 通过（原有逻辑）"""
+        svc = _make_service()
+        p = svc.create_plan(
+            "tpl-001",
+            "测试",
+            config={"asset_ids": ["asset-001"]},
+        )
+        svc.transition_status(p.id, EditPlanStatus.EDITING)
+        clip1 = svc.create_clip(p.id, "intro", 0)
+        svc.create_clip(p.id, "main", 1)
+        svc.assign_asset(clip1.id, "asset-001")
+
+        # 至少一个 clip 有素材 → 通过（不触发自动修复）
+        can, reason = svc.can_generate(p.id)
+        assert can is True
+
+    def test_auto_repair_multiple_assets_distributed(self):
+        """多个素材按顺序分配给多个 clips"""
+        svc = _make_service()
+        p = svc.create_plan(
+            "tpl-001",
+            "测试",
+            config={"asset_ids": ["asset-A", "asset-B", "asset-C"]},
+        )
+        svc.transition_status(p.id, EditPlanStatus.EDITING)
+        svc.create_clip(p.id, "intro", 0)
+        svc.create_clip(p.id, "main", 1)
+        svc.create_clip(p.id, "outro", 2)
+
+        can, reason = svc.can_generate(p.id)
+        assert can is True
+
+        clips = svc.list_clips(p.id)
+        assert clips[0].asset_id == "asset-A"
+        assert clips[1].asset_id == "asset-B"
+        assert clips[2].asset_id == "asset-C"
+
+    def test_auto_repair_idempotent_on_second_call(self):
+        """第二次调用 can_generate 不会重复分配（已有素材则跳过自动修复）"""
+        svc = _make_service()
+        p = svc.create_plan(
+            "tpl-001",
+            "测试",
+            config={"asset_ids": ["asset-001"]},
+        )
+        svc.transition_status(p.id, EditPlanStatus.EDITING)
+        svc.create_clip(p.id, "intro", 0)
+
+        # 第一次调用触发自动修复
+        can1, _ = svc.can_generate(p.id)
+        assert can1 is True
+
+        # 第二次调用应该直接通过，不再触发修复
+        can2, reason2 = svc.can_generate(p.id)
+        assert can2 is True
+        assert reason2 == ""