import logging import logging.handlers from pathlib import Path from typing import Optional import time from datetime import datetime, timedelta from app.config import settings class CustomTimedRotatingFileHandler(logging.handlers.TimedRotatingFileHandler): """自定义时间轮转文件处理器，支持指定轮转时间""" def __init__(self, filename, hour=0, minute=0, when='midnight', interval=1, backupCount=7, encoding='utf-8'): self.target_hour = hour self.target_minute = minute self.target_when = when self.target_interval = interval # 先调用父类初始化 super().__init__(filename, when=when, interval=interval, backupCount=backupCount, encoding=encoding) # 重新计算轮转时间 self.computeRollover() def computeRollover(self, currentTime=None): """计算下一次轮转时间，设置为每天的指定时间""" if currentTime is None: currentTime = time.time() # 获取当前时间 current_time = datetime.fromtimestamp(currentTime) # 创建今天的目标时间 target_time = current_time.replace(hour=self.target_hour, minute=self.target_minute, second=0, microsecond=0) # 如果今天的目标时间已过，设置为明天 if current_time >= target_time: target_time += timedelta(days=1) # 转换为时间戳 self.rolloverAt = target_time.timestamp() def doRollover(self): # 调用父类轮转方法 super().doRollover() # 重新计算下次轮转时间 self.computeRollover() print(f"✅ 日志轮转完成，下次: {datetime.fromtimestamp(self.rolloverAt).strftime('%Y-%m-%d %H:%M:%S')}") class LoggerManager: """日志管理器""" _loggers = {} _initialized = False @classmethod def setup_logging(cls): """设置日志系统""" if cls._initialized: return # 创建日志目录 log_dir = Path(settings.log_file).parent log_dir.mkdir(parents=True, exist_ok=True) # 配置根日志器 root_logger = logging.getLogger() root_logger.setLevel(getattr(logging, settings.log_level.upper())) # 清除现有处理器 root_logger.handlers.clear() # 创建控制台处理器 console_handler = logging.StreamHandler() console_handler.setLevel(getattr(logging, settings.log_level.upper())) # 控制台使用普通格式化器 console_formatter = logging.Formatter( fmt=settings.log_format, datefmt=settings.log_date_format ) console_handler.setFormatter(console_formatter) # 创建文件处理器（按时间轮转，每天0点） file_handler = CustomTimedRotatingFileHandler( filename=settings.log_file, hour=0, # 0点 minute=0, # 0分 when='midnight', # 每天轮转 interval=1, # 每天一次 backupCount=settings.log_backup_count, encoding='utf-8', ) # 创建后设置 suffix 属性 file_handler.suffix = "%Y%m%d" # 设置文件后缀格式，如 app.log.20231215 file_handler.setLevel(settings.log_level) # 文件记录所有级别 # 文件使用普通格式化器 file_formatter = logging.Formatter( fmt=settings.log_format, datefmt=settings.log_date_format ) file_handler.setFormatter(file_formatter) # 添加处理器到根日志器 root_logger.addHandler(console_handler) root_logger.addHandler(file_handler) # 设置第三方库的日志级别 logging.getLogger("uvicorn").setLevel(logging.INFO) logging.getLogger("uvicorn.access").setLevel(logging.WARNING) logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING) logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("watchfiles").setLevel(logging.WARNING) cls._initialized = True # 记录日志系统启动 logger = cls.get_logger("LoggerManager") logger.info("🚀 日志系统初始化完成") logger.info(f"📁 日志文件: {settings.log_file}") logger.info(f"📊 日志级别: {settings.log_level}") @classmethod def get_logger(cls, name: str) -> logging.Logger: """获取指定名称的日志器""" if not cls._initialized: cls.setup_logging() if name not in cls._loggers: cls._loggers[name] = logging.getLogger(name) return cls._loggers[name] def get_logger(name: Optional[str] = None) -> logging.Logger: """获取日志器的便捷函数""" if name is None: # 自动获取调用者的模块名 import inspect frame = inspect.currentframe().f_back name = frame.f_globals.get('__name__', 'unknown') return LoggerManager.get_logger(name) # 预定义的日志器 app_logger = get_logger("app") api_logger = get_logger("api") db_logger = get_logger("database") external_api_logger = get_logger("external_api") # 各模块专用日志器 main_logger = get_logger("main") routes_logger = get_logger("routes") celery_logger = get_logger("celery") celery_tasks_logger = get_logger("celery_tasks") callback_service_logger = get_logger("callback_service") redis_logger = get_logger("redis") # 通用文件日志器设置函数 def setup_celery_file_logger(logger_name: str, log_file: str, level: str = "INFO"): """为Celery组件设置专用的文件日志器，同时输出到控制台和文件""" # 创建日志目录 log_dir = Path(log_file).parent log_dir.mkdir(parents=True, exist_ok=True) # 获取日志器 logger = get_logger(logger_name) logger.setLevel(getattr(logging, level.upper())) # 清除现有处理器（避免重复添加） logger.handlers.clear() # 创建控制台处理器 console_handler = logging.StreamHandler() console_handler.setLevel(getattr(logging, level.upper())) # 控制台格式化器 console_formatter = logging.Formatter( fmt='%(asctime)s - %(name)s - %(levelname)s - %(message)s', datefmt='%H:%M:%S' ) console_handler.setFormatter(console_formatter) # 创建文件处理器 file_handler = CustomTimedRotatingFileHandler( filename=log_file, hour=0, # 0点 minute=0, # 0分 when='midnight', interval=1, backupCount=7, encoding='utf-8', ) file_handler.suffix = "%Y%m%d" file_handler.setLevel(getattr(logging, level.upper())) # 文件格式化器 file_formatter = logging.Formatter( fmt='%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) file_handler.setFormatter(file_formatter) # 添加处理器（先控制台，后文件） logger.addHandler(console_handler) logger.addHandler(file_handler) logger.propagate = False # 防止重复输出到父日志器 return logger # 各模块专用日志器映射 CELERY_LOGGERS = { "celery_tasks": ("logs/celery_tasks.log", "INFO"), "celery": ("logs/celery_app.log", "INFO"), "celery.beat": ("logs/celery_app.log", "INFO"), # Beat日志输出到celery_app.log "celery.worker": ("logs/celery_app.log", "INFO"), # Worker日志输出到celery_app.log } # 初始化所有Celery文件日志器 _celery_file_loggers = {} for logger_name, (log_file, level) in CELERY_LOGGERS.items(): _celery_file_loggers[logger_name] = setup_celery_file_logger(logger_name, log_file, level) # 向后兼容的默认日志器 logger = app_logger # 通用获取器函数 def get_celery_file_logger(logger_name: str): """获取指定名称的Celery文件日志器""" return _celery_file_loggers.get(logger_name, get_logger(logger_name)) # 简化的便捷函数 def get_main_logger(): return main_logger def get_routes_logger(): return routes_logger def get_celery_logger(): return get_celery_file_logger("celery") def get_celery_tasks_logger(): return get_celery_file_logger("celery_tasks") def get_callback_service_logger(): return callback_service_logger def get_redis_logger(): return redis_logger def get_celery_beat_logger(): return get_celery_file_logger("celery.beat") def get_celery_worker_logger(): return get_celery_file_logger("celery.worker")