Spaces:
Sleeping
Sleeping
File size: 5,487 Bytes
00b00eb |
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 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 |
"""
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)
|