From 353dfed5023b0c3e44b412699a5242811b401af8 Mon Sep 17 00:00:00 2001 From: chengyongru <2755839590@qq.com> Date: Sat, 1 Aug 2026 12:09:54 +0800 Subject: [PATCH] fix(whatsapp): route media by detected MIME type --- nanobot/channels/whatsapp/runtime.py | 27 ++++++- .../whatsapp/tests/test_whatsapp_channel.py | 77 ++++++++++++++++++- 2 files changed, 98 insertions(+), 6 deletions(-) diff --git a/nanobot/channels/whatsapp/runtime.py b/nanobot/channels/whatsapp/runtime.py index e576bfe99..e25a73e99 100644 --- a/nanobot/channels/whatsapp/runtime.py +++ b/nanobot/channels/whatsapp/runtime.py @@ -39,6 +39,7 @@ class _NeonizeAPI(NamedTuple): MessageEv: Any PairStatusEv: Any build_jid: Any + detect_mime: Any class _MediaInfo(NamedTuple): @@ -52,6 +53,8 @@ class _MediaInfo(NamedTuple): _NEONIZE_API: _NeonizeAPI | None = None _JID_RE = re.compile(r"^(?P[^@]+)@(?P[^@]+)$") _LEGACY_BRIDGE_CONFIG_FIELDS = ("bridgeUrl", "bridgeToken", "bridge_url", "bridge_token") +# OGG is intentionally excluded: WhatsApp accepts only mono Opus, which MIME sniffing cannot prove. +_DIRECT_AUDIO_MIMETYPES = {"audio/aac", "audio/amr", "audio/mp4", "audio/mpeg"} def _default_database_path() -> Path: @@ -68,9 +71,14 @@ def _load_neonize() -> _NeonizeAPI: return _NEONIZE_API try: + import magic from neonize.aioze.client import NewAClient from neonize.aioze.events import ConnectedEv, DisconnectedEv, MessageEv, PairStatusEv from neonize.utils.jid import build_jid + + detect_mime = getattr(magic, "from_file", None) + if not callable(detect_mime): + raise ImportError("python-magic does not expose from_file") except ImportError as exc: raise RuntimeError( "WhatsApp dependencies not installed. Run: nanobot plugins enable whatsapp" @@ -83,6 +91,7 @@ def _load_neonize() -> _NeonizeAPI: MessageEv=MessageEv, PairStatusEv=PairStatusEv, build_jid=build_jid, + detect_mime=detect_mime, ) return _NEONIZE_API @@ -418,13 +427,12 @@ class WhatsAppChannel(BaseChannel): async def _send_media(self, client: Any, to: Any, media_path: str) -> None: path = str(Path(media_path).expanduser()) - mime, _ = mimetypes.guess_type(path) - mimetype = mime or "application/octet-stream" + mimetype = self._detect_mimetype(path) if mimetype.startswith("image/"): await client.send_image(to, path) elif mimetype.startswith("video/"): await client.send_video(to, path) - elif mimetype.startswith("audio/"): + elif mimetype in _DIRECT_AUDIO_MIMETYPES: await client.send_audio(to, path) else: await client.send_document( @@ -434,6 +442,19 @@ class WhatsAppChannel(BaseChannel): mimetype=mimetype, ) + def _detect_mimetype(self, path: str) -> str: + try: + detected = _load_neonize().detect_mime(path, mime=True) + except Exception as exc: + self.logger.debug("Failed to inspect WhatsApp media {}: {}", path, exc) + detected = None + + if isinstance(detected, str) and "/" in detected: + return detected.partition(";")[0].strip().lower() + + guessed, _ = mimetypes.guess_type(path) + return guessed or "application/octet-stream" + def _register_handlers( self, client: Any, diff --git a/nanobot/channels/whatsapp/tests/test_whatsapp_channel.py b/nanobot/channels/whatsapp/tests/test_whatsapp_channel.py index 86dd0668c..e25bc7963 100644 --- a/nanobot/channels/whatsapp/tests/test_whatsapp_channel.py +++ b/nanobot/channels/whatsapp/tests/test_whatsapp_channel.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import mimetypes import sys import types from types import SimpleNamespace @@ -78,7 +79,10 @@ def _make_channel(config: dict | None = None) -> WhatsAppChannel: return ch -def _patch_neonize_api(monkeypatch) -> None: +def _patch_neonize_api(monkeypatch, detect_mime=None) -> None: + detect_mime = detect_mime or ( + lambda path, *, mime: mimetypes.guess_type(path)[0] or "application/octet-stream" + ) monkeypatch.setattr( whatsapp_module, "_NEONIZE_API", @@ -89,6 +93,7 @@ def _patch_neonize_api(monkeypatch) -> None: MessageEv=object(), PairStatusEv=object(), build_jid=lambda user, server="s.whatsapp.net": (user, server), + detect_mime=detect_mime, ), ) @@ -213,14 +218,14 @@ async def test_send_media_dispatches_by_mimetype(monkeypatch) -> None: channel="whatsapp", chat_id="12345@s.whatsapp.net", content="", - media=["photo.jpg", "clip.mp4", "voice.ogg", "report.pdf"], + media=["photo.jpg", "clip.mp4", "voice.mp3", "report.pdf"], ) ) jid = ("12345", "s.whatsapp.net") client.send_image.assert_awaited_once_with(jid, "photo.jpg") client.send_video.assert_awaited_once_with(jid, "clip.mp4") - client.send_audio.assert_awaited_once_with(jid, "voice.ogg") + client.send_audio.assert_awaited_once_with(jid, "voice.mp3") client.send_document.assert_awaited_once_with( jid, "report.pdf", @@ -229,6 +234,72 @@ async def test_send_media_dispatches_by_mimetype(monkeypatch) -> None: ) +@pytest.mark.asyncio +async def test_send_mislabeled_audio_as_document(monkeypatch) -> None: + _patch_neonize_api(monkeypatch, detect_mime=lambda path, *, mime: "audio/x-wav") + client = SimpleNamespace( + send_message=AsyncMock(), + send_image=AsyncMock(), + send_video=AsyncMock(), + send_audio=AsyncMock(), + send_document=AsyncMock(), + ) + ch = _make_channel() + ch._client = client + ch._connected = True + + await ch.send( + OutboundMessage( + channel="whatsapp", + chat_id="12345@s.whatsapp.net", + content="", + media=["recording.mpeg"], + ) + ) + + jid = ("12345", "s.whatsapp.net") + client.send_document.assert_awaited_once_with( + jid, + "recording.mpeg", + filename="recording.mpeg", + mimetype="audio/x-wav", + ) + client.send_video.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_send_unsupported_ogg_audio_as_document(monkeypatch) -> None: + _patch_neonize_api(monkeypatch, detect_mime=lambda path, *, mime: "audio/ogg") + client = SimpleNamespace( + send_message=AsyncMock(), + send_image=AsyncMock(), + send_video=AsyncMock(), + send_audio=AsyncMock(), + send_document=AsyncMock(), + ) + ch = _make_channel() + ch._client = client + ch._connected = True + + await ch.send( + OutboundMessage( + channel="whatsapp", + chat_id="12345@s.whatsapp.net", + content="", + media=["voice.ogg"], + ) + ) + + jid = ("12345", "s.whatsapp.net") + client.send_document.assert_awaited_once_with( + jid, + "voice.ogg", + filename="voice.ogg", + mimetype="audio/ogg", + ) + client.send_audio.assert_not_awaited() + + @pytest.mark.asyncio async def test_send_when_disconnected_raises() -> None: ch = _make_channel()