fix: download telegram photo/sticker/document/video to local temp path

Fixes #9448: these components stored Telegram's file URL directly,
which generic message components can't resolve as a local file.

Author: wcqqq1214

astrbot/core/platform/sources/telegram/tg_adapter.py

@@ -436,6 +436,26 @@ async def message_handler(
         if abm:
             await self.handle_msg(abm)
 
+    async def _download_to_temp(self, file_path: str) -> str:
+        """Download a Telegram file to a local temp path.
+
+        Telegram's file_path can be a relative path (self-hosted Bot API server
+        or reverse proxy setups), and the generic message components only
+        understand local paths or http(s) URLs. So we always fetch the file
+        ourselves instead of passing file_path straight into the component.
+
+        Args:
+            file_path: The file_path returned by Telegram's getFile API.
+
+        Returns:
+            Local absolute path of the downloaded file.
+        """
+        file_basename = os.path.basename(file_path)
+        temp_dir = get_astrbot_temp_path()
+        temp_path = os.path.join(temp_dir, f"{uuid.uuid4().hex}_{file_basename}")
+        await download_file(file_path, path=temp_path)
+        return temp_path
+
     async def convert_message(
         self,
         update: Update,
@@ -591,13 +611,27 @@ def _apply_caption() -> None:
         elif update.message.photo:
             photo = update.message.photo[-1]  # get the largest photo
             file = await photo.get_file()
-            message.message.append(Comp.Image(file=file.file_path, url=file.file_path))
-            _apply_caption()
+            file_path = file.file_path
+            if file_path is None:
+                logger.warning(
+                    "Telegram photo file_path is None, cannot save the file."
+                )
+            else:
+                temp_path = await self._download_to_temp(file_path)
+                message.message.append(Comp.Image(file=temp_path, url=temp_path))
+                _apply_caption()
 
         elif update.message.sticker:
             # 将sticker当作图片处理
             file = await update.message.sticker.get_file()
-            message.message.append(Comp.Image(file=file.file_path, url=file.file_path))
+            file_path = file.file_path
+            if file_path is None:
+                logger.warning(
+                    "Telegram sticker file_path is None, cannot save the file."
+                )
+            else:
+                temp_path = await self._download_to_temp(file_path)
+                message.message.append(Comp.Image(file=temp_path, url=temp_path))
             if update.message.sticker.emoji:
                 sticker_text = f"Sticker: {update.message.sticker.emoji}"
                 message.message_str = sticker_text
@@ -612,8 +646,9 @@ def _apply_caption() -> None:
                     f"Telegram document file_path is None, cannot save the file {file_name}.",
                 )
             else:
+                temp_path = await self._download_to_temp(file_path)
                 message.message.append(
-                    Comp.File(file=file_path, name=file_name, url=file_path)
+                    Comp.File(file=temp_path, name=file_name, url=temp_path)
                 )
                 _apply_caption()
 
@@ -626,7 +661,8 @@ def _apply_caption() -> None:
                     f"Telegram video file_path is None, cannot save the file {file_name}.",
                 )
             else:
-                message.message.append(Comp.Video(file=file_path, path=file.file_path))
+                temp_path = await self._download_to_temp(file_path)
+                message.message.append(Comp.Video(file=temp_path, path=temp_path))
                 _apply_caption()
 
         return message

tests/test_telegram_adapter.py

@@ -156,7 +156,7 @@ async def test_telegram_reply_without_quote_text_uses_full_message(quote_text):
 
 
 @pytest.mark.asyncio
-async def test_telegram_document_caption_populates_message_text_and_plain():
+async def test_telegram_document_caption_populates_message_text_and_plain(tmp_path):
     TelegramPlatformAdapter = _load_telegram_adapter()
     adapter = TelegramPlatformAdapter(
         make_platform_config("telegram"),
@@ -172,8 +172,16 @@ async def test_telegram_document_caption_populates_message_text_and_plain():
         caption="@alice 请总结这份文档",
         caption_entities=[mention],
     )
+    convert_message_globals = adapter.convert_message.__func__.__globals__
 
-    result = await adapter.convert_message(update, _build_context())
+    with patch.dict(
+        convert_message_globals,
+        {
+            "get_astrbot_temp_path": MagicMock(return_value=str(tmp_path)),
+            "download_file": AsyncMock(),
+        },
+    ):
+        result = await adapter.convert_message(update, _build_context())
 
     assert result is not None
     assert result.message_str == "@alice 请总结这份文档"
@@ -189,7 +197,7 @@ async def test_telegram_document_caption_populates_message_text_and_plain():
 
 
 @pytest.mark.asyncio
-async def test_telegram_video_caption_populates_message_text_and_plain():
+async def test_telegram_video_caption_populates_message_text_and_plain(tmp_path):
     TelegramPlatformAdapter = _load_telegram_adapter()
     adapter = TelegramPlatformAdapter(
         make_platform_config("telegram"),
@@ -203,8 +211,16 @@ async def test_telegram_video_caption_populates_message_text_and_plain():
         video=video,
         caption="这段视频讲了什么",
     )
+    convert_message_globals = adapter.convert_message.__func__.__globals__
 
-    result = await adapter.convert_message(update, _build_context())
+    with patch.dict(
+        convert_message_globals,
+        {
+            "get_astrbot_temp_path": MagicMock(return_value=str(tmp_path)),
+            "download_file": AsyncMock(),
+        },
+    ):
+        result = await adapter.convert_message(update, _build_context())
 
     assert result is not None
     assert result.message_str == "这段视频讲了什么"
@@ -215,6 +231,133 @@ async def test_telegram_video_caption_populates_message_text_and_plain():
     )
 
 
+@pytest.mark.asyncio
+async def test_telegram_document_downloads_to_local_temp_path(tmp_path):
+    """#9448: document 组件必须拿到本地路径，而不是原始 Telegram file_path/URL。"""
+    TelegramPlatformAdapter = _load_telegram_adapter()
+    adapter = TelegramPlatformAdapter(
+        make_platform_config("telegram"),
+        {},
+        asyncio.Queue(),
+    )
+    document = create_mock_file("https://api.telegram.org/file/test/report.md")
+    document.file_name = "report.md"
+    update = create_mock_update(message_text=None, document=document)
+    convert_message_globals = adapter.convert_message.__func__.__globals__
+    mock_download = AsyncMock()
+
+    with patch.dict(
+        convert_message_globals,
+        {
+            "get_astrbot_temp_path": MagicMock(return_value=str(tmp_path)),
+            "download_file": mock_download,
+        },
+    ):
+        result = await adapter.convert_message(update, _build_context())
+
+    assert result is not None
+    file_comp = next(c for c in result.message if isinstance(c, Comp.File))
+    assert file_comp.name == "report.md"
+    assert file_comp.file_.startswith(str(tmp_path))
+    assert file_comp.file_ != "https://api.telegram.org/file/test/report.md"
+    mock_download.assert_awaited_once()
+    assert (
+        mock_download.await_args.args[0]
+        == "https://api.telegram.org/file/test/report.md"
+    )
+
+
+@pytest.mark.asyncio
+async def test_telegram_video_downloads_to_local_temp_path(tmp_path):
+    """#9448: video 组件必须拿到本地路径，而不是原始 Telegram file_path/URL。"""
+    TelegramPlatformAdapter = _load_telegram_adapter()
+    adapter = TelegramPlatformAdapter(
+        make_platform_config("telegram"),
+        {},
+        asyncio.Queue(),
+    )
+    video = create_mock_file("https://api.telegram.org/file/test/lesson.mp4")
+    video.file_name = "lesson.mp4"
+    update = create_mock_update(message_text=None, video=video)
+    convert_message_globals = adapter.convert_message.__func__.__globals__
+    mock_download = AsyncMock()
+
+    with patch.dict(
+        convert_message_globals,
+        {
+            "get_astrbot_temp_path": MagicMock(return_value=str(tmp_path)),
+            "download_file": mock_download,
+        },
+    ):
+        result = await adapter.convert_message(update, _build_context())
+
+    assert result is not None
+    video_comp = next(c for c in result.message if isinstance(c, Comp.Video))
+    assert video_comp.file.startswith(str(tmp_path))
+    assert video_comp.path.startswith(str(tmp_path))
+    assert video_comp.file != "https://api.telegram.org/file/test/lesson.mp4"
+
+
+@pytest.mark.asyncio
+async def test_telegram_photo_downloads_to_local_temp_path(tmp_path):
+    """#9448: photo 组件必须拿到本地路径，而不是原始 Telegram file_path/URL。"""
+    TelegramPlatformAdapter = _load_telegram_adapter()
+    adapter = TelegramPlatformAdapter(
+        make_platform_config("telegram"),
+        {},
+        asyncio.Queue(),
+    )
+    photo = create_mock_file("https://api.telegram.org/file/test/photo.jpg")
+    update = create_mock_update(message_text=None, photo=[photo])
+    convert_message_globals = adapter.convert_message.__func__.__globals__
+    mock_download = AsyncMock()
+
+    with patch.dict(
+        convert_message_globals,
+        {
+            "get_astrbot_temp_path": MagicMock(return_value=str(tmp_path)),
+            "download_file": mock_download,
+        },
+    ):
+        result = await adapter.convert_message(update, _build_context())
+
+    assert result is not None
+    image_comp = next(c for c in result.message if isinstance(c, Comp.Image))
+    assert image_comp.file.startswith(str(tmp_path))
+    assert image_comp.file != "https://api.telegram.org/file/test/photo.jpg"
+
+
+@pytest.mark.asyncio
+async def test_telegram_sticker_downloads_to_local_temp_path(tmp_path):
+    """#9448: sticker 组件必须拿到本地路径，而不是原始 Telegram file_path/URL。"""
+    TelegramPlatformAdapter = _load_telegram_adapter()
+    adapter = TelegramPlatformAdapter(
+        make_platform_config("telegram"),
+        {},
+        asyncio.Queue(),
+    )
+    sticker = create_mock_file("https://api.telegram.org/file/test/sticker.webp")
+    sticker.emoji = "😀"
+    update = create_mock_update(message_text=None, sticker=sticker)
+    convert_message_globals = adapter.convert_message.__func__.__globals__
+    mock_download = AsyncMock()
+
+    with patch.dict(
+        convert_message_globals,
+        {
+            "get_astrbot_temp_path": MagicMock(return_value=str(tmp_path)),
+            "download_file": mock_download,
+        },
+    ):
+        result = await adapter.convert_message(update, _build_context())
+
+    assert result is not None
+    image_comp = next(c for c in result.message if isinstance(c, Comp.Image))
+    assert image_comp.file.startswith(str(tmp_path))
+    assert image_comp.file != "https://api.telegram.org/file/test/sticker.webp"
+    assert result.message_str == "Sticker: 😀"
+
+
 @pytest.mark.asyncio
 async def test_telegram_voice_message_creates_record_component(tmp_path):
     TelegramPlatformAdapter = _load_telegram_adapter()