Skip to content
Closed
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
6 changes: 6 additions & 0 deletions gateway/platforms/feishu.py
Original file line number Diff line number Diff line change
Expand Up @@ -4865,6 +4865,12 @@ def _resolve_outbound_file_routing(
) -> tuple[str, str]:
ext = Path(file_path).suffix.lower()

# If the caller explicitly requests audio, honor it. Feishu accepts
# non-Opus audio uploaded as file_type=opus without codec validation,
# so MP3/WAV/M4A all render as voice bubbles when routed this way.
if requested_message_type == "audio":
return "opus", "audio"

if ext in _FEISHU_OPUS_UPLOAD_EXTENSIONS:
return "opus", "audio"

Expand Down
52 changes: 52 additions & 0 deletions tests/gateway/test_feishu.py
Original file line number Diff line number Diff line change
Expand Up @@ -2491,6 +2491,58 @@ async def _direct(func, *args, **kwargs):
self.assertEqual(captured["message_request"].request_body.msg_type, "audio")
self.assertEqual(captured["message_request"].request_body.content, '{"file_key": "file_audio_123"}')

def test_send_voice_routes_mp3_as_opus(self):
"""MP3 files should be uploaded as file_type=opus when sent via send_voice,
even though .mp3 is not in _FEISHU_OPUS_UPLOAD_EXTENSIONS. Feishu accepts
non-Opus audio uploaded as opus without codec validation."""
from gateway.config import PlatformConfig
from gateway.platforms.feishu import FeishuAdapter

adapter = FeishuAdapter(PlatformConfig())
captured = {}

class _FileAPI:
def create(self, request):
captured["upload_request"] = request
return SimpleNamespace(
success=lambda: True,
data=SimpleNamespace(file_key="file_mp3_456"),
)

class _MessageAPI:
def create(self, request):
captured["message_request"] = request
return SimpleNamespace(
success=lambda: True,
data=SimpleNamespace(message_id="om_mp3_msg"),
)

adapter._client = SimpleNamespace(
im=SimpleNamespace(
v1=SimpleNamespace(
file=_FileAPI(),
message=_MessageAPI(),
)
)
)

async def _direct(func, *args, **kwargs):
return func(*args, **kwargs)

with tempfile.NamedTemporaryFile("wb", suffix=".mp3", delete=False) as tmp:
tmp.write(b"mp3")
audio_path = tmp.name

try:
with patch("gateway.platforms.feishu.asyncio.to_thread", side_effect=_direct):
result = asyncio.run(adapter.send_voice(chat_id="oc_chat", audio_path=audio_path))
finally:
os.unlink(audio_path)

self.assertTrue(result.success)
self.assertEqual(captured["upload_request"].request_body.file_type, "opus")
self.assertEqual(captured["message_request"].request_body.msg_type, "audio")

@patch.dict(os.environ, {}, clear=True)
def test_build_post_payload_extracts_title_and_links(self):
from gateway.config import PlatformConfig
Expand Down
29 changes: 29 additions & 0 deletions tests/tools/test_tts_opus_routing.py
Original file line number Diff line number Diff line change
Expand Up @@ -68,3 +68,32 @@ def fake_convert(path: str) -> str:
assert result["voice_compatible"] is True
assert result["media_tag"] == f"[[audio_as_voice]]\nMEDIA:{opus}"
convert.assert_called_once_with(str(out))


def test_minimax_feishu_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", "feishu")
monkeypatch.setattr(tts_tool, "_load_tts_config", lambda: {"provider": "minimax"})
monkeypatch.setattr(
tts_tool,
"_generate_minimax_tts",
lambda _text, _output, _cfg: (_ := Path(_output).write_bytes(b"mp3")) or _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))
2 changes: 1 addition & 1 deletion tools/tts_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -2064,7 +2064,7 @@ def text_to_speech_tool(
# and needs ffmpeg for conversion.
from gateway.session_context import get_session_env
platform = get_session_env("HERMES_SESSION_PLATFORM", "").lower()
want_opus = (platform == "telegram")
want_opus = platform in {"telegram", "feishu"}

# Determine output path
if output_path:
Expand Down