223 lines
7.3 KiB
Python
223 lines
7.3 KiB
Python
"""
|
||
日志配置工具
|
||
"""
|
||
import os
|
||
import logging
|
||
import logging.handlers
|
||
from datetime import datetime
|
||
from pythonjsonlogger import jsonlogger
|
||
|
||
def setup_logging(app):
|
||
"""设置应用日志"""
|
||
log_level = app.config.get('LOG_LEVEL', 'INFO')
|
||
log_dir = app.config.get('LOG_DIR', 'logs')
|
||
|
||
# 确保日志目录存在
|
||
os.makedirs(log_dir, exist_ok=True)
|
||
|
||
# 设置根日志器
|
||
root_logger = logging.getLogger()
|
||
root_logger.setLevel(getattr(logging, log_level))
|
||
|
||
# 清除现有的处理器
|
||
root_logger.handlers = []
|
||
|
||
# 控制台处理器
|
||
console_handler = logging.StreamHandler()
|
||
console_handler.setLevel(getattr(logging, log_level))
|
||
console_formatter = ColoredFormatter(
|
||
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||
)
|
||
console_handler.setFormatter(console_formatter)
|
||
root_logger.addHandler(console_handler)
|
||
|
||
# 文件处理器 - 一般日志
|
||
file_handler = logging.handlers.RotatingFileHandler(
|
||
os.path.join(log_dir, 'app.log'),
|
||
maxBytes=10485760, # 10MB
|
||
backupCount=10
|
||
)
|
||
file_handler.setLevel(logging.INFO)
|
||
file_formatter = logging.Formatter(
|
||
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||
)
|
||
file_handler.setFormatter(file_formatter)
|
||
root_logger.addHandler(file_handler)
|
||
|
||
# 错误日志文件
|
||
error_handler = logging.handlers.RotatingFileHandler(
|
||
os.path.join(log_dir, 'error.log'),
|
||
maxBytes=10485760,
|
||
backupCount=10
|
||
)
|
||
error_handler.setLevel(logging.ERROR)
|
||
error_handler.setFormatter(file_formatter)
|
||
root_logger.addHandler(error_handler)
|
||
|
||
# JSON格式日志（用于日志分析）
|
||
json_handler = logging.handlers.RotatingFileHandler(
|
||
os.path.join(log_dir, 'app.json.log'),
|
||
maxBytes=10485760,
|
||
backupCount=10
|
||
)
|
||
json_formatter = CustomJsonFormatter()
|
||
json_handler.setFormatter(json_formatter)
|
||
json_handler.setLevel(logging.INFO)
|
||
root_logger.addHandler(json_handler)
|
||
|
||
# 研究任务专用日志
|
||
research_logger = logging.getLogger('research')
|
||
research_handler = logging.handlers.RotatingFileHandler(
|
||
os.path.join(log_dir, 'research.log'),
|
||
maxBytes=10485760,
|
||
backupCount=10
|
||
)
|
||
research_handler.setFormatter(file_formatter)
|
||
research_logger.addHandler(research_handler)
|
||
research_logger.setLevel(logging.DEBUG)
|
||
|
||
# 设置第三方库的日志级别
|
||
logging.getLogger('urllib3').setLevel(logging.WARNING)
|
||
logging.getLogger('requests').setLevel(logging.WARNING)
|
||
logging.getLogger('openai').setLevel(logging.WARNING)
|
||
|
||
app.logger.info(f"日志系统初始化完成，级别: {log_level}")
|
||
|
||
class ColoredFormatter(logging.Formatter):
|
||
"""带颜色的控制台日志格式化器"""
|
||
|
||
COLORS = {
|
||
'DEBUG': '\033[36m', # 青色
|
||
'INFO': '\033[32m', # 绿色
|
||
'WARNING': '\033[33m', # 黄色
|
||
'ERROR': '\033[31m', # 红色
|
||
'CRITICAL': '\033[35m', # 紫色
|
||
}
|
||
RESET = '\033[0m'
|
||
|
||
def format(self, record):
|
||
log_color = self.COLORS.get(record.levelname, self.RESET)
|
||
record.levelname = f"{log_color}{record.levelname}{self.RESET}"
|
||
return super().format(record)
|
||
|
||
class CustomJsonFormatter(jsonlogger.JsonFormatter):
|
||
"""自定义JSON日志格式化器"""
|
||
|
||
def add_fields(self, log_record, record, message_dict):
|
||
super().add_fields(log_record, record, message_dict)
|
||
|
||
# 添加额外字段
|
||
log_record['timestamp'] = datetime.utcnow().isoformat()
|
||
log_record['level'] = record.levelname
|
||
log_record['logger'] = record.name
|
||
|
||
# 添加异常信息
|
||
if record.exc_info:
|
||
log_record['exception'] = self.formatException(record.exc_info)
|
||
|
||
# 添加额外的上下文信息
|
||
if hasattr(record, 'session_id'):
|
||
log_record['session_id'] = record.session_id
|
||
if hasattr(record, 'subtopic_id'):
|
||
log_record['subtopic_id'] = record.subtopic_id
|
||
if hasattr(record, 'user_id'):
|
||
log_record['user_id'] = record.user_id
|
||
|
||
def get_logger(name: str) -> logging.Logger:
|
||
"""获取指定名称的日志器"""
|
||
return logging.getLogger(name)
|
||
|
||
def log_performance(func):
|
||
"""性能日志装饰器"""
|
||
import functools
|
||
import time
|
||
|
||
@functools.wraps(func)
|
||
def wrapper(*args, **kwargs):
|
||
logger = logging.getLogger(func.__module__)
|
||
start_time = time.time()
|
||
|
||
try:
|
||
result = func(*args, **kwargs)
|
||
elapsed_time = time.time() - start_time
|
||
logger.info(
|
||
f"{func.__name__} 执行成功，耗时: {elapsed_time:.3f}秒",
|
||
extra={'performance': {'function': func.__name__, 'duration': elapsed_time}}
|
||
)
|
||
return result
|
||
except Exception as e:
|
||
elapsed_time = time.time() - start_time
|
||
logger.error(
|
||
f"{func.__name__} 执行失败，耗时: {elapsed_time:.3f}秒，错误: {str(e)}",
|
||
extra={'performance': {'function': func.__name__, 'duration': elapsed_time}},
|
||
exc_info=True
|
||
)
|
||
raise
|
||
|
||
return wrapper
|
||
|
||
def log_api_call(service_name: str):
|
||
"""API调用日志装饰器"""
|
||
def decorator(func):
|
||
import functools
|
||
|
||
@functools.wraps(func)
|
||
def wrapper(*args, **kwargs):
|
||
logger = logging.getLogger('api_calls')
|
||
|
||
# 记录请求
|
||
logger.info(
|
||
f"调用 {service_name} API: {func.__name__}",
|
||
extra={
|
||
'api_service': service_name,
|
||
'api_method': func.__name__,
|
||
'args': str(args)[:200], # 限制长度
|
||
'kwargs': str(kwargs)[:200]
|
||
}
|
||
)
|
||
|
||
try:
|
||
result = func(*args, **kwargs)
|
||
logger.info(
|
||
f"{service_name} API 调用成功: {func.__name__}",
|
||
extra={
|
||
'api_service': service_name,
|
||
'api_method': func.__name__,
|
||
'success': True
|
||
}
|
||
)
|
||
return result
|
||
except Exception as e:
|
||
logger.error(
|
||
f"{service_name} API 调用失败: {func.__name__} - {str(e)}",
|
||
extra={
|
||
'api_service': service_name,
|
||
'api_method': func.__name__,
|
||
'success': False,
|
||
'error': str(e)
|
||
},
|
||
exc_info=True
|
||
)
|
||
raise
|
||
|
||
return wrapper
|
||
return decorator
|
||
|
||
class SessionLoggerAdapter(logging.LoggerAdapter):
|
||
"""带会话ID的日志适配器"""
|
||
|
||
def process(self, msg, kwargs):
|
||
if 'extra' not in kwargs:
|
||
kwargs['extra'] = {}
|
||
|
||
if hasattr(self, 'session_id'):
|
||
kwargs['extra']['session_id'] = self.session_id
|
||
|
||
return msg, kwargs
|
||
|
||
def get_session_logger(session_id: str, logger_name: str = 'research') -> SessionLoggerAdapter:
|
||
"""获取带会话ID的日志器"""
|
||
logger = logging.getLogger(logger_name)
|
||
adapter = SessionLoggerAdapter(logger, {})
|
||
adapter.session_id = session_id
|
||
return adapter |