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

如何抑制Telegram Bot断连时的Traceback并优化错误输出

解决方案

1. 核心思路

  • 日志分流:将含Traceback的详细日志写入文件,终端仅输出去重后的关键错误信息
  • 过滤重复错误:自定义日志过滤器拦截重复的NetworkError
  • 调整依赖库日志:限制Telegram及HTTP客户端的日志输出级别,避免冗余信息

2. 完整代码实现

import logging
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes

# 自定义过滤器:拦截重复的NetworkError
class DuplicateErrorFilter(logging.Filter):
    def __init__(self):
        self.last_error = None

    def filter(self, record):
        # 仅针对NetworkError做去重判断
        if record.exc_info and record.exc_info[0] is telegram.error.NetworkError:
            error_content = str(record.exc_info[1])
            current_error = (record.exc_info[0], error_content)
            if current_error == self.last_error:
                return False
            self.last_error = current_error
        return True

def main():
    # 重置根日志器,清除默认配置
    root_logger = logging.getLogger()
    root_logger.setLevel(logging.DEBUG)
    root_logger.handlers.clear()

    # 1. 文件日志处理器:记录完整错误(含Traceback)
    file_handler = logging.FileHandler('bot_error.log')
    file_handler.setLevel(logging.DEBUG)
    file_formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s\n%(exc_text)s'
    )
    file_handler.setFormatter(file_formatter)

    # 2. 终端日志处理器:仅输出去重后的ERROR级信息,无Traceback
    stream_handler = logging.StreamHandler()
    stream_handler.setLevel(logging.ERROR)
    stream_formatter = logging.Formatter(
        '%(asctime)s - %(levelname)s - %(message)s'
    )
    stream_handler.setFormatter(stream_formatter)
    stream_handler.addFilter(DuplicateErrorFilter())

    # 将处理器添加到根日志器
    root_logger.addHandler(file_handler)
    root_logger.addHandler(stream_handler)

    # 限制依赖库日志输出级别,避免冗余
    logging.getLogger('telegram').setLevel(logging.ERROR)
    logging.getLogger('httpx').setLevel(logging.WARNING)  # v20+版本依赖httpx

    # 初始化Bot应用
    app = ApplicationBuilder().token("你的BotToken").build()

    # 示例命令
    async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
        await update.message.reply_text('Hello!')

    app.add_handler(CommandHandler("start", start))
    app.run_polling()

if __name__ == '__main__':
    main()

3. 关键说明

  • 日志分流:文件处理器使用DEBUG级别并保留%(exc_text)s,完整记录错误栈;终端处理器用ERROR级别且省略错误栈,只展示关键信息。
  • 重复错误过滤:通过DuplicateErrorFilter记录最近一次的NetworkError内容,重复触发时直接拦截,确保终端只显示一次。
  • 依赖日志控制:将telegram库日志设为ERROR,避免它自行打印非必要信息;httpx设为WARNING,过滤HTTP调试日志。

内容的提问来源于stack exchange,提问作者luisito_36

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.06.14 11:08:22