控制台日志

代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import logging

# 日志文件名
filename = 'test.log'


def loggers():
    # 创建一个logging对象
    logger = logging.getLogger()

    # 再创建一个handler用于输出到控制台
    ch = logging.StreamHandler()

    # 定义输出格式
    datafmt = "%Y-%m-%d %H:%M:%S"
    fmt = '%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s'
    formatter = logging.Formatter(fmt, datafmt)

    # 定义日志输出层级
    logger.setLevel(logging.DEBUG)

    # 定义控制台输出层级
    logger.setLevel(logging.DEBUG)

    # 为控制台操作符绑定格式
    ch.setFormatter(formatter)

    # 给logger对象绑定文件操作符
    logger.addHandler(ch)

    return logging


if __name__ == "__main__":
    log = loggers()
    log.debug('Debug状态')
    log.info('输入状态')
    log.warning('警告级别错误')
    log.error('产生错误信息')
    log.critical('产生严重错误')

效果图

文件日志

代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
import os
import logging

# 日志文件名
filename = 'test.log'


def do_file():
    # 获取当前文件夹
    current_path = os.path.dirname(__file__)
    # 获取当前文件夹的上一层文件
    base_path = os.path.dirname(current_path)

    # 拼接路径
    path = os.path.join(base_path, filename)
    print(f'{filename} -- 文件路径: {path}')

    """如果文件不存在就创建"""
    if not os.path.exists(path):
        try:
            fh = open(path, 'w')
        except IOError:
            print(f"Error: 创建 {filename} 文件成功！！！")
        else:
            print(f"创建 {filename} 文件成功！！！")
            fh.close()
    else:
        print(f"{filename} 文件已存在！！！")


def loggers():
    # 创建一个logging对象
    logger = logging.getLogger()

    # 创建一个handler，用于写入日志文件
    fh = logging.FileHandler(filename, mode='a', encoding='utf-8')

    # 再创建一个handler用于输出到控制台
    ch = logging.StreamHandler()

    # 定义输出格式
    datafmt = "%Y-%m-%d %H:%M:%S"
    fmt = '%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s'
    formatter = logging.Formatter(fmt, datafmt)

    # 定义日志输出层级
    logger.setLevel(logging.DEBUG)

    # 为文件、控制台操作符定义日志输出层级
    # fh.setLevel(logging.DEBUG)
    # fh.setLevel(logging.INFO)

    # 为文件、控制台操作符绑定格式
    fh.setFormatter(formatter)
    ch.setFormatter(formatter)

    # 给logger对象绑定文件、控制台操作符
    logger.addHandler(fh)
    logger.addHandler(ch)

    return logging


if __name__ == "__main__":
    log = loggers()
    log.debug('Debug状态')
    log.info('输入状态')
    log.warning('警告级别错误')
    log.error('产生错误信息')
    log.critical('产生严重错误')

效果图

  • 控制台

  • 文件

按时间切割日志

代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
import logging
import os.path
import time

# 日志文件名
folder_name = "logs"


def my_logging():
    # 第一步，创建一个logger
    log = logging.getLogger()
    # Log等级总开关
    log.setLevel(logging.DEBUG)

    # 第二步，创建一个handler，用于写入日志文件
    # (修改此处实现按 秒 分钟 小时 日 月 年 切分)
    # rq = time.strftime('%Y%m%d%H%M', time.localtime(time.time()))  # 按 分钟 切分日志
    rq = time.strftime('%Y%m%d%H%M%S', time.localtime(time.time()))

    # 获取当前路径
    log_path = os.path.dirname(os.getcwd()) + os.sep + folder_name + os.sep

    """如果文件夹不存在就创建"""
    os.makedirs(log_path.strip(), exist_ok=True)

    # 拼接日志名
    logfile = log_path + rq + '.log'
    fh = logging.FileHandler(filename=logfile, mode='w', encoding='utf-8')
    fh.setLevel(logging.DEBUG)  # 输出到file的log等级的开关
    # 创建一个handler，用于输出到控制台
    ch = logging.StreamHandler()
    ch.setLevel(logging.DEBUG)  # 输出到console的log等级的开关

    # 第三步，定义handler的输出格式
    datafmt = "%Y-%m-%d %H:%M:%S"
    fmt = '%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s'
    formatter = logging.Formatter(fmt, datafmt)
    fh.setFormatter(formatter)
    ch.setFormatter(formatter)

    # 第四步，将logger添加到handler里面
    log.addHandler(fh)
    log.addHandler(ch)

    return log


if __name__ == '__main__':
    logger = my_logging()
    logger.debug('Debug状态')
    logger.info('输入状态')
    logger.warning('警告级别错误')
    logger.error('产生错误信息')
    logger.critical('产生严重错误')

效果演示

按时间切割自动日志

说明

使用的是 TimedRotatingFileHandler( filename [, when [, interval [, backupCount] ] ] ) 函数

  • filename 输出日志的文件名
  • when 一个字符串，定义了日志切分的间隔时间单位
    • S：Second 秒
    • M：Minutes 分钟
    • H：Hour 小时
    • D：Days 天
    • W：Week day（0 = Monday）
    • midnight：Roll over at midnight
  • interval 间隔时间单位的个数，指等待多少个 when 的时间后 Logger 会自动重建新闻继续进行日志记录
  • backupCount 保留日志的文件个数，超过删除最早的日志

代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
import logging
import os
from logging import handlers

# 日志文件夹
folder_name = "logs"
# 临时日志文件名
loggerName = "log"
# 日志等级
level = "debug"


# 创建日志文件夹
def do_folder():
    # 获取当前文件夹
    current_path = os.path.dirname(__file__)

    # 获取当前文件夹的上一层文件
    base_path = os.path.abspath(os.path.join(current_path, ".."))

    # 拼接路径
    log_folder = base_path + os.sep + folder_name + os.sep
    # print(f'日志文件夹名: {folder_name} -- {log_folder} \n')
    """如果文件夹不存在就创建"""
    os.makedirs(log_folder.strip(), exist_ok=True)

    return log_folder


# 日志功能
def my_logging():
    level_relations = {
        'debug': logging.DEBUG,
        'info': logging.INFO,
        'warning': logging.WARNING,
        'error': logging.ERROR,
        'critical': logging.CRITICAL
    }

    # 获取 日志文件夹 路径
    log_folder = do_folder()
    log_file = log_folder + loggerName

    # 创建一个handler用于写入日志文件
    # when 时间段
    # backupCount 日志的最大个数, 超出删除最早的日志
    # delay 是否加入缓存, 不加缓存有错误(未尝试)
    fh = handlers.TimedRotatingFileHandler(filename=log_file, when='S', encoding='utf-8',
                                           backupCount=3, delay=True)

    fh.suffix = "%Y-%m-%d_%H-%M-%S.log"

    # 创建一个handler用于输出到控制台
    ch = logging.StreamHandler()

    # 定义输出格式
    date_fmt = '%Y-%m-%d %H:%M:%S'
    text_fmt = '%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s'
    # text_fmt = '%(asctime)s - %(pathname)s - [line:%(lineno)d] - %(levelname)s > %(message)s'
    format_str = logging.Formatter(text_fmt, date_fmt)

    # 为文件、控制台操作符绑定格式
    fh.setFormatter(format_str)
    ch.setFormatter(format_str)

    log = logging.getLogger()
    log.setLevel(level_relations.get(level))  # 设置 日志 等级

    # 给log对象绑定文件、控制台操作符
    log.addHandler(ch)
    log.addHandler(fh)

    return log


if __name__ == '__main__':
    logger = my_logging()
    logger.debug('Debug状态')
    logger.info('输入状态')
    logger.warning('警告级别错误')
    logger.error('产生错误信息')
    logger.critical('产生严重错误')

4.3 效果演示

封装按时间切割日志函数

说明

  • 使用的函数也是 TimedRotatingFileHandler
  • 只不过我不喜欢生成日志的名字，自己封装了一下

代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
import logging
import os
import re
from logging.handlers import TimedRotatingFileHandler

# 文件夹名
folderName = "logs"
# 临时日志文件名
loggerName = "log"
# 日志等级
level = "debug"

# 日志打印等级
LEVELS = {
    'debug': logging.DEBUG,
    'info': logging.INFO,
    'warning': logging.WARNING,
    'error': logging.ERROR,
    'critical': logging.CRITICAL
}


# 创建日志文件夹
def create_folder():
    # 获取当前文件夹
    current_path = os.path.dirname(__file__)

    # 获取当前文件夹的上一层文件
    base_path = os.path.abspath(os.path.join(current_path, ".."))

    # 拼接路径
    log_dir = base_path + os.sep + folderName + os.sep
    # print(f'日志文件夹名: {folderName} -- {log_dir} \n')

    """如果文件夹不存在就创建"""
    os.makedirs(log_dir.strip(), exist_ok=True)

    return log_dir


# 继承 TimedRotatingFileHandler
class MyTimedRotatingFileHandler(TimedRotatingFileHandler):
    """重写 getFilesToDelete 方法"""

    def getFilesToDelete(self):
        dirName = create_folder()
        # 获取文件下的所有的文件(不包含 临时日志文件)
        fileNames = os.listdir(dirName)
        result = []
        for fileName in fileNames:
            result.append(os.path.join(dirName, fileName))
        if len(result) < self.backupCount:
            result = []
        else:
            result.sort()
            result = result[:len(result) - self.backupCount]
        return result


# 日志方法
def my_logger():
    # 获取 日志文件夹 路径
    log_folder = create_folder()
    log_file = log_folder + loggerName  # 拼接 日志名

    # 创建一个handler，用于写入日志文件 (每分钟生成1个，最多保留3个日志)
    # when 时间段 [ S:秒 | M:分钟 | H:小时 | D:天 | W:周(0=Monday) | midnight:午夜 ]
    # backupCount 日志的最大个数, 超出删除最早的日志
    # delay 是否加入缓存, 不加缓存有错误(未尝试)
    fh = MyTimedRotatingFileHandler(filename=log_file, when='S', backupCount=3, encoding='utf-8', delay=True)
    fh.suffix = "%Y-%m-%d_%H-%M-%S.log"  # 日志文件的后缀名
    fh.extMatch = re.compile(r"^\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}.log$")  # 匹配日志文件名

    # 修改日志文件名 [ log.2021-10-06_23-22-35.log --> 2021-10-06_23-22-35.log ]
    def namer(filename):
        return filename.replace(loggerName + ".", "")

    fh.namer = namer

    # 再创建一个handler，用于输出到控制台
    ch = logging.StreamHandler()

    # 单独设置日志等级
    # fh.setLevel(logging.DEBUG)  # 输出到file的log等级的开关
    # ch.setLevel(logging.WARNING)  # 输出到console的log等级的开关

    # 定义输出格式
    datefmt = '%Y-%m-%d %H:%M:%S'
    format = '%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s \n'
    # format = '%(asctime)s - %(pathname)s - [line:%(lineno)d] - %(levelname)s > %(message)s \n'
    format_str = logging.Formatter(format, datefmt)

    # 为文件、控制台操作符绑定格式
    fh.setFormatter(format_str)
    ch.setFormatter(format_str)

    # 创建一个logging对象
    log = logging.getLogger()
    log.setLevel(LEVELS.get(level))  # 定义日志输出层级

    # 给log添加handler
    log.addHandler(fh)
    log.addHandler(ch)

    return log


if __name__ == '__main__':
    logger = my_logger()
    logger.debug('Debug状态')
    logger.info('输入状态')
    logger.warning('警告级别错误')
    logger.error('产生错误信息')
    logger.critical('产生严重错误')

效果演示

带线程锁的切割日志函数

说明

  • 虽然 TimedRotatingFileHandler 函数有线程安全的功能，但是在多线程下会出错，查看了网上了多进程的方法，没有弄出来！！！
  • 于是改变了思路，使用线程锁（多进程单例）！！！

代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
import os
import re
import threading

import logging
import time

from logging.handlers import TimedRotatingFileHandler


# 创建日志文件夹
def create_folder(folder_name):
    # 获取当前文件夹
    current_path = os.path.dirname(__file__)

    # 获取当前文件夹的上一层文件
    base_path = os.path.abspath(os.path.join(current_path, ".."))

    # 拼接路径
    log_dir = base_path + os.sep + folder_name + os.sep
    # print(f'日志文件夹名: {folder_name} -- {log_dir} \n')

    """如果文件夹不存在就创建"""
    os.makedirs(log_dir.strip(), exist_ok=True)

    return log_dir


# 继承 TimedRotatingFileHandler
class MyTimedRotatingFileHandler(TimedRotatingFileHandler):
    """重写 getFilesToDelete 方法"""

    def getFilesToDelete(self):
        dirName = os.path.abspath(os.path.join(self.baseFilename, ".."))
        # 获取文件下的所有的文件(不包含 临时日志文件)
        fileNames = os.listdir(dirName)
        result = []
        for fileName in fileNames:
            result.append(os.path.join(dirName, fileName))
        if len(result) < self.backupCount:
            result = []
        else:
            result.sort()
            result = result[:len(result) - self.backupCount]
        return result


# log日志
class MyLogger:
    # 使用线程锁保证单例模式线程安全
    _instance_lock = threading.Lock()

    def start(self, *args, **kwargs):
        # 接收参数
        self.folderName = (kwargs["folderName"] if kwargs.get("folderName") else "logs")  # 日志文件夹名
        self.logname = (kwargs["tempName"] if kwargs.get("tempName") else "log")  # 临时日志文件名
        self.level = (kwargs["level"] if kwargs.get("level") else "debug")  # 日志级别
        self.datefmt = (kwargs["datefmt"] if kwargs.get("datefmt") else None)
        self.fmt = (kwargs["format"] if kwargs.get("format")
                    else "%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s")  # 格式化
        self.console = (kwargs["console"] if kwargs.get("console") else True)  # 是否输出到控制台
        self.file = (kwargs["file"] if kwargs.get("file")
                     else (create_folder(self.folderName) + self.logname))  # 保存的文件名
        self.when = (kwargs["when"] if kwargs.get("when") else "D")  # 日志按时间切分(默认天)
        self.backCount = (kwargs["backCount"] if kwargs.get("backCount") else 30)  # 备份文件最大个数,超过删除

        # 设置loggin参数
        self.level_relations = {  # 日志级别
            'debug': logging.DEBUG,
            'info': logging.INFO,
            'warning': logging.WARNING,
            'error': logging.ERROR,
            'critical': logging.CRITICAL
        }
        self.logger = logging.getLogger(__name__)
        self.format = logging.Formatter(self.fmt, self.datefmt)

        if not self.level_relations.get(self.level):
            self.level = "debug"
        self.logger.setLevel(self.level_relations[self.level])
        if self.console:
            streamHandler = logging.StreamHandler()
            streamHandler.setFormatter(self.format)
            self.logger.addHandler(streamHandler)
        if self.file:  # log存档,按时间切割存档
            timeHandler = MyTimedRotatingFileHandler(
                filename=self.file,
                when=self.when,
                backupCount=self.backCount,
                encoding="utf-8"
            )
            timeHandler.suffix = "%Y-%m-%d_%H-%M-%S.log"  # 日志文件的后缀名
            timeHandler.extMatch = re.compile(r"^\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}.log$")  # 匹配日志文件名

            # 修改日志文件名 [ log.2021-10-06_23-22-35.log --> 2021-10-06_23-22-35.log ]
            def namer(filename):
                return filename.replace(self.logname + ".", "")

            timeHandler.namer = namer
            timeHandler.setFormatter(self.format)
            self.logger.addHandler(timeHandler)

    # 单例模式实现
    def __new__(cls, *args, **kwargs):
        # 单例模式，单个进程内永远只有一个MyLogger对象
        # 实例化先走__new__再走__init__
        if not hasattr(MyLogger, "_instance"):  # 先检查该类有没有 _instance属性
            with MyLogger._instance_lock:  # 线程安全所 同一时间只有一个线程访问这里
                if not hasattr(MyLogger, "_instance"):  # 防止等待过程已经生成新对象
                    MyLogger._instance = object.__new__(cls)  # 走__new__实例化对象并赋值给_instance
                    MyLogger._instance.start(*args, **kwargs)  # 手动调用start

                time.sleep(1)  # 防止刚实例化就写入有可能导致数据丢失,等待一下
                return MyLogger._instance.logger


if __name__ == '__main__':
    logger = MyLogger(
        folderName="logs",  # 日志文件夹名
        tempName="log",  # 日志文件夹名
        # level="debug",  # 日志等级 (默认值为 "debug")
        datefmt='%Y-%m-%d %H:%M:%S',  # 日志时间格式 (默认值带毫秒)
        # fmt="%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s",  # 日志格式
        # console=True,  # 将日志输出到控制台
        when="S",  # 按 时间段 切分日志 (默认值为 "D")
        backCount=3  # 日志保留的最大个数, 超出将删除最早的日志
    )
    logger.debug('Debug状态')
    logger.info('输入状态')
    logger.warning('警告级别错误')
    logger.error('产生错误信息')
    logger.critical('产生严重错误')

效果演示

推荐使用的日志模块


参考网站：【鸣谢】

TimedRotatingFileHandler 函数用法

Python logger 模块

python 模块 多线程 单例_python Logger 模块单例模式

最方便最好看最好用的 python 日志