在需要添加日志的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")
评论 (0)