fix(line): route inbound media to matching caches

plugins/platforms/line/adapter.py

@@ -92,7 +92,10 @@
     MessageEvent,
     MessageType,
     SendResult,
+    cache_audio_from_bytes,
+    cache_document_from_bytes,
     cache_image_from_bytes,
+    cache_video_from_bytes,
 )
 from gateway.config import Platform
 
@@ -502,7 +505,7 @@ async def loading(self, chat_id: str, seconds: int = 60) -> None:
         except Exception as exc:  # best-effort; never raise
             logger.debug("LINE loading indicator failed: %s", exc)
 
-    async def fetch_content(self, message_id: str) -> bytes:
+    async def fetch_content(self, message_id: str) -> Tuple[bytes, str]:
         """Download an inbound media message's binary content."""
         import aiohttp
         url = LINE_CONTENT_URL_FMT.format(message_id=message_id)
@@ -511,7 +514,8 @@ async def fetch_content(self, message_id: str) -> bytes:
             async with session.get(url, headers={"Authorization": f"Bearer {self._token}"}) as resp:
                 if resp.status >= 400:
                     raise RuntimeError(f"LINE content {resp.status}")
-                return await resp.read()
+                content_type = resp.headers.get("Content-Type", "")
+                return await resp.read(), content_type
 
     async def get_bot_user_id(self) -> Optional[str]:
         """Fetch this channel's own userId so we can filter self-messages."""
@@ -957,7 +961,8 @@ async def _handle_message_event(self, event: Dict[str, Any]) -> None:
             local_path = await self._download_media(message_id, msg_type)
             if local_path:
                 media_urls.append(local_path)
-                media_types.append(msg_type)
+                mime, _ = mimetypes.guess_type(local_path)
+                media_types.append(mime or "application/octet-stream")
             text = f"[{msg_type}]"
         elif msg_type == "sticker":
             keywords = msg.get("keywords") or []
@@ -1056,18 +1061,34 @@ async def _download_media(self, message_id: str, msg_type: str) -> Optional[str]
         if not self._client or not message_id:
             return None
         try:
-            data = await self._client.fetch_content(message_id)
+            data, content_type = await self._client.fetch_content(message_id)
         except Exception as exc:
             logger.warning("LINE: failed to fetch %s content for %s: %s", msg_type, message_id, exc)
             return None
-        ext = {
-            "image": ".jpg",
-            "audio": ".m4a",
-            "video": ".mp4",
-            "file": ".bin",
-        }.get(msg_type, ".bin")
+
+        ext = None
+        if content_type:
+            clean_ct = content_type.split(";")[0].strip()
+            ext = mimetypes.guess_extension(clean_ct)
+
+        if not ext:
+            ext = {
+                "image": ".jpg",
+                "audio": ".m4a",
+                "video": ".mp4",
+                "file": ".bin",
+            }.get(msg_type, ".bin")
+
         try:
-            return cache_image_from_bytes(data, ext=ext)
+            if msg_type == "image":
+                return cache_image_from_bytes(data, ext=ext)
+            if msg_type == "audio":
+                return cache_audio_from_bytes(data, ext=ext)
+            if msg_type == "video":
+                return cache_video_from_bytes(data, ext=ext)
+
+            filename = f"file_{message_id}{ext}"
+            return cache_document_from_bytes(data, filename)
         except Exception as exc:
             logger.warning("LINE: failed to cache %s payload: %s", msg_type, exc)
             return None

tests/gateway/test_line_plugin.py

@@ -19,7 +19,7 @@
 import hmac
 import base64
 import json
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -674,3 +674,86 @@ def test_every_line_type_maps_to_correct_enum(self):
     def test_unknown_type_falls_back_to_text(self):
         MessageType = _line.MessageType
         assert _line._LINE_MESSAGE_TYPES.get("flex", MessageType.TEXT) == MessageType.TEXT
+
+
+class TestInboundMediaCacheRouting:
+
+    def _adapter(self):
+        from gateway.config import PlatformConfig
+        cfg = PlatformConfig(
+            enabled=True,
+            extra={
+                "channel_access_token": "tok",
+                "channel_secret": "sec",
+            },
+        )
+        ad = LineAdapter(cfg)
+        ad.handle_message = AsyncMock()
+        return ad
+
+    def test_inbound_media_records_mime_type_from_cached_path(self):
+        ad = self._adapter()
+        ad._download_media = AsyncMock(return_value="/tmp/test_image.jpg")
+
+        event = {
+            "type": "message",
+            "replyToken": "rt",
+            "source": {"type": "user", "userId": "U1"},
+            "message": {"id": "m1", "type": "image"},
+        }
+        asyncio.run(ad._handle_message_event(event))
+        ad.handle_message.assert_awaited_once()
+        event_obj = ad.handle_message.call_args[0][0]
+        assert event_obj.media_urls == ["/tmp/test_image.jpg"]
+        assert event_obj.media_types == ["image/jpeg"]
+
+    def test_download_media_image_uses_image_cache(self):
+        ad = self._adapter()
+        ad._client = AsyncMock()
+        gif_bytes = b"GIF89a\x01\x00"
+        ad._client.fetch_content.return_value = (gif_bytes, "image/gif")
+
+        with patch.object(_line, "cache_image_from_bytes", return_value="/tmp/fake.gif") as mock_cache:
+            path = asyncio.run(ad._download_media("m1", "image"))
+            assert path == "/tmp/fake.gif"
+            mock_cache.assert_called_once_with(gif_bytes, ext=".gif")
+
+    def test_download_media_audio_uses_audio_cache(self):
+        ad = self._adapter()
+        ad._client = AsyncMock()
+        ad._client.fetch_content.return_value = (b"fake audio data", "audio/mpeg")
+
+        with patch.object(_line, "cache_audio_from_bytes", return_value="/tmp/fake.mp3") as mock_cache:
+            path = asyncio.run(ad._download_media("m2", "audio"))
+            assert path == "/tmp/fake.mp3"
+            mock_cache.assert_called_once_with(b"fake audio data", ext=".mp3")
+
+    def test_download_media_video_uses_video_cache(self):
+        ad = self._adapter()
+        ad._client = AsyncMock()
+        ad._client.fetch_content.return_value = (b"fake video data", "video/mp4")
+
+        with patch.object(_line, "cache_video_from_bytes", return_value="/tmp/fake.mp4") as mock_cache:
+            path = asyncio.run(ad._download_media("m3", "video"))
+            assert path == "/tmp/fake.mp4"
+            mock_cache.assert_called_once_with(b"fake video data", ext=".mp4")
+
+    def test_download_media_file_uses_document_cache_with_mime_extension(self):
+        ad = self._adapter()
+        ad._client = AsyncMock()
+        ad._client.fetch_content.return_value = (b"fake doc pdf data", "application/pdf")
+
+        with patch.object(_line, "cache_document_from_bytes", return_value="/tmp/fake.pdf") as mock_cache:
+            path = asyncio.run(ad._download_media("m4", "file"))
+            assert path == "/tmp/fake.pdf"
+            mock_cache.assert_called_once_with(b"fake doc pdf data", "file_m4.pdf")
+
+    def test_download_media_file_falls_back_to_bin_extension(self):
+        ad = self._adapter()
+        ad._client = AsyncMock()
+        ad._client.fetch_content.return_value = (b"fake bin data", "")
+
+        with patch.object(_line, "cache_document_from_bytes", return_value="/tmp/fake.bin") as mock_cache:
+            path = asyncio.run(ad._download_media("m5", "file"))
+            assert path == "/tmp/fake.bin"
+            mock_cache.assert_called_once_with(b"fake bin data", "file_m5.bin")