Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wcqqq1214 wants to merge 2 commits into
base: master
Choose a base branch
from
+244 −9
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2e14309
fix: download telegram photo/sticker/document/video to local temp path
wcqqq1214 Jul 30, 2026
bd3c231
fix: reuse local path from local-mode Bot API server instead of downl…
wcqqq1214 Jul 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 46 additions & 5 deletions astrbot/core/platform/sources/telegram/tg_adapter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -436,6 +436,31 @@ 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.

On the cloud Bot API (and non-local self-hosted servers), get_file()
returns an http(s) URL, which we fetch into a temp file so downstream
components get a readable local path. But when the Bot API server runs
in local mode, get_file() returns an absolute local path instead; that
file already lives on disk, so we return it as-is rather than feeding a
local path into download_file() (an aiohttp GET, which cannot resolve
it).

Args:
file_path: The file_path returned by Telegram's getFile API.

Returns:
Local absolute path of the file.
"""
if os.path.isfile(file_path):
return file_path
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,
Expand Down Expand Up @@ -591,13 +616,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
Expand All @@ -612,8 +651,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()

Expand All @@ -626,7 +666,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
Expand Down
202 changes: 198 additions & 4 deletions tests/test_telegram_adapter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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):

@sourcery-ai sourcery-ai Bot Jul 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion (testing): Avoid reaching into __func__.__globals__ in tests; patch the module symbols instead for clarity and robustness.

Using convert_message.__func__.__globals__ with patch.dict to override get_astrbot_temp_path/download_file is brittle and hard to follow. Prefer patching these at the module level (e.g. patch("astrbot.core.platform.sources.telegram.tg_adapter.get_astrbot_temp_path", ...)) or via pytest’s monkeypatch fixture so the tests remain stable if convert_message is refactored and are easier to understand.

Suggested implementation:

@pytest.mark.asyncio
async def test_telegram_document_caption_populates_message_text_and_plain(tmp_path):
    TelegramPlatformAdapter = _load_telegram_adapter()
    adapter = TelegramPlatformAdapter(
        make_platform_config("telegram"),
        caption="@alice 请总结这份文档",
        caption_entities=[mention],
    )

    # Patch the module symbols directly instead of reaching into convert_message.__func__.__globals__
    with patch(
        "astrbot.core.platform.sources.telegram.tg_adapter.get_astrbot_temp_path",
        return_value=tmp_path,
    ), patch(
        "astrbot.core.platform.sources.telegram.tg_adapter.download_file"

The with patch(... download_file block should be completed to mirror whatever behavior the original patch.dict(... {"download_file": ...}) provided. For example, you likely need to:

  1. Provide an appropriate side_effect or return_value for download_file matching the expectations of convert_message in this test (e.g., creating a temporary file in tmp_path and returning its path, or returning a bytes-like object).
  2. Remove the now-unused mapping body that followed the original patch.dict(...) call (the { ... }), since the two patch calls fully replace the globals for get_astrbot_temp_path and download_file.
    Adjust the patch arguments if the actual module import path for the Telegram adapter differs from astrbot.core.platform.sources.telegram.tg_adapter.

TelegramPlatformAdapter = _load_telegram_adapter()
adapter = TelegramPlatformAdapter(
make_platform_config("telegram"),
Expand All @@ -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 请总结这份文档"
Expand All @@ -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"),
Expand All @@ -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 == "这段视频讲了什么"
Expand All @@ -215,6 +231,184 @@ async def test_telegram_video_caption_populates_message_text_and_plain():
)


@pytest.mark.asyncio
async def test_download_to_temp_returns_local_path_without_downloading(tmp_path):
"""#9448: a local-mode Bot API server returns an absolute local path, which
should be reused as-is instead of being downloaded as a URL."""
TelegramPlatformAdapter = _load_telegram_adapter()
adapter = TelegramPlatformAdapter(
make_platform_config("telegram"),
{},
asyncio.Queue(),
)
local_file = tmp_path / "photos" / "file_10.jpg"
local_file.parent.mkdir(parents=True)
local_file.write_bytes(b"local-bytes")
convert_message_globals = adapter._download_to_temp.__func__.__globals__
mock_download = AsyncMock()

with patch.dict(convert_message_globals, {"download_file": mock_download}):
result = await adapter._download_to_temp(str(local_file))

assert result == str(local_file)
mock_download.assert_not_awaited()


@pytest.mark.asyncio
async def test_download_to_temp_downloads_remote_url(tmp_path):
"""A remote URL should still be downloaded to the temp dir, not returned as-is."""
TelegramPlatformAdapter = _load_telegram_adapter()
adapter = TelegramPlatformAdapter(
make_platform_config("telegram"),
{},
asyncio.Queue(),
)
url = "https://api.telegram.org/file/bot123/photos/file_10.jpg"
convert_message_globals = adapter._download_to_temp.__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._download_to_temp(url)

assert result.startswith(str(tmp_path))
assert result != url
mock_download.assert_awaited_once()
assert mock_download.await_args.args[0] == url


@pytest.mark.asyncio
async def test_telegram_document_downloads_to_local_temp_path(tmp_path):
"""#9448: the document component must hold a local path, not the raw 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: the video component must hold a local path, not the raw 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: the photo component must hold a local path, not the raw 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: the sticker component must hold a local path, not the raw 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()
Expand Down
Loading