diff --git a/apps/api/app/config.py b/apps/api/app/config.py index 893e50ff0..68ec61b5d 100755 --- a/apps/api/app/config.py +++ b/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", diff --git a/apps/api/app/services/ai_service.py b/apps/api/app/services/ai_service.py index 4af575745..24dd1b645 100755 --- a/apps/api/app/services/ai_service.py +++ b/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: diff --git a/apps/worker/worker_app/tasks/ai_tasks.py b/apps/worker/worker_app/tasks/ai_tasks.py old mode 100644 new mode 100755 index 2a8323367..560acdf4c --- a/apps/worker/worker_app/tasks/ai_tasks.py +++ b/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 封面生成 ────────────────────────────────────────────────────────────── diff --git a/packages/shared/ai_client.py b/packages/shared/ai_client.py new file mode 100755 index 000000000..467c3cace --- /dev/null +++ b/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 diff --git a/packages/shared/config.py b/packages/shared/config.py index 694bfa44f..92ad6b821 100755 --- a/packages/shared/config.py +++ b/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 diff --git a/tests/unit/test_ai_service.py b/tests/unit/test_ai_service.py index 5050aedf2..f83e408ae 100755 --- a/tests/unit/test_ai_service.py +++ b/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) diff --git a/tests/unit/test_ai_tasks.py b/tests/unit/test_ai_tasks.py new file mode 100755 index 000000000..1d447767f --- /dev/null +++ b/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()