跳转至

General 通用工具模块

提供通用工具和功能,包括高级日志记录、多进程支持、print重定向等

📋 目录

✨ 功能特性

  • 📝 高级日志: 基于loguru的强大日志记录功能
  • 🔄 多进程支持: 线程安全的日志记录
  • 🖨️ Print重定向: 自动捕获print输出到日志
  • 📊 日志分级: 支持多种日志级别和过滤
  • 📁 文件管理: 自动日志文件轮转和管理
  • 🎨 彩色输出: 美观的控制台彩色日志
  • 🛡️ 异常捕获: 自动异常记录和堆栈跟踪

🚀 快速开始

基本使用

from sindre.general.logs import CustomLogger

# 创建基本日志记录器
logger = CustomLogger(
    logger_name="MyApp",
    level="INFO",
    console_output=True,
    file_output=True
).get_logger()

# 记录日志
logger.info("应用程序启动")
logger.warning("发现潜在问题")
logger.error("发生错误")

高级配置

from sindre.general.logs import CustomLogger

# 创建高级日志记录器
logger = CustomLogger(
    logger_name="AdvancedApp",
    level="DEBUG",
    log_dir="logs",
    console_output=True,
    file_output=True,
    capture_print=True
).get_logger()

# 记录不同级别的日志
logger.debug("调试信息")
logger.info("普通信息")
logger.success("成功信息")
logger.warning("警告信息")
logger.error("错误信息")
logger.critical("严重错误")

🔧 核心功能

CustomLogger 类

class CustomLogger:
    """自定义日志记录器类"""

    def __init__(
        self,
        logger_name=None,
        level="DEBUG",
        log_dir="logs",
        console_output=True,
        file_output=False,
        capture_print=False,
        filter_log=None
    ):
        """
        初始化日志记录器

        Args:
            logger_name: 日志记录器名称
            level: 日志级别 (TRACE, DEBUG, INFO, SUCCESS, WARNING, ERROR, CRITICAL)
            log_dir: 日志文件存储目录
            console_output: 是否启用控制台输出
            file_output: 是否启用文件输出
            capture_print: 是否捕获print输出
            filter_log: 自定义日志过滤函数
        """

    def get_logger(self):
        """获取配置好的日志记录器实例"""
        return self.logger

📖 使用指南

1. 基本日志记录

简单日志记录

from sindre.general.logs import CustomLogger

# 创建基本日志记录器
logger = CustomLogger(
    logger_name="SimpleApp",
    level="INFO",
    console_output=True
).get_logger()

# 记录不同级别的日志
logger.debug("这是调试信息")
logger.info("这是普通信息")
logger.warning("这是警告信息")
logger.error("这是错误信息")
logger.critical("这是严重错误")

# 格式化日志
name = "张三"
age = 25
logger.info(f"用户 {name} 年龄 {age} 岁")
logger.info("用户 {} 年龄 {} 岁", name, age)  # loguru风格

文件日志记录

from sindre.general.logs import CustomLogger

# 创建文件日志记录器
logger = CustomLogger(
    logger_name="FileApp",
    level="DEBUG",
    log_dir="logs",
    console_output=True,
    file_output=True
).get_logger()

# 日志会自动保存到文件
logger.info("这条信息会同时显示在控制台和保存到文件")
logger.error("错误信息也会被记录")

# 查看生成的日志文件
# - logs/run_YYYY-MM-DD.log (运行日志,每天轮转)
# - logs/error.log (错误日志,10MB轮转)

2. Print重定向

捕获Print输出

from sindre.general.logs import CustomLogger

# 创建支持print重定向的日志记录器
logger = CustomLogger(
    logger_name="PrintCapture",
    level="INFO",
    console_output=True,
    capture_print=True
).get_logger()

# 普通的print语句会被自动捕获
print("这条print语句会被捕获并记录到日志中")
print("包含行号信息的print输出")

# 日志输出示例:
# 2024-01-01 12:00:00 | INFO     | PrintCapture | main.py:main:15 - Print(line 15): 这条print语句会被捕获并记录到日志中

混合使用

from sindre.general.logs import CustomLogger

logger = CustomLogger(
    logger_name="MixedApp",
    level="INFO",
    console_output=True,
    capture_print=True
).get_logger()

# 使用logger记录
logger.info("使用logger记录的信息")

# 使用print(会被自动捕获)
print("使用print输出的信息")

# 两者都会显示在日志中,但格式略有不同

3. 异常捕获

自动异常捕获

from sindre.general.logs import CustomLogger

logger = CustomLogger(
    logger_name="ExceptionApp",
    level="INFO",
    console_output=True
).get_logger()

# 使用装饰器自动捕获异常
@logger.catch
def risky_function():
    """可能出错的函数"""
    result = 10 / 0
    return result

# 调用函数,异常会被自动记录
try:
    risky_function()
except Exception:
    pass

# 手动捕获异常
try:
    raise ValueError("这是一个测试异常")
except Exception:
    logger.exception("捕获到异常")

4. 日志过滤

自定义过滤器

from sindre.general.logs import CustomLogger

# 定义过滤函数
def info_only_filter(record):
    """只显示INFO级别的日志"""
    return record["level"].name == "INFO"

def exclude_sensitive_filter(record):
    """排除包含敏感信息的日志"""
    sensitive_words = ["password", "token", "secret"]
    message = record["message"].lower()
    return not any(word in message for word in sensitive_words)

# 创建带过滤器的日志记录器
logger = CustomLogger(
    logger_name="FilteredApp",
    level="DEBUG",
    console_output=True,
    filter_log=info_only_filter
).get_logger()

# 只有INFO级别的日志会被显示
logger.debug("这条调试信息不会显示")
logger.info("这条信息会显示")
logger.warning("这条警告不会显示")

🚀 高级功能

1. 多进程日志

from sindre.general.logs import CustomLogger
import multiprocessing as mp

def worker_process(logger_name, process_id):
    """工作进程函数"""
    logger = CustomLogger(
        logger_name=logger_name,
        level="INFO",
        log_dir="logs",
        console_output=False,
        file_output=True
    ).get_logger()

    logger.info(f"进程 {process_id} 开始工作")
    # 执行一些工作...
    logger.info(f"进程 {process_id} 完成工作")

# 主进程
if __name__ == "__main__":
    main_logger = CustomLogger(
        logger_name="MainApp",
        level="INFO",
        console_output=True,
        file_output=True
    ).get_logger()

    main_logger.info("启动多进程任务")

    # 创建多个进程
    processes = []
    for i in range(4):
        p = mp.Process(target=worker_process, args=("WorkerApp", i))
        processes.append(p)
        p.start()

    # 等待所有进程完成
    for p in processes:
        p.join()

    main_logger.info("所有进程完成")

2. 日志轮转

from sindre.general.logs import CustomLogger

# 日志会自动轮转
logger = CustomLogger(
    logger_name="RotationApp",
    level="INFO",
    log_dir="logs",
    file_output=True
).get_logger()

# 运行日志:每天00:00轮转
# 错误日志:达到10MB时轮转
for i in range(1000):
    logger.info(f"这是第 {i} 条日志信息")
    if i % 100 == 0:
        logger.error(f"这是第 {i} 条错误信息")

3. 性能监控

from sindre.general.logs import CustomLogger
import time

logger = CustomLogger(
    logger_name="PerformanceApp",
    level="INFO",
    console_output=True
).get_logger()

# 使用上下文管理器记录执行时间
@logger.catch
def performance_test():
    start_time = time.time()

    # 模拟耗时操作
    time.sleep(2)

    end_time = time.time()
    execution_time = end_time - start_time

    logger.info(f"操作完成,耗时: {execution_time:.2f}秒")

    # 模拟异常
    if execution_time > 1.5:
        raise TimeoutError("操作超时")

# 执行性能测试
performance_test()

4. 结构化日志

from sindre.general.logs import CustomLogger
import json

logger = CustomLogger(
    logger_name="StructuredApp",
    level="INFO",
    console_output=True
).get_logger()

# 记录结构化数据
user_data = {
    "user_id": 12345,
    "username": "张三",
    "action": "login",
    "timestamp": "2024-01-01T12:00:00Z"
}

logger.info("用户操作: {}", json.dumps(user_data, ensure_ascii=False))

# 记录业务事件
def log_business_event(event_type, data):
    """记录业务事件"""
    event = {
        "event_type": event_type,
        "data": data,
        "timestamp": time.time()
    }
    logger.info("业务事件: {}", json.dumps(event, ensure_ascii=False))

# 使用示例
log_business_event("user_login", {"user_id": 12345, "ip": "192.168.1.1"})
log_business_event("order_created", {"order_id": "ORD001", "amount": 99.99})

⚙️ 配置选项

日志级别

# 支持的日志级别(从低到高)
levels = [
    "TRACE",      # 最详细的调试信息
    "DEBUG",      # 调试信息
    "INFO",       # 一般信息
    "SUCCESS",    # 成功信息
    "WARNING",    # 警告信息
    "ERROR",      # 错误信息
    "CRITICAL"    # 严重错误
]

# 设置日志级别
logger = CustomLogger(
    logger_name="LevelApp",
    level="WARNING",  # 只显示WARNING及以上级别的日志
    console_output=True
).get_logger()

日志格式

# 默认日志格式
log_format = (
    "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
    "<level>{level: <8}</level> | "
    "<blue>{extra[name]: <8}</blue> | "
    "<cyan>{file}</cyan>:<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
    "<level>{message}</level>"
)

# 输出示例:
# 2024-01-01 12:00:00 | INFO     | MyApp     | main.py:main:15 - 应用程序启动

文件配置

# 日志文件配置
logger = CustomLogger(
    logger_name="FileConfigApp",
    level="INFO",
    log_dir="custom_logs",  # 自定义日志目录
    console_output=True,
    file_output=True
).get_logger()

# 生成的文件:
# - custom_logs/run_2024-01-01.log (运行日志)
# - custom_logs/error.log (错误日志)

❓ 常见问题

Q1: 如何禁用特定模块的日志?

A: 使用enable/disable方法:

from sindre.general.logs import CustomLogger

logger = CustomLogger("MyApp").get_logger()

# 禁用当前模块的日志
logger.disable(__name__)
logger.info("这条日志不会显示")

# 重新启用
logger.enable(__name__)
logger.info("这条日志会显示")

Q2: 如何自定义日志格式?

A: 修改CustomLogger类的_configure_logger方法:

# 在CustomLogger类中修改log_format
log_format = (
    "{time:YYYY-MM-DD HH:mm:ss} | "
    "{level} | "
    "{name} | "
    "{message}"
)

Q3: 如何处理大量日志?

A: 使用日志轮转和过滤:

logger = CustomLogger(
    logger_name="HighVolumeApp",
    level="WARNING",  # 只记录重要日志
    log_dir="logs",
    file_output=True
).get_logger()

# 日志会自动轮转,避免文件过大

Q4: 如何在不同进程间共享日志?

A: 使用文件输出和进程安全的配置:

logger = CustomLogger(
    logger_name="MultiProcessApp",
    level="INFO",
    log_dir="logs",
    console_output=False,  # 避免控制台冲突
    file_output=True       # 使用文件输出
).get_logger()

Q5: 如何捕获第三方库的日志?

A: 使用print重定向和异常捕获:

logger = CustomLogger(
    logger_name="ThirdPartyApp",
    level="INFO",
    console_output=True,
    capture_print=True  # 捕获print输出
).get_logger()

# 第三方库的print输出会被捕获
import some_third_party_library
some_third_party_library.some_function()

Q6: 如何调试日志配置问题?

A: 使用简单的配置进行测试:

# 最简单的配置
logger = CustomLogger(
    logger_name="DebugApp",
    level="DEBUG",
    console_output=True,
    file_output=False
).get_logger()

# 逐步添加功能
logger.info("测试基本功能")
logger.debug("测试调试级别")

Q7: 如何优化日志性能?

A: 使用适当的配置:

logger = CustomLogger(
    logger_name="PerformanceApp",
    level="INFO",  # 避免过多DEBUG日志
    console_output=True,
    file_output=False,  # 如果不需要文件输出
    capture_print=False  # 如果不需要捕获print
).get_logger()

Q8: 如何处理日志文件权限问题?

A: 确保目录权限正确:

import os

# 确保日志目录存在且有写权限
log_dir = "logs"
os.makedirs(log_dir, exist_ok=True)

logger = CustomLogger(
    logger_name="PermissionApp",
    log_dir=log_dir,
    file_output=True
).get_logger()

📊 性能基准

操作 日志数量 时间 内存使用
控制台输出 10,000条 ~2s ~10MB
文件输出 10,000条 ~5s ~20MB
Print重定向 10,000条 ~3s ~15MB
异常捕获 1,000次 ~1s ~5MB

🔗 相关链接


如有问题,请查看 常见问题 或提交 Issue