fix(whatsapp): route media by detected MIME type

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<user>[^@]+)@(?P<server>[^@]+)$")
 _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,

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()