-
Notifications
You must be signed in to change notification settings - Fork 0
/
logger_util.py
84 lines (67 loc) · 2.62 KB
/
logger_util.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
import logging
import logging.handlers
import multiprocessing
import os
import sys
from pathlib import Path
class MultiProcesses_Logger_Util:
def __init__(self, file_name: str) -> None:
self.queue = multiprocessing.Manager().Queue(-1)
self.file_name = file_name
self.process = multiprocessing.get_context('fork').Process(
target=self._process_target,
)
self.process.start()
self.redirect_main_process_log_to_queue()
def get_log_file_path(self) -> str:
log_file_dir = os.path.join(Path(__file__).parent, '..', 'logs')
Path(log_file_dir).mkdir(parents=True, exist_ok=True)
log_file_path = os.path.join(log_file_dir, f'{self.file_name}.log')
return log_file_path
def redirect_main_process_log_to_queue(self):
main_process_id = os.getpid()
def filter(record: logging.LogRecord):
if record.process == main_process_id:
self.queue.put_nowait(record)
return None
root = logging.getLogger()
root.setLevel(logging.INFO)
handler = logging.Handler()
handler.addFilter(filter)
root.addHandler(handler)
def init_logger_configure(self):
root = logging.getLogger()
root.setLevel(logging.INFO)
file_handler = logging.handlers.RotatingFileHandler(
self.get_log_file_path(),
mode='w',
# maxBytes=5000,
# backupCount=0,
encoding='utf-8',
)
file_handler.setLevel(logging.INFO)
file_handler_formatter = logging.Formatter(
fmt='%(asctime)s | %(name)s | %(levelname)s | %(filename)s:%(lineno)d | %(process)d | %(message)s',
datefmt="%m-%d %H:%M:%S"
)
file_handler.setFormatter(file_handler_formatter)
root.addHandler(file_handler)
std_handler = logging.StreamHandler(sys.stdout)
std_handler.setLevel(logging.DEBUG)
std_handler_formatter = logging.Formatter(
fmt='%(asctime)s | %(name)s | %(levelname)s | %(filename)s:%(lineno)d | %(process)d | %(message)s',
datefmt="%m-%d %H:%M:%S"
)
std_handler.setFormatter(std_handler_formatter)
root.addHandler(std_handler)
def _process_target(self):
self.init_logger_configure()
while True:
record: logging.LogRecord = self.queue.get()
if record is None:
break
logger = logging.getLogger(record.name)
logger.handle(record)
def close(self):
self.process.join(timeout=0)
self.process.terminate()