Add CAMB AI as a TTS provider

.env.template

@@ -30,6 +30,14 @@ TTS_API_KEY="your_tts_key"
 # OpenAI TTS 模型
 TTS_MODEL="tts-1"                          
 
+# CAMB AI TTS 配置（当 USE_CAMBAI_TTS=true 时需配置）
+USE_CAMBAI_TTS=false
+CAMBAI_API_KEY=""
+CAMBAI_SPEECH_MODEL="mars-flash"
+CAMBAI_LANGUAGE="en-us"
+# CAMB AI 声音映射（JSON格式，将角色映射到 CAMB AI voice ID）
+CAMBAI_VOICE_MAP={}
+
 # 微软 TTS 代理配置
 # 当USE_OPENAI_TTS_MODEL为false时，且在国内网络环境使用时，需要配置代理
 HTTPS_PROXY="http://localhost:7890"

requirements.in

@@ -21,4 +21,5 @@ python-multipart
 bcrypt==4.0.1
 alembic
 tenacity
-loguru
+loguru
+camb-sdk

server/core/config.py

@@ -41,6 +41,13 @@ class Settings(BaseSettings):
     TTS_BASE_URL: str
     TTS_API_KEY: str
     TTS_MODEL: str
+
+    # CAMB AI TTS配置
+    USE_CAMBAI_TTS: bool = False
+    CAMBAI_API_KEY: str = ""
+    CAMBAI_SPEECH_MODEL: str = "mars-flash"
+    CAMBAI_LANGUAGE: str = "en-us"
+    CAMBAI_VOICE_MAP: Dict[str, int] = {}
 
     # 代理配置
     HTTPS_PROXY: str | None = None
@@ -96,6 +103,11 @@ class ConfigManager:
         'TTS_BASE_URL',
         'TTS_API_KEY',
         'TTS_MODEL',
+        'USE_CAMBAI_TTS',
+        'CAMBAI_API_KEY',
+        'CAMBAI_SPEECH_MODEL',
+        'CAMBAI_LANGUAGE',
+        'CAMBAI_VOICE_MAP',
         'ANCHOR_TYPE_MAP',
         'HTTPS_PROXY',
         'ALLOW_REGISTRATION',

server/services/cambai_tts.py

@@ -0,0 +1,43 @@
+import os
+import tempfile
+
+from camb.client import CambAI, save_stream_to_file
+from camb.types import StreamTtsOutputConfiguration
+from core.config import settings
+from core.logging import log
+
+
+class CambAITTSService:
+    def __init__(self):
+        self.client = CambAI(api_key=settings.CAMBAI_API_KEY)
+        self.voice_mapping = settings.CAMBAI_VOICE_MAP
+
+    def generate_speech(self, text: str, voice: str, response_format: str = "mp3") -> str:
+        voice_id = self._resolve_voice_id(voice)
+        speech_model = settings.CAMBAI_SPEECH_MODEL
+
+        temp_output = tempfile.NamedTemporaryFile(delete=False, suffix=f".{response_format}")
+        try:
+            stream = self.client.text_to_speech.tts(
+                text=text,
+                language=settings.CAMBAI_LANGUAGE if hasattr(settings, 'CAMBAI_LANGUAGE') else "en-us",
+                voice_id=voice_id,
+                speech_model=speech_model,
+                output_configuration=StreamTtsOutputConfiguration(format=response_format),
+            )
+            save_stream_to_file(stream, temp_output.name)
+            return temp_output.name
+        except Exception as e:
+            log.error(f"CAMB AI TTS generation failed: {str(e)}")
+            if os.path.exists(temp_output.name):
+                os.remove(temp_output.name)
+            raise
+
+    def _resolve_voice_id(self, voice: str) -> int:
+        if voice in self.voice_mapping:
+            return self.voice_mapping[voice]
+        try:
+            return int(voice)
+        except (ValueError, TypeError):
+            log.warning(f"Unknown voice '{voice}', using default voice 147320")
+            return 147320

server/services/task/steps/audio.py

@@ -11,6 +11,7 @@
 from services.edgetts import EdgeTTSService
 from openai import OpenAI
 from core.config import settings
+from services.cambai_tts import CambAITTSService
 from core.logging import log
 from services.task.utils.progress_tracker import ProgressTracker
 
@@ -36,6 +37,7 @@ def __init__(
             base_url=settings.TTS_BASE_URL,
             api_key=settings.TTS_API_KEY
         )
+        self.cambai_tts = CambAITTSService() if settings.USE_CAMBAI_TTS else None
 
     def _verify_audio_file(self, file_path: str) -> bool:
         """验证音频文件是否有效"""
@@ -60,18 +62,24 @@ def _generate_audio_with_retry(self, item: dict, file_path: str, anchor_type: st
         """生成音频文件，支持重试"""
         for attempt in range(max_retries):
             try:
-                if settings.USE_OPENAI_TTS_MODEL:
+                if settings.USE_CAMBAI_TTS:
+                    temp_audio_file = self.cambai_tts.generate_speech(item['content'], anchor_type)
+                    if not temp_audio_file or not os.path.exists(temp_audio_file):
+                        raise Exception("CAMB AI TTS 生成失败")
+
+                    shutil.move(temp_audio_file, file_path)
+                elif settings.USE_OPENAI_TTS_MODEL:
                     audio_content = self._sync_openai_tts_request(item['content'], anchor_type)
                     if audio_content is None:
                         raise Exception("OpenAI TTS 返回空内容")
-                    
+
                     with open(file_path, 'wb') as f:
                         f.write(audio_content)
                 else:
                     temp_audio_file = self.edge_tts.generate_speech(item['content'], anchor_type)
                     if not temp_audio_file or not os.path.exists(temp_audio_file):
                         raise Exception("Edge TTS 生成失败")
-                    
+
                     shutil.move(temp_audio_file, file_path)
 
                 # 验证生成的音频文件

server/tests/test_cambai_tts.py

@@ -0,0 +1,168 @@
+import os
+import sys
+from pathlib import Path
+
+# Add server/ to path so 'services', 'core' etc. are importable
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+import pytest
+import tempfile
+from unittest.mock import Mock, patch, MagicMock
+
+
+@pytest.fixture
+def mock_settings():
+    with patch("services.cambai_tts.settings") as mock_s:
+        mock_s.CAMBAI_API_KEY = "test-key"
+        mock_s.CAMBAI_SPEECH_MODEL = "mars-flash"
+        mock_s.CAMBAI_LANGUAGE = "en-us"
+        mock_s.CAMBAI_VOICE_MAP = {"host_en": 147320, "guest_en": 147321}
+        yield mock_s
+
+
+@pytest.fixture
+def cambai_service(mock_settings):
+    with patch("services.cambai_tts.CambAI") as mock_camb_cls:
+        mock_client = Mock()
+        mock_camb_cls.return_value = mock_client
+        from services.cambai_tts import CambAITTSService
+        service = CambAITTSService()
+        service._mock_client = mock_client
+        yield service
+
+
+class TestVoiceMapping:
+    def test_resolve_voice_id_from_map(self, cambai_service):
+        assert cambai_service._resolve_voice_id("host_en") == 147320
+        assert cambai_service._resolve_voice_id("guest_en") == 147321
+
+    def test_resolve_voice_id_fallback_to_int(self, cambai_service):
+        assert cambai_service._resolve_voice_id("99999") == 99999
+
+    def test_resolve_voice_id_fallback_to_default(self, cambai_service):
+        assert cambai_service._resolve_voice_id("unknown_voice") == 147320
+
+
+class TestGenerateSpeech:
+    def test_success_returns_temp_file_path(self, cambai_service):
+        mock_stream = iter([b"fake-audio-bytes"])
+        cambai_service._mock_client.text_to_speech.tts.return_value = mock_stream
+
+        with patch("services.cambai_tts.save_stream_to_file") as mock_save:
+            result = cambai_service.generate_speech("Hello world", "host_en")
+
+            assert isinstance(result, str)
+            assert result.endswith(".mp3")
+            mock_save.assert_called_once()
+            cambai_service._mock_client.text_to_speech.tts.assert_called_once()
+
+    def test_uses_correct_speech_model(self, cambai_service, mock_settings):
+        mock_settings.CAMBAI_SPEECH_MODEL = "mars-pro"
+        cambai_service._mock_client.text_to_speech.tts.return_value = iter([b"audio"])
+
+        with patch("services.cambai_tts.save_stream_to_file"):
+            cambai_service.generate_speech("test", "host_en")
+
+        call_kwargs = cambai_service._mock_client.text_to_speech.tts.call_args.kwargs
+        assert call_kwargs["speech_model"] == "mars-pro"
+
+    def test_uses_correct_language(self, cambai_service, mock_settings):
+        mock_settings.CAMBAI_LANGUAGE = "ja-jp"
+        cambai_service._mock_client.text_to_speech.tts.return_value = iter([b"audio"])
+
+        with patch("services.cambai_tts.save_stream_to_file"):
+            cambai_service.generate_speech("test", "host_en")
+
+        call_kwargs = cambai_service._mock_client.text_to_speech.tts.call_args.kwargs
+        assert call_kwargs["language"] == "ja-jp"
+
+    def test_failure_cleans_up_temp_file(self, cambai_service):
+        cambai_service._mock_client.text_to_speech.tts.side_effect = Exception("API error")
+
+        with pytest.raises(Exception, match="API error"):
+            cambai_service.generate_speech("Hello", "host_en")
+
+    def test_custom_response_format(self, cambai_service):
+        cambai_service._mock_client.text_to_speech.tts.return_value = iter([b"audio"])
+
+        with patch("services.cambai_tts.save_stream_to_file"):
+            result = cambai_service.generate_speech("test", "host_en", response_format="wav")
+
+        assert result.endswith(".wav")
+
+
+class TestAudioStepIntegration:
+    def test_audio_step_uses_cambai_when_enabled(self):
+        with patch("services.task.steps.audio.settings") as mock_s, \
+             patch("services.task.steps.audio.CambAITTSService") as mock_cls, \
+             patch("services.task.steps.audio.EdgeTTSService"), \
+             patch("services.task.steps.audio.OpenAI"), \
+             patch("services.task.steps.audio.ProgressTracker"):
+            mock_s.USE_CAMBAI_TTS = True
+            mock_s.USE_OPENAI_TTS_MODEL = False
+            mock_s.TTS_BASE_URL = "http://test"
+            mock_s.TTS_API_KEY = "test"
+
+            mock_cambai = Mock()
+            mock_cambai.generate_speech.return_value = "/tmp/test.mp3"
+            mock_cls.return_value = mock_cambai
+
+            from services.task.steps.audio import AudioStep
+            step = AudioStep(
+                level="beginner",
+                lang="en",
+                progress_tracker=Mock(),
+                context_manager=Mock()
+            )
+            assert step.cambai_tts is not None
+
+    def test_audio_step_skips_cambai_when_disabled(self):
+        with patch("services.task.steps.audio.settings") as mock_s, \
+             patch("services.task.steps.audio.CambAITTSService") as mock_cls, \
+             patch("services.task.steps.audio.EdgeTTSService"), \
+             patch("services.task.steps.audio.OpenAI"), \
+             patch("services.task.steps.audio.ProgressTracker"):
+            mock_s.USE_CAMBAI_TTS = False
+            mock_s.USE_OPENAI_TTS_MODEL = False
+            mock_s.TTS_BASE_URL = "http://test"
+            mock_s.TTS_API_KEY = "test"
+
+            from services.task.steps.audio import AudioStep
+            step = AudioStep(
+                level="beginner",
+                lang="en",
+                progress_tracker=Mock(),
+                context_manager=Mock()
+            )
+            assert step.cambai_tts is None
+
+
+@pytest.mark.integration
+class TestCambAIIntegration:
+    @pytest.fixture(autouse=True)
+    def skip_without_key(self):
+        if not os.environ.get("CAMB_API_KEY"):
+            pytest.skip("CAMB_API_KEY not set")
+
+    def test_real_api_generates_audio(self):
+        from camb.client import CambAI, save_stream_to_file
+        from camb.types import StreamTtsOutputConfiguration
+
+        client = CambAI(api_key=os.environ["CAMB_API_KEY"])
+        with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f:
+            filepath = f.name
+
+        try:
+            stream = client.text_to_speech.tts(
+                text="Integration test for lingopod.",
+                language="en-us",
+                voice_id=147320,
+                speech_model="mars-flash",
+                output_configuration=StreamTtsOutputConfiguration(format="mp3"),
+            )
+            save_stream_to_file(stream, filepath)
+            assert os.path.exists(filepath)
+            assert os.path.getsize(filepath) > 0
+        finally:
+            if os.path.exists(filepath):
+                os.remove(filepath)