Python 有着内置的日志输出模块:logging 使用也很方便,但我们今天不说这个,我们用文件读写模块,实现自己的日志输出模块;这样在项目中,可以存在更高的自由度及更高的扩展性;
先来看看日志输出的文件效果:根目录 \Logs\ 按日期输出 \2024-04-28\ 按日志类型输出txt文件
日志类型可自定义
首先定义 Log基类 :LogRecord
class LogRecord:#初始化日志模块def __init__(self):self._total_log_dic = {} #字典项用来保存 日志类型:日志内容self._log_path = os.path.join(os.getcwd(), "LogRecord") #日志输出路径self._locker = threading.Lock() #线程锁self.create_auto_save_thread() #创建自定运行的线程#创建自定运行的线程def create_auto_save_thread(self):def auto_save():while True:self.write_in_txt_once() #将缓存字典中的日志信息都输出到文件中threading.Event().wait(3)#定义线程运行auto_save方法thread = threading.Thread(target=auto_save)thread.daemon = Truethread.start()#设置日志输出根目录,保存的日志文件目录数-多出的会被删除def set_log_path(self, path, remain_count=7):self._log_path = pathself._remain_count = remain_count#添加日志信息到缓存字典中def record(self, logt, content):if logt not in self._total_log_dic:self._total_log_dic[logt] = deque()self._total_log_dic[logt].append(f"\r\n【{datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')}】:{content}")#将缓存字典中的数据,输出到日志文件中def write_in_txt_once(self):with self._locker:for item in self._total_log_dic:if len(self._total_log_dic[item]) == 0:continuequeue = self._total_log_dic[item]v_list = []while len(queue) > 0:v_list.append(queue.popleft())if len(v_list) > 0:self.write_in_txt(v_list, item)#写文件的操作方法def write_in_txt(self, logs, logs_name):if not hasattr(self, "is_deleted") or not self.is_deleted:try:self.delete_director(self._log_path, self._remain_count)except:passself.is_deleted = Truetry:directory_name = datetime.now().strftime("%Y-%m-%d")log_path = os.path.join(self._log_path, directory_name)if not os.path.exists(log_path):os.makedirs(log_path)single_path = os.path.join(log_path, f"{logs_name}.txt")with open(single_path, "a", encoding="utf-8") as f:for line in logs:f.write(line + "")except:pass#删除保存天数外的目录文件def delete_director(self, path, remain_count):if not os.path.exists(path):returndirs = [d for d in os.listdir(path) if os.path.isdir(os.path.join(path, d))]if len(dirs) > remain_count:for i in range(len(dirs) - remain_count):shutil.rmtree(os.path.join(path, dirs[i]))#强制触发将缓存字典中的数据写入到日志文件中def save(self):self.write_in_txt_once()
接下来对LogRecord基类进行一次的封装:LogHelper
#对LogRecord基类进行一次的封装
class LogHelper:def __init__(self):self.log_helper = LogRecord()#初始化日志模块,输出根目录,保存天数等def set_path(self, path, remain_count=7):self.log_helper.set_log_path(path, remain_count)#添加日志记录到缓存字典def record(self, logt, content):self.log_helper.record(logt, content)#将缓存字典中的日志信息都输出到文件中def save(self):self.log_helper.save()
定义 日志类型:LogType
#日志类型-也对应相应的文件名
class LogType:Send = "Send"Info = "Info"Error = "Error"Click = "Click"Start = "Start"Web = "Web"Debug = "Debug"
接下来就对在代码中使用的输出模块进行功能定义: LogOperateBase
import osclass LogOperateBase:cur_process_id = -1_default_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "Logs")_log_helper = LogHelper()@staticmethoddef init_log(id, path="", remain_count=7):LogOperateBase.cur_process_id = idif not path:path = LogOperateBase._default_pathLogOperateBase._log_helper.set_path(path, remain_count)@staticmethoddef save():LogOperateBase._log_helper.save()
最后定义日志输出类,即代码中实际使用的对象:LogOperate
import traceback
from Modules.Tools.Logs import LogHelper,LogType
from Modules.Tools.LogOperateBase import LogOperateBaseclass LogOperate(LogOperateBase):@staticmethoddef debug(content):if __debug__:LogOperateBase._log_helper.record(LogType.Debug, content)@staticmethoddef info(message):LogOperateBase._log_helper.record(LogType.Info, message)@staticmethoddef web(message):LogOperateBase._log_helper.record(LogType.Web, message)@staticmethoddef start(message):LogOperateBase._log_helper.record(logt=LogType.Start,content= f"【{LogOperateBase.cur_process_id}】{message}")@staticmethoddef error_msg(message):LogOperateBase._log_helper.record(LogType.Error, message)@staticmethoddef error(message, ex):try:log = f"{message}\r\n{ex.__class__.__name__}{str(ex)}\r\n{traceback.format_exc()}\r\n-----------------------------------------------------------------------"LogOperateBase._log_helper.record(LogType.Error, log)except:pass@staticmethoddef send_log(message):LogOperateBase._log_helper.record(LogType.Send, message)@staticmethoddef click_log(message):LogOperateBase._log_helper.record(LogType.Click, message)@staticmethoddef general(log_type, message):LogOperateBase._log_helper.record(log_type, message)
最后就能在代码中这样使用了:
LogOperate.init_log(os.getpid(),Config.log_path)LogOperate.start("开始启动程序")LogOperate.info("开始启动程序")LogOperate.web("开始启动程序")LogOperate.click_log("开始启动程序")LogOperate.debug("开始启动程序")LogOperate.error_msg("开始启动程序")LogOperate.send_log("开始启动程序")LogOperate.start("开始启动程序1")LogOperate.info("开始启动程序1")LogOperate.web("开始启动程序1")LogOperate.click_log("开始启动程序1")LogOperate.debug("开始启动程序1")LogOperate.error_msg("开始启动程序1")LogOperate.send_log("开始启动程序1")LogOperate.save()