import logging import logging.config import os from datetime import datetime def setup_logging( level: str = None, log_file: str = None, enable_file_logging: bool = False ): """ 设置日志配置 Args: level: 日志级别 (DEBUG, INFO, WARNING, ERROR, CRITICAL) log_file: 日志文件名 enable_file_logging: 是否启用文件日志 """ # 如果没有指定级别,从环境变量获取,默认INFO if level is None: level = os.getenv('LOG_LEVEL', 'INFO').upper() # 默认日志格式 log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' date_format = '%Y-%m-%d %H:%M:%S' # 基本配置 config = { 'version': 1, 'disable_existing_loggers': False, 'formatters': { 'standard': { 'format': log_format, 'datefmt': date_format }, 'detailed': { 'format': '%(asctime)s - %(name)s - %(levelname)s - %(module)s - %(funcName)s - %(lineno)d - %(message)s', 'datefmt': date_format } }, 'handlers': { 'console': { 'level': level, 'class': 'logging.StreamHandler', 'formatter': 'standard', 'stream': 'ext://sys.stdout' } }, 'root': { 'level': level, 'handlers': ['console'] }, 'loggers': { # 针对特定模块的日志配置 'memory_integration': { 'level': level, 'handlers': ['console'], 'propagate': False }, 'chat_service': { 'level': level, 'handlers': ['console'], 'propagate': False }, 'rag_pipeline': { 'level': level, 'handlers': ['console'], 'propagate': False }, 'uvicorn': { 'level': 'INFO', 'handlers': ['console'], 'propagate': False }, 'fastapi': { 'level': 'INFO', 'handlers': ['console'], 'propagate': False } } } # 如果启用文件日志 if enable_file_logging: if log_file is None: # 默认日志文件名 log_dir = 'logs' os.makedirs(log_dir, exist_ok=True) log_file = os.path.join( log_dir, f"app_{datetime.now().strftime('%Y%m%d')}.log" ) # 添加文件处理器 config['handlers']['file'] = { 'level': 'DEBUG', # 文件日志记录所有级别 'class': 'logging.handlers.RotatingFileHandler', 'filename': log_file, 'maxBytes': 10485760, # 10MB 'backupCount': 5, 'formatter': 'detailed', 'encoding': 'utf-8' } # 为所有logger添加文件处理器 for logger_name in config['loggers']: config['loggers'][logger_name]['handlers'].append('file') config['root']['handlers'].append('file') # 应用配置 logging.config.dictConfig(config) def get_logger(name: str) -> logging.Logger: """获取指定名称的logger""" return logging.getLogger(name)