apps/api/app/config.py

@@ -109,14 +109,6 @@ class Settings(BaseSettings):
     # 渲染引擎选择：legacy=旧VideoComposeService，unified=新UnifiedRenderService
     RENDER_ENGINE: str = "legacy"
 
-    # 豆包大模型配置（火山引擎方舟平台）
-    # 未配置 API Key 时自动降级为本地模拟生成
-    DOUBAO_API_KEY: str = ""
-    DOUBAO_MODEL: str = "doubao-seed-1-6-250615"
-    DOUBAO_BASE_URL: str = "https://ark.cn-beijing.volces.com/api/v3"
-    DOUBAO_TIMEOUT: int = 30
-    DOUBAO_MAX_RETRIES: int = 2
-
     model_config = SettingsConfigDict(
         env_file=".env",
         env_file_encoding="utf-8",

apps/api/app/services/ai_service.py

@@ -16,11 +16,9 @@ import json
 import logging
 import math
 import random
-import time
 from typing import Any, Dict, List, Optional
 
-import httpx
-from app.config import get_settings
+from packages.shared.ai_client import get_doubao_client
 
 logger = logging.getLogger(__name__)
 
@@ -58,85 +56,6 @@ TITLE_STYLES = {
 }
 
 
-# ── 豆包 AI 客户端 ──────────────────────────────────────────────────────────
-
-
-class DoubaoAIClient:
-    """豆包大模型 API 客户端.
-
-    使用火山引擎方舟平台的 OpenAI 兼容接口。
-    未配置 API Key 时，is_available 返回 False，调用方应降级处理。
-    """
-
-    def __init__(self) -> None:
-        settings = get_settings()
-        self.api_key: str = settings.DOUBAO_API_KEY
-        self.model: str = settings.DOUBAO_MODEL
-        self.base_url: str = settings.DOUBAO_BASE_URL.rstrip("/")
-        self.timeout: int = settings.DOUBAO_TIMEOUT
-        self.max_retries: int = settings.DOUBAO_MAX_RETRIES
-
-    @property
-    def is_available(self) -> bool:
-        """是否可用（配置了 API Key）."""
-        return bool(self.api_key)
-
-    def _chat_completion(
-        self,
-        messages: List[Dict[str, str]],
-        temperature: float = 0.7,
-        max_tokens: int = 1024,
-    ) -> Optional[str]:
-        """调用豆包 Chat Completion 接口.
-
-        Returns:
-            模型返回的文本内容，失败返回 None
-        """
-        if not self.is_available:
-            return None
-
-        url = f"{self.base_url}/chat/completions"
-        headers = {
-            "Authorization": f"Bearer {self.api_key}",
-            "Content-Type": "application/json",
-        }
-        payload = {
-            "model": self.model,
-            "messages": messages,
-            "temperature": temperature,
-            "max_tokens": max_tokens,
-        }
-
-        last_error: Optional[Exception] = None
-        for attempt in range(self.max_retries + 1):
-            try:
-                response = httpx.post(
-                    url,
-                    headers=headers,
-                    json=payload,
-                    timeout=self.timeout,
-                )
-                response.raise_for_status()
-                data = response.json()
-                content = data["choices"][0]["message"]["content"]
-                return content.strip()
-            except Exception as e:
-                last_error = e
-                if attempt < self.max_retries:
-                    wait = 0.5 * (2**attempt)
-                    logger.warning(
-                        "豆包API调用失败，%s秒后重试 (第%d/%d次): %s",
-                        wait,
-                        attempt + 1,
-                        self.max_retries + 1,
-                        e,
-                    )
-                    time.sleep(wait)
-
-        logger.error("豆包API调用最终失败: %s", last_error)
-        return None
-
-
 # ── 智能标题生成 ─────────────────────────────────────────────────────────────
 
 
@@ -252,7 +171,7 @@ def generate_smart_titles(
     count = max(3, min(10, count))  # 3-10 个
     description = (description or "").strip()
 
-    client = DoubaoAIClient()
+    client = get_doubao_client()
     if not client.is_available:
         logger.info("豆包API未配置，使用本地降级生成标题")
         titles = _generate_titles_fallback(description, style, count)
@@ -281,7 +200,7 @@ def generate_smart_titles(
         {"role": "user", "content": user_prompt},
     ]
 
-    result = client._chat_completion(
+    result = client.chat_completion(
         messages=messages,
         temperature=0.8,
         max_tokens=512,
@@ -474,7 +393,7 @@ def semantic_match_assets(
     if not assets:
         return {"matches": [], "source": "fallback", "description": description, "total": 0}
 
-    client = DoubaoAIClient()
+    client = get_doubao_client()
     if not client.is_available:
         logger.info("豆包API未配置，使用本地降级做素材语义匹配")
         matched = _semantic_match_fallback(description, assets)
@@ -524,7 +443,7 @@ def semantic_match_assets(
         {"role": "user", "content": user_prompt},
     ]
 
-    result = client._chat_completion(
+    result = client.chat_completion(
         messages=messages,
         temperature=0.3,
         max_tokens=1024,
@@ -593,7 +512,7 @@ class AIService:
     """AI 服务统一入口，便于后续扩展更多能力."""
 
     def __init__(self) -> None:
-        self._client = DoubaoAIClient()
+        self._client = get_doubao_client()
 
     @property
     def is_available(self) -> bool:

apps/worker/worker_app/tasks/ai_tasks.py

@@ -10,12 +10,14 @@
 
 from __future__ import annotations
 
+import json
 import logging
 import random
 import time
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 from packages.domain.config_schemas import DEFAULT_EDIT_PLAN_CONFIG
+from packages.shared.ai_client import get_doubao_client
 
 logger = logging.getLogger(__name__)
 
@@ -23,17 +25,16 @@ logger = logging.getLogger(__name__)
 # ── AI 推荐片段方案 ──────────────────────────────────────────────────────────
 
 
-def _call_ai_recommend_service(
+def _fallback_recommend_clips(
     plan_id: str,
     template_id: str,
     asset_ids: List[str],
     editing_mode: str,
     target_duration: float,
 ) -> Dict[str, Any]:
-    """调用 AI 推荐服务（stub）
+    """本地降级推荐方案（原 stub 逻辑）.
 
-    TODO: 接入真实 AI 服务，分析素材内容并生成推荐方案。
-    当前返回基于模板规则的模拟推荐数据。
+    当豆包 API 不可用或调用失败时使用，基于模板规则生成模拟推荐数据。
     """
     # 模拟 AI 分析耗时
     time.sleep(0.5)
@@ -87,6 +88,7 @@ def _call_ai_recommend_service(
             "config": {},
         }
     )
+    order += 1
 
     # 生成推荐 config
     config = DEFAULT_EDIT_PLAN_CONFIG.copy()
@@ -101,6 +103,163 @@ def _call_ai_recommend_service(
     }
 
 
+def _parse_recommend_response(
+    content: str,
+    asset_ids: List[str],
+    target_duration: float,
+) -> Optional[Dict[str, Any]]:
+    """解析豆包返回的推荐方案.
+
+    期望返回结构：
+    {
+        "clips": [
+            {"clip_type": "intro/showcase/outro", "order": 0,
+             "text_content": "...", "duration": 3.0,
+             "transition_effect": "fade/cut", "asset_id": "...",
+             "start_time": 0.0, "config": {}}
+        ],
+        "title": "视频标题",
+        "confidence": 0.85
+    }
+    """
+    if not content:
+        return None
+
+    try:
+        cleaned = content.strip()
+        if cleaned.startswith("```"):
+            cleaned = cleaned.strip("`")
+            if cleaned.lower().startswith("json"):
+                cleaned = cleaned[4:]
+            cleaned = cleaned.strip()
+
+        data = json.loads(cleaned)
+        if not isinstance(data, dict):
+            return None
+
+        clips_data = data.get("clips", [])
+        if not isinstance(clips_data, list) or len(clips_data) == 0:
+            return None
+
+        clips: List[Dict[str, Any]] = []
+        for i, clip in enumerate(clips_data):
+            if not isinstance(clip, dict):
+                continue
+            asset_id = str(clip.get("asset_id", ""))
+            # 校验 asset_id 是否在输入列表中
+            if asset_id and asset_id not in asset_ids:
+                asset_id = ""
+            clips.append({
+                "clip_type": clip.get("clip_type", "showcase"),
+                "order": clip.get("order", len(clips)),
+                "text_content": str(clip.get("text_content", "")),
+                "duration": max(1.0, min(30.0, float(clip.get("duration", 3.0)))),
+                "transition_effect": clip.get("transition_effect", "cut"),
+                "asset_id": asset_id,
+                "start_time": max(0.0, float(clip.get("start_time", 0.0))),
+                "config": clip.get("config", {}) or {},
+            })
+
+        if not clips:
+            return None
+
+        # 按 order 排序
+        clips.sort(key=lambda c: c["order"])
+        # 重新编号 order 保证连续
+        for i, clip in enumerate(clips):
+            clip["order"] = i
+
+        config = DEFAULT_EDIT_PLAN_CONFIG.copy()
+        title = data.get("title", "")
+        if title:
+            config["title"]["text"] = str(title)
+            config["title"]["ai_auto"] = True
+
+        confidence = float(data.get("confidence", 0.7))
+        confidence = max(0.0, min(1.0, confidence))
+
+        total_duration = round(sum(c["duration"] for c in clips), 1)
+
+        return {
+            "clips": clips,
+            "config": config,
+            "total_duration": total_duration,
+            "confidence": round(confidence, 2),
+        }
+
+    except (json.JSONDecodeError, ValueError, TypeError, KeyError):
+        return None
+
+
+def _call_ai_recommend_service(
+    plan_id: str,
+    template_id: str,
+    asset_ids: List[str],
+    editing_mode: str,
+    target_duration: float,
+) -> Dict[str, Any]:
+    """调用 AI 推荐服务生成片段编排方案.
+
+    优先使用豆包大模型生成，失败或未配置时降级为本地规则生成。
+    """
+    client = get_doubao_client()
+    if not client.is_available:
+        logger.info("豆包API未配置，使用本地降级生成AI推荐方案")
+        return _fallback_recommend_clips(plan_id, template_id, asset_ids, editing_mode, target_duration)
+
+    # 构建 prompt
+    system_prompt = (
+        "你是一个专业的视频剪辑导演助手。"
+        "根据提供的素材列表和目标时长，设计一个完整的视频片段编排方案。\n"
+        "要求：\n"
+        "1. 片段类型分为三类：intro（开场）、showcase（展示）、outro（结尾）\n"
+        "2. 每个片段包含：clip_type、order、text_content（字幕/标题文字）、"
+        "duration（时长秒）、transition_effect（转场效果：fade/cut/dissolve）、"
+        "asset_id（使用的素材ID）、start_time（素材起始时间秒）\n"
+        "3. 总时长接近 target_duration，每个素材至少用一次\n"
+        "4. 转场效果合理分配，不要全用cut\n"
+        "5. 返回纯JSON，不要其他文字\n"
+        "返回格式：{\"clips\": [...], \"title\": \"视频标题\", \"confidence\": 0.85}"
+    )
+
+    assets_desc = "\n".join([f"  - 素材ID: {aid}" for i, aid in enumerate(asset_ids[:30])])
+    user_prompt = (
+        f"剪辑计划ID: {plan_id}\n"
+        f"模板ID: {template_id}\n"
+        f"剪辑模式: {editing_mode}\n"
+        f"目标时长: {target_duration}秒\n"
+        f"素材列表（共{len(asset_ids)}个）：\n{assets_desc}\n\n"
+        f"请设计完整的片段编排方案："
+    )
+
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": user_prompt},
+    ]
+
+    result = client.chat_completion(
+        messages=messages,
+        temperature=0.7,
+        max_tokens=2048,
+    )
+
+    if result:
+        parsed = _parse_recommend_response(result, asset_ids, target_duration)
+        if parsed and len(parsed["clips"]) >= 2:
+            logger.info(
+                "豆包AI推荐生成成功: plan_id=%s clips=%d duration=%.1f confidence=%.2f",
+                plan_id,
+                len(parsed["clips"]),
+                parsed["total_duration"],
+                parsed["confidence"],
+            )
+            return parsed
+        logger.warning("豆包AI推荐返回解析失败，降级到本地方案: %s", result[:100])
+
+    # 降级
+    return _fallback_recommend_clips(plan_id, template_id, asset_ids, editing_mode, target_duration)
+
+
 # ── AI 封面生成 ──────────────────────────────────────────────────────────────
 
 

packages/shared/ai_client.py

@@ -0,0 +1,117 @@
+"""豆包大模型 API 客户端（共享层）.
+
+API 和 Worker 两边共用。基于火山引擎方舟平台的 OpenAI 兼容接口。
+
+使用方式：
+    from packages.shared.ai_client import get_doubao_client
+
+    client = get_doubao_client()
+    if client.is_available:
+        result = client.chat_completion(messages=[...])
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from typing import Any, Dict, List, Optional
+
+import httpx
+from packages.shared.config import get_shared_settings
+
+logger = logging.getLogger(__name__)
+
+
+class DoubaoClient:
+    """豆包大模型 API 客户端.
+
+    封装 OpenAI 兼容的 Chat Completion 接口，支持自动重试。
+    未配置 API Key 时 is_available 为 False，调用方应降级处理。
+    """
+
+    def __init__(self) -> None:
+        settings = get_shared_settings()
+        self.api_key: str = settings.doubao_api_key
+        self.model: str = settings.doubao_model
+        self.base_url: str = settings.doubao_base_url.rstrip("/")
+        self.timeout: int = settings.doubao_timeout
+        self.max_retries: int = settings.doubao_max_retries
+
+    @property
+    def is_available(self) -> bool:
+        """是否可用（配置了 API Key）."""
+        return bool(self.api_key)
+
+    def chat_completion(
+        self,
+        messages: List[Dict[str, str]],
+        temperature: float = 0.7,
+        max_tokens: int = 1024,
+    ) -> Optional[str]:
+        """调用 Chat Completion 接口.
+
+        Args:
+            messages: 对话消息列表，[{"role": "user"/"system"/"assistant", "content": "..."}]
+            temperature: 采样温度，0-2，默认0.7
+            max_tokens: 最大生成token数，默认1024
+
+        Returns:
+            模型返回的文本内容，失败返回 None
+        """
+        if not self.is_available:
+            return None
+
+        url = f"{self.base_url}/chat/completions"
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json",
+        }
+        payload: Dict[str, Any] = {
+            "model": self.model,
+            "messages": messages,
+            "temperature": temperature,
+            "max_tokens": max_tokens,
+        }
+
+        last_error: Optional[Exception] = None
+        for attempt in range(self.max_retries + 1):
+            try:
+                response = httpx.post(
+                    url,
+                    headers=headers,
+                    json=payload,
+                    timeout=self.timeout,
+                )
+                response.raise_for_status()
+                data = response.json()
+                content = data["choices"][0]["message"]["content"]
+                return content.strip()
+            except Exception as e:
+                last_error = e
+                if attempt < self.max_retries:
+                    wait = 0.5 * (2**attempt)
+                    logger.warning(
+                        "豆包API调用失败，%.1fs后重试 (第%d/%d次): %s",
+                        wait,
+                        attempt + 1,
+                        self.max_retries + 1,
+                        e,
+                    )
+                    time.sleep(wait)
+
+        logger.error("豆包API调用最终失败: %s", last_error)
+        return None
+
+
+# ── 单例 ─────────────────────────────────────────────────────────────────────
+
+
+_client: Optional[DoubaoClient] = None
+
+
+def get_doubao_client() -> DoubaoClient:
+    """获取豆包客户端单例."""
+    global _client
+    if _client is None:
+        _client = DoubaoClient()
+    return _client

packages/shared/config.py

@@ -39,6 +39,13 @@ class SharedSettings(BaseSettings):
     # 音色克隆模型名（固定为 voice-enrollment）
     cosyvoice_clone_model: str = "voice-enrollment"
 
+    # 豆包大模型（火山引擎方舟）
+    doubao_api_key: str = ""
+    doubao_model: str = "doubao-seed-1-6-250615"
+    doubao_base_url: str = "https://ark.cn-beijing.volces.com/api/v3"
+    doubao_timeout: int = 30
+    doubao_max_retries: int = 2
+
     # Environment
     environment: str = "development"
     auto_create_schema: bool = False

tests/unit/test_ai_service.py

@@ -18,7 +18,6 @@ from unittest.mock import MagicMock, patch
 sys.path.insert(0, "apps/api")
 
 from app.services.ai_service import (  # noqa: E402
-    DoubaoAIClient,
     TITLE_STYLES,
     _generate_titles_fallback,
     _parse_titles_from_response,
@@ -29,48 +28,34 @@ from app.services.ai_service import (  # noqa: E402
 )
 
 
-class TestDoubaoAIClient(unittest.TestCase):
-    """豆包客户端基础测试."""
+class TestAIClientAvailability(unittest.TestCase):
+    """AI客户端可用性检测（通过mock get_doubao_client）."""
 
-    def test_client_availability_without_key(self):
-        """未配置 API Key 时不可用."""
-        with patch("app.services.ai_service.get_settings") as mock_settings:
-            mock_settings.return_value = MagicMock(
-                DOUBAO_API_KEY="",
-                DOUBAO_MODEL="test-model",
-                DOUBAO_BASE_URL="https://test.com",
-                DOUBAO_TIMEOUT=30,
-                DOUBAO_MAX_RETRIES=2,
-            )
-            client = DoubaoAIClient()
-            self.assertFalse(client.is_available)
+    def test_generate_fallback_when_client_unavailable(self):
+        """客户端不可用时走降级."""
+        mock_client = MagicMock()
+        mock_client.is_available = False
+        mock_client.chat_completion = MagicMock(return_value=None)
 
-    def test_client_availability_with_key(self):
-        """配置了 API Key 时可用."""
-        with patch("app.services.ai_service.get_settings") as mock_settings:
-            mock_settings.return_value = MagicMock(
-                DOUBAO_API_KEY="sk-test-123",
-                DOUBAO_MODEL="test-model",
-                DOUBAO_BASE_URL="https://test.com",
-                DOUBAO_TIMEOUT=30,
-                DOUBAO_MAX_RETRIES=2,
-            )
-            client = DoubaoAIClient()
-            self.assertTrue(client.is_available)
+        with patch("app.services.ai_service.get_doubao_client", return_value=mock_client):
+            result = generate_smart_titles("测试内容", "viral", 5)
+            self.assertEqual(result["source"], "fallback")
+            self.assertEqual(len(result["titles"]), 5)
+            # 不可用时不应调用 chat_completion
+            mock_client.chat_completion.assert_not_called()
 
-    def test_chat_completion_not_available_returns_none(self):
-        """不可用时调用返回 None."""
-        with patch("app.services.ai_service.get_settings") as mock_settings:
-            mock_settings.return_value = MagicMock(
-                DOUBAO_API_KEY="",
-                DOUBAO_MODEL="test-model",
-                DOUBAO_BASE_URL="https://test.com",
-                DOUBAO_TIMEOUT=30,
-                DOUBAO_MAX_RETRIES=2,
-            )
-            client = DoubaoAIClient()
-            result = client._chat_completion([{"role": "user", "content": "hi"}])
-            self.assertIsNone(result)
+    def test_generate_calls_client_when_available(self):
+        """客户端可用时调用API."""
+        mock_client = MagicMock()
+        mock_client.is_available = True
+        mock_client.chat_completion = MagicMock(
+            return_value=json.dumps(["AI标题1", "AI标题2", "AI标题3", "AI标题4", "AI标题5"])
+        )
+
+        with patch("app.services.ai_service.get_doubao_client", return_value=mock_client):
+            result = generate_smart_titles("测试", "viral", 5)
+            self.assertEqual(result["source"], "doubao")
+            mock_client.chat_completion.assert_called_once()
 
 
 class TestTitleParsing(unittest.TestCase):
@@ -168,14 +153,9 @@ class TestGenerateSmartTitles(unittest.TestCase):
 
     def test_generate_without_api_key_fallback(self):
         """无 API Key 时走降级路径."""
-        with patch("app.services.ai_service.get_settings") as mock_settings:
-            mock_settings.return_value = MagicMock(
-                DOUBAO_API_KEY="",
-                DOUBAO_MODEL="test",
-                DOUBAO_BASE_URL="https://test.com",
-                DOUBAO_TIMEOUT=30,
-                DOUBAO_MAX_RETRIES=2,
-            )
+        mock_client = MagicMock()
+        mock_client.is_available = False
+        with patch("app.services.ai_service.get_doubao_client", return_value=mock_client):
             result = generate_smart_titles("测试视频内容", "viral", 5)
             self.assertEqual(result["source"], "fallback")
             self.assertEqual(result["style"], "viral")
@@ -183,27 +163,17 @@ class TestGenerateSmartTitles(unittest.TestCase):
 
     def test_generate_invalid_style_defaults_to_viral(self):
         """无效风格默认 viral."""
-        with patch("app.services.ai_service.get_settings") as mock_settings:
-            mock_settings.return_value = MagicMock(
-                DOUBAO_API_KEY="",
-                DOUBAO_MODEL="test",
-                DOUBAO_BASE_URL="https://test.com",
-                DOUBAO_TIMEOUT=30,
-                DOUBAO_MAX_RETRIES=2,
-            )
+        mock_client = MagicMock()
+        mock_client.is_available = False
+        with patch("app.services.ai_service.get_doubao_client", return_value=mock_client):
             result = generate_smart_titles("测试", "invalid_style", 5)
             self.assertEqual(result["style"], "viral")
 
     def test_generate_count_bounds(self):
         """数量边界处理."""
-        with patch("app.services.ai_service.get_settings") as mock_settings:
-            mock_settings.return_value = MagicMock(
-                DOUBAO_API_KEY="",
-                DOUBAO_MODEL="test",
-                DOUBAO_BASE_URL="https://test.com",
-                DOUBAO_TIMEOUT=30,
-                DOUBAO_MAX_RETRIES=2,
-            )
+        mock_client = MagicMock()
+        mock_client.is_available = False
+        with patch("app.services.ai_service.get_doubao_client", return_value=mock_client):
             # 小于最小值
             result = generate_smart_titles("测试", "viral", 1)
             self.assertEqual(len(result["titles"]), 3)
@@ -213,70 +183,39 @@ class TestGenerateSmartTitles(unittest.TestCase):
 
     def test_generate_with_api_success(self):
         """API 调用成功路径."""
-        with patch("app.services.ai_service.get_settings") as mock_settings:
-            mock_settings.return_value = MagicMock(
-                DOUBAO_API_KEY="sk-test-123",
-                DOUBAO_MODEL="test-model",
-                DOUBAO_BASE_URL="https://test.com",
-                DOUBAO_TIMEOUT=30,
-                DOUBAO_MAX_RETRIES=0,
-            )
-            mock_response = MagicMock()
-            mock_response.status_code = 200
-            mock_response.json.return_value = {
-                "choices": [
-                    {
-                        "message": {
-                            "content": json.dumps(["AI标题1", "AI标题2", "AI标题3", "AI标题4", "AI标题5"])
-                        }
-                    }
-                ]
-            }
-            mock_response.raise_for_status = MagicMock()
-
-            with patch("httpx.post", return_value=mock_response):
-                result = generate_smart_titles("测试视频", "viral", 5)
-                self.assertEqual(result["source"], "doubao")
-                self.assertEqual(len(result["titles"]), 5)
-                self.assertIn("AI标题1", result["titles"])
+        mock_client = MagicMock()
+        mock_client.is_available = True
+        mock_client.chat_completion = MagicMock(
+            return_value=json.dumps(["AI标题1", "AI标题2", "AI标题3", "AI标题4", "AI标题5"])
+        )
+        with patch("app.services.ai_service.get_doubao_client", return_value=mock_client):
+            result = generate_smart_titles("测试视频", "viral", 5)
+            self.assertEqual(result["source"], "doubao")
+            self.assertEqual(len(result["titles"]), 5)
+            self.assertIn("AI标题1", result["titles"])
 
     def test_generate_with_api_failure_fallback(self):
         """API 调用失败时降级."""
-        with patch("app.services.ai_service.get_settings") as mock_settings:
-            mock_settings.return_value = MagicMock(
-                DOUBAO_API_KEY="sk-test-123",
-                DOUBAO_MODEL="test-model",
-                DOUBAO_BASE_URL="https://test.com",
-                DOUBAO_TIMEOUT=1,
-                DOUBAO_MAX_RETRIES=0,
-            )
-            with patch("httpx.post", side_effect=Exception("API Error")):
-                result = generate_smart_titles("测试视频", "viral", 5)
-                self.assertEqual(result["source"], "fallback")
-                self.assertEqual(len(result["titles"]), 5)
+        mock_client = MagicMock()
+        mock_client.is_available = True
+        mock_client.chat_completion = MagicMock(return_value=None)
+        with patch("app.services.ai_service.get_doubao_client", return_value=mock_client):
+            result = generate_smart_titles("测试视频", "viral", 5)
+            self.assertEqual(result["source"], "fallback")
+            self.assertEqual(len(result["titles"]), 5)
 
     def test_generate_api_returns_unparseable_fallback(self):
         """API 返回无法解析时降级."""
-        with patch("app.services.ai_service.get_settings") as mock_settings:
-            mock_settings.return_value = MagicMock(
-                DOUBAO_API_KEY="sk-test-123",
-                DOUBAO_MODEL="test-model",
-                DOUBAO_BASE_URL="https://test.com",
-                DOUBAO_TIMEOUT=30,
-                DOUBAO_MAX_RETRIES=0,
-            )
-            mock_response = MagicMock()
-            mock_response.status_code = 200
-            # 返回无法解析的内容（只有一个标题且格式异常）
-            mock_response.json.return_value = {
-                "choices": [{"message": {"content": "一段文字说明，不是标题列表"}}]
-            }
-            mock_response.raise_for_status = MagicMock()
-
-            with patch("httpx.post", return_value=mock_response):
-                result = generate_smart_titles("测试视频", "viral", 5)
-                # 只有1个有效标题，不足2个触发降级
-                self.assertEqual(result["source"], "fallback")
+        mock_client = MagicMock()
+        mock_client.is_available = True
+        # 返回无法解析的内容（只有一个标题且格式异常）
+        mock_client.chat_completion = MagicMock(
+            return_value="一段文字说明，不是标题列表"
+        )
+        with patch("app.services.ai_service.get_doubao_client", return_value=mock_client):
+            result = generate_smart_titles("测试视频", "viral", 5)
+            # 只有1个有效标题，不足2个触发降级
+            self.assertEqual(result["source"], "fallback")
 
 
 class TestTitleStyles(unittest.TestCase):
@@ -431,14 +370,9 @@ class TestSemanticMatchAssets(unittest.TestCase):
 
     def test_fallback_mode_without_api_key(self):
         """无API Key时走降级."""
-        with patch("app.services.ai_service.get_settings") as mock_settings:
-            mock_settings.return_value = MagicMock(
-                DOUBAO_API_KEY="",
-                DOUBAO_MODEL="test",
-                DOUBAO_BASE_URL="https://test.com",
-                DOUBAO_TIMEOUT=30,
-                DOUBAO_MAX_RETRIES=0,
-            )
+        mock_client = MagicMock()
+        mock_client.is_available = False
+        with patch("app.services.ai_service.get_doubao_client", return_value=mock_client):
             result = semantic_match_assets("海边", self._make_assets())
             self.assertEqual(result["source"], "fallback")
             self.assertEqual(result["total"], 3)
@@ -452,70 +386,41 @@ class TestSemanticMatchAssets(unittest.TestCase):
 
     def test_top_k_limit(self):
         """top_k 限制返回数量."""
-        with patch("app.services.ai_service.get_settings") as mock_settings:
-            mock_settings.return_value = MagicMock(
-                DOUBAO_API_KEY="",
-                DOUBAO_MODEL="test",
-                DOUBAO_BASE_URL="https://test.com",
-                DOUBAO_TIMEOUT=30,
-                DOUBAO_MAX_RETRIES=0,
-            )
+        mock_client = MagicMock()
+        mock_client.is_available = False
+        with patch("app.services.ai_service.get_doubao_client", return_value=mock_client):
             result = semantic_match_assets("测试", self._make_assets(), top_k=2)
             self.assertEqual(len(result["matches"]), 2)
 
     def test_with_doubao_success(self):
         """豆包调用成功路径."""
-        with patch("app.services.ai_service.get_settings") as mock_settings:
-            mock_settings.return_value = MagicMock(
-                DOUBAO_API_KEY="sk-test-123",
-                DOUBAO_MODEL="test-model",
-                DOUBAO_BASE_URL="https://test.com",
-                DOUBAO_TIMEOUT=30,
-                DOUBAO_MAX_RETRIES=0,
-            )
-            mock_response = MagicMock()
-            mock_response.status_code = 200
-            mock_response.json.return_value = {
-                "choices": [{
-                    "message": {
-                        "content": json.dumps({"a1": 0.9, "a2": 0.5, "a3": 0.2})
-                    }
-                }]
-            }
-            mock_response.raise_for_status = MagicMock()
-
-            with patch("httpx.post", return_value=mock_response):
-                result = semantic_match_assets("风景视频", self._make_assets())
-                self.assertEqual(result["source"], "doubao")
-                self.assertEqual(len(result["matches"]), 3)
-                # 按分数降序，a1最高
-                self.assertEqual(result["matches"][0]["id"], "a1")
-                self.assertAlmostEqual(result["matches"][0]["match_score"], 0.9)
+        mock_client = MagicMock()
+        mock_client.is_available = True
+        mock_client.chat_completion = MagicMock(
+            return_value=json.dumps({"a1": 0.9, "a2": 0.5, "a3": 0.2})
+        )
+        with patch("app.services.ai_service.get_doubao_client", return_value=mock_client):
+            result = semantic_match_assets("风景视频", self._make_assets())
+            self.assertEqual(result["source"], "doubao")
+            self.assertEqual(len(result["matches"]), 3)
+            # 按分数降序，a1最高
+            self.assertEqual(result["matches"][0]["id"], "a1")
+            self.assertAlmostEqual(result["matches"][0]["match_score"], 0.9)
 
     def test_with_doubao_failure_fallback(self):
         """豆包调用失败降级."""
-        with patch("app.services.ai_service.get_settings") as mock_settings:
-            mock_settings.return_value = MagicMock(
-                DOUBAO_API_KEY="sk-test-123",
-                DOUBAO_MODEL="test-model",
-                DOUBAO_BASE_URL="https://test.com",
-                DOUBAO_TIMEOUT=1,
-                DOUBAO_MAX_RETRIES=0,
-            )
-            with patch("httpx.post", side_effect=Exception("API Error")):
-                result = semantic_match_assets("测试", self._make_assets())
-                self.assertEqual(result["source"], "fallback")
+        mock_client = MagicMock()
+        mock_client.is_available = True
+        mock_client.chat_completion = MagicMock(return_value=None)
+        with patch("app.services.ai_service.get_doubao_client", return_value=mock_client):
+            result = semantic_match_assets("测试", self._make_assets())
+            self.assertEqual(result["source"], "fallback")
 
     def test_each_match_has_required_fields(self):
         """每个匹配结果都有必要字段."""
-        with patch("app.services.ai_service.get_settings") as mock_settings:
-            mock_settings.return_value = MagicMock(
-                DOUBAO_API_KEY="",
-                DOUBAO_MODEL="test",
-                DOUBAO_BASE_URL="https://test.com",
-                DOUBAO_TIMEOUT=30,
-                DOUBAO_MAX_RETRIES=0,
-            )
+        mock_client = MagicMock()
+        mock_client.is_available = False
+        with patch("app.services.ai_service.get_doubao_client", return_value=mock_client):
             result = semantic_match_assets("测试", self._make_assets())
             for item in result["matches"]:
                 self.assertIn("id", item)

tests/unit/test_ai_tasks.py

@@ -0,0 +1,359 @@
+"""Worker AI 任务单元测试.
+
+测试覆盖：
+- AI推荐（豆包调用成功/失败/降级）
+- 推荐响应解析（多种格式）
+- 封面生成降级
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+import unittest
+from unittest.mock import MagicMock, patch
+
+sys.path.insert(0, "apps/worker")
+sys.path.insert(0, "packages")
+
+from worker_app.tasks.ai_tasks import (  # noqa: E402
+    _fallback_recommend_clips,
+    _parse_recommend_response,
+    run_ai_recommend,
+    run_generate_cover,
+)
+
+
+class TestFallbackRecommend(unittest.TestCase):
+    """降级推荐方案测试."""
+
+    def test_fallback_returns_expected_structure(self):
+        """降级推荐返回正确结构."""
+        result = _fallback_recommend_clips(
+            plan_id="plan-1",
+            template_id="tpl-1",
+            asset_ids=["a1", "a2", "a3"],
+            editing_mode="one_take",
+            target_duration=30.0,
+        )
+        self.assertIn("clips", result)
+        self.assertIn("config", result)
+        self.assertIn("total_duration", result)
+        self.assertIn("confidence", result)
+
+    def test_fallback_clips_structure(self):
+        """每个片段都有必要字段."""
+        result = _fallback_recommend_clips(
+            plan_id="plan-1",
+            template_id="tpl-1",
+            asset_ids=["a1", "a2"],
+            editing_mode="one_take",
+            target_duration=20.0,
+        )
+        clips = result["clips"]
+        self.assertTrue(len(clips) >= 3)  # intro + showcase + outro
+        for clip in clips:
+            self.assertIn("clip_type", clip)
+            self.assertIn("order", clip)
+            self.assertIn("text_content", clip)
+            self.assertIn("duration", clip)
+            self.assertIn("transition_effect", clip)
+            self.assertIn("asset_id", clip)
+            self.assertIn("start_time", clip)
+            self.assertIn("config", clip)
+
+    def test_fallback_first_is_intro_last_is_outro(self):
+        """第一个是开场，最后一个是结尾."""
+        result = _fallback_recommend_clips(
+            plan_id="plan-1",
+            template_id="tpl-1",
+            asset_ids=["a1", "a2", "a3"],
+            editing_mode="one_take",
+            target_duration=30.0,
+        )
+        clips = result["clips"]
+        self.assertEqual(clips[0]["clip_type"], "intro")
+        self.assertEqual(clips[-1]["clip_type"], "outro")
+
+    def test_fallback_order_sequential(self):
+        """order 连续递增."""
+        result = _fallback_recommend_clips(
+            plan_id="plan-1",
+            template_id="tpl-1",
+            asset_ids=["a1", "a2"],
+            editing_mode="one_take",
+            target_duration=30.0,
+        )
+        for i, clip in enumerate(result["clips"]):
+            self.assertEqual(clip["order"], i)
+
+    def test_fallback_empty_assets(self):
+        """空素材列表也能生成."""
+        result = _fallback_recommend_clips(
+            plan_id="plan-1",
+            template_id="tpl-1",
+            asset_ids=[],
+            editing_mode="one_take",
+            target_duration=10.0,
+        )
+        self.assertTrue(len(result["clips"]) >= 2)
+
+    def test_fallback_confidence_in_range(self):
+        """置信度在0-1之间."""
+        result = _fallback_recommend_clips(
+            plan_id="plan-1",
+            template_id="tpl-1",
+            asset_ids=["a1"],
+            editing_mode="one_take",
+            target_duration=10.0,
+        )
+        self.assertGreaterEqual(result["confidence"], 0.0)
+        self.assertLessEqual(result["confidence"], 1.0)
+
+
+class TestRecommendResponseParsing(unittest.TestCase):
+    """推荐响应解析测试."""
+
+    def _asset_ids(self):
+        return ["a1", "a2", "a3"]
+
+    def test_parse_valid_response(self):
+        """解析正常响应."""
+        data = {
+            "clips": [
+                {"clip_type": "intro", "order": 0, "text_content": "开场",
+                 "duration": 3.0, "transition_effect": "fade",
+                 "asset_id": "a1", "start_time": 0.0, "config": {}},
+                {"clip_type": "showcase", "order": 1, "text_content": "展示",
+                 "duration": 5.0, "transition_effect": "cut",
+                 "asset_id": "a2", "start_time": 1.0, "config": {}},
+                {"clip_type": "outro", "order": 2, "text_content": "结尾",
+                 "duration": 2.0, "transition_effect": "fade",
+                 "asset_id": "", "start_time": 0.0, "config": {}},
+            ],
+            "title": "精彩视频",
+            "confidence": 0.85,
+        }
+        result = _parse_recommend_response(
+            json.dumps(data), self._asset_ids(), 30.0
+        )
+        self.assertIsNotNone(result)
+        self.assertEqual(len(result["clips"]), 3)
+        self.assertEqual(result["clips"][0]["clip_type"], "intro")
+        self.assertEqual(result["confidence"], 0.85)
+        self.assertIn("精彩视频", result["config"].get("title", {}).get("text", ""))
+
+    def test_parse_markdown_code_block(self):
+        """解析markdown代码块."""
+        data = {"clips": [
+            {"clip_type": "showcase", "order": 0, "text_content": "t",
+             "duration": 3, "transition_effect": "cut",
+             "asset_id": "a1", "start_time": 0, "config": {}}
+        ], "confidence": 0.7}
+        content = "```json\n" + json.dumps(data) + "\n```"
+        result = _parse_recommend_response(content, self._asset_ids(), 30.0)
+        self.assertIsNotNone(result)
+        self.assertEqual(len(result["clips"]), 1)
+
+    def test_parse_empty_content(self):
+        """空内容返回None."""
+        result = _parse_recommend_response("", self._asset_ids(), 30.0)
+        self.assertIsNone(result)
+
+    def test_parse_invalid_json(self):
+        """无效JSON返回None."""
+        result = _parse_recommend_response("不是json", self._asset_ids(), 30.0)
+        self.assertIsNone(result)
+
+    def test_parse_no_clips(self):
+        """无clips字段返回None."""
+        result = _parse_recommend_response(
+            json.dumps({"title": "abc"}), self._asset_ids(), 30.0
+        )
+        self.assertIsNone(result)
+
+    def test_parse_filters_invalid_asset_ids(self):
+        """过滤不在输入列表中的asset_id."""
+        data = {"clips": [
+            {"clip_type": "showcase", "order": 0, "text_content": "t",
+             "duration": 3, "transition_effect": "cut",
+             "asset_id": "fake-id", "start_time": 0, "config": {}}
+        ], "confidence": 0.7}
+        result = _parse_recommend_response(
+            json.dumps(data), self._asset_ids(), 30.0
+        )
+        self.assertIsNotNone(result)
+        # 非法asset_id被清空
+        self.assertEqual(result["clips"][0]["asset_id"], "")
+
+    def test_parse_clamps_duration(self):
+        """时长被限制在合理范围."""
+        data = {"clips": [
+            {"clip_type": "showcase", "order": 0, "text_content": "t",
+             "duration": 100, "transition_effect": "cut",
+             "asset_id": "a1", "start_time": 0, "config": {}}
+        ]}
+        result = _parse_recommend_response(
+            json.dumps(data), self._asset_ids(), 30.0
+        )
+        self.assertIsNotNone(result)
+        self.assertLessEqual(result["clips"][0]["duration"], 30.0)
+
+    def test_parse_reorders_clips(self):
+        """clips按order排序并重新编号."""
+        data = {"clips": [
+            {"clip_type": "showcase", "order": 5, "text_content": "b",
+             "duration": 3, "transition_effect": "cut",
+             "asset_id": "a2", "start_time": 0, "config": {}},
+            {"clip_type": "intro", "order": 0, "text_content": "a",
+             "duration": 3, "transition_effect": "fade",
+             "asset_id": "a1", "start_time": 0, "config": {}},
+        ]}
+        result = _parse_recommend_response(
+            json.dumps(data), self._asset_ids(), 30.0
+        )
+        self.assertIsNotNone(result)
+        # 第一个应该是order=0的intro
+        self.assertEqual(result["clips"][0]["clip_type"], "intro")
+        # order被重新编号为连续
+        self.assertEqual(result["clips"][0]["order"], 0)
+        self.assertEqual(result["clips"][1]["order"], 1)
+
+    def test_parse_confidence_clamped(self):
+        """confidence被限制在0-1."""
+        data = {"clips": [
+            {"clip_type": "showcase", "order": 0, "text_content": "t",
+             "duration": 3, "transition_effect": "cut",
+             "asset_id": "a1", "start_time": 0, "config": {}}
+        ], "confidence": 2.5}
+        result = _parse_recommend_response(
+            json.dumps(data), self._asset_ids(), 30.0
+        )
+        self.assertIsNotNone(result)
+        self.assertLessEqual(result["confidence"], 1.0)
+
+
+class TestRunAIRecommend(unittest.TestCase):
+    """run_ai_recommend 集成测试."""
+
+    def test_fallback_when_client_unavailable(self):
+        """客户端不可用时走降级."""
+        mock_client = MagicMock()
+        mock_client.is_available = False
+        mock_client.chat_completion = MagicMock(return_value=None)
+
+        with patch("worker_app.tasks.ai_tasks.get_doubao_client", return_value=mock_client):
+            result = run_ai_recommend(
+                plan_id="plan-1",
+                template_id="tpl-1",
+                asset_ids=["a1", "a2"],
+                editing_mode="one_take",
+                target_duration=20.0,
+            )
+            self.assertIn("clips", result)
+            self.assertIn("total_duration", result)
+            mock_client.chat_completion.assert_not_called()
+
+    def test_doubao_success(self):
+        """豆包调用成功路径."""
+        mock_client = MagicMock()
+        mock_client.is_available = True
+        mock_response = {
+            "clips": [
+                {"clip_type": "intro", "order": 0, "text_content": "开场",
+                 "duration": 3.0, "transition_effect": "fade",
+                 "asset_id": "a1", "start_time": 0.0, "config": {}},
+                {"clip_type": "outro", "order": 1, "text_content": "结尾",
+                 "duration": 2.0, "transition_effect": "fade",
+                 "asset_id": "a2", "start_time": 0.0, "config": {}},
+            ],
+            "title": "AI生成标题",
+            "confidence": 0.9,
+        }
+        mock_client.chat_completion = MagicMock(return_value=json.dumps(mock_response))
+
+        with patch("worker_app.tasks.ai_tasks.get_doubao_client", return_value=mock_client):
+            result = run_ai_recommend(
+                plan_id="plan-1",
+                template_id="tpl-1",
+                asset_ids=["a1", "a2"],
+                editing_mode="one_take",
+                target_duration=30.0,
+            )
+            self.assertEqual(result["confidence"], 0.9)
+            self.assertEqual(len(result["clips"]), 2)
+            mock_client.chat_completion.assert_called_once()
+
+    def test_doubao_failure_fallback(self):
+        """豆包调用失败降级."""
+        mock_client = MagicMock()
+        mock_client.is_available = True
+        mock_client.chat_completion = MagicMock(return_value=None)
+
+        with patch("worker_app.tasks.ai_tasks.get_doubao_client", return_value=mock_client):
+            result = run_ai_recommend(
+                plan_id="plan-1",
+                template_id="tpl-1",
+                asset_ids=["a1"],
+                editing_mode="one_take",
+                target_duration=10.0,
+            )
+            # 降级后有结果
+            self.assertTrue(len(result["clips"]) >= 2)
+            mock_client.chat_completion.assert_called_once()
+
+    def test_doubao_unparseable_fallback(self):
+        """豆包返回无法解析时降级."""
+        mock_client = MagicMock()
+        mock_client.is_available = True
+        mock_client.chat_completion = MagicMock(return_value="一堆废话不是json")
+
+        with patch("worker_app.tasks.ai_tasks.get_doubao_client", return_value=mock_client):
+            result = run_ai_recommend(
+                plan_id="plan-1",
+                template_id="tpl-1",
+                asset_ids=["a1"],
+                editing_mode="one_take",
+                target_duration=10.0,
+            )
+            # 降级后有结果
+            self.assertTrue(len(result["clips"]) >= 2)
+
+
+class TestGenerateCover(unittest.TestCase):
+    """封面生成测试（降级路径）."""
+
+    def test_ai_frame_type(self):
+        """AI封面模式返回预期结构."""
+        result = run_generate_cover(
+            plan_id="plan-1",
+            asset_ids=["a1"],
+            cover_type="ai_frame",
+        )
+        self.assertIn("type", result)
+        self.assertEqual(result["type"], "ai_frame")
+        self.assertIn("image_url", result)
+
+    def test_manual_type(self):
+        """手动选帧模式."""
+        result = run_generate_cover(
+            plan_id="plan-1",
+            asset_ids=["a1"],
+            cover_type="manual",
+            frame_time=5.0,
+        )
+        self.assertEqual(result["type"], "manual")
+        self.assertEqual(result["frame_time"], 5.0)
+
+    def test_upload_type(self):
+        """上传封面模式."""
+        result = run_generate_cover(
+            plan_id="plan-1",
+            asset_ids=["a1"],
+            cover_type="upload",
+        )
+        self.assertEqual(result["type"], "upload")
+
+
+if __name__ == "__main__":
+    unittest.main()