248 lines
8.6 KiB
Python
248 lines
8.6 KiB
Python
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") |