""" Contains logging functions to initialize and configure loggers, facilitating efficient tracking of ETL processes. Also Provides methods to log messages with different log levels and additional context information such as channel name and client name """ import os import streamlit as st import logging, logging.handlers from datetime import datetime, timezone def delete_old_log_files(): """ Deletes all log files in the 'logs' directory that are older than the current month. Returns: None """ # Set the logs directory to 'logs' within the current working directory logs_directory = os.path.join(os.getcwd(), "logs") # Ensure the logs directory exists if not os.path.exists(logs_directory): print(f"Directory {logs_directory} does not exist.") return # Get the current year and month in UTC current_utc_time = datetime.now(timezone.utc) current_year_month = current_utc_time.strftime("%Y%m") # Iterate through all files in the logs directory for filename in os.listdir(logs_directory): # We assume the log files are in the format "eid_YYYYMM.log" if filename.endswith(".log"): # Extract the date portion from the filename try: file_date = filename.split("_")[-1].replace(".log", "") # Compare the file date with the current date if file_date < current_year_month: # Construct the full file path file_path = os.path.join(logs_directory, filename) # Delete the file os.remove(file_path) except IndexError: # If the filename doesn't match the expected pattern, skip it pass def create_log_file_name(emp_id, project_name): """ Generates a log file name using the format eid_YYYYMMDD.log based on UTC time. Returns: str: The generated log file name. """ # Get the current UTC time utc_now = datetime.now(timezone.utc) # Format the date as YYYYMMDD date_str = utc_now.strftime("%Y%m%d") # Create the file name with the format eid_YYYYMMDD.log log_file_name = f"{emp_id}_{project_name}_{date_str}.log" return log_file_name def get_logger(log_file): """ Initializes and configures a logger. If the log file already exists, it appends to it. Returns the same logger instance if called multiple times. Args: log_file (str): The path to the log file where logs will be written. Returns: logging.Logger: Configured logger instance. """ # Define the logger name logger_name = os.path.basename(log_file) logger = logging.getLogger(logger_name) # If the logger already has handlers, return it to prevent duplicate handlers if logger.hasHandlers(): return logger # Set the logging level logging_level = logging.INFO logger.setLevel(logging_level) # Ensure the logs directory exists and is writable logs_dir = os.path.dirname(log_file) if not os.path.exists(logs_dir): os.makedirs(logs_dir, exist_ok=True) # Change the directory permissions to 0777 (0o777 in Python 3) os.chmod(logs_dir, 0o777) # Create a file handler to append to the log file file_handler = logging.FileHandler(log_file) file_handler.setFormatter( logging.Formatter("%(asctime)s %(levelname)s %(message)s", "%Y-%m-%d %H:%M:%S") ) logger.addHandler(file_handler) # Create a stream handler to print to console stream_handler = logging.StreamHandler() stream_handler.setFormatter( logging.Formatter("%(asctime)s %(levelname)s %(message)s", "%Y-%m-%d %H:%M:%S") ) logger.addHandler(stream_handler) return logger def log_message(log_type, message, page_name): """ Logs a message with the specified log type and additional context information. Args: log_type (str): The type of log ('info', 'error', 'warning', 'debug'). message (str): The message to log. page_name (str): The name of the page associated with the message. Returns: None """ # Retrieve the employee ID and project name from session state emp_id = st.session_state["emp_id"] project_name = st.session_state["project_name"] # Create log file name using a function that generates a name based on the current date and EID log_file_name = create_log_file_name(emp_id, project_name) # Construct the full path to the log file within the "logs" within the current working directory file_path = os.path.join(os.getcwd(), "logs", log_file_name) # Generate and store the logger instance in session state logger = get_logger(file_path) # Construct the log message with all required context information log_message = ( f"USER_EID: {emp_id} ; " f"PROJECT_NAME: {project_name} ; " f"PAGE_NAME: {page_name} ; " f"MESSAGE: {message}" ) # Log the message with the appropriate log level based on the log_type argument if log_type == "info": logger.info(log_message) elif log_type == "error": logger.error(log_message) elif log_type == "warning": logger.warning(log_message) else: logger.debug(log_message)