Python开发Telegram机器人 如何实现每条新消息仅响应一次
问题根源
重复触发响应的核心原因是调用getUpdates接口时未传入offset参数:Telegram 服务端会默认留存所有未被确认处理的更新消息,你每次循环都拉取全量更新、固定取最后一条内容,自然会反复拿到已经处理过的/text命令,触发重复回复。
解决方案
Telegram Bot API 的getUpdates接口通过offset参数标记已处理的消息位点:
- 传入的
offset值为「上一次已处理消息的update_id + 1」时,服务端只会返回该位点之后的新消息,不会再返回已经处理过的旧内容,从根源上避免重复触发 - 额外传入
timeout参数开启长轮询,可以减少空请求的频次,降低资源消耗和被接口限流的概率
你原有代码还存在几个容易出问题的点,需要一并修正:
- 不要把
chat_id作为固定参数传入处理函数,chat_id应该从实际收到的消息体中读取,避免写死导致发错会话 - 不要每次处理命令时单独调用一次
getUpdates,会和主循环的拉取逻辑冲突,把拉取、处理逻辑统一放在主循环即可 - 原正则
^/text*会匹配/tex、/textttt这类不符合预期的内容,匹配固定命令直接用字符串开头判断即可,不需要复杂正则 - 每次拉取到新消息后要遍历所有返回的更新逐一处理,固定取最后一条会漏掉用户连续发送的消息
修正后可直接运行的代码
# TELEGRAM BOT USING THE API import requests as req from dotenv import load_dotenv import os config = load_dotenv(".env") teletoken = os.environ.get("telegram_token") uri = f"https://api.telegram.org/bot{teletoken}/" # 拼接Telegram API请求地址 def method(api_method): return f"{uri}{api_method}" if __name__ == "__main__": # 记录已处理到的最新更新ID,初始值0表示从启动后的新消息开始拉取 last_update_id = 0 while True: # 传入offset标记已处理位点,timeout=30开启30秒长轮询 resp = req.get( method("getUpdates"), params={ "offset": last_update_id + 1, "timeout": 30 } ) updates_data = resp.json() # 接口请求异常直接跳过本轮 if not updates_data.get("ok"): continue update_list = updates_data.get("result", []) # 遍历所有新收到的更新逐一处理 for update in update_list: # 先更新处理位点,避免下次重复拉取本条消息 last_update_id = update["update_id"] # 过滤非普通消息类型的更新(如入群提醒、频道动态等) if "message" not in update: continue msg = update["message"] chat_id = msg["chat"]["id"] msg_text = msg.get("text", "") # 匹配/text命令 if msg_text.startswith("/text"): req.post( method("sendMessage"), data={ "chat_id": chat_id, "text": "Example text" } )
补充说明
- 位点更新时机建议放在拿到单条更新后的第一时间,哪怕后续消息处理逻辑报错,下次循环也不会重复拉到这条消息,避免死循环卡单
- 长轮询模式下脚本不会疯狂发起无效请求,有新消息时会立刻收到,没有新消息时会等待最多30秒再发起下一次请求,稳定性比原来的无限空轮询高很多
内容的提问来源于stack exchange,提问作者Raxabi
相关产品推荐
相关产品推荐

