Python脚本asyncio异常：无法获取Telegram Bot响应的跨环境问题

背景 #

我编写了一个基于Telethon的Python脚本，用于和Telegram Bot交互，在本地Windows和一台Ubuntu机器上运行都正常，但在另一台Ubuntu机器上出现异常。

代码片段 #

import platform
import asyncio
import re
from telethon import TelegramClient


if platform.system() == "Windows":
  asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())


client = TelegramClient("dtrend", api_id, api_hash)


async def main():

  await client.send_message(dtrend_bot_id, "/delete")
  delete_response = await get_last_message()
  await handle_start(delete_response)

  await handle_main()


async def handle_main():

  network_select_message = await get_last_message()
  await select_option(network_select_message, networks[network], 0)
...

async def get_last_message():
    history = await client.get_messages(bot_id)
    return history[0]

with client:
    client.loop.run_until_complete(main())

故障现象 #

发送消息或选择Bot的消息选项后，脚本无法获取到Bot的最新响应。比如调用get_last_message()时，拿到的不是预期的最新回复。

已尝试操作 #

• 在本地执行python3 -m pip install --upgrade asyncio telethon升级依赖后问题解决，但目标机器执行同样操作后问题依旧。
• 尝试使用Telethon的对话上下文，只能获取send_message的响应，无法获取选项选择后的响应。
• 搭建虚拟环境测试，发现即使asyncio版本相同，不同机器上安装的模块内容存在差异，问题仍存在。

解决方案 #

1. 统一asyncio安装包 #

由于同版本asyncio在不同机器上内容有差异，可能是安装源导致的。可以从本地导出asyncio的wheel包，到目标机器离线安装：

• 本地执行：pip wheel asyncio==<你的本地版本号>，生成对应的.whl文件。
• 将wheel文件传到目标机器，执行：pip install <wheel文件名>

2. 显式设置Ubuntu的事件循环策略 #

代码中仅针对Windows设置了事件循环策略，Ubuntu上也可能存在兼容性问题，尝试添加：

import asyncio
from asyncio import DefaultEventLoopPolicy

# 在初始化TelegramClient之前添加
asyncio.set_event_loop_policy(DefaultEventLoopPolicy())

或者尝试使用uvloop优化事件循环：
先安装依赖：

pip install uvloop

然后在代码中设置：

import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

3. 优化消息获取逻辑，添加重试等待 #

当前get_last_message()直接取历史第一条，可能因为Bot响应还未同步导致拿不到最新消息，添加轮询重试：

async def get_last_message():
    # 最多等待5秒，每0.5秒轮询一次
    for _ in range(10):
        history = await client.get_messages(bot_id, limit=1)
        if history:
            return history[0]
        await asyncio.sleep(0.5)
    raise TimeoutError("无法获取Bot的响应")

同时确认bot_id是目标Bot的正确ID或用户名，避免错误。

4. 使用事件监听替代历史消息查询 #

改用Telethon的wait_event监听来自Bot的新消息，比直接查历史更可靠：

from telethon.events import NewMessage

async def wait_for_bot_response():
    async with client.conversation(bot_id) as conv:
        # 等待来自目标Bot的新消息，超时10秒
        response_event = await conv.wait_event(NewMessage(from_users=bot_id), timeout=10)
        return response_event.message

然后在需要获取响应的地方，替换get_last_message()为这个函数即可。

内容的提问来源于stack exchange，提问作者youjustmy