diff --git a/nanobot/channels/telegram.py b/nanobot/channels/telegram.py index a5942da..34c4a3b 100644 --- a/nanobot/channels/telegram.py +++ b/nanobot/channels/telegram.py @@ -534,7 +534,8 @@ class TelegramChannel(BaseChannel): getattr(media_file, "file_name", None), ) media_dir = get_media_dir("telegram") - file_path = media_dir / f"{media_file.file_id[:16]}{ext}" + unique_id = getattr(media_file, "file_unique_id", media_file.file_id) + file_path = media_dir / f"{unique_id}{ext}" await file.download_to_drive(str(file_path)) path_str = str(file_path) if media_type in ("voice", "audio"): diff --git a/tests/test_telegram_channel.py b/tests/test_telegram_channel.py index c96f5e4..4c34469 100644 --- a/tests/test_telegram_channel.py +++ b/tests/test_telegram_channel.py @@ -446,6 +446,56 @@ async def test_download_message_media_returns_path_when_download_succeeds( assert "[image:" in parts[0] +@pytest.mark.asyncio +async def test_download_message_media_uses_file_unique_id_when_available( + monkeypatch, tmp_path +) -> None: + media_dir = tmp_path / "media" / "telegram" + media_dir.mkdir(parents=True) + monkeypatch.setattr( + "nanobot.channels.telegram.get_media_dir", + lambda channel=None: media_dir if channel else tmp_path / "media", + ) + + downloaded: dict[str, str] = {} + + async def _download_to_drive(path: str) -> None: + downloaded["path"] = path + + channel = TelegramChannel( + TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]), + MessageBus(), + ) + app = _FakeApp(lambda: None) + app.bot.get_file = AsyncMock( + return_value=SimpleNamespace(download_to_drive=_download_to_drive) + ) + channel._app = app + + msg = SimpleNamespace( + photo=[ + SimpleNamespace( + file_id="file-id-that-should-not-be-used", + file_unique_id="stable-unique-id", + mime_type="image/jpeg", + file_name=None, + ) + ], + voice=None, + audio=None, + document=None, + video=None, + video_note=None, + animation=None, + ) + + paths, parts = await channel._download_message_media(msg) + + assert downloaded["path"].endswith("stable-unique-id.jpg") + assert paths == [str(media_dir / "stable-unique-id.jpg")] + assert parts == [f"[image: {media_dir / 'stable-unique-id.jpg'}]"] + + @pytest.mark.asyncio async def test_on_message_attaches_reply_to_media_when_available(monkeypatch, tmp_path) -> None: """When user replies to a message with media, that media is downloaded and attached to the turn."""