自分用に、バイナリファイルやテキストファイルを簡単に記録するロガークラスを作ったのでメモしておく。
Loggerクラスを別のPythonファイルにimportしてあげて、main関数の例の形で利用するだけ。
Loggerのlog_nameに指定した文字列がログファイルに使われるようにしてある。
後は用途に合わせてカスタマイズすること。
コード
from concurrent.futures.thread import _worker
import threading
import queue
from time import sleep
def main():
# test & sample code
# ロガーオブジェクトを作成して、スレッドを開始する
logger = Logger(log_name="sample")
logger.start()
for i in range(0, 100):
print("add " + str(i))
logger.add(str(i)) # ログ文字列を追加
sleep(1)
class Logger():
def __init__(self, log_name):
self.q = queue.Queue()
self.log_name = log_name
self.isRunning = False
# ロガーの起動
def start(self):
self.isRunning = True
th = threading.Thread(target=self._worker)
th.start()
# ロガーの終了
def end(self):
self.isRunning = False
# ログファイル書き出しワーカスレッドの関数
def _worker(self):
# fw = open("./work/" + self.log_name + ".dat", 'wb') # バイナリファイルの場合
fw = open("./work/" + self.log_name + ".txt", 'w', encoding='utf-8') # テキストファイルの場合
def write_process():
item = self.q.get()
fw.write(item) # 改行したいときはfw.writelines()
fw.flush()
self.q.task_done()
print("start worker")
while self.isRunning:
write_process()
while not self.q.empty(): # end呼び出し後にキューに残っている情報の書き出しを行う
write_process()
print("end worker")
# ログ追加
def add(self, msg):
if not self.isRunning:
return
self.q.put(msg)
if __name__ == "__main__":
main()