117 lines
4.2 KiB
Python
117 lines
4.2 KiB
Python
__author__ = 'Wendell Jones'
|
|
|
|
import logging
|
|
from logging.handlers import RotatingFileHandler
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
import os
|
|
from glob import glob
|
|
import math as m
|
|
|
|
|
|
class Logger:
|
|
def __init__(self):
|
|
pass
|
|
|
|
def __del__(self):
|
|
pass
|
|
|
|
|
|
class LPrint:
|
|
def __init__(self, log_dir, force_rotate):
|
|
self.log_dir = log_dir
|
|
self.logfile = os.path.join(log_dir, 'dbsync.log')
|
|
self.setup_logfile()
|
|
self.cleanup_old_logs()
|
|
self.check_log_size(force_rotate)
|
|
|
|
def setup_logfile(self):
|
|
"""Ensure the logfile exists and is ready for use."""
|
|
os.makedirs(self.log_dir, exist_ok=True)
|
|
with open(self.logfile, 'a'):
|
|
os.utime(self.logfile, None)
|
|
|
|
def cleanup_old_logs(self):
|
|
"""Remove log files older than 14 days."""
|
|
logfile_list = glob(f'{self.logfile}.*')
|
|
date_limit = datetime.today() - timedelta(days=14)
|
|
for log in logfile_list:
|
|
timestamp = datetime.fromtimestamp(os.path.getmtime(log))
|
|
if timestamp < date_limit:
|
|
os.remove(log)
|
|
|
|
def check_log_size(self, force_rotate):
|
|
"""Check the log file size and rotate if necessary."""
|
|
log_size = os.path.getsize(self.logfile)
|
|
print(f'Logfile Size: {log_size} ({self.humanbytes(log_size)})')
|
|
if force_rotate or log_size > 50_000_000: # 10 MB
|
|
self.rotate_logs()
|
|
|
|
def rotate_logs(self):
|
|
"""Rotate the log file."""
|
|
from logs.logrotater import LogRotate # Assuming logrotater is a custom module
|
|
rotater = LogRotate(prefix=self.logfile, verbose=True)
|
|
try:
|
|
rotater.rotate()
|
|
except FileExistsError as e:
|
|
print(f'Unable to overwrite archive log: {e}. Deleting old logs.')
|
|
for log in glob(f'{self.logfile}.*'):
|
|
try:
|
|
os.remove(log)
|
|
except Exception as e:
|
|
print(f'Unable to delete {log}: {e}')
|
|
|
|
def start_log(self):
|
|
"""Initialize logging."""
|
|
logging.basicConfig(filename=self.logfile, level=logging.DEBUG)
|
|
|
|
def logprint(self, loglevel, logmessage):
|
|
"""Log a message with a timestamp."""
|
|
logstamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
log_message = f'{logstamp}: {logmessage}'
|
|
loglevel = loglevel.lower()
|
|
if loglevel == 'debug':
|
|
logging.debug(log_message)
|
|
elif loglevel == 'warning':
|
|
logging.warning(log_message)
|
|
elif loglevel == 'info':
|
|
logging.info(log_message)
|
|
else:
|
|
logging.error(f'Unknown log level: {loglevel}. Message: {logmessage}')
|
|
|
|
def screenprint(self, loglevel, logmessage):
|
|
"""Log a message with a timestamp."""
|
|
logstamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
log_message = f'{logstamp}: {logmessage}'
|
|
loglevel = loglevel.lower()
|
|
if loglevel == 'debug':
|
|
logging.debug(log_message)
|
|
print(f"{log_message}")
|
|
elif loglevel == 'warning':
|
|
logging.warning(log_message)
|
|
print(f"{log_message}")
|
|
elif loglevel == 'info':
|
|
logging.info(log_message)
|
|
print(f"{log_message}")
|
|
else:
|
|
logging.error(f'Unknown log level: {loglevel}. Message: {logmessage}')
|
|
|
|
def create_rotating_log(self, logpath, max_bytes=10_000_000, backup_count=5):
|
|
"""Create a rotating log handler."""
|
|
logger = logging.getLogger("Rotating Log")
|
|
logger.setLevel(logging.INFO)
|
|
handler = RotatingFileHandler(logpath, maxBytes=max_bytes, backupCount=backup_count)
|
|
logger.addHandler(handler)
|
|
|
|
@staticmethod
|
|
def humanbytes(size, binary=False, precision=2):
|
|
"""Convert bytes to a human-readable format."""
|
|
multiples = ["B", "k{}B", "M{}B", "G{}B", "T{}B", "P{}B", "E{}B", "Z{}B", "Y{}B"]
|
|
base = 1024 if binary else 1000
|
|
multiple = m.trunc(m.log2(size) / m.log2(base)) if size > 0 else 0
|
|
value = size / m.pow(base, multiple) if size > 0 else 0
|
|
suffix = multiples[multiple].format("i" if binary else "")
|
|
return f"{value:.{precision}f} {suffix}"
|
|
|
|
def __del__(self):
|
|
pass |