apv

- 🐍 advanced python logging 📔
git clone git://git.acid.vegas/apv.git
apv.py (7968B)
      1 #!/usr/bin/env python3
      2 # Advanced Python Logging - Developed by acidvegas in Python (https://git.acid.vegas/apv)
      3 # apv.py
      4 
      5 import gzip
      6 import json
      7 import logging
      8 import logging.handlers
      9 import os
     10 import socket
     11 import sys
     12 
     13 
     14 class LogColors:
     15     '''ANSI color codes for log messages'''
     16 
     17     NOTSET    = '\033[97m'         # White text
     18     DEBUG     = '\033[96m'         # Cyan
     19     INFO      = '\033[92m'         # Green
     20     WARNING   = '\033[93m'         # Yellow
     21     ERROR     = '\033[91m'         # Red
     22     CRITICAL  = '\033[97m\033[41m' # White on Red
     23     FATAL     = '\033[97m\033[41m' # Same as CRITICAL
     24     DATE      = '\033[90m'         # Dark Grey
     25     MODULE    = '\033[95m'         # Pink
     26     FUNCTION  = '\033[94m'         # Blue
     27     LINE      = '\033[33m'         # Orange
     28     RESET     = '\033[0m'
     29     SEPARATOR = '\033[90m'         # Dark Grey
     30 
     31 
     32 class ConsoleFormatter(logging.Formatter):
     33     '''A formatter for the consolethat supports colored output'''
     34     
     35     def __init__(self, datefmt: str = None, details: bool = False):
     36         super().__init__(datefmt=datefmt)
     37         self.details = details
     38 
     39 
     40     def format(self, record: logging.LogRecord) -> str:
     41         '''
     42         Format a log record for the console
     43         
     44         :param record: The log record to format
     45         '''
     46 
     47         # Get the color for the log level
     48         color = getattr(LogColors, record.levelname, LogColors.RESET)
     49         
     50         # Format the log level
     51         log_level = f'{color}{record.levelname:<8}{LogColors.RESET}'
     52 
     53         # Get the log message
     54         message = record.getMessage()
     55 
     56         # Format the timestamp
     57         asctime = f'{LogColors.DATE}{self.formatTime(record, self.datefmt)}'
     58 
     59         # Get the separator
     60         separator = f'{LogColors.SEPARATOR} ┃ {LogColors.RESET}'
     61         details   = f'{LogColors.MODULE}{record.module}{separator}{LogColors.FUNCTION}{record.funcName}{separator}{LogColors.LINE}{record.lineno}{separator}' if self.details else ''
     62         
     63         return f'{asctime}{separator}{log_level}{separator}{details}{message}'
     64 
     65 
     66 class JsonFormatter(logging.Formatter):
     67     '''Formatter for JSON output'''
     68     
     69     def __init__(self, datefmt: str = None):
     70         super().__init__(datefmt=datefmt)
     71 
     72 
     73     def format(self, record: logging.LogRecord) -> str:
     74         '''
     75         Format a log record for JSON output
     76         
     77         :param record: The log record to format
     78         '''
     79 
     80         # Create a dictionary to store the log record
     81         log_dict = {
     82             '@timestamp'    : self.formatTime(record, self.datefmt),
     83             'level'        : record.levelname,
     84             'message'      : record.getMessage(),
     85             'process_id'   : record.process,
     86             'process_name' : record.processName,
     87             'thread_id'    : record.thread,
     88             'thread_name'  : record.threadName,
     89             'logger_name'  : record.name,
     90             'filename'     : record.filename,
     91             'line_number'  : record.lineno,
     92             'function'     : record.funcName,
     93             'module'       : record.module,
     94             'hostname'     : socket.gethostname()
     95         }
     96 
     97         # Add the exception if it exists
     98         if record.exc_info:
     99             log_dict['exception'] = self.formatException(record.exc_info)
    100 
    101         # Add any custom attributes that start with an underscore
    102         custom_attrs = {k: v for k, v in record.__dict__.items() if k.startswith('_') and not k.startswith('__')}
    103         log_dict.update(custom_attrs)
    104 
    105         return json.dumps(log_dict)
    106 
    107 
    108 class GZipRotatingFileHandler(logging.handlers.RotatingFileHandler):
    109     '''RotatingFileHandler that compresses rotated log files'''
    110     
    111     def rotation_filename(self, default_name: str) -> str:
    112         return default_name + '.gz'
    113 
    114     def rotate(self, source: str, dest: str):
    115         with open(source, 'rb') as src, gzip.open(dest, 'wb') as dst:
    116             dst.write(src.read())
    117 
    118 
    119 class LoggerSetup:
    120     def __init__(self, level: str = 'INFO', date_format: str = '%Y-%m-%d %H:%M:%S', log_to_disk: bool = False, max_log_size: int = 10*1024*1024, max_backups: int = 7, log_file_name: str = 'app', json_log: bool = False, show_details: bool = False, compress_backups: bool = False):
    121         '''
    122         Initialize the LoggerSetup with provided parameters
    123         
    124         :param level: The logging level (e.g., 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL')
    125         :param date_format: The date format for log messages
    126         :param log_to_disk: Whether to log to disk
    127         :param max_log_size: The maximum size of log files before rotation
    128         :param max_backups: The maximum number of backup log files to keep
    129         :param log_file_name: The base name of the log file
    130         :param json_log: Whether to log in JSON format
    131         :param show_details: Whether to show detailed log messages
    132         :param compress_backups: Whether to compress old log files using gzip
    133         '''
    134 
    135         self.level            = level
    136         self.date_format      = date_format
    137         self.log_to_disk      = log_to_disk
    138         self.max_log_size     = max_log_size
    139         self.max_backups      = max_backups
    140         self.log_file_name    = log_file_name
    141         self.json_log         = json_log
    142         self.show_details     = show_details
    143         self.compress_backups = compress_backups
    144 
    145 
    146     def setup(self):
    147         '''Set up logging with various handlers and options'''
    148 
    149         # Clear existing handlers
    150         logging.getLogger().handlers.clear()
    151         logging.getLogger().setLevel(logging.DEBUG)
    152 
    153         # Convert the level string to a logging level object
    154         level_num = getattr(logging, self.level.upper(), logging.INFO)
    155 
    156         # Setup console handler
    157         self.setup_console_handler(level_num)
    158 
    159         # Setup file handler if enabled
    160         if self.log_to_disk:
    161             self.setup_file_handler(level_num)
    162 
    163 
    164     def setup_console_handler(self, level_num: int):
    165         '''
    166         Set up the console handler
    167         
    168         :param level_num: The logging level number
    169         '''
    170 
    171         # Create the console handler
    172         console_handler = logging.StreamHandler()
    173         console_handler.setLevel(level_num)
    174         
    175         # Create the formatter
    176         formatter = JsonFormatter(datefmt=self.date_format) if self.json_log else ConsoleFormatter(datefmt=self.date_format, details=self.show_details)
    177         console_handler.setFormatter(formatter)
    178         
    179         # Add the handler to the root logger
    180         logging.getLogger().addHandler(console_handler)
    181 
    182 
    183     def setup_file_handler(self, level_num: int):
    184         '''
    185         Set up the file handler
    186         
    187         :param level_num: The logging level number
    188         '''
    189 
    190         # Create logs directory if it doesn't exist
    191         logs_dir = os.path.join(sys.path[0], 'logs')
    192         os.makedirs(logs_dir, exist_ok=True)
    193 
    194         # Set up log file path
    195         file_extension = '.json' if self.json_log else '.log'
    196         log_file_path  = os.path.join(logs_dir, f'{self.log_file_name}{file_extension}')
    197 
    198         # Create the rotating file handler
    199         handler_class = GZipRotatingFileHandler if self.compress_backups else logging.handlers.RotatingFileHandler
    200         file_handler  = handler_class(log_file_path, maxBytes=self.max_log_size, backupCount=self.max_backups)
    201         file_handler.setLevel(level_num)
    202 
    203         # Set up the appropriate formatter
    204         formatter = JsonFormatter(datefmt=self.date_format) if self.json_log else logging.Formatter(fmt='%(asctime)s ┃ %(levelname)-8s ┃ %(module)s ┃ %(funcName)s ┃ %(lineno)d ┃ %(message)s', datefmt=self.date_format)
    205         file_handler.setFormatter(formatter)
    206 
    207         logging.getLogger().addHandler(file_handler)
    208 
    209 
    210 def setup_logging(**kwargs):
    211     '''Set up logging with various handlers and options'''
    212 
    213     # Create a LoggerSetup instance with the provided keyword arguments
    214     logger_setup = LoggerSetup(**kwargs)
    215 
    216     # Set up the logging system
    217     logger_setup.setup()