packages/shared/ai_service.py

@@ -13,6 +13,8 @@ import random
 import time
 from typing import Any, Dict, List, Optional
 
+import requests as http_requests
+
 from packages.domain.config_schemas import DEFAULT_EDIT_PLAN_CONFIG
 from packages.shared.ai_client import get_doubao_client
 
@@ -398,6 +400,26 @@ def _call_ai_cover_service(
 
         client = get_mediakit_client()
         if client.is_available:
+            # 先检查视频 URL 是否可访问，避免 MediaKit 下载失败后超时
+            try:
+                head_resp = http_requests.head(primary_video_url, timeout=10, allow_redirects=True)
+                if head_resp.status_code != 200:
+                    logger.error(
+                        "封面视频URL不可访问: plan_id=%s url=%s status=%d",
+                        plan_id,
+                        primary_video_url,
+                        head_resp.status_code,
+                    )
+                    raise RuntimeError(
+                        f"封面生成失败: 预览视频URL不可访问 (HTTP {head_resp.status_code})。"
+                        f"请重新生成预览视频后再试。"
+                    )
+            except http_requests.RequestException as e:
+                logger.error("封面视频URL连通性检查失败: plan_id=%s url=%s error=%s", plan_id, primary_video_url, e)
+                raise RuntimeError(
+                    f"封面生成失败: 无法访问预览视频 ({e.__class__.__name__})。" f"请重新生成预览视频后再试。"
+                ) from e
+
             try:
                 logger.info("调用 MediaKit 抽帧: plan_id=%s video=%s", plan_id, primary_video_url[:80])
                 frames = client.extract_frames(

tests/unit/test_mediakit_cover.py

@@ -142,9 +142,13 @@ class TestMediaKitClient:
 class TestAICoverService:
     """AI 封面服务测试."""
 
+    @patch("packages.shared.ai_service.http_requests.head")
     @patch("packages.shared.mediakit_client.get_mediakit_client")
-    def test_call_ai_cover_with_mediakit_success(self, mock_get_client):
+    def test_call_ai_cover_with_mediakit_success(self, mock_get_client, mock_head):
         """MediaKit 抽帧成功."""
+        # Mock HEAD request to verify URL is accessible
+        mock_head.return_value.status_code = 200
+
         mock_client = Mock()
         mock_client.is_available = True
         mock_client.extract_frames.return_value = [{"image_url": "https://example.com/frame.jpg", "timestamp": 3.5}]
@@ -166,9 +170,34 @@ class TestAICoverService:
 
         mock_client.extract_frames.assert_called_once()
 
+    @patch("packages.shared.ai_service.http_requests.head")
     @patch("packages.shared.mediakit_client.get_mediakit_client")
-    def test_call_ai_cover_with_mediakit_failure_raises(self, mock_get_client):
+    def test_call_ai_cover_video_url_unreachable(self, mock_get_client, mock_head):
+        """视频 URL 不可访问时抛出 RuntimeError."""
+        # Mock HEAD request to return 404
+        mock_head.return_value.status_code = 404
+
+        mock_client = Mock()
+        mock_client.is_available = True
+        mock_get_client.return_value = mock_client
+
+        from packages.shared.ai_service import _call_ai_cover_service
+
+        with pytest.raises(RuntimeError, match="预览视频URL不可访问"):
+            _call_ai_cover_service(
+                plan_id="plan-123",
+                asset_ids=["asset-1"],
+                cover_type="ai_frame",
+                primary_video_url="https://example.com/nonexistent.mp4",
+            )
+
+    @patch("packages.shared.ai_service.http_requests.head")
+    @patch("packages.shared.mediakit_client.get_mediakit_client")
+    def test_call_ai_cover_with_mediakit_failure_raises(self, mock_get_client, mock_head):
         """MediaKit 失败时抛出 RuntimeError（不再降级到 stub）."""
+        # Mock HEAD request to return 200 (URL is accessible, but MediaKit fails)
+        mock_head.return_value.status_code = 200
+
         mock_client = Mock()
         mock_client.is_available = True
         mock_client.extract_frames.side_effect = Exception("API error")
@@ -243,9 +272,12 @@ class TestAICoverService:
                 primary_video_url="https://example.com/video.mp4",
             )
 
+    @patch("packages.shared.ai_service.http_requests.head")
     @patch("packages.shared.mediakit_client.get_mediakit_client")
-    def test_call_ai_cover_empty_frames_raises(self, mock_get_client):
+    def test_call_ai_cover_empty_frames_raises(self, mock_get_client, mock_head):
         """MediaKit 返回空帧列表时抛出 RuntimeError（不再降级）."""
+        mock_head.return_value.status_code = 200  # URL accessible
+
         mock_client = Mock()
         mock_client.is_available = True
         mock_client.extract_frames.return_value = []