import logging import os from pathlib import Path from logging.handlers import RotatingFileHandler from datetime import datetime # ========================================== # 日志系统配置 # ========================================== LOG_BASE_DIR = "./logs" SYSTEM_LOG_DIR = os.path.join(LOG_BASE_DIR, "system") AI_GROQ_LOG_DIR = os.path.join(LOG_BASE_DIR, "ai", "groq") AI_CODEX_LOG_DIR = os.path.join(LOG_BASE_DIR, "ai", "codex") # 日志格式 LOG_FORMAT = "[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s" DATE_FORMAT = "%Y-%m-%d %H:%M:%S" # 日志轮转配置 MAX_BYTES = 10 * 1024 * 1024 # 10MB BACKUP_COUNT = 5 # ========================================== def setup_directories(): """创建所有必要的日志目录""" for directory in [SYSTEM_LOG_DIR, AI_GROQ_LOG_DIR, AI_CODEX_LOG_DIR]: Path(directory).mkdir(parents=True, exist_ok=True) def get_system_logger(module_name): """ 获取系统日志记录器 :param module_name: 模块名称 (如 'monitor', 'upload') :return: logger对象 """ setup_directories() logger = logging.getLogger(f"system.{module_name}") logger.setLevel(logging.DEBUG) # 避免重复添加handler if logger.handlers: return logger # 文件handler log_file = os.path.join(SYSTEM_LOG_DIR, f"{module_name}.log") file_handler = RotatingFileHandler( log_file, maxBytes=MAX_BYTES, backupCount=BACKUP_COUNT, encoding='utf-8' ) file_handler.setLevel(logging.DEBUG) # 控制台handler console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) # 设置格式 formatter = logging.Formatter(LOG_FORMAT, DATE_FORMAT) file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger def get_ai_logger(ai_type, task_name=""): """ 获取AI调用的独立日志记录器 :param ai_type: AI类型 ('groq' 或 'codex') :param task_name: 任务名称 (如 'transcribe', 'songs') :return: logger对象和日志文件路径 """ setup_directories() # 生成唯一的日志文件名 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") if task_name: log_filename = f"{task_name}_{timestamp}.log" else: log_filename = f"{timestamp}.log" # 根据AI类型选择目录 if ai_type == "groq": log_dir = AI_GROQ_LOG_DIR elif ai_type == "codex": log_dir = AI_CODEX_LOG_DIR else: raise ValueError(f"未知的AI类型: {ai_type}") log_file = os.path.join(log_dir, log_filename) # 创建独立的logger logger_name = f"ai.{ai_type}.{timestamp}" logger = logging.getLogger(logger_name) logger.setLevel(logging.DEBUG) # 清除已有的handlers logger.handlers.clear() # 只使用文件handler file_handler = logging.FileHandler(log_file, encoding='utf-8') file_handler.setLevel(logging.DEBUG) formatter = logging.Formatter(LOG_FORMAT, DATE_FORMAT) file_handler.setFormatter(formatter) logger.addHandler(file_handler) # 防止日志传播到父logger logger.propagate = False return logger, log_file def log_exception(logger, exception, context=""): """ 记录异常信息的辅助函数 :param logger: logger对象 :param exception: 异常对象 :param context: 上下文信息 """ if context: logger.error(f"{context}: {type(exception).__name__}: {str(exception)}", exc_info=True) else: logger.error(f"{type(exception).__name__}: {str(exception)}", exc_info=True)