mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-31 19:16:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

fix(media): recognize m2a audio attachments

Browse source
This commit is contained in:
luyifan 2026-07-24 19:40:38 +08:00 • committed by Teknium
parent 971cb9b886
commit 4ae27548d6
8 changed files with 139 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

23
gateway/platforms/base.py
View file

@ -26,9 +26,18 @@ from utils import normalize_proxy_url
logger = logging.getLogger(__name__)
# Audio file extensions Hermes recognizes for native audio delivery.
# Kept in sync with tools/send_message_tool.py and cron/scheduler.py via
# should_send_media_as_audio() below.
_AUDIO_EXTS = frozenset({'.ogg', '.opus', '.mp3', '.wav', '.m4a', '.flac'})
# Keep Telegram's narrower attachment/voice sets below separate: formats such
# as MPEG-2 Layer II are audio to Hermes but unsupported by sendAudio/sendVoice.
_AUDIO_MIME_TYPES = {
".ogg": "audio/ogg",
".opus": "audio/opus",
".mp3": "audio/mpeg",
".m2a": "audio/mpeg",
".wav": "audio/wav",
".m4a": "audio/m4a",
".flac": "audio/flac",
}
_AUDIO_EXTS = frozenset(_AUDIO_MIME_TYPES)
# Telegram's Bot API sendAudio only accepts MP3 / M4A. Other audio
# formats either need to go through sendVoice (Opus/OGG) or must be
# delivered as a regular document.
@ -1499,7 +1508,7 @@ MEDIA_DELIVERY_EXTS: Tuple[str, ...] = (
# Video (embed inline where supported)
".mp4", ".mov", ".avi", ".mkv", ".webm", ".3gp",
# Audio (delivered as voice/audio where supported)
".mp3", ".wav", ".ogg", ".opus", ".m4a", ".flac",
".mp3", ".m2a", ".wav", ".ogg", ".opus", ".m4a", ".flac",
# Documents (uploaded as file attachments)
".pdf", ".docx", ".doc", ".odt", ".rtf", ".txt", ".md", ".epub",
# Spreadsheets / data
@ -1857,7 +1866,7 @@ def cache_media_bytes(
or default_kind == "image"
)
is_video = mime.startswith("video/") or ext in SUPPORTED_VIDEO_TYPES or default_kind == "video"
is_audio = mime.startswith("audio/") or default_kind == "audio"
is_audio = mime.startswith("audio/") or ext in _AUDIO_EXTS or default_kind == "audio"
if is_image:
img_ext = ext if ext in SUPPORTED_IMAGE_DOCUMENT_TYPES else ".jpg"
@ -1874,9 +1883,9 @@ def cache_media_bytes(
return CachedMedia(to_agent_visible_cache_path(path), SUPPORTED_VIDEO_TYPES.get(vid_ext, "video/mp4"), "video", display)
if is_audio:
aud_ext = ext if ext in {".ogg", ".mp3", ".wav", ".m4a", ".opus", ".flac"} else ".ogg"
aud_ext = ext if ext in _AUDIO_EXTS else ".ogg"
path = cache_audio_from_bytes(data, ext=aud_ext)
out_mime = mime if mime.startswith("audio/") else f"audio/{aud_ext.lstrip('.')}"
out_mime = mime if mime.startswith("audio/") else _AUDIO_MIME_TYPES[aud_ext]
return CachedMedia(to_agent_visible_cache_path(path), out_mime, "audio", display)
# Any other file type is cached and surfaced to the agent as a local path

33
plugins/platforms/telegram/adapter.py
View file

@ -8310,6 +8310,8 @@ class TelegramAdapter(BasePlatformAdapter):
event.message_type = MessageType.PHOTO
elif cached.kind == "video":
event.message_type = MessageType.VIDEO
elif cached.kind == "audio":
event.message_type = MessageType.AUDIO
event.text = self._append_observed_note(event.text, cached.context_note())
logger.info("[Telegram] Cached observed group %s at %s", cached.kind, cached.path)
@ -8353,6 +8355,8 @@ class TelegramAdapter(BasePlatformAdapter):
event.message_type = MessageType.PHOTO
elif cached.kind == "video":
event.message_type = MessageType.VIDEO
elif cached.kind == "audio":
event.message_type = MessageType.AUDIO
event.text = self._append_observed_note(
event.text,
f"[Replied-to {cached.kind} '{cached.display_name}' saved at: {cached.path}]",
@ -9103,11 +9107,30 @@ class TelegramAdapter(BasePlatformAdapter):
file_obj = await doc.get_file()
doc_bytes = await file_obj.download_as_bytearray()
raw_bytes = bytes(doc_bytes)
cached_path = cache_document_from_bytes(raw_bytes, original_filename or f"document{ext or '.bin'}")
mime_type = SUPPORTED_DOCUMENT_TYPES.get(ext) or doc.mime_type or "application/octet-stream"
event.media_urls = [cached_path]
event.media_types = [mime_type]
logger.info("[Telegram] Cached user document at %s (%s)", cached_path, mime_type)
from gateway.platforms.base import cache_media_bytes
cached = cache_media_bytes(
raw_bytes,
filename=original_filename or f"document{ext or '.bin'}",
mime_type=doc_mime,
)
if cached is None:
event.text = (
f"Document '{original_filename or doc_mime or ext or 'unknown'}' "
"could not be cached."
)
await self.handle_message(event)
return
event.media_urls = [cached.path]
event.media_types = [cached.media_type]
if cached.kind == "audio":
event.message_type = MessageType.AUDIO
logger.info(
"[Telegram] Cached user %s at %s (%s)",
cached.kind,
cached.path,
cached.media_type,
)
# For text-readable files, inject content into event.text (capped
# at 100 KB). Gate on a text-like extension/MIME — NOT a blind

14
tests/gateway/test_document_cache.py
View file

@ -28,6 +28,9 @@ def _redirect_cache(tmp_path, monkeypatch):
monkeypatch.setattr(
"gateway.platforms.base.DOCUMENT_CACHE_DIR", tmp_path / "doc_cache"
)
monkeypatch.setattr(
"gateway.platforms.base.AUDIO_CACHE_DIR", tmp_path / "audio_cache"
)
# ---------------------------------------------------------------------------
@ -211,6 +214,17 @@ class TestCacheMediaBytes:
assert result.kind == "video"
assert result.media_type == "video/mp4"
def test_m2a_routes_to_mpeg_audio_without_mime_hint(self):
from gateway.platforms.base import cache_media_bytes
result = cache_media_bytes(b"mpeg-audio", filename="clip.m2a", mime_type="")
assert result is not None
assert result.kind == "audio"
assert result.media_type == "audio/mpeg"
assert result.path.endswith(".m2a")
assert os.path.exists(result.path)
def test_mime_only_resolves_extension(self):
from gateway.platforms.base import cache_media_bytes
result = cache_media_bytes(b"col1,col2\n1,2", filename="", mime_type="text/csv")

2
tests/gateway/test_extract_local_files.py
View file

@ -99,7 +99,7 @@ class TestBasicDetection:
def test_audio_extensions(self):
"""Audio files are detected and routed by the gateway dispatch."""
for ext in (".mp3", ".wav", ".ogg", ".m4a", ".flac"):
for ext in (".mp3", ".m2a", ".wav", ".ogg", ".m4a", ".flac"):
text = f"Audio at /tmp/sound{ext} ready"
paths, _ = _extract(text)
assert len(paths) == 1, f"Failed for {ext}"

7
tests/gateway/test_platform_base.py
View file

@ -1625,7 +1625,7 @@ class TestShouldSendMediaAsAudio:
def test_non_telegram_platforms_route_all_audio(self):
from gateway.platforms.base import should_send_media_as_audio
for ext in (".mp3", ".m4a", ".wav", ".flac", ".ogg", ".opus"):
for ext in (".mp3", ".m2a", ".m4a", ".wav", ".flac", ".ogg", ".opus"):
assert should_send_media_as_audio("discord", ext) is True
assert should_send_media_as_audio("slack", ext) is True
@ -1639,6 +1639,11 @@ class TestShouldSendMediaAsAudio:
assert should_send_media_as_audio("telegram", ".wav") is False
assert should_send_media_as_audio("telegram", ".flac") is False
def test_telegram_m2a_falls_through_to_document(self):
from gateway.platforms.base import should_send_media_as_audio
assert should_send_media_as_audio("telegram", ".m2a") is False
def test_telegram_ogg_opus_only_when_voice_flagged(self):
from gateway.platforms.base import should_send_media_as_audio
assert should_send_media_as_audio("telegram", ".ogg", is_voice=True) is True

46
tests/gateway/test_telegram_documents.py
View file

@ -150,6 +150,9 @@ def _redirect_cache(tmp_path, monkeypatch):
monkeypatch.setattr(
"gateway.platforms.base.VIDEO_CACHE_DIR", tmp_path / "video_cache"
)
monkeypatch.setattr(
"gateway.platforms.base.AUDIO_CACHE_DIR", tmp_path / "audio_cache"
)
# ---------------------------------------------------------------------------
@ -202,6 +205,27 @@ class TestDocumentDownloadBlock:
assert os.path.exists(event.media_urls[0])
assert event.media_types == ["application/pdf"]
@pytest.mark.asyncio
async def test_m2a_document_is_cached_as_audio(self, adapter):
audio_bytes = b"mpeg-audio"
file_obj = _make_file_obj(audio_bytes)
doc = _make_document(
file_name="clip.m2a",
mime_type="application/octet-stream",
file_size=len(audio_bytes),
file_obj=file_obj,
)
msg = _make_message(document=doc)
update = _make_update(msg)
await adapter._handle_media_message(update, MagicMock())
event = adapter.handle_message.call_args[0][0]
assert event.message_type == MessageType.AUDIO
assert event.media_types == ["audio/mpeg"]
assert event.media_urls[0].endswith(".m2a")
assert os.path.exists(event.media_urls[0])
@pytest.mark.asyncio
async def test_supported_txt_injects_content(self, adapter):
content = b"Hello from a text file"
@ -629,6 +653,28 @@ class TestSendVoice:
connected_adapter._bot.send_document.assert_awaited_once()
connected_adapter._bot.send_audio.assert_not_awaited()
@pytest.mark.asyncio
async def test_m2a_falls_back_to_document(self, connected_adapter, tmp_path):
"""MPEG-2 Layer II is audio, but Telegram sendAudio does not accept M2A."""
audio_file = tmp_path / "clip.m2a"
audio_file.write_bytes(b"mpeg-audio")
mock_msg = MagicMock()
mock_msg.message_id = 104
connected_adapter._bot.send_voice = AsyncMock()
connected_adapter._bot.send_audio = AsyncMock()
connected_adapter._bot.send_document = AsyncMock(return_value=mock_msg)
result = await connected_adapter.send_voice(
chat_id="12345",
audio_path=str(audio_file),
)
assert result.success is True
connected_adapter._bot.send_document.assert_awaited_once()
connected_adapter._bot.send_audio.assert_not_awaited()
connected_adapter._bot.send_voice.assert_not_awaited()
@pytest.mark.asyncio
async def test_mp3_routes_to_send_audio(self, connected_adapter, tmp_path):
"""MP3 is Telegram-sendAudio-compatible."""

27
tests/tools/test_send_message_tool.py
View file

@ -683,6 +683,33 @@ class TestSendTelegramMediaDelivery:
bot.send_audio.assert_awaited_once()
bot.send_voice.assert_not_awaited()
def test_sends_m2a_as_document(self, tmp_path, monkeypatch):
audio_path = tmp_path / "clip.m2a"
audio_path.write_bytes(b"mpeg-audio")
bot = MagicMock()
bot.send_message = AsyncMock()
bot.send_photo = AsyncMock()
bot.send_video = AsyncMock()
bot.send_voice = AsyncMock()
bot.send_audio = AsyncMock()
bot.send_document = AsyncMock(return_value=SimpleNamespace(message_id=9))
_install_telegram_mock(monkeypatch, bot)
result = asyncio.run(
_send_telegram(
"token",
"12345",
"",
media_files=[(str(audio_path), False)],
)
)
assert result["success"] is True
bot.send_document.assert_awaited_once()
bot.send_audio.assert_not_awaited()
bot.send_voice.assert_not_awaited()
def test_missing_media_returns_error_without_leaking_raw_tag(self, monkeypatch):
bot = MagicMock()
bot.send_message = AsyncMock()

2
tools/send_message_tool.py
View file

@ -63,7 +63,7 @@ _EMAIL_TARGET_RE = re.compile(r"^\s*[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2
_HOME_CHANNEL_ENV_OVERRIDES = {"email": "EMAIL_HOME_ADDRESS"}
_IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".gif"}
_VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".3gp"}
_AUDIO_EXTS = {".ogg", ".opus", ".mp3", ".wav", ".m4a", ".flac"}
_AUDIO_EXTS = {".ogg", ".opus", ".mp3", ".m2a", ".wav", ".m4a", ".flac"}
_VOICE_EXTS = {".ogg", ".opus"}
# Telegram's Bot API sendAudio only accepts MP3 / M4A. Other audio
# formats either route through sendVoice (Opus/OGG) or fall back to