diff --git a/tests/tools/test_tts_speed.py b/tests/tools/test_tts_speed.py index d9274bb84d79..fde0045447b6 100644 --- a/tests/tools/test_tts_speed.py +++ b/tests/tools/test_tts_speed.py @@ -136,12 +136,12 @@ def _hex_response(payload_audio: bytes = b"\x00\x01\x02\x03"): class TestMinimaxTtsT2aV2: """Default path: base_url contains 't2a_v2'.""" - def _run(self, tts_config, tmp_path, monkeypatch, response=None): + def _run(self, tts_config, tmp_path, monkeypatch, response=None, suffix=".mp3"): monkeypatch.setenv("MINIMAX_API_KEY", "test-key") resp = response if response is not None else _hex_response() with patch("requests.post", return_value=resp) as mock_post: from tools.tts_tool import _generate_minimax_tts - output = _generate_minimax_tts("Hello", str(tmp_path / "out.mp3"), tts_config) + output = _generate_minimax_tts("Hello", str(tmp_path / f"out{suffix}"), tts_config) return mock_post, output def test_nested_payload(self, tmp_path, monkeypatch): @@ -157,6 +157,12 @@ def test_nested_payload(self, tmp_path, monkeypatch): # Don't send flat top-level voice_id alongside nested voice_setting. assert "voice_id" not in payload + def test_ogg_output_requests_opus(self, tmp_path, monkeypatch): + """Native Opus is requested when the output path is an OGG file.""" + mock_post, _ = self._run({}, tmp_path, monkeypatch, suffix=".ogg") + payload = mock_post.call_args[1]["json"] + assert payload["audio_setting"]["format"] == "opus" + def test_decodes_hex_audio(self, tmp_path, monkeypatch): """t2a_v2 hex-encoded audio is decoded and written verbatim.""" _, output = self._run({}, tmp_path, monkeypatch) @@ -207,6 +213,74 @@ def test_api_error_raises(self, tmp_path, monkeypatch): self._run({}, tmp_path, monkeypatch, response=resp) +class TestMinimaxTelegramOutput: + def test_explicit_mp3_path_is_rewritten_to_native_ogg(self, tmp_path): + """Gateway temp paths must not force Telegram MiniMax replies to MP3.""" + import json + + from tools.tts_tool import text_to_speech_tool + + requested_paths = [] + + def fake_generate(_text, output_path, _config): + requested_paths.append(output_path) + with open(output_path, "wb") as f: + f.write(b"opus") + + with patch( + "tools.tts_tool._load_tts_config", + return_value={"provider": "minimax"}, + ), patch( + "gateway.session_context.get_session_env", + return_value="telegram", + ), patch( + "tools.tts_tool._generate_minimax_tts", + side_effect=fake_generate, + ): + result = json.loads(text_to_speech_tool( + "Hello", output_path=str(tmp_path / "reply.mp3") + )) + + assert requested_paths == [str(tmp_path / "reply.ogg")] + assert result["file_path"] == str(tmp_path / "reply.ogg") + assert result["voice_compatible"] is True + + +class TestElevenLabsTelegramOutput: + def test_explicit_mp3_path_is_rewritten_to_native_ogg(self, tmp_path): + """Gateway temp paths must not force Telegram ElevenLabs replies to MP3.""" + import json + + from tools.tts_tool import text_to_speech_tool + + requested_paths = [] + + def fake_generate(_text, output_path, _config): + requested_paths.append(output_path) + with open(output_path, "wb") as f: + f.write(b"opus") + + with patch( + "tools.tts_tool._load_tts_config", + return_value={"provider": "elevenlabs"}, + ), patch( + "gateway.session_context.get_session_env", + return_value="telegram", + ), patch( + "tools.tts_tool._import_elevenlabs", + ), patch( + "tools.tts_tool._generate_elevenlabs", + side_effect=fake_generate, + ): + result = json.loads(text_to_speech_tool( + "Hello", output_path=str(tmp_path / "reply.mp3") + )) + + assert requested_paths == [str(tmp_path / "reply.ogg")] + assert result["file_path"] == str(tmp_path / "reply.ogg") + assert result["voice_compatible"] is True + + class TestMinimaxTtsLegacyTextToSpeech: """Legacy path: caller pins base_url to the old text_to_speech endpoint.""" diff --git a/tools/tts_tool.py b/tools/tts_tool.py index cab2cc584ab0..13628b5f6e7e 100644 --- a/tools/tts_tool.py +++ b/tools/tts_tool.py @@ -1245,7 +1245,7 @@ def _generate_minimax_tts(text: str, output_path: str, tts_config: Dict[str, Any "audio_setting": { "sample_rate": sample_rate, "bitrate": bitrate, - "format": "mp3", + "format": "opus" if output_path.endswith(".ogg") else "mp3", "channel": 1, }, } @@ -1834,6 +1834,18 @@ def _generate_kittentts(text: str, output_path: str, tts_config: Dict[str, Any]) # =========================================================================== # Main tool function # =========================================================================== +def _supports_opus_output(provider: str, tts_config: Dict[str, Any]) -> bool: + """Return whether *provider* can write an Opus OGG output path.""" + if provider in {"elevenlabs", "openai", "mistral", "gemini"}: + return True + if provider == "minimax": + base_url = tts_config.get("minimax", {}).get( + "base_url", DEFAULT_MINIMAX_BASE_URL + ) + return "t2a_v2" in base_url + return False + + def text_to_speech_tool( text: str, output_path: Optional[str] = None, @@ -1884,6 +1896,7 @@ def text_to_speech_tool( from gateway.session_context import get_session_env platform = get_session_env("HERMES_SESSION_PLATFORM", "").lower() want_opus = (platform == "telegram") + supports_opus_output = _supports_opus_output(provider, tts_config) # Determine output path if output_path: @@ -1913,6 +1926,8 @@ def text_to_speech_tool( file_path = _configured_command_tts_output_path( file_path, command_provider_config ) + elif want_opus and supports_opus_output and file_path.suffix.lower() == ".mp3": + file_path = file_path.with_suffix(".ogg") else: timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") out_dir = Path(DEFAULT_OUTPUT_DIR) @@ -1922,7 +1937,7 @@ def text_to_speech_tool( file_path = out_dir / f"tts_{timestamp}.{fmt}" # Use .ogg for Telegram with providers that support native Opus output, # otherwise fall back to .mp3 (Edge TTS will attempt ffmpeg conversion later). - elif want_opus and provider in {"openai", "elevenlabs", "mistral", "gemini"}: + elif want_opus and supports_opus_output: file_path = out_dir / f"tts_{timestamp}.ogg" else: file_path = out_dir / f"tts_{timestamp}.mp3" @@ -2110,7 +2125,7 @@ def text_to_speech_tool( if opus_path: file_str = opus_path voice_compatible = True - elif provider in {"elevenlabs", "openai", "mistral", "gemini"}: + elif supports_opus_output: voice_compatible = want_opus and file_str.endswith(".ogg") file_size = os.path.getsize(file_str) diff --git a/website/docs/user-guide/features/tts.md b/website/docs/user-guide/features/tts.md index 96c33d745b91..271ff011a02b 100644 --- a/website/docs/user-guide/features/tts.md +++ b/website/docs/user-guide/features/tts.md @@ -141,7 +141,7 @@ Telegram voice bubbles require Opus/OGG audio format: - **OpenAI, ElevenLabs, and Mistral** produce Opus natively — no extra setup - **Edge TTS** (default) outputs MP3 and needs **ffmpeg** to convert: -- **MiniMax TTS** outputs MP3 and needs **ffmpeg** to convert for Telegram voice bubbles +- **MiniMax TTS** produces Opus natively with the default `t2a_v2` endpoint; legacy endpoints output MP3 and need **ffmpeg** - **Google Gemini TTS** outputs raw PCM and uses **ffmpeg** to encode Opus directly for Telegram voice bubbles - **xAI TTS** outputs MP3 and needs **ffmpeg** to convert for Telegram voice bubbles - **NeuTTS** outputs WAV and also needs **ffmpeg** to convert for Telegram voice bubbles