import logging
import sys
from datetime import datetime
from typing import Union
from app.common.signal_bus import signalBus
class Logger:
"""
Logger class for logging
"""
def __init__(self):
"""
:param logger_signal: Logger Box signal
"""
# Init logger box signal, logs and logger
# logger box signal is used to output log to logger box
self.logs = ""
self.logger_signal = signalBus.loggerSignal
self.logger = logging.getLogger("KAFFIO_Logger")
formatter = logging.Formatter("%(levelname)8s |%(category)s | %(message)s ")
handler1 = logging.StreamHandler(stream=sys.stdout)
handler1.setFormatter(formatter)
self.logger.setLevel(logging.INFO)
self.logger.addHandler(handler1)
self.text = ['INFO', 'SUCCESS', 'ERROR', 'SKIPPED', 'REPLACED', 'RENAMED', 'REMOVED']
# Status Text: INFO, SUCCESS, ERROR, SKIPPED, REPLACED, RENAMED, REMOVED
self.status = [' INFO', ' SUCCESS', ' ERROR',
' SKIPPED', ' REPLACED', ' RENAMED', ' REMOVED']
# Status Color: Blue, Red, Green, Orange,
self.statusColor = ['#2d8cf0', '#ed3f14', '#f90', '#f90', '#f90', '#f90', '#00c12b']
# Status HTML: status
self.statusHtml = [
f'{status}'
for _color, status in zip(self.statusColor, self.status)]
def __out__(self, category: str, message: str, level: int = 1, raw_print=False) -> None:
"""
Output log
:param message: log message
:param level: log level
:return: None
"""
# If raw_print is True, output log to logger box
if raw_print:
self.logs += message
self.logger_signal.emit(message)
return
while len(logging.root.handlers) > 0:
logging.root.handlers.pop()
# If logger box is not None, output log to logger box
# else output log to console
if self.logger_signal is not None:
message = message.replace('\n', '
').replace(' ', ' ')
adding = (f'''