import os import logging from datetime import datetime from logging import Logger from pandas import DataFrame, Series class LogUtil(object): LOG_ROOT_DIR = "_logs" @staticmethod def create_logger(name: str, std_out: bool = True, file_path: str = None, level=logging.DEBUG) -> Logger: log_format = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(name) logger.setLevel(level) if std_out: std_handler = logging.StreamHandler() std_handler.setLevel(level) std_handler.setFormatter(log_format) logger.addHandler(std_handler) if file_path is not None: file_handler = logging.FileHandler(file_path, encoding="utf-8") file_handler.setLevel(level) file_handler.setFormatter(log_format) logger.addHandler(file_handler) return logger @staticmethod def create_timestamped_log_dir(sub_folder: str) -> str: now_timestamp = datetime.utcnow().strftime("%Y%m%d%H%M%S") dir_path = "{}/{}/{}/".format(LogUtil.LOG_ROOT_DIR, sub_folder, now_timestamp) os.makedirs(dir_path) return dir_path class LoggingMixin(object): def __init__(self, logger_name: str, log_file: str = None): self.logger = LogUtil.create_logger(logger_name, file_path=log_file) class PandasUtil(object): @staticmethod def append_column(data: DataFrame, column_name: str, values: Series): column = {column_name: values} return data.assign(**column)