import os import time import logging from typing import Union class ColoredFormatter(logging.Formatter): """ Class to prettify logger and command line output """ MAPPING = { "DEBUG": 37, # white "INFO": 36, # cyan "WARNING": 33, # yellow "ERROR": 31, # red "CRITICAL": 41, # white on red bg } PREFIX = "\033[" SUFFIX = "\033[0m" def __init__(self, pattern): logging.Formatter.__init__(self, pattern) def format(self, record): seq = self.MAPPING.get(record.levelname, 37) # default white record.levelname = "{0}{1}m{2}{3}" \ .format(self.PREFIX, seq, record.levelname, self.SUFFIX) return logging.Formatter.format(self, record) class MainLogger: def __init__(self, name: str): self.main_logger = logging.getLogger(name) def generate_file_path(self, filename: str): folder_path = os.path.dirname(os.path.abspath(os.path.join(__file__ ,"../.."))) folder_path = os.path.join(folder_path, f"logs/{time.strftime('%d-%m-%Y_%H-00')}/") if not os.path.exists(folder_path): os.makedirs(folder_path) file_path = os.path.join(folder_path, filename) return file_path def configure_main_logger(self, filemode: str = "w+", logging_level: int = logging.INFO) -> logging.Logger: file_path = self.generate_file_path("converter.log") file_handler = logging.FileHandler(file_path, mode=filemode) self.main_logger.addHandler(file_handler) logger_format = logging.Formatter(fmt="%(asctime)s - %(levelname)s - %(message)s " "[%(filename)s:%(lineno)d in %(funcName)s]") file_handler.setFormatter(logger_format) self.main_logger.setLevel(logging_level) return self.main_logger class BookLogger(MainLogger): def __init__(self, name: str): """ Method for Logger configuration. Logger will write to file. Parameters ---------- name: str name of the Logger book_id: Union[int, str] id of the book main_logger: Logger main logger of the converter filemode: str mode of opening log file. logging_level: int logging level: 10 - debug, 20 - info, 30 - warning, 40 - error, 50 - critical logging_format: str format of record in log file """ super().__init__(name) self.book_logger = logging.getLogger(name) self.book_logger.propagate = False def configure_book_logger(self, book_id: Union[int, str], filemode: str = "w+", logging_level: int = logging.INFO): file_path = self.generate_file_path(f"{book_id}.log") book_logger_format: str = "%(asctime)s - %(levelname)s - %(message)s" \ " [%(filename)s:%(lineno)d in %(funcName)s]" file_handler = logging.FileHandler(file_path, mode=filemode) self.book_logger.addHandler(file_handler) file_format = logging.Formatter(book_logger_format) file_handler.setFormatter(file_format) stream_handler = logging.StreamHandler() stream_format = ColoredFormatter(book_logger_format) stream_handler.setFormatter(stream_format) self.book_logger.addHandler(stream_handler) self.book_logger.setLevel(logging_level) def log(self, message: str, logging_level: int = 20): """ Method for logging. Parameters ---------- message: str body of the message logging_level: int level of logging """ self.book_logger.log(msg=message, level=logging_level, stacklevel=2) def log_error_to_main_log(self, message: str = ""): """ Method for logging error to main log file. """ if self.main_logger: if not message: message = f"Error in book conversion. Check log file." self.main_logger.error(message) class BookStatusWrapper: """Class sets/updates statuses of Converter on Platform""" def __init__(self, access, logger_object: BookLogger, book_id: int = 0): self.access = access self.logger_object = logger_object self.book_id = book_id def set_status(self, status: str): str_2_status = { "[PROCESS]": self.access.PROCESS, "[GENERATE]": self.access.GENERATE, "[ERROR]": self.access.ERROR } try: if self.access: self.access.update_status(self.book_id, str_2_status[status]) self.logger_object.log(f"Status has been updated to {status}.") except Exception as exc: self.logger_object.log( f"Can't update status of the book {status}.", logging.ERROR) self.logger_object.log_error_to_main_log() raise exc def set_processing(self): self.set_status("[PROCESS]") def set_generating(self): self.set_status("[GENERATE]") def set_error(self): self.set_status("[ERROR]")