pylogging-ext

一个支持 异步、批量、重试、多通道通知 的 logging 扩展库。

不只是 logging handler

安装

pip install pylogging-ext

可选依赖

• cls：腾讯云 CLS 支持

  pip install pylogging-ext[cls] 

• socks：SOCKS 代理支持

  pip install pylogging-ext[socks]

使用

代码方式

import logging
from pylogging_ext import TelegramHandler

logger = logging.getLogger('demo')
logger.setLevel(logging.INFO)

handler = TelegramHandler(
    token='your-bot-token',
    chat_id='your-chat-id',
)

logger.addHandler(handler)

logger.error('发生错误了')

dictConfig

import logging
from logging.config import dictConfig

LOGGING_CONFIG = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'standard': {
            'format': '%(asctime)s [%(levelname)s] %(name)s: %(message)s',
            'datefmt': '%Y-%m-%d %H:%M:%S'
        },
    },
    'handlers': {
        'console': {
            'level': 'INFO',
            'class': 'logging.StreamHandler',
            'formatter': 'standard'
        },
        'telegram': {
            'level': 'WARNING',
            'class': 'pylogging_ext.TelegramHandler',
            'token': 'your-bot-token',
            'chat_id': 'your-chat-id',
            'formatter': 'standard'
        },
    },
    'root': {
        'handlers': ['console', 'telegram'],
        'level': 'INFO'
    }
}

dictConfig(LOGGING_CONFIG)
logger = logging.getLogger('root')

logger.error('发生错误了')

fileConfig

定义 logging.conf

[loggers]
keys=root

[handlers]
keys=consoleHandler,telegramHandler

[formatters]
keys=standardFormatter

[logger_root]
level=INFO
handlers=consoleHandler,telegramHandler

[handler_consoleHandler]
class=logging.StreamHandler
level=INFO
formatter=standardFormatter
args=(sys.stdout,)

[handler_telegramHandler]
class=pylogging_ext.TelegramHandler
level=WARNING
formatter=standardFormatter
args=('your-bot-token', 'your-chat-id')

[formatter_standardFormatter]
format=%(asctime)s [%(levelname)s] %(name)s: %(message)s
datefmt=%Y-%m-%d %H:%M:%S

import logging
from logging.config import fileConfig

fileConfig('logging.conf')
logger = logging.getLogger('root')

logger.error('发生错误了')

API

Class: BaseHandler

通用异步 Handler 基类，所有通知 Handler 均继承自此类。

BaseHandler(...)

• batch_size 批量发送大小
• flush_interval 刷新间隔 (秒)
• queue_size 队列大小
• retry 重试次数

特性：

• 异步发送（非阻塞）
• 批量处理
• 指数退避重试
• 优雅关闭

腾讯云 CLS

Class: TencentCloudCLSHandler

from pylogging_ext import TencentCloudCLSHandler

handler = TencentCloudCLSHandler(
    endpoint='ap-shanghai.cls.tencentcs.com',
    secret_id='your-secret-id',
    secret_key='your-secret-key',
    topic_id='your-topic-id',
)

参数

• endpoint CLS接入地址
• secret_id 腾讯云密钥
• secret_key 腾讯云密钥
• topic_id 日志主题 ID

微信机器人

Class: WeChatHandler

from pylogging_ext import WeChatHandler

handler = WeChatHandler(
    webhook='https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx'
)

参数

• webhook 企业微信机器人地址

Telegram

Class: TelegramHandler

from pylogging_ext import TelegramHandler

handler = TelegramHandler(
    token='your-bot-token',
    chat_id='your-chat-id',
    proxies={
        'http': 'http://127.0.0.1:7890',
        'https': 'http://127.0.0.1:7890',
    }
)

参数

• token Telegram Bot Token
• chat_id 聊天 ID
• proxies 代理配置（可选）
• ssl_verify SSL校验 (默认 True)

钉钉机器人

Class: DingTalkHandler

from pylogging_ext import DingTalkHandler

handler = DingTalkHandler(
    access_token='your-access-token',
    secret='your-secret',
    at_mobiles=['18600000000'],
)

参数

• access_token Webhook URL 中的 access_token
• secret 签名密钥（开启加签时必填）
• at_mobiles 被@人的手机号（可选）
• is_at_all @所有人是 True，否则为 False (默认 False)

飞书机器人

Class: FeiShuHandler

from pylogging_ext import FeiShuHandler

handler = FeiShuHandler(
    webhook='https://open.feishu.cn/open-apis/bot/v2/hook/xxx',
    secret='your-secret',
    at_users=['ou_18eac8********17ad4f02e8bbbb']
)

参数

• webhook 飞书机器人地址
• secret 签名密钥（开启加签时必填）
• at_users 用户的 Open ID 或 User ID，@ 所有人时，填 all（可选）

支持的处理器

• 腾讯云 CLS
• 企业微信
• Telegram
• 钉钉
• 飞书

持续扩展中...