You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.27 03:27:32