Python产品数据管道需终端与文件双输出的彩色日志工具

首先纠正一个误区：Python标准库的logging模块完全支持同时输出日志到终端和文件，只是默认没有配置彩色终端输出而已。下面分两种方案给出实现方式，你可以根据需求选择或自行封装。

一、原生logging模块实现双端输出+终端彩色 #

1. 自定义彩色终端Handler #

通过继承logging.StreamHandler，重写format方法来添加颜色控制码：

import logging

class ColorStreamHandler(logging.StreamHandler):
    # 定义日志级别对应的颜色
    COLORS = {
        logging.DEBUG: '\033[94m',    # 蓝色
        logging.INFO: '\033[92m',     # 绿色
        logging.WARNING: '\033[93m',  # 黄色
        logging.ERROR: '\033[91m',    # 红色
        logging.CRITICAL: '\033[95m'  # 紫色
    }
    RESET = '\033[0m'

    def format(self, record):
        # 给日志级别名称添加颜色
        record.levelname = f"{self.COLORS.get(record.levelno, self.RESET)}{record.levelname}{self.RESET}"
        return super().format(record)

2. 配置双Handler（终端+文件） #

创建日志器，同时添加彩色终端Handler和文件Handler：

def setup_logger(name="data_pipeline"):
    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)
    logger.handlers.clear()  # 避免重复添加Handler

    # 1. 彩色终端Handler
    console_handler = ColorStreamHandler()
    console_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
    console_handler.setFormatter(console_formatter)
    logger.addHandler(console_handler)

    # 2. 文件Handler（无颜色，纯文本）
    file_handler = logging.FileHandler('pipeline.log', encoding='utf-8')
    file_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(module)s:%(lineno)d - %(message)s')
    file_handler.setFormatter(file_formatter)
    logger.addHandler(file_handler)

    return logger

3. 使用示例 #

logger = setup_logger()
logger.debug("调试信息，终端显示蓝色")
logger.info("普通信息，终端显示绿色")
logger.warning("警告信息，终端显示黄色")
logger.error("错误信息，终端显示红色")
logger.critical("严重错误，终端显示紫色")

二、第三方库简化实现 #

如果不想自己写Handler，可以用成熟的第三方库，以下两个比较常用：

1. colorlog #

基于标准logging扩展，直接支持彩色终端输出，同时可以配合FileHandler实现文件写入：

import logging
import colorlog

def setup_colorlog():
    logger = logging.getLogger("data_pipeline")
    logger.setLevel(logging.DEBUG)
    logger.handlers.clear()

    # 彩色终端配置
    console_handler = colorlog.StreamHandler()
    console_formatter = colorlog.ColoredFormatter(
        '%(log_color)s%(asctime)s - %(levelname)s - %(message)s',
        log_colors={
            'DEBUG': 'blue',
            'INFO': 'green',
            'WARNING': 'yellow',
            'ERROR': 'red',
            'CRITICAL': 'purple'
        }
    )
    console_handler.setFormatter(console_formatter)
    logger.addHandler(console_handler)

    # 文件Handler
    file_handler = logging.FileHandler('pipeline.log', encoding='utf-8')
    file_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(module)s:%(lineno)d - %(message)s')
    file_handler.setFormatter(file_formatter)
    logger.addHandler(file_handler)

    return logger

2. loguru #

更简洁的日志库，默认支持彩色终端输出，一行代码就能同时输出到文件：

from loguru import logger

# 配置：同时输出到终端（彩色）和文件（无颜色）
logger.add(
    "pipeline.log",
    format="{time} - {level} - {module}:{line} - {message}",
    level="DEBUG"
)

# 使用示例
logger.debug("调试信息")
logger.info("普通信息")
logger.warning("警告信息")
logger.error("错误信息")
logger.critical("严重错误")

三、封装统一类（按需自定义） #

如果你想封装成自己的统一日志类，可以参考以下结构：

import logging
from typing import Optional

class PipelineLogger:
    def __init__(self, name: str = "data_pipeline", log_file: str = "pipeline.log"):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(logging.DEBUG)
        self.logger.handlers.clear()

        # 初始化彩色终端Handler
        self._setup_console_handler()
        # 初始化文件Handler
        self._setup_file_handler(log_file)

    def _setup_console_handler(self):
        class ColorStreamHandler(logging.StreamHandler):
            COLORS = {
                logging.DEBUG: '\033[94m',
                logging.INFO: '\033[92m',
                logging.WARNING: '\033[93m',
                logging.ERROR: '\033[91m',
                logging.CRITICAL: '\033[95m'
            }
            RESET = '\033[0m'

            def format(self, record):
                record.levelname = f"{self.COLORS.get(record.levelno, self.RESET)}{record.levelname}{self.RESET}"
                return super().format(record)

        console_handler = ColorStreamHandler()
        formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
        console_handler.setFormatter(formatter)
        self.logger.addHandler(console_handler)

    def _setup_file_handler(self, log_file: str):
        file_handler = logging.FileHandler(log_file, encoding='utf-8')
        formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(module)s:%(lineno)d - %(message)s')
        file_handler.setFormatter(formatter)
        self.logger.addHandler(file_handler)

    def debug(self, msg: str):
        self.logger.debug(msg)

    def info(self, msg: str):
        self.logger.info(msg)

    def warning(self, msg: str):
        self.logger.warning(msg)

    def error(self, msg: str):
        self.logger.error(msg)

    def critical(self, msg: str):
        self.logger.critical(msg)

# 使用示例
pl = PipelineLogger()
pl.info("数据管道启动")
pl.error("数据读取失败")

内容的提问来源于stack exchange，提问作者B1smuth