如何处理异步TeleBot自身异常?自定义控制台错误输出问询
自定义异步TeleBot的全局异常捕获与控制台输出
针对异步版TeleBot,要替换默认错误提示、自定义控制台输出并覆盖所有异常场景,可通过以下几种方式实现:
方法1:替换默认日志处理器
TeleBot默认通过logging模块输出日志,自定义日志处理器可以完全接管所有日志输出,包括Token错误这类API请求异常:
import logging import asyncio from telebot.async_telebot import AsyncTeleBot # 自定义日志处理器 class CustomTeleBotLogger(logging.Handler): def emit(self, record): # 针对不同级别日志自定义输出格式 if record.levelno == logging.ERROR: msg = record.getMessage() if "401" in msg: print(f"[自定义错误] 机器人Token无效,请检查API密钥!") else: print(f"[自定义错误] 运行异常: {msg}") elif record.levelno == logging.INFO: print(f"[自定义信息] {msg}") # 重置TeleBot的日志配置 logger = logging.getLogger('TeleBot') logger.handlers.clear() logger.addHandler(CustomTeleBotLogger()) logger.setLevel(logging.INFO) # 初始化异步机器人 bot = AsyncTeleBot("你的错误Token") async def main(): @bot.message_handler(commands=['start']) async def start_cmd(message): await bot.send_message(message.chat.id, "Hello!") try: await bot.polling(non_stop=True) except Exception as e: print(f"[启动异常] {str(e)}") if __name__ == "__main__": asyncio.run(main())
方法2:注册全局异常处理器
异步TeleBot提供了ExceptionHandler基类,通过自定义子类并绑定到机器人实例,可以捕获所有运行过程中的API异常及其他错误:
import asyncio from telebot.async_telebot import AsyncTeleBot from telebot.asyncio_handler_backends import ExceptionHandler from telebot.apihelper import ApiTelegramException # 自定义异常处理器 class CustomExceptionHandler(ExceptionHandler): async def handle(self, exception): if isinstance(exception, ApiTelegramException): if exception.error_code == 401: print(f"[自定义错误] Token无效,未授权访问Telegram API!") else: print(f"[API错误] 错误码: {exception.error_code}, 描述: {exception.description}") else: print(f"[未知异常] {str(exception)}") # 绑定处理器到机器人 bot = AsyncTeleBot("你的错误Token") bot.exception_handler = CustomExceptionHandler() async def main(): @bot.message_handler(commands=['start']) async def start_cmd(message): await bot.send_message(message.chat.id, "Hello!") await bot.polling(non_stop=True) if __name__ == "__main__": asyncio.run(main())
方法3:启动阶段直接捕获异常
如果需要在机器人启动时就捕获Token错误,可以在外层try-except块包裹polling逻辑:
import asyncio from telebot.async_telebot import AsyncTeleBot async def main(): bot = AsyncTeleBot("你的错误Token") @bot.message_handler(commands=['start']) async def start_cmd(message): await bot.send_message(message.chat.id, "Hello!") try: await bot.polling(non_stop=True) except Exception as e: err_msg = str(e) if "401" in err_msg: print(f"[启动错误] 无效的Token,请检查后重试!") else: print(f"[启动失败] {err_msg}") if __name__ == "__main__": asyncio.run(main())
最佳实践
要覆盖所有异常场景,推荐结合方法1+方法2:日志处理器接管所有日志输出,异常处理器捕获运行时的具体异常,确保没有遗漏任何错误场景。
内容的提问来源于stack exchange,提问作者Alexandr Nikitin
相关产品推荐
相关产品推荐

