apps/api/app/api/routes/generation_cover.py

@@ -31,8 +31,6 @@ logger = logging.getLogger(__name__)
 router = APIRouter(tags=["Generation"])
 
 
-
-
 # ── Schemas ──────────────────────────────────────────────────────────────
 
 
@@ -65,7 +63,6 @@ class GenerateCoverResponse(BaseModel):
 # ── Route ────────────────────────────────────────────────────────────────
 
 
-
 def _persist_cover_frame(frame_url: str, plan_id: str) -> str:
     """下载 MediaKit 返回的临时帧图并转存到 OSS covers/ 路径。"""
     import tempfile
@@ -238,7 +235,7 @@ def generate_cover(
             )
 
     # 使用裸 URL（rendered/* 已配置公开读）；找不到渲染视频时不立即报错，
-    # 因为步骤 E 可以直接从源素材抽帧（前端纯 Canvas/WebCodecs 预览无后端渲染产物）
+    # 因为步骤 E 可以直接从源素材抽帧（历史数据或 Worker 抽帧失败时的兜底）
     primary_video_url = None
     if rendered_storage_key:
         plan_svc.update_plan_config(plan_id, {"rendered_storage_key": rendered_storage_key})
@@ -252,6 +249,7 @@ def generate_cover(
                 primary_video_url = storage_svc.get_url(rendered_storage_key)
             if primary_video_url:
                 import re as _re
+
                 primary_video_url = _re.sub(r"(?<!:)//", "/", primary_video_url)
             logger.info(
                 "获取预览视频URL用于封面生成: plan_id=%s url=%s",
@@ -341,9 +339,7 @@ def generate_cover(
             if isinstance(_candidates, list) and _candidates:
                 _first = _candidates[0]
                 if isinstance(_first, dict):
-                    cover_url_from_task = (
-                        _first.get("image_url") or _first.get("url") or ""
-                    )
+                    cover_url_from_task = _first.get("image_url") or _first.get("url") or ""
                     if cover_url_from_task:
                         logger.info(
                             "[封面生成] 统一管道封面(步骤D-cover_candidates): plan_id=%s url=%s",
@@ -368,6 +364,9 @@ def generate_cover(
                         video_url=primary_video_url,
                         strategy="SpecifiedFrames",
                         max_frames=1,
+                        poll_interval=2.0,
+                        max_poll_attempts=5,
+                        max_retries=0,
                     )
                     if snapshots:
                         raw = snapshots[0].get("image_url") or snapshots[0].get("url") or ""
@@ -385,8 +384,8 @@ def generate_cover(
                     exc_info=True,
                 )
 
-        # 步骤 E2：前端纯 Canvas/WebCodecs 预览（无后端渲染任务）时，
-        # 直接从用户选择的第一个视频素材中抽取封面帧。
+        # 步骤 E2：当 A/B/C/D/E1 均未命中（如历史预览任务无 cover_url）时，
+        # 直接从用户选择的第一个视频素材中抽取封面帧作为兜底。API 请求内短超时，不阻塞。
         if not cover_url_from_task and body.asset_ids:
             from packages.adapters.sqlalchemy_impl.asset_repository import (
                 SQLAlchemyAssetRepository,
@@ -419,6 +418,9 @@ def generate_cover(
                             video_url=src_url,
                             strategy="SpecifiedFrames",
                             max_frames=1,
+                            poll_interval=2.0,
+                            max_poll_attempts=5,
+                            max_retries=0,
                         )
                         if snapshots:
                             raw = snapshots[0].get("image_url") or snapshots[0].get("url") or ""