本文主要分析 TimedRotatingFileHandler 在实际使用中 backupCount 设置未生效的问题。源码分析显示，文件删除依赖于后缀 suffix 的正则匹配，如果自定义了 suffix 格式，必须同步更新 extMatch 的正则表达式（保证正则表达式可以正常匹配到你新格式的日志文件）。

现象

本文为了测试日志轮转删除，将轮转间隔设为秒（S），设置文件的 backupCount 为 5，同时设置新的日志文件后缀格式为 %Y%m%d%H%M%S.log。

示例如下：

import logging
import os
from logging import handlers

os.makedirs("./logs", exist_ok=True)


def _logging(**kwargs):
    level = kwargs.pop('level', logging.DEBUG)
    filename = kwargs.pop('filename', 'default.log')
    datefmt = kwargs.pop('datefmt', '%Y-%m-%d %H:%M:%S')
    format = kwargs.pop('format', '[%(asctime)s,%(msecs)d][%(module)s][%(levelname)s] %(message)s')

    log = logging.getLogger(filename)
    format_str = logging.Formatter(format, datefmt)

    th = handlers.TimedRotatingFileHandler(filename=filename, when='S', backupCount=5, encoding="utf-8")
    th.suffix = "%Y%m%d%H%M%S.log"
    th.setFormatter(format_str)
    th.setLevel(level)

    log.addHandler(th)
    log.setLevel(level)
    return log


if __name__ == '__main__':
    logger = _logging(filename="./logs/test.log")
    logger.info("Hello world")

然而运行几遍以后，发现产生的日志文件个数已经远远超过了设置的 backupCount 的个数，但旧日期的日志文件却始终没有被删除。

原因

查看源码发现，文件删除依赖于suffix 正则匹配。

而 S 秒默认对应的 extMatch 是下面这个：

extMatch = r"^\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}(\.\w+)?$"

 默认的 extMatch 和新的后缀格式匹配不上，导致没有轮滚删除旧的文件。

解决

原因明了以后，解决方案也就明朗了。

设置新的匹配新日志格式的正则表达式就可以了（因为是初始化以后再进行的 extMatch 替换，所以替换时需要先对正则表达式的字符串用 re.compile 进行编译）：

extMatch = re.compile(r"^\d{4}\d{2}\d{2}\d{2}\d{2}\d{2}(\.\w+)?$", re.ASCII)

完整示例如下： 

import logging
import os
import re
from logging import handlers

os.makedirs("./logs", exist_ok=True)


def _logging(**kwargs):
    level = kwargs.pop('level', logging.DEBUG)
    filename = kwargs.pop('filename', 'default.log')
    datefmt = kwargs.pop('datefmt', '%Y-%m-%d %H:%M:%S')
    format = kwargs.pop('format', '[%(asctime)s,%(msecs)d][%(module)s][%(levelname)s] %(message)s')

    log = logging.getLogger(filename)
    format_str = logging.Formatter(format, datefmt)

    th = handlers.TimedRotatingFileHandler(filename=filename, when='S', backupCount=5, encoding="utf-8")
    th.suffix = "%Y%m%d%H%M%S.log"
    th.extMatch = re.compile(r"^\d{4}\d{2}\d{2}\d{2}\d{2}\d{2}(\.\w+)?$", re.ASCII)
    th.setFormatter(format_str)
    th.setLevel(level)

    log.addHandler(th)
    log.setLevel(level)
    return log


if __name__ == '__main__':
    logger = _logging(filename="./logs/test.log")
    logger.info("Hello world")

这个时候，产生的日志文件就可以正常轮转删除了。

日志脱敏

哈哈，再顺便来个日志脱敏的小函数，可以指定要脱敏的字段，会将对应字段的 value 替换为等长度的 * 字符串。

logger.py

import copy
import json
import logging
import os
import re
import traceback
from logging import handlers

os.makedirs("./logs", exist_ok=True)


def _logging(**kwargs):
    level = kwargs.pop('level', logging.DEBUG)
    filename = kwargs.pop('filename', 'default.log')
    datefmt = kwargs.pop('datefmt', '%Y-%m-%d %H:%M:%S')
    format = kwargs.pop('format', '[%(asctime)s,%(msecs)d][%(module)s][%(levelname)s] %(message)s')

    log = logging.getLogger(filename)
    format_str = logging.Formatter(format, datefmt)

    th = handlers.TimedRotatingFileHandler(filename=filename, when='S', backupCount=5, encoding="utf-8")
    th.suffix = "%Y%m%d%H%M%S.log"
    th.extMatch = re.compile(r"^\d{4}\d{2}\d{2}\d{2}\d{2}\d{2}(\.\w+)?$", re.ASCII)
    th.setFormatter(format_str)
    th.setLevel(level)

    log.addHandler(th)
    log.setLevel(level)
    return log


def do_desensitization(data, flags: list):
    data = copy.copy(data)
    try:
        if not isinstance(data, str):
            data = json.dumps(data, ensure_ascii=False)  # 非字符串类型数据都先统一转 json 字符串处理
        for flag in flags:
            match_pattern = f'([\'|\"]{flag}[\'|\"].*?:.*?[\'|\"](.*?)[\'|\"])'
            res = re.findall(match_pattern, data)
            for item in res:
                full_match = item[0]  # 获取全匹配结果
                value_match = item[1]  # 获取 value 匹配结果
                value_desensitization = '*' * len(value_match)  # 构造等长字符串
                full_match_desensitization = full_match.replace(value_match, value_desensitization)  # 脱敏替换
                data = data.replace(full_match, full_match_desensitization)  # 脱敏替换
        return data
    except Exception as e:
        traceback.print_exc()
        return data


if __name__ == '__main__':
    logger = _logging(filename="./logs/test.log")
    logger.info("Hello world")
    data = {'name': "Looking", "key": "value", 'test': {"name": "test name", "new_key": "new_value"}}
    result = do_desensitization(data, ["name", "key"])
    logger.info(result)

test.log

[2024-09-12 19:06:08,24][logger][INFO] Hello world
[2024-09-12 19:06:08,25][logger][INFO] {"name": "*******", "key": "*****", "test": {"name": "*********", "new_key": "new_value"}}