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

@@ -48,9 +48,9 @@ def _build_asset_analyses(
         asset_repo = SQLAlchemyAssetRepository(db)
         storage_svc = get_shared_storage_service()
 
-        # 查找素材并获取下载 URL
+        # 查找素材并获取下载 URL（使用并行列表保持索引对应，避免 URL 重复导致映射覆盖）
         video_urls: list[str] = []
-        url_to_asset_id: dict[str, str] = {}
+        valid_asset_ids: list[str] = []
 
         for aid in asset_ids[:10]:  # MediaKit 单次最多 10 个视频
             asset = asset_repo.get(aid)
@@ -64,7 +64,7 @@ def _build_asset_analyses(
                 url = storage_svc.get_download_url(asset.storage_key)
                 if url:
                     video_urls.append(url)
-                    url_to_asset_id[url] = aid
+                    valid_asset_ids.append(aid)
             except Exception as e:
                 logger.warning("获取素材URL失败: asset_id=%s error=%s", aid, str(e))
 
@@ -79,23 +79,26 @@ def _build_asset_analyses(
             "控制在100字以内。"
         )
 
+        # 限制轮询参数以适配 API 网关超时（nginx 60s）
+        # 视频理解最多 30s（poll_interval=2s * max_poll_attempts=15）
+        # 剩余 30s 留给 LLM 调用
         contents = client.analyze_videos(
             video_urls=video_urls,
             prompt=prompt,
             level="Economy",
+            poll_interval=2.0,
+            max_poll_attempts=15,
         )
 
         if not contents:
             logger.warning("MediaKit 视频理解未返回结果")
             return {}
 
-        # 将结果映射回 asset_id
+        # 将结果映射回 asset_id（通过索引对应）
         analyses: dict[str, str] = {}
         for i, content in enumerate(contents):
-            if i < len(video_urls) and content:
-                asset_id = url_to_asset_id.get(video_urls[i])
-                if asset_id:
-                    analyses[asset_id] = content
+            if i < len(valid_asset_ids) and content:
+                analyses[valid_asset_ids[i]] = content
 
         logger.info(
             "MediaKit 视频理解完成: total=%d analyzed=%d",