程序员经验分享-hwhale

Python-logging详解（彩色日志扩展，多进程安全等）

2023-10-26

目录

简介

日志级别

记录器

处理器

格式器

多线程与多进程安全

代码

导入及全局变量

函数

实验及结果

参考

简介

日志是工程中不可缺少的一部分，国家等保2.0也规定，至少保留日志180天。对于程序员来说，日志也方便进行记录及排错。

logging是Python自带的日志模块，主要有以下几个部分：

  • 记录器暴露了应用程序代码直接使用的接口。

  • 处理器将日志记录（由记录器创建）发送到适当的目标。

  • 过滤器提供了更细粒度的功能，用于确定要输出的日志记录。

  • 格式器指定最终输出中日志记录的样式。

日志级别

日志级别

级别

数值

CRITICAL

50

ERROR

40

WARNING

30

INFO

20

DEBUG

10

NOTSET

0

import logging之后可以从里面找到，用于区分日志记录的级别，通过级别可以确定是否输出。

记录器

setLevel(level)

给记录器设置阈值为 level 。日志等级小于 level 会被忽略。严重性为 level 或更高的日志消息将由该记录器的任何一个或多个处理器发出，除非将处理器的级别设置为比 level 更高的级别。

创建记录器时，级别默认设置为 NOTSET。

addHandler(hdlr)

将指定的处理器 hdlr 添加到此记录器。

addFilter(filter)

将指定的过滤器 filter 添加到此记录器。

debug(msg, *args, **kwargs)

在此记录器上记录 DEBUG 级别的消息。 

info(msg, *args, **kwargs)

在此记录器上记录 INFO 级别的消息。

warning(msg, *args, **kwargs)

在此记录器上记录 WARNING 级别的消息。

error(msg, *args, **kwargs)

在此记录器上记录 ERROR 级别的消息。

critical(msg, *args, **kwargs)

在此记录器上记录 CRITICAL 级别的消息。

log(level, msg, *args, **kwargs)

在此记录器上记录 level 整数代表的级别的消息。

处理器

StreamHandler

它可将日志记录输出发送到数据流例如 sys.stdout, sys.stderr 或任何文件类对象（或者更精确地说，任何支持 write() 和 flush() 方法的对象）。

FileHandler

它可将日志记录输出到磁盘文件中。

格式器

Formatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)

属性名称

格式

描述

args

此属性不需要用户进行格式化。

合并到 msg 以产生 message 的包含参数的元组，或是其中的值将被用于合并的字典（当只有一个参数且其类型为字典时）。

asctime

%(asctime)s

被创建的供人查看时间值。 默认形式为 '2003-07-08 16:49:45,896' （逗号之后的数字为时间的毫秒部分）。

created

%(created)f

被创建的时间。

exc_info

此属性不需要用户进行格式化。

异常元组（例如 sys.exc_info）或者如未发生异常则为 None。

filename

%(filename)s

pathname 的文件名部分。

funcName

%(funcName)s

函数名包括调用日志记录.

levelname

%(levelname)s

消息文本记录级别（'DEBUG'，'INFO'，'WARNING'，'ERROR'，'CRITICAL'）。

levelno

%(levelno)s

消息数字的记录级别 (DEBUG, INFO, WARNING, ERROR, CRITICAL).

lineno

%(lineno)d

发出日志记录调用所在的源行号（如果可用）。

message

%(message)s

记入日志的消息，即 msg % args 的结果。

module -- 模块

%(module)s

模块 (filename 的名称部分)。

msecs

%(msecs)d

被创建的时间的毫秒部分。

msg

此属性不需要用户进行格式化。

在原始日志记录调用中传入的格式字符串。 与 args 合并以产生 message，或是一个任意对象。

name

%(name)s

用于记录调用的日志记录器名称。

pathname

%(pathname)s

发出日志记录调用的源文件的完整路径名（如果可用）。

process

%(process)d

进程ID（如果可用）

processName

%(processName)s

进程名（如果可用）

relativeCreated

%(relativeCreated)d

以毫秒数表示的 LogRecord 被创建的时间，即相对于 logging 模块被加载时间的差值。

stack_info

此属性不需要用户进行格式化。

当前线程中从堆栈底部起向上直到包括日志记录调用并引发创建当前记录堆栈帧创建的堆栈帧信息（如果可用）。

thread

%(thread)d

线程ID（如果可用）

threadName

%(threadName)s

线程名（如果可用）

一般常用的就是时间，级别，内容，可以在团队内自行规定msg的格式作为补充。

多线程与多进程安全

logging是多线程安全的，所以添加多进程安全即可

代码

"""
-*- coding:utf-8 -*-
@File: log.py
@Author:frank yu
@DateTime: 2021.10.12 20:56
@Contact: frankyu112058@gmail.com
@Description:
"""
import logging
import os
import random
import sys
import multiprocessing
import time
from shutil import copyfile
from colorlog import ColoredFormatter

# log directory
LOG_DIR = "./logs"
# level fot stdout
STD_LEVEL = logging.WARN
# level for file
FILE_LEVEL = logging.DEBUG

CRITICAL = 50
FATAL = CRITICAL
ERROR = 40
WARNING = 30
WARN = WARNING
INFO = 20
DEBUG = 10
NOTSET = 0


class Log:
    def __init__(self, log_dir=LOG_DIR):
        """
        function:init create log file, judge if log file is too bigger
        :param log_dir: directory of logs
        """
        log_path = log_dir + "/log"
        if not os.path.exists(log_dir):
            os.mkdir(log_dir)
            os.chmod(log_dir, 0o777)
        if not os.path.exists(log_path):
            f = open(log_path, mode='w', encoding='utf-8')
            f.close()
            os.chmod(log_path, 0o777)
        # if log file is more than 1MB, copy to a file and clear log file
        if os.path.getsize(log_path) / 1048576 > 1:
            # print(os.path.getsize(log_path))
            copyfile(log_path, log_dir + "/log" + str(time.time()).replace(".", ""))
            with open(log_path, 'w') as f:
                f.truncate()
                f.close()
        self.logger_format = logging.Formatter('%(asctime)s - %(levelname)s: %(message)s')
        self.c_logger_format = ColoredFormatter(fmt='%(log_color)s%(asctime)s'
                                                    ' - %(log_color)s%(levelname)s: %(log_color)s%(message)s',
                                                reset=True,
                                                secondary_log_colors={},
                                                style='%'
                                                )
        self.logger = logging.getLogger(str(random.random()))
        self.logger.handlers.clear()
        self.logger.setLevel(logging.DEBUG)

        self.filehandler = logging.FileHandler(log_path, mode='a')
        self.filehandler.setLevel(FILE_LEVEL)
        self.filehandler.setFormatter(self.logger_format)

        self.stdouthandler = logging.StreamHandler(sys.stdout)
        self.stdouthandler.setLevel(STD_LEVEL)
        self.stdouthandler.setFormatter(self.c_logger_format)

        self.logger.addHandler(self.stdouthandler)
        self.logger.addHandler(self.filehandler)

        self.__lock = multiprocessing.Lock()

    def __log(self, msg=None, level=logging.INFO):
        if level == logging.DEBUG:
            self.logger.debug(msg)
        elif level == logging.INFO:
            self.logger.info(msg)
        elif level == logging.WARNING:
            self.logger.warning(msg)
        elif level == logging.ERROR:
            self.logger.error(msg)
        elif level == logging.CRITICAL:
            self.logger.critical(msg)

    def log_show_store(self, msg, level):
        """
        function: show logs on screen and store logs into log file
        :param msg:message of log
        :param level:level of log
        return: None
        """
        self.__lock.acquire()
        self.logger.addHandler(self.stdouthandler)
        self.logger.addHandler(self.filehandler)
        self.__log(msg=msg, level=level)
        self.__lock.release()

    def log_show(self, msg, level):
        """
        function:show msg on the screen
        :param msg:message of log
        :param level: level of log
        """
        self.__lock.acquire()
        self.logger.removeHandler(self.filehandler)
        self.logger.addHandler(self.stdouthandler)
        self.__log(msg, level=level)
        self.logger.removeHandler(self.stdouthandler)
        self.logger.addHandler(self.filehandler)
        self.__lock.release()

    def log_store(self, msg, level):
        """
        function:store msg into log file
        :param msg:message of log
        :param level: level of log
        """
        self.__lock.acquire()
        self.logger.removeHandler(self.stdouthandler)
        self.logger.addHandler(self.filehandler)
        self.__log(msg, level=level)
        self.logger.removeHandler(self.filehandler)
        self.logger.addHandler(self.stdouthandler)
        self.__lock.release()

    def log_wrapper(self, *args):
        """
        function: record arg and result of functions
        :param args:
        :return: wrapper
        """
        self.log_store(' '.join(args), FILE_LEVEL)

        def wrapper(func):
            def inner(*args, **kwargs):
                self.log_store(getattr(func, "__name__") + " call " + str(args) + str(kwargs), 10)
                ret = func(*args, **kwargs)
                self.log_store(getattr(func, "__name__") + " return " + str(ret), 10)
                return ret

            return inner

        return wrapper

导入及全局变量

colorlog需要安装一下，可以控制台输出彩色的

multiprocessing用于进程安全，加锁

设置了几个全局变量，方便修改

LOG_DIR是日志保存的目录

STD_LEVEL是控制台输出的级别，只输出WARN及以上的

FILE_LEVEL是保存日志文件的级别，一般调试时使用DEBUG，生产环境时使用INFO

函数

__init__

  1. 创建目录和日志文件，如果不存在。若大于1M，复制为当前日期时间的日志文件
  2. 设置了控制台和文件的格式器，控制台是彩色的
  3. 设置了控制台和文件的处理器

log_show_store

保存到日志文件并打印到控制台

log_show

仅打印到控制台

log_store

仅保存到日志文件

log_wrapper

日志装饰器，用于添加到重要函数上，记录函数调用参数及返回值

实验及结果

"""
-*- coding:utf-8 -*-
@File: main.py
@Author:frank yu
@DateTime: 2021.10.12 20:56
@Contact: frankyu112058@gmail.com
@Description:
"""
from utils.log import Log, DEBUG, WARN, INFO
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

log = Log()


@log.log_wrapper("wrapper for test")
def test(a, b, c=5):
    log.log_show('main test 1 ' + str(a), level=DEBUG)
    log.log_show('main test 2 ' + str(b), level=WARN)
    log.log_store('main test 3 ' + str(a), level=DEBUG)
    log.log_store('main test 4 ' + str(b), level=WARN)
    log.log_show_store('main test 5 ' + str(a + b), level=INFO)
    return a + b + c


def even(num):
    log.log_store('main even 1 ' + str(num), level=INFO)


def odd(num):
    log.log_store('main odd 1 ' + str(num), level=INFO)


# multi thread test
def multi_thread():
    exe = ThreadPoolExecutor(4)
    for i in range(10):
        if i & 1 == 0:
            exe.submit(even, i)
        else:
            exe.submit(odd, i)


# multi process test
def multi_process():
    exe = ProcessPoolExecutor(4)
    for i in range(10):
        if i & 1 == 0:
            exe.submit(even, i)
        else:
            exe.submit(odd, i)


if __name__ == '__main__':
    test(3, 4, 6)
    multi_thread()
    multi_process()

msg这里传的是 模块 函数 函数行数 消息 ，程序出错方便定位。

test测试了日志级别和日志装饰器

multi_thread测试了多线程

multi_process测试了多进程

结果如下：

测试结果

可以看到控制台仅输出了WARNING级别，当然，如果有更高级别也会输出。

可以看到test函数的调用及参数和返回结果

可以看到多线程和多进程状态下结果没有丢失

更新

用了段时间，彩色的花里胡哨，很少用到，formatter有点少，而且封装后不能定位打印日志的文件，还是直接使用logger吧。

log.py

"""
-*- coding:utf-8 -*-
@File: log.py
@Author:frank yu
@DateTime: 2023.08.17 15:07
@Contact: frankyu112058@gmail.com
@Description:
日志封装
1. 可以多线程、多进程使用
2. 默认只有控制台handler，可以通过set_std、set_file、set_std_file设置，尤其是想保存到文件时
3. 其他地方使用只import log对象即可
"""
import functools
import logging
import os
import random
import sys
import multiprocessing
import time
from shutil import copyfile

# log directory
LOG_DIR = "./logs"
# level for stdout
STD_LEVEL = logging.DEBUG
# level for file
FILE_LEVEL = logging.DEBUG


class Log:
    def __init__(self, log_dir=LOG_DIR):
        """
        function:init create log file, judge if log file is too bigger
        :param log_dir: directory of logs
        """
        log_path = log_dir + "/log"
        if not os.path.exists(log_dir):
            os.mkdir(log_dir)
            os.chmod(log_dir, 0o777)
        if not os.path.exists(log_path):
            f = open(log_path, mode='w', encoding='utf-8')
            f.close()
            os.chmod(log_path, 0o777)
        # if log file is more than 1MB, copy to a file and clear log file
        if os.path.getsize(log_path) / 1048576 > 1:
            # print(os.path.getsize(log_path))
            copyfile(log_path, log_dir + "/log" + str(time.time()).replace(".", ""))
            with open(log_path, 'w') as f:
                f.truncate()
                f.close()
        self.logger_format = logging.Formatter('[%(asctime)s %(filename)s:%(funcName)s:%(lineno)d %(levelname)-8s] %('
                                               'message)s')

        self.logger = logging.getLogger(str(random.random()))
        self.logger.handlers.clear()
        self.logger.setLevel(logging.DEBUG)

        self.filehandler = logging.FileHandler(log_path, mode='a')
        self.filehandler.setLevel(FILE_LEVEL)
        self.filehandler.setFormatter(self.logger_format)

        self.stdouthandler = logging.StreamHandler(sys.stdout)
        self.stdouthandler.setLevel(STD_LEVEL)
        self.stdouthandler.setFormatter(self.logger_format)

        self.logger.addHandler(self.stdouthandler)

        self.__lock = multiprocessing.Lock()

    def set_std_file(self):
        """
        function: set std and file handler
        return: None
        """
        self.__lock.acquire()
        self.logger.addHandler(self.stdouthandler)
        self.logger.addHandler(self.filehandler)
        self.__lock.release()

    def set_std(self):
        """
        function: set std handler
        """
        self.__lock.acquire()
        self.logger.removeHandler(self.filehandler)
        self.logger.addHandler(self.stdouthandler)
        self.__lock.release()

    def set_file(self):
        """
        function: set file handler
        """
        self.__lock.acquire()
        self.logger.removeHandler(self.stdouthandler)
        self.logger.addHandler(self.filehandler)
        self.__lock.release()

    def log_wrapper(self, func):
        """
        function: record arg and result of functions
        :param func: 自动获取名字，不需要填写
        :return: wrapper
        """

        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            self.logger.debug(getattr(func, "__name__") + " call " + str(args) + str(kwargs))
            ret = func(*args, **kwargs)
            self.logger.debug(getattr(func, "__name__") + " return " + str(ret))
            return ret

        return wrapper


log = Log()

参考

logging-Python 的日志记录工具

更多python相关内容：【python总结】python学习框架梳理

有问题请下方评论，转载请注明出处，并附有原文链接，谢谢！如有侵权，请及时联系。如果您感觉有所收获，自愿打赏，可选择支付宝18833895206（小于），您的支持是我不断更新的动力。

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

python

日志

Python-logging详解（彩色日志扩展，多进程安全等） 的相关文章

随机推荐

  • UA到底是什么

    欢迎关注勤于奋 每天12点准时更新国外LEAD相关技术 今天这篇文章也是比较基础的文章 其实有很多新手还是搞不懂 所以我特意来写篇文章聊聊这个UA 到底是什么 在我们国外LEAD当中起到的作用 如果搞不懂 很多东西你都没有办法灵活运用 首先

  • 19款资源整合类网站推荐：每一个网站都堪称以一敌百

    强烈推荐这19个资源聚合网站 每一个网站都足以堪称 以一敌百 因为每一个网站都聚合了相当多不同类型 不同领域的网站资源 并且做了分类导航方便大家查找使用 所以 与其收藏那么多零碎的网址 不如收藏下面这些网站资源 比格张 发现更好的资源 一个

  • java8的日期工具类（获取当前时间 相隔天数 小时 分钟 秒等处理）

    package com example list test import java text ParseException import java text SimpleDateFormat import java time import

  • tvm 入门（二）

    代码是一个tvm入门的例子 以向量相加为例 使用tvm的流程是 1 描述串行的向量相加是怎么做的 2 描述并行的时候 怎么对计算单元做划分 3 编译目标函数 本文所示代码可以看到用tvm生成的cuda代码 4 把编译生成的内容保存成文件 加

  • VMware Workstation 未能启动 VMware Authorization Service。您可以尝试手动启动 VMware Authorization Service。如果此问题仍然存

    报错界面 解决方法一 管理员身份运行即可 解决方法二 打开服务 找到VMware Authorization Service右键选择然后点启动 然后发现又报了下面的错误 不慌这时候打开属性把启动类型改成手动即可

  • String、StringBuffer和StringBuilder的异同点

    String StringBuffer StringBuilder三者的异同 String 不可变的字符序列 底层使用char 存储 StringBuffer 可变的字符序列 线程安全的 效率低 底层使用char 存储 StringBuil

  • OpenAI的人工智能语音识别模型Whisper详解及使用

    1 whisper介绍 拥有ChatGPT语言模型的OpenAI公司 开源了 Whisper 自动语音识别系统 OpenAI 强调 Whisper 的语音识别能力已达到人类水准 Whisper是一个通用的语音识别模型 它使用了大量的多语言和

  • 安装mingw出现download failed和unable to continue

    利用mingw get setup安装mingw总是出现download failed和unable to continue错误 截图如下 错误原因 因为服务器在外网 可能是对方服务器不稳定 连接出错等问题导致 解决办法 下载对应等离线安装

  • 11信号学习之sigaction函数及使用其实现信号捕捉案例(信号最重要的一节)

    概述 注意 在我关于信号的文章中 我所说的系统的mask的意思实际上是进程的mask 每个进程的mask都是唯一的 所以我就将其称为系统的 但不能理解为每个进程的mask都是共用的 1 sigaction函数 1 上一篇我们说的signal

  • 基于STM32的DMX512开发

    首先基本了解一下DMX512的基本协议 一 DMX512协议 DMX 是Digital MultipleX 的缩写 意为多路数字传输 DMX512控制协议是美国舞台灯光协会 usITT 于1990年发布的灯光控制器与灯具设备进行数据传输的工

  • 复习使用git（二）

    删除远程分支 git push origin delete 分支名 撤销修改 撤销工作区的修改 已修改 但尚未添加 add 使用 git restore 文件名 撤销工作区的修改 Note git checkout 文件名 checkout

  • 基于STM32F407的SDCard读写操作及USB挂载（HAL库）

    基于STM32F407的SDCard读写操作及USB挂载 HAL库 本来在上一篇SD卡读写也都OK了 后来想着挂载SD卡做U盘 就去查了下资料 结果基本全是HAL库的 原来没用过HAL库 于是本着好奇的心态去下了 说实话 确实看起来简单多了

  • 【Spring Boot】——集成JSON工具

    前言 json是现在非常流程的数据交换格式 所以对于被开发人员来说如何更好了解java对象和json格式之间的转换是至关重要的 接下来我们来好好说一说 一 什么是JSON 摘自百度百科 JSON JavaScript Object Nota

  • android 涨潮动画加载_潮汐apk客户端-潮汐android最新版APP下载v2.0.1.1 免费版-腾牛安卓网...

    潮汐一个非常有特色的静心平台 它可以让你在工作困乏时聆听几分钟就可以更加高效的工作 让你的工作效率达到更好的专注 它可以让你的心情更加温和 让你的每一天不再那么浮躁和枯燥无味 让你的心情越来越舒畅 喜欢的朋友快来腾牛网下载吧 软件特色 保持

  • Linux实验报告【全集】

    若对你有帮助 记得点赞 关注我哦 实验目录 Linux常用命令 linux下的shell编程 Linux下的c编程 Linux下的API编程 每个实验的图片都比较多 一开始实验基本都是书上的例题 后面会变难 做实验时往往会想 为什么每届都做

  • 开发环境配置：服务器训练模型工具tmux基础使用

    开发环境配置 服务器训练模型工具tmux基础使用 1 tumx可以做什么 2 Ubuntu安装tmux 3 常用命令 3 1 新建会话并进入 3 2 退出会话 3 3 查看存在的会话 3 4 重新进入会话 3 5 销毁会话 1 tumx可以

  • C#检测是否在单元测试环境

    true则为单元测试环境下 false不是 bool unitTest System Environment StackTrace IndexOf NUnit Framework StringComparison CurrentCultur

  • ShareSDK集成第三方登录和分享的步骤

    转自 http my oschina net u 1024921 blog 170588 之前用过这个几次了 而每次都没有记录一下具体的步骤 这次就写一下吧 1 去ShareSDK下载官方的SDK 2 现在他们的服务特别人性化 解压SDK之

  • SiamFC：利用全卷积孪生网络进行视频跟踪

    目录 论文下载地址 代码下载地址 论文作者 模型讲解 模型结构 模型输入 损失函数 训练过程 结果分析 传送门 论文下载地址 SiamFC论文地址 SiamFC论文百度网盘下载地址 提取码 7309 SiamFC论文翻译 水印 百度网盘下载

  • Python-logging详解（彩色日志扩展，多进程安全等）

    目录 简介 日志级别 记录器 处理器 格式器 多线程与多进程安全 代码 导入及全局变量 函数 实验及结果 参考 简介 日志是工程中不可缺少的一部分 国家等保2 0也规定 至少保留日志180天 对于程序员来说 日志也方便进行记录及排错 log

热门标签