File size: 2,540 Bytes
60016b2 |
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 |
import logging
from datetime import datetime
from collections import deque
from threading import Lock
DEBUG = False # 可以从环境变量中获取
LOG_FORMAT_DEBUG = '%(asctime)s - %(levelname)s - [%(key)s]-%(request_type)s-[%(model)s]-%(status_code)s: %(message)s - %(error_message)s'
LOG_FORMAT_NORMAL = '[%(asctime)s] [%(levelname)s] [%(key)s]-%(request_type)s-[%(model)s]-%(status_code)s: %(message)s'
# 配置 logger
logger = logging.getLogger("my_logger")
logger.setLevel(logging.DEBUG)
# 控制台处理器
console_handler = logging.StreamHandler()
console_formatter = logging.Formatter('%(message)s')
console_handler.setFormatter(console_formatter)
logger.addHandler(console_handler)
# 日志缓存，用于在网页上显示最近的日志
class LogManager:
def __init__(self, max_logs=100):
self.logs = deque(maxlen=max_logs) # 使用双端队列存储最近的日志
self.lock = Lock()
def add_log(self, log_entry):
with self.lock:
self.logs.append(log_entry)
def get_recent_logs(self, count=50):
with self.lock:
return list(self.logs)[-count:]
# 创建日志管理器实例
log_manager = LogManager()
def format_log_message(level, message, extra=None):
extra = extra or {}
log_values = {
'asctime': datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
'levelname': level,
'key': extra.get('key', 'N/A'),
'request_type': extra.get('request_type', 'N/A'),
'model': extra.get('model', 'N/A'),
'status_code': extra.get('status_code', 'N/A'),
'error_message': extra.get('error_message', ''),
'message': message
}
log_format = LOG_FORMAT_DEBUG if DEBUG else LOG_FORMAT_NORMAL
formatted_log = log_format % log_values
# 将格式化后的日志添加到日志管理器
log_entry = {
'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
'level': level,
'key': extra.get('key', 'N/A'),
'request_type': extra.get('request_type', 'N/A'),
'model': extra.get('model', 'N/A'),
'status_code': extra.get('status_code', 'N/A'),
'message': message,
'error_message': extra.get('error_message', ''),
'formatted': formatted_log
}
log_manager.add_log(log_entry)
return formatted_log
def log(level: str, message: str, **extra):
"""简化日志记录的统一函数"""
msg = format_log_message(level.upper(), message, extra=extra)
getattr(logger, level.lower())(msg) |