Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions scripts/release.py
Original file line number Diff line number Diff line change
Expand Up @@ -174,6 +174,7 @@
"zyrixtrex@gmail.com": "Zyrixtrex", # PR #26754 salvage (avoid duplicate text after auto-TTS)
"264138787+nftpoetrist@users.noreply.github.com": "nftpoetrist", # PR #25856 salvage (escape slash-confirm preview)
"197455947+samahn0601@users.noreply.github.com": "samahn0601", # PR #27887 salvage (retry wrapped connect timeouts)
"gonzes7@gmail.com": "aqilaziz", # PR #26406 salvage (preserve native audio outside Telegram)
"282919977+eliteworkstation94-ai@users.noreply.github.com": "eliteworkstation94-ai", # PR #28157 salvage (group reply session splits)
"androidhtml@yandex.com": "hllqkb",
"25840394+Bongulielmi@users.noreply.github.com": "Bongulielmi",
Expand Down
70 changes: 70 additions & 0 deletions tests/tools/test_tts_opus_routing.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
import json
from pathlib import Path
from unittest.mock import Mock

import pytest

from gateway.session_context import _UNSET, _VAR_MAP
from tools import tts_tool


def _reset_session_context() -> None:
for var in _VAR_MAP.values():
var.set(_UNSET)


@pytest.fixture(autouse=True)
def _clean_session_platform(monkeypatch):
_reset_session_context()
monkeypatch.delenv("HERMES_SESSION_PLATFORM", raising=False)
yield
_reset_session_context()


async def _write_edge_output(_text: str, output_path: str, _tts_config: dict) -> str:
Path(output_path).write_bytes(b"mp3")
return output_path


def test_edge_cli_preserves_native_mp3(tmp_path, monkeypatch):
out = tmp_path / "speech.mp3"
convert = Mock()

monkeypatch.setattr(tts_tool, "_load_tts_config", lambda: {"provider": "edge"})
monkeypatch.setattr(tts_tool, "_import_edge_tts", lambda: object())
monkeypatch.setattr(tts_tool, "_generate_edge_tts", _write_edge_output)
monkeypatch.setattr(tts_tool, "_convert_to_opus", convert)

result = json.loads(tts_tool.text_to_speech_tool("hello", output_path=str(out)))

assert result["success"] is True
assert result["file_path"] == str(out)
assert result["voice_compatible"] is False
assert result["media_tag"] == f"MEDIA:{out}"
convert.assert_not_called()


def test_edge_telegram_converts_to_opus_voice(tmp_path, monkeypatch):
out = tmp_path / "speech.mp3"
opus = tmp_path / "speech.ogg"

def fake_convert(path: str) -> str:
assert path == str(out)
opus.write_bytes(b"ogg")
return str(opus)

convert = Mock(side_effect=fake_convert)

monkeypatch.setenv("HERMES_SESSION_PLATFORM", "telegram")
monkeypatch.setattr(tts_tool, "_load_tts_config", lambda: {"provider": "edge"})
monkeypatch.setattr(tts_tool, "_import_edge_tts", lambda: object())
monkeypatch.setattr(tts_tool, "_generate_edge_tts", _write_edge_output)
monkeypatch.setattr(tts_tool, "_convert_to_opus", convert)

result = json.loads(tts_tool.text_to_speech_tool("hello", output_path=str(out)))

assert result["success"] is True
assert result["file_path"] == str(opus)
assert result["voice_compatible"] is True
assert result["media_tag"] == f"[[audio_as_voice]]\nMEDIA:{opus}"
convert.assert_called_once_with(str(out))
14 changes: 10 additions & 4 deletions tools/tts_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -1830,8 +1830,10 @@ def text_to_speech_tool(
"error": f"TTS generation produced no output (provider: {provider})"
}, ensure_ascii=False)

# Try Opus conversion for Telegram compatibility
# Edge TTS outputs MP3, NeuTTS/KittenTTS output WAV — all need ffmpeg conversion
# Try Opus conversion for Telegram compatibility.
# Edge TTS outputs MP3, NeuTTS/KittenTTS output WAV. Keep those native
# formats for local/CLI playback and only convert when the current
# platform actually needs Opus voice delivery.
voice_compatible = False
if command_provider_config is not None:
# Command providers are documents by default. Voice-bubble
Expand All @@ -1843,13 +1845,17 @@ def text_to_speech_tool(
if opus_path:
file_str = opus_path
voice_compatible = file_str.endswith(".ogg")
elif provider in {"edge", "neutts", "minimax", "xai", "kittentts", "piper"} and not file_str.endswith(".ogg"):
elif (
want_opus
and provider in {"edge", "neutts", "minimax", "xai", "kittentts", "piper"}
and not file_str.endswith(".ogg")
):
opus_path = _convert_to_opus(file_str)
if opus_path:
file_str = opus_path
voice_compatible = True
elif provider in {"elevenlabs", "openai", "mistral", "gemini"}:
voice_compatible = file_str.endswith(".ogg")
voice_compatible = want_opus and file_str.endswith(".ogg")

file_size = os.path.getsize(file_str)
logger.info("TTS audio saved: %s (%s bytes, provider: %s)", file_str, f"{file_size:,}", provider)
Expand Down
Loading