一个比较好用的python logger脚本
标签搜索

一个比较好用的python logger脚本

irx999
2024-11-05 / 0 评论 / 49 阅读 / 正在检测是否收录...

在需要添加日志的python文件中

#注入
from modules.logger import get_logger
logger = get_logger(__name__,"sql.log")
#日志写入
logger.warning("%s的%s的最大值为None,返回默认的None",\
                            table_name,select_columns)
""" 日志模块 """

def get_logger(name, log_name="wr.log", debug_mode=True, notification=True):
    """
    获取日志对象
    :param name: 日志名称 填入 "__name__"
    :param log_name: 日志文件名称

    usage:

    from modules.logger import get_logger
    logger = get_logger(__name__)
    logger.info("这是一个info 的log信息有%s,%s等信息", param1, param2,)

    """
    # 获取log文件路径
    main__working_dir = getattr(sys, "_MEIPASS", os.getcwd())
    log_dir = os.path.join(main__working_dir, "logs")
    if not os.path.exists(log_dir):
        os.makedirs(log_dir)

    # 创建logger
    logger = logging.getLogger(name)

    # 设置日志格式
    formatter = logging.Formatter(
        "%(asctime)s - %(levelname)s -"
        "[%(filename)s:%(lineno)d] -%(funcName)s  - %(message)s"
    )

    # 创建一个滚动文件处理器,每个日志文件最大大小为5M,保存5个旧日志文件
    rf_handler = RotatingFileHandler(
        os.path.join(log_dir, log_name),
        maxBytes=5 * 1024 * 1024,
        backupCount=5,
        encoding="utf-8",
    )

    rf_handler.setFormatter(formatter)
    logger.addHandler(rf_handler)

    if debug_mode:
        # 如果debug模式打开,则输出到控制台
        print("Debug mode is on, log will be output to console.")
        logger.setLevel(logging.DEBUG)
        console_handler = logging.StreamHandler()
        console_handler.setFormatter(formatter)
        logger.addHandler(console_handler)

    elif notification:
        # 如果错误类型是error,则发送邮件通知管理员
        notificatio_handler = NotificationHandler()
        notificatio_handler.setFormatter(formatter)
        logger.addHandler(notificatio_handler)
    return logger


class NotificationHandler(logging.Handler):
    """信息通知"""
    def emit(self, record):
        if record.levelno >= logging.ERROR:
            error_message = self.format(record)
            dingrobot(error_message)


def dingrobot(
    msg: list | str = "当你看到这个说明你没有填写msg",
    msgtype="text",
    access_token="access_token",
    safe_word=">-<",
) -> bool:
    """
    钉钉机器人推送
    :param access_token: 钉钉机器人的access_token
    :param msg: 推送消息,可以是字符串或列表,列表第一个元素为标题,第二个元素为内容
    :param msgtype: 推送消息类型,可选值为text或markdown
    :return: bool
    """

    requests = importlib.import_module("requests")
    json = importlib.import_module("json")

    match msgtype:
        case "text":
            data = {"text": {"content": f"{safe_word}---{msg}"}, "msgtype": "text"}

        case "markdown":
            data = {
                "markdown": {"text": f"{safe_word}---{msg[1]}", "title": msg[0]},
                "msgtype": "markdown",
            }
    headers = {"Content-Type": "application/json"}
    rebot_url = f"https://oapi.dingtalk.com/robot/send?access_token={access_token}"
    robot = requests.post(
        url=rebot_url, data=json.dumps(data), headers=headers, timeout=5
    )
    if robot.status_code != 200:
        return False
    return True


if __name__ == "__main__":
    LOGGER = get_logger(__name__)
    LOGGER.debug("这是一个ERROR 的log信息有%s,%s等信息", "param1", "param2")
1

评论 (0)

取消