⚡ 基于 Loguru 的 Python 日志模块 - 开箱即用,支持多种告警通知
Project description
📚 目录
🎯 简介
XQCLog 是一个基于 Loguru 开发的自用 Python 日志模块,专注于提供开箱即用、功能强大的日志解决方案。
为什么选择 XQCLog?
相比直接使用 Loguru,XQCLog 提供了以下增强功能:
| 特性 | Loguru 原生 | XQCLog |
|---|---|---|
| 基础日志功能 | ✅ | ✅ |
| 环境自适应配置 | ❌ 需手动配置 | ✅ preset="auto" 一键适配 |
| 配置文件支持 | ❌ 需自己实现 | ✅ 原生支持 YAML/JSON |
| 场景预设 | ❌ | ✅ Web/爬虫/数据处理等预设 |
| 多渠道告警 | ❌ 需自己集成 | ✅ 钉钉/企业微信/邮件开箱即用 |
| 灵活告警控制 | ❌ | ✅ alert 参数精确控制 |
| 告警策略 | ❌ | ✅ 并行/顺序/故障转移/优先级 |
| 装饰器支持 | ❌ 需自己实现 | ✅ @log_execution @timer 等 |
| 结构化日志 | ❌ | ✅ HTTP/数据库/API 等专用方法 |
✨ 功能特性
🎯 核心功能
- 📊 多级别日志 - TRACE、DEBUG、INFO、SUCCESS、WARNING、ERROR、CRITICAL
- 🎨 彩色输出 - 控制台日志彩色显示,提升可读性
- 🗂️ 智能分割 - 按级别、大小、时间自动分割日志文件
- 🔄 自动轮转 - 支持按大小或时间轮转,自动压缩旧日志
- ⚡ 异步写入 - 不影响应用性能的异步日志写入
- 🔍 异常追踪 - 详细的异常堆栈和变量诊断信息
🔔 智能告警
支持的通知方式:
| 通知方式 | 特性 |
|---|---|
| 钉钉机器人 | ✅ Webhook ✅ 签名认证 ✅ @指定人 |
| 企业微信群机器人 | ✅ Webhook ✅ Markdown 格式 ✅ @指定人 |
| 企业微信应用 | ✅ 企业应用消息 ✅ Token 缓存 |
| 邮件通知 | ✅ SMTP ✅ SSL/TLS ✅ HTML 格式 ✅ 抄送 |
| 自定义通知器 | ✅ 可扩展任何第三方平台 |
告警控制机制:
# 三层控制优先级(从高到低)
1. alert 参数 # 最高优先级,强制控制
├─ alert=True → 强制发送(无论配置)
├─ alert=False → 强制不发送(无论配置)
└─ alert=None → 使用配置判断
2. alert_levels 配置 # 根据日志级别判断
├─ ["ERROR", "CRITICAL"] → 只有这些级别发送
├─ None → 默认不发送
└─ [] → 不发送
3. 默认行为 # 不发送
发送策略:
| 策略 | 说明 | 适用场景 |
|---|---|---|
| parallel | 同时发送到所有通知器 | 多渠道同时通知 |
| sequential | 按顺序发送到所有通知器 | 需要保证顺序 |
| failover | 轮询发送直到成功 | 高可靠性需求 |
| priority | 按优先级发送 | 分级通知 |
🎭 场景预设
| 预设 | 适用场景 | 日志级别 | 特点 |
|---|---|---|---|
| auto | 自动识别 | 根据 ENV 环境变量 | 自动适配开发/测试/生产 |
| development | 开发环境 | DEBUG | 详细诊断信息 |
| testing | 测试环境 | INFO | 适度日志 |
| production | 生产环境 | WARNING | 仅警告和错误 |
| web | Web 应用 | INFO | 自动分割日志 |
| crawler | 爬虫应用 | INFO | 异步写入优化 |
| data | 数据处理 | INFO | 大容量日志支持 |
📦 安装
使用 pip
pip install xqclog
使用 Poetry(推荐)
poetry add xqclog
从源码安装
git clone https://github.com/xiaoqiangclub/xqclog.git
cd xqclog
poetry install
🚀 快速开始
1️⃣ 最简单的使用(10 秒上手)
from xqclog import logger, init_logger
# 默认 DEBUG 级别,彩色输出
logger.info("Hello, XQCLog!")
logger.success("操作成功 ✅")
logger.warning("这是一个警告 ⚠️")
logger.error("发生错误 ❌")
2️⃣ 使用预设配置
from xqclog import logger, init_logger
# 自动根据环境变量配置(推荐)
# export ENV=production
init_logger(preset="auto")
# 或使用特定预设
init_logger(preset="web") # Web 应用
init_logger(preset="crawler") # 爬虫应用
init_logger(preset="production") # 生产环境
logger.info("使用预设配置")
3️⃣ 添加告警通知
from xqclog import init_logger, logger, LogConfig
config = LogConfig(
log_level="INFO",
notifiers=[
{
"type": "dingtalk",
"webhook": "https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN",
"alert_levels": ["ERROR", "CRITICAL"], # 只有这两个级别会发送
}
]
)
init_logger(config)
logger.info("普通日志,不会发送通知")
logger.error("错误日志,会发送到钉钉") # ✅ 发送通知
4️⃣ 灵活控制告警
from xqclog import init_logger, logger
init_logger(
notifiers=[
{
"type": "dingtalk",
"webhook": "https://...",
"alert_levels": ["ERROR", "CRITICAL"],
}
]
)
# 场景1:强制发送(即使 INFO 不在 alert_levels 中)
logger.info("重要通知:系统升级完成", alert=True) # ✅ 强制发送
# 场景2:强制不发送(即使 ERROR 在 alert_levels 中)
logger.error("已知错误,已处理", alert=False) # ❌ 强制不发送
# 场景3:使用默认逻辑(根据 alert_levels 判断)
logger.error("未处理的错误") # ✅ 根据配置发送
5️⃣ 使用配置文件
# 复制配置模板
cp logging.yaml.example logging.yaml
# 编辑配置文件,替换为真实的参数
vim logging.yaml
from xqclog import init_logger, logger
# 从配置文件加载
init_logger(config_file="logging.yaml")
logger.info("从配置文件加载的日志系统")
📖 详细文档
基础配置
LogConfig 完整参数
from xqclog import LogConfig
config = LogConfig(
# ========== 基础配置 ==========
log_level="INFO", # 日志级别:TRACE/DEBUG/INFO/SUCCESS/WARNING/ERROR/CRITICAL
log_dir="logs", # 日志目录
log_file="app.log", # 日志文件名
# ========== 日志轮转 ==========
rotation="100 MB", # 轮转条件:
# - 按大小:"100 MB", "1 GB"
# - 按时间:"1 week", "1 day"
# - 按时刻:"00:00"(每天零点)
retention="30 days", # 保留时间:"30 days", "1 week", "2 months"
compression="zip", # 压缩格式:zip/gz/bz2/xz
# ========== 输出控制 ==========
console_output=True, # 是否输出到控制台
file_output=True, # 是否输出到文件
auto_split=False, # 是否按级别自动分割文件
# ========== 格式化 ==========
encoding="utf-8", # 文件编码
colorize=True, # 控制台彩色输出
format_string=None, # 自定义日志格式
# ========== 性能优化 ==========
enqueue=True, # 异步写入(推荐开启)
# ========== 调试信息 ==========
backtrace=True, # 显示详细堆栈
diagnose=True, # 显示变量诊断(生产环境建议关闭)
# ========== 告警配置 ==========
alert_strategy="parallel", # 发送策略:parallel/sequential/failover/priority
alert_retry=3, # 失败重试次数
alert_retry_delay=1.0, # 重试延迟(秒)
alert_timeout=5.0, # 发送超时(秒)
notifiers=[], # 通知器配置列表
)
三种初始化方式
from xqclog import init_logger, LogConfig
# 方式1:使用配置对象
config = LogConfig(log_level="INFO", log_dir="logs/myapp")
init_logger(config)
# 方式2:使用关键字参数
init_logger(log_level="DEBUG", log_dir="logs", auto_split=True)
# 方式3:使用配置文件
init_logger(config_file="logging.yaml")
预设配置
自动环境识别
from xqclog import init_logger, logger
# 自动根据 ENV 或 ENVIRONMENT 环境变量识别
# export ENV=production
init_logger(preset="auto")
logger.info("自动适配环境配置")
环境识别规则:
| 环境变量值 | 使用的预设 |
|---|---|
prod, production |
production |
test, testing |
testing |
| 其他 | development |
各预设详情
| 预设 | log_level | log_dir | rotation | retention | 特殊配置 |
|---|---|---|---|---|---|
| development | DEBUG | logs/dev | 50 MB | 7 days | diagnose=True |
| testing | INFO | logs/test | 100 MB | 14 days | - |
| production | WARNING | logs/prod | 500 MB | 90 days | console_output=False, compression=gz |
| web | INFO | logs/web | 200 MB | 30 days | auto_split=True |
| crawler | INFO | logs/crawler | 100 MB | 14 days | enqueue=True |
| data | INFO | logs/data | 1 GB | 60 days | compression=gz |
使用示例
from xqclog import init_logger
# 开发环境
init_logger(preset="development")
# 生产环境
init_logger(preset="production")
# Web 应用
init_logger(preset="web")
配置文件
配置文件模板
查看 logging.yaml.example 获取完整配置模板。
基础配置示例:
# logging.yaml
logging:
# 基础配置
log_level: INFO
log_dir: logs
log_file: app.log
rotation: 100 MB
retention: 30 days
# 告警配置
notifiers:
- type: dingtalk
webhook: "https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN"
alert_levels:
- ERROR
- CRITICAL
从配置文件加载
from xqclog import init_logger, logger
# 加载 YAML 配置
init_logger(config_file="logging.yaml")
# 或加载 JSON 配置
init_logger(config_file="logging.json")
logger.info("配置文件加载成功")
保存配置到文件
from xqclog import logger, init_logger
# 初始化
init_logger(log_level="INFO", auto_split=True)
# 保存当前配置
logger.save_config("my_config.yaml")
日志级别
可用级别
from xqclog import logger, init_logger
init_logger(log_level="DEBUG")
logger.trace("追踪信息") # TRACE - 最详细的调试信息
logger.debug("调试信息") # DEBUG - 调试信息
logger.info("普通信息") # INFO - 一般信息
logger.success("成功信息") # SUCCESS - 操作成功
logger.warning("警告信息") # WARNING - 警告
logger.error("错误信息") # ERROR - 错误
logger.critical("严重错误") # CRITICAL - 严重错误
日志级别说明
| 级别 | 用途 | 是否应该配置告警 |
|---|---|---|
| TRACE | 非常详细的调试 | ❌ 否 |
| DEBUG | 调试信息 | ❌ 否 |
| INFO | 一般信息 | ❌ 否(特殊情况用 alert=True) |
| SUCCESS | 成功操作 | ❌ 否 |
| WARNING | 警告信息 | ⚠️ 谨慎(容易产生告警疲劳) |
| ERROR | 错误信息 | ✅ 推荐 |
| CRITICAL | 严重错误 | ✅ 强烈推荐 |
动态修改级别
from xqclog import logger, init_logger
init_logger(log_level="INFO")
logger.debug("不会显示")
logger.info("会显示")
# 动态修改
logger.set_level("DEBUG")
logger.debug("现在会显示了")
日志分割
按级别自动分割
from xqclog import init_logger, logger
init_logger(
log_level="DEBUG",
log_dir="logs",
auto_split=True # 👈 启用自动分割
)
logger.debug("写入 debug.log")
logger.info("写入 info.log")
logger.warning("写入 warning.log")
logger.error("写入 error.log")
logger.critical("写入 critical.log")
生成的文件结构:
logs/
├── debug.log
├── info.log
├── success.log
├── warning.log
├── error.log
└── critical.log
按大小轮转
init_logger(rotation="100 MB") # 达到 100MB 时轮转
按时间轮转
# 每天轮转
init_logger(rotation="1 day")
# 每周轮转
init_logger(rotation="1 week")
# 每天零点轮转
init_logger(rotation="00:00")
# 每天中午轮转
init_logger(rotation="12:00")
日志压缩和保留
init_logger(
rotation="100 MB",
retention="30 days", # 只保留 30 天
compression="zip" # 压缩为 zip 格式
)
告警通知
告警触发机制
XQCLog 提供三层告警控制机制:
控制优先级(从高到低):
1. alert 参数(最高优先级)
├─ alert=True → 强制发送通知
├─ alert=False → 强制不发送
└─ alert=None → 使用下一级判断
2. alert_levels 配置
├─ alert_levels=None → 默认不发送
├─ alert_levels=[] → 不发送
└─ level in alert_levels → 发送
3. 默认行为
└─ 不发送
工作流程:
记录日志
↓
是否设置 alert 参数?
├─ 是 → alert=True → ✅ 发送
│ alert=False → ❌ 不发送
└─ 否 → 检查 alert_levels
├─ None → ❌ 不发送
├─ 级别在列表中 → ✅ 发送
└─ 级别不在列表中 → ❌ 不发送
通知器配置
1. 钉钉机器人
from xqclog import init_logger, LogConfig
config = LogConfig(
log_level="INFO",
notifiers=[
{
"type": "dingtalk",
"webhook": "https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN",
"secret": "YOUR_SECRET", # 可选:签名密钥
"at_mobiles": ["13800138000"], # 可选:@的手机号
"at_all": False, # 可选:是否@所有人
"alert_levels": ["ERROR", "CRITICAL"],
"timeout": 5, # 可选:超时时间
"enabled": True, # 可选:是否启用
"priority": 100, # 可选:优先级
}
]
)
init_logger(config)
2. 企业微信群机器人
config = LogConfig(
notifiers=[
{
"type": "weixin_webhook",
"webhook": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY",
"mentioned_list": ["user1", "user2"], # 可选:@的成员ID
"mentioned_mobile_list": ["13800138000"], # 可选:@的手机号
"alert_levels": ["ERROR", "CRITICAL"],
"priority": 90,
}
]
)
3. 企业微信应用
config = LogConfig(
notifiers=[
{
"type": "weixin_app",
"corpid": "ww1234567890abcdef", # 必填:企业ID
"corpsecret": "YOUR_SECRET", # 必填:应用Secret
"agentid": 1000002, # 必填:应用AgentId
"touser": "@all", # 可选:接收用户
"toparty": "", # 可选:接收部门
"totag": "", # 可选:接收标签
"alert_levels": ["ERROR", "CRITICAL"],
"priority": 80,
}
]
)
4. 邮件通知
config = LogConfig(
notifiers=[
{
"type": "email",
"smtp_host": "smtp.qq.com", # 必填:SMTP服务器
"smtp_port": 465, # 可选:端口
"smtp_user": "your@qq.com", # 必填:SMTP用户名
"smtp_password": "your_password", # 必填:SMTP密码/授权码
"use_ssl": True, # 可选:使用SSL
"use_tls": False, # 可选:使用TLS
"from_addr": "your@qq.com", # 可选:发件人
"from_name": "XQCLog告警", # 可选:发件人名称
"to_addrs": ["admin@example.com"], # 必填:收件人列表
"cc_addrs": ["manager@example.com"], # 可选:抄送列表
"subject_prefix": "[生产告警]", # 可选:主题前缀
"alert_levels": ["ERROR", "CRITICAL"],
"priority": 70,
}
]
)
常用邮箱 SMTP 配置:
| 邮箱 | SMTP 服务器 | SSL 端口 | TLS 端口 | 说明 |
|---|---|---|---|---|
| QQ 邮箱 | smtp.qq.com | 465 | 587 | 需要使用授权码 |
| 163 邮箱 | smtp.163.com | 465 | 994 | 需开启SMTP服务 |
| Gmail | smtp.gmail.com | 465 | 587 | 需应用专用密码 |
| Outlook | smtp.office365.com | 587 | 587 | - |
发送策略
parallel(并行发送)- 默认
同时发送到所有通知器:
config = LogConfig(
alert_strategy="parallel", # 默认策略
notifiers=[...]
)
适用场景: 多渠道同时通知
sequential(顺序发送)
按顺序发送到所有通知器:
config = LogConfig(
alert_strategy="sequential",
notifiers=[...]
)
适用场景: 需要保证顺序
failover(故障转移)
轮询发送直到成功:
config = LogConfig(
alert_strategy="failover",
alert_retry=3,
alert_retry_delay=2.0,
notifiers=[
{"type": "dingtalk", ...}, # 优先尝试
{"type": "weixin_app", ...}, # 失败则尝试
{"type": "email", ...}, # 最后尝试
]
)
适用场景: 高可靠性需求
priority(优先级发送)
按优先级发送:
config = LogConfig(
alert_strategy="priority",
notifiers=[
{
"type": "weixin_app",
"priority": 100, # 最高优先级
"alert_levels": ["CRITICAL"],
},
{
"type": "dingtalk",
"priority": 50,
"alert_levels": ["ERROR", "CRITICAL"],
},
]
)
适用场景: 分级通知
alert 参数详解
基本用法
from xqclog import init_logger, logger
init_logger(
notifiers=[
{
"type": "dingtalk",
"webhook": "https://...",
"alert_levels": ["ERROR", "CRITICAL"],
}
]
)
# 场景1:强制发送(无论配置)
logger.info("重要通知", alert=True) # ✅ 发送
# 场景2:强制不发送(无论配置)
logger.error("已知错误", alert=False) # ❌ 不发送
# 场景3:根据配置(默认行为)
logger.error("未知错误") # ✅ 发送(ERROR在alert_levels中)
logger.warning("警告") # ❌ 不发送(WARNING不在alert_levels中)
alert_levels=None 的用法
设置为 None 表示默认不发送,只在手动指定时发送:
config = LogConfig(
notifiers=[
{
"type": "email",
"smtp_host": "smtp.qq.com",
"smtp_port": 465,
"smtp_user": "alert@qq.com",
"smtp_password": "password",
"use_ssl": True,
"to_addrs": ["ceo@company.com"],
"alert_levels": None, # 👈 默认不发送
}
]
)
init_logger(config)
# 所有日志默认都不发送邮件
logger.error("普通错误") # ❌ 不发送
logger.critical("严重错误") # ❌ 不发送
# 只有手动指定才发送
logger.critical("系统崩溃", alert=True) # ✅ 发送邮件
适用场景: 重要通知渠道(如邮件),避免频繁打扰,只在关键时刻手动触发
结构化日志中使用 alert
所有结构化日志方法都支持 alert 参数:
# HTTP 请求日志
logger.log_request(
method="POST",
url="/api/pay",
status=500,
duration=5.0,
alert=True # 支付接口错误需要告警
)
# 数据库查询日志
logger.log_db_query(
query="UPDATE orders SET status=?",
duration=10.5,
rows=1000,
alert=True # 慢查询告警
)
# API 调用日志
logger.log_api_call(
api_name="支付宝支付",
duration=2.0,
success=True,
alert=True, # 支付成功也通知
amount=1000.0
)
# 业务日志
logger.log_business(
event="用户注册",
level="INFO",
alert=True, # 重要业务事件
user_id=12345
)
实际应用示例
示例1:支付业务
def process_payment(order_id: str, amount: float):
try:
result = payment_api.pay(order_id, amount)
if result.success:
logger.info(f"支付成功: {order_id}")
return True
else:
# 支付失败可能是正常情况(余额不足等)
logger.error(
f"支付失败: {order_id}, 原因: {result.reason}",
alert=False # 👈 不发送告警
)
return False
except NetworkError:
# 网络错误需要人工介入
logger.error(
f"支付网络异常: {order_id}",
alert=True # 👈 强制发送告警
)
return False
示例2:多通知器分级
config = LogConfig(
alert_strategy="parallel",
notifiers=[
# 钉钉 - 常规错误
{
"type": "dingtalk",
"webhook": "https://...",
"alert_levels": ["ERROR", "CRITICAL"],
},
# 企业微信 - 只通知严重错误
{
"type": "weixin_app",
"corpid": "...",
"corpsecret": "...",
"agentid": 1000002,
"touser": "DevLead",
"alert_levels": ["CRITICAL"],
},
# 邮件 - 只在手动触发时发送
{
"type": "email",
"smtp_host": "smtp.qq.com",
"smtp_port": 465,
"smtp_user": "alert@qq.com",
"smtp_password": "password",
"use_ssl": True,
"to_addrs": ["ceo@company.com"],
"alert_levels": None, # 默认不发送
}
]
)
init_logger(config)
# 普通错误
logger.error("数据处理失败")
# → 钉钉:✅ 发送
# → 企业微信:❌ 不发送
# → 邮件:❌ 不发送
# 严重错误
logger.critical("数据库宕机")
# → 钉钉:✅ 发送
# → 企业微信:✅ 发送
# → 邮件:❌ 不发送
# 核心业务故障(需CEO知晓)
logger.critical("订单系统瘫痪", alert=True)
# → 钉钉:✅ 发送
# → 企业微信:✅ 发送
# → 邮件:✅ 发送(alert=True强制发送)
装饰器
@log_execution - 记录函数执行
from xqclog import logger, init_logger
from xqclog.decorators import log_execution
init_logger()
@log_execution(
level="INFO", # 日志级别
log_args=True, # 是否记录参数
log_result=True, # 是否记录返回值
log_time=True # 是否记录执行时间
)
def calculate(a: int, b: int) -> int:
return a + b
result = calculate(10, 20)
# 输出:
# 执行函数: __main__.calculate | 参数: args=(10, 20), kwargs={}
# 函数执行成功: __main__.calculate | 耗时: 0.0001秒 | 返回值: 30
@catch_errors - 捕获异常
from xqclog.decorators import catch_errors
@catch_errors(
level="ERROR", # 日志级别
reraise=False, # 是否重新抛出异常
default_return=None # 异常时的默认返回值
)
def risky_function(x: int) -> int:
if x < 0:
raise ValueError("x 不能为负数")
return x * 2
result = risky_function(-1) # 返回 None,不抛出异常
@timer - 计时器
from xqclog.decorators import timer
@timer(name="数据处理", level="INFO")
def process_data():
import time
time.sleep(1)
return "完成"
result = process_data()
# 输出:
# ⏱️ 开始计时: 数据处理
# ⏱️ 数据处理 执行完成,耗时: 1.0001秒
结构化日志
HTTP 请求日志
from xqclog import logger, init_logger
init_logger()
logger.log_request(
method="GET",
url="/api/users",
status=200,
duration=0.123,
# 可选参数
user_id=12345,
ip="192.168.1.1",
alert=False # 成功请求不需要告警
)
# 输出:🌐 GET /api/users - 200 - 0.123s
数据库查询日志
logger.log_db_query(
query="SELECT * FROM users WHERE id = ?",
duration=0.015,
rows=100,
db_name="main",
table="users"
)
# 输出:💾 数据库查询 - 0.015s - 100 rows
# SELECT * FROM users WHERE id = ?
API 调用日志
logger.log_api_call(
api_name="支付宝支付",
duration=1.234,
success=True,
order_id="ORD123456",
amount=99.9
)
# 输出:📡 API调用: 支付宝支付 - ✅ 成功 - 1.234s
性能指标日志
logger.log_performance(
metric_name="API响应时间",
value=123.45,
unit="ms",
endpoint="/api/users"
)
# 输出:📊 性能指标: API响应时间 = 123.45ms
业务日志
logger.log_business(
event="用户注册",
level="INFO",
user_id=12345,
username="张三",
email="zhangsan@example.com"
)
# 输出:💼 业务事件: 用户注册
上下文管理
计时器上下文
from xqclog import logger, init_logger
init_logger()
with logger.timer("数据处理", level="INFO"):
import time
time.sleep(1)
# 输出:
# ⏱️ 开始: 数据处理
# ⏱️ 完成: 数据处理,耗时: 1.0001秒
绑定上下文信息
# 方式1:bind 方法
user_logger = logger.bind(user_id=12345, username="张三")
user_logger.info("用户登录")
user_logger.info("查看订单")
# 方式2:contextualize 上下文管理器
with logger.contextualize(request_id="REQ-12345", ip="192.168.1.1"):
logger.info("处理请求")
logger.info("返回响应")
💼 完整示例
示例1:生产环境配置
from xqclog import init_logger, logger, LogConfig
# 生产环境完整配置
config = LogConfig(
# 基础配置
log_level="INFO",
log_dir="/var/log/myapp",
log_file="app.log",
# 日志轮转
rotation="500 MB",
retention="90 days",
compression="gz",
# 输出控制
console_output=False, # 生产环境不输出控制台
file_output=True,
auto_split=True, # 按级别分割
# 性能优化
enqueue=True,
diagnose=False, # 生产环境关闭诊断
# 告警配置
alert_strategy="priority",
alert_retry=3,
alert_retry_delay=2.0,
notifiers=[
# 邮件 - 只在手动触发时发送
{
"type": "email",
"smtp_host": "smtp.company.com",
"smtp_port": 465,
"smtp_user": "alert@company.com",
"smtp_password": "password",
"use_ssl": True,
"to_addrs": ["ceo@company.com"],
"alert_levels": None, # 默认不发送
"priority": 100,
},
# 企业微信 - ERROR和CRITICAL
{
"type": "weixin_app",
"corpid": "wwxxx",
"corpsecret": "secret",
"agentid": 1000002,
"touser": "DevTeam",
"alert_levels": ["ERROR", "CRITICAL"],
"priority": 90,
},
# 钉钉 - ERROR和CRITICAL
{
"type": "dingtalk",
"webhook": "https://...",
"alert_levels": ["ERROR", "CRITICAL"],
"priority": 80,
},
]
)
init_logger(config)
# 应用启动
logger.info("应用启动", version="1.0.0")
# 业务处理
logger.error("一般错误") # 发送到企业微信和钉钉
logger.critical("系统崩溃", alert=True) # 发送到所有渠道包括邮件
示例2:Flask Web 应用
from flask import Flask, request
from xqclog import logger, init_logger
# 使用 Web 预设
init_logger(preset="web")
app = Flask(__name__)
@app.before_request
def log_request_info():
logger.bind(
request_id=request.headers.get('X-Request-ID', 'N/A'),
ip=request.remote_addr
).info(f"收到请求: {request.method} {request.path}")
@app.after_request
def log_response_info(response):
logger.log_request(
method=request.method,
url=request.path,
status=response.status_code,
duration=0.1,
ip=request.remote_addr
)
return response
@app.route('/api/users')
def get_users():
with logger.timer("查询用户"):
users = [{"id": 1, "name": "张三"}]
logger.log_db_query(
query="SELECT * FROM users",
duration=0.015,
rows=len(users)
)
return {"users": users}
if __name__ == '__main__':
app.run()
示例3:数据处理任务
import pandas as pd
from xqclog import logger, init_logger
from xqclog.decorators import log_execution, timer
# 使用数据处理预设
init_logger(preset="data", log_level="INFO", auto_split=True)
@log_execution(log_args=True, log_time=True)
def load_data(file_path: str) -> pd.DataFrame:
"""加载数据"""
with logger.timer("加载数据"):
df = pd.read_csv(file_path)
logger.info(f"加载 {len(df)} 行数据")
return df
@timer(name="数据处理")
def process_data(df: pd.DataFrame) -> pd.DataFrame:
"""处理数据"""
logger.info("开始数据清洗")
df_clean = df.dropna()
logger.log_performance(
metric_name="数据清洗完成率",
value=len(df_clean) / len(df) * 100,
unit="%"
)
return df_clean
def main():
logger.log_business("开始数据处理任务")
df = load_data("input.csv")
df_processed = process_data(df)
df_processed.to_csv("output.csv", index=False)
logger.log_business("数据处理完成", alert=True) # 完成后通知
if __name__ == '__main__':
main()
📚 API 参考
LogConfig 类
主要参数:
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| log_level | str | "INFO" | 日志级别 |
| log_dir | str | "logs" | 日志目录 |
| log_file | str | "app.log" | 日志文件名 |
| rotation | str | "100 MB" | 轮转条件 |
| retention | str | "30 days" | 保留时间 |
| auto_split | bool | False | 自动分割 |
| alert_strategy | str | "parallel" | 告警策略 |
| alert_retry | int | 3 | 重试次数 |
| notifiers | list | [] | 通知器列表 |
方法:
# 从文件加载
config = LogConfig.from_file("logging.yaml")
# 从字典创建
config = LogConfig.from_dict({"log_level": "INFO"})
# 转换为字典
config_dict = config.to_dict()
XQCLogger 类
日志记录方法(均支持 alert 参数):
logger.trace(message, *args, alert=None, **kwargs)
logger.debug(message, *args, alert=None, **kwargs)
logger.info(message, *args, alert=None, **kwargs)
logger.success(message, *args, alert=None, **kwargs)
logger.warning(message, *args, alert=None, **kwargs)
logger.error(message, *args, alert=None, **kwargs)
logger.critical(message, *args, alert=None, **kwargs)
logger.exception(message, *args, alert=None, **kwargs)
结构化日志方法:
logger.log_request(method, url, status, duration, alert=None, **extra)
logger.log_db_query(query, duration, rows=None, alert=None, **extra)
logger.log_api_call(api_name, duration, success=True, alert=None, **extra)
logger.log_performance(metric_name, value, unit="ms", alert=None, **extra)
logger.log_business(event, level="INFO", alert=None, **extra)
上下文管理:
logger.timer(name, level="INFO")
logger.bind(**kwargs)
logger.contextualize(**kwargs)
配置管理:
logger.set_level(level)
logger.get_config()
logger.save_config(config_file)
💡 最佳实践
1. 环境配置
使用预设配置适配不同环境:
import os
from xqclog import init_logger
# 推荐:使用 auto 预设
init_logger(preset="auto")
# 或手动判断
env = os.getenv("ENV", "development")
if env == "production":
init_logger(preset="production")
else:
init_logger(preset="development")
2. 告警配置建议
# ✅ 推荐:只告警需要人工介入的问题
alert_levels = ["ERROR", "CRITICAL"]
# ⚠️ 谨慎:可能产生告警疲劳
alert_levels = ["WARNING", "ERROR", "CRITICAL"]
# ✅ 推荐:重要通知用 alert_levels=None
{
"type": "email",
"to_addrs": ["ceo@company.com"],
"alert_levels": None # 默认不发送,手动触发
}
3. 避免告警疲劳
# ❌ 不推荐:所有错误都告警
logger.error("用户输入错误")
# ✅ 推荐:区分需要人工介入的错误
logger.error("用户输入错误", alert=False) # 不告警
logger.error("数据库连接失败", alert=True) # 需要告警
4. 结构化日志
# ✅ 推荐:使用结构化方法
logger.log_request(method="GET", url="/api/users", status=200, duration=0.1)
# ❌ 不推荐:使用普通字符串
logger.info("GET /api/users 200 0.1s")
5. 异常处理
try:
risky_operation()
except Exception as e:
# ✅ 推荐:使用 exception(自动记录堆栈)
logger.exception("操作失败")
# ❌ 不推荐:使用 error(丢失堆栈)
logger.error(f"操作失败: {e}")
6. 性能优化
# 生产环境推荐配置
init_logger(
enqueue=True, # 异步写入
diagnose=False, # 关闭诊断
backtrace=True, # 保留异常追踪
)
❓ 常见问题
Q1: 如何控制哪些日志发送通知?
A: 有三种方式(优先级从高到低):
- 使用 alert 参数(最高优先级)
logger.error("错误", alert=False) # 强制不发送
logger.info("重要", alert=True) # 强制发送
- 配置 alert_levels
notifiers=[{
"type": "dingtalk",
"alert_levels": ["ERROR", "CRITICAL"] # 只有这两个级别发送
}]
- 设置 alert_levels=None
notifiers=[{
"type": "email",
"alert_levels": None # 默认不发送,只在 alert=True 时发送
}]
Q2: alert 参数在哪些方法中可用?
A: 所有日志方法都支持:
# 基础方法
logger.info("...", alert=True)
logger.error("...", alert=True)
# 结构化日志
logger.log_request(..., alert=True)
logger.log_db_query(..., alert=True)
logger.log_api_call(..., alert=True)
logger.log_business(..., alert=True)
Q3: 如何在多个模块中使用?
A: XQCLog 使用单例模式,直接导入即可:
# module_a.py
from xqclog import logger
logger.info("模块 A")
# module_b.py
from xqclog import logger
logger.info("模块 B")
# main.py
from xqclog import init_logger
import module_a, module_b
init_logger() # 只需初始化一次
Q4: 如何测试告警是否正常?
A:
from xqclog import logger, init_logger
init_logger(
notifiers=[{
"type": "dingtalk",
"webhook": "https://...",
"alert_levels": ["ERROR"]
}]
)
# 手动触发测试
logger.error("【测试】告警系统测试,请忽略")
Q5: QQ 邮箱发送失败?
A: QQ 邮箱需要使用授权码:
- 登录 QQ 邮箱
- 设置 -> 账户 -> 开启 SMTP 服务
- 获取授权码
- 使用授权码作为
smtp_password(不是 QQ 密码)
Q6: 日志文件太大怎么办?
A: 使用日志轮转和压缩:
init_logger(
rotation="100 MB", # 超过100MB轮转
retention="30 days", # 只保留30天
compression="gz" # 压缩旧日志
)
Q7: 企业微信通知收不到?
A: 检查以下几点:
- 确认
corpid、corpsecret、agentid正确 - 确认应用有发送消息权限
- 确认接收人在通讯录中
- 查看日志中的错误信息
📝 更新日志
🆕 最新更新 v0.0.3
✨ 开箱即用
- 🚀 现在无需手动初始化,直接导入即可使用!
from xqclog import logger
logger.debug("调试信息")
logger.info("普通信息")
logger.warning("警告信息")
logger.error("错误信息")
# 就这么简单!无需配置,默认 DEBUG 级别,彩色输出
v0.0.2 (2025-11-21)
🐛 Bug 修复
- 修复日志调用者名称显示问题
- 所有日志方法添加
opt(depth=1)参数,确保正确识别真实的调用者 - 修复前:所有日志显示为
xqclog.logger - 修复后:正确显示调用者的模块名、函数名和行号
- 所有日志方法添加
✨ 新功能
- 支持标准库 logging 格式
- 新增
logging_format参数,兼容标准库logging的格式字符串 - 自动将 logging 格式转换为 loguru 格式
- 便于从标准库 logging 迁移
- 新增
# 使用 logging 格式初始化
from xqclog import init_logger
logger = init_logger(
logging_format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
v0.0.1 (2025-11-18)
🎉 首次发布
核心功能:
- ✨ 基于 Loguru 增强封装
- ✨ 7种预设配置(auto/development/testing/production/web/crawler/data)
- ✨ 配置文件支持(YAML/JSON)
- ✨ 日志自动分割、轮转、压缩
- ✨ 彩色控制台输出
- ✨ 异步日志写入
告警通知:
- ✨ 钉钉机器人通知(支持签名、@人)
- ✨ 企业微信 Webhook 通知
- ✨ 企业微信应用通知(支持 Token 缓存)
- ✨ 邮件通知(支持 SSL/TLS、HTML 格式、抄送)
- ✨ 自定义通知器支持
灵活告警控制:
- ✨
alert参数精确控制发送 - ✨
alert_levels支持None(默认不发送) - ✨ 三层优先级控制机制
发送策略:
- ✨ parallel(并行发送)
- ✨ sequential(顺序发送)
- ✨ failover(故障转移)
- ✨ priority(优先级发送)
- ✨ 自动重试机制
增强功能:
- ✨ 装饰器支持(@log_execution、@catch_errors、@timer)
- ✨ 结构化日志(HTTP、数据库、API、性能、业务)
- ✨ 上下文管理(timer、bind、contextualize)
- ✨ 完整的类型注解
- ✨ 中文友好的文档
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file xqclog-0.0.3.tar.gz.
File metadata
- Download URL: xqclog-0.0.3.tar.gz
- Upload date:
- Size: 45.0 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: poetry/2.1.3 CPython/3.13.3 Windows/11
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
35e3e8c600bd27fd29836effc3ff58ef905d89b4aa5a42cb87ca7a51cebc287b
|
|
| MD5 |
ecab376eca8f404198836963fa77d8c8
|
|
| BLAKE2b-256 |
2f9c73b8318b5e73c4371d05ca1acc7b421046491af31dfeeaad1ad170549fe8
|
File details
Details for the file xqclog-0.0.3-py3-none-any.whl.
File metadata
- Download URL: xqclog-0.0.3-py3-none-any.whl
- Upload date:
- Size: 44.2 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: poetry/2.1.3 CPython/3.13.3 Windows/11
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
2ab873962b808652abac31bcf4afcb62d8711097cf4f2508676f9794b57ad036
|
|
| MD5 |
3fc40810ff6c558bc44e32b47ed35cef
|
|
| BLAKE2b-256 |
1be05a964b4969eb4026935e99e72be7780aaaea840fb12b0c172e5bc14110db
|