如何处理Telegram机器人的403 'Forbidden: bot was blocked by the user'错误并保障正常运行
如何处理Telegram机器人的403 'Forbidden: bot was blocked by the user'错误并保障正常运行
我太懂你这个烦恼了——明明没人用机器人,却突然蹦出403错误,而且处理完之后程序还不对劲。别着急,咱们一步步拆解问题,把它搞定。
首先得明白为啥会出现这个错误:除了用户主动拉黑机器人,还有可能是之前的会话里用户已经拉黑了,但你的机器人还在尝试给这个用户发消息;或者轮询的时候碰到了被拉黑用户的更新请求,直接触发了403。你之前的错误处理虽然能不让程序崩溃,但没从根源上避免后续的异常操作,所以才会出问题。
接下来咱们改进你的代码,从两个层面入手:细粒度的消息处理异常捕获和更稳健的轮询重启逻辑。
1. 在消息处理函数里单独捕获403错误
很多时候403是在给用户发消息时触发的,而不是轮询阶段。所以咱们在处理每个用户消息的函数里,单独加异常处理,一旦发现用户拉黑了机器人,就标记这个用户,后续不再给他发消息,从根源上避免重复触发错误。
示例代码:
@bot.message_handler(func=lambda message: True) def handle_chat_message(message): current_user_id = message.from_user.id try: # 这里写你的匿名聊天业务逻辑,比如从MySQL匹配聊天对象、发送消息 target_user_id = get_matched_user(current_user_id) # 假设这是你的匹配函数 bot.send_message(target_user_id, message.text) except telebot.apihelper.ApiException as e: if e.result.status_code == 403: print(f"用户 {current_user_id} 已经拉黑机器人,标记为无效用户") # 同步更新MySQL,把这个用户标记为已拉黑,后续跳过处理 cursor.execute("UPDATE chat_users SET is_blocked = 1 WHERE user_id = %s", (current_user_id,)) db.commit() else: # 其他API错误直接抛出,让上层处理 raise e
2. 优化主轮询的错误处理逻辑
你原来的while循环虽然能捕获错误,但碰到异常后直接继续polling,可能会让机器人处于异常状态。咱们给不同的错误加对应的处理,比如延迟重启、针对限流的等待,确保机器人能平稳恢复。
改进后的主循环代码:
import time import telebot import mysql.connector from telebot import apihelper # 初始化机器人和数据库连接 bot = telebot.TeleBot("你的机器人Token") db = mysql.connector.connect( host="localhost", user="你的MySQL用户名", password="你的MySQL密码", database="你的数据库名" ) cursor = db.cursor() def get_matched_user(user_id): # 这里写你的用户匹配逻辑,从MySQL查询对应的聊天对象 pass while True: try: print("机器人启动轮询,等待消息...") bot.polling(none_stop=True, interval=0, timeout=20) except apihelper.ApiException as e: status_code = e.result.status_code reason = e.result.reason print(f"API错误触发:{status_code} - {reason}") if status_code == 403: # 碰到拉黑错误,延迟5秒再重启轮询,避免频繁重试 time.sleep(5) elif status_code == 429: # 处理Telegram的限流,按返回的Retry-After头等待 retry_wait = int(e.result.headers.get('Retry-After', 10)) print(f"请求太频繁啦,等待{retry_wait}秒后再试") time.sleep(retry_wait) else: # 其他API错误,延迟3秒重启 time.sleep(3) except mysql.connector.Error as err: if err.errno == mysql.connector.errorcode.ER_ACCESS_DENIED_ERROR: print(f"MySQL权限不对:{err},10秒后重试") time.sleep(10) elif err.errno == mysql.connector.errorcode.ER_BAD_DB_ERROR: print(f"数据库找不到:{err},请手动检查后重启机器人") break # 这个错误没法自动修复,直接退出循环 else: print(f"MySQL出错了:{err},5秒后重试") time.sleep(5) except Exception as e: print(f"碰到未知错误:{e},5秒后重启") time.sleep(5)
额外的小建议
- 定期清理MySQL里的已拉黑用户,或者在查询用户时先检查
is_blocked状态,避免无效的消息发送操作 - 可以开启Telebot的日志功能,更清楚地看到错误发生的细节:
import logging telebot.logger.setLevel(logging.INFO) - 偶尔检查一下机器人的Token是否有效,有没有被Telegram撤销,这也可能导致莫名的403错误
备注:内容来源于stack exchange,提问作者Vladyslav
相关产品推荐
相关产品推荐

