244 lines
6.9 KiB
Python
244 lines
6.9 KiB
Python
"""
|
|
日志管理工具
|
|
基于loguru实现的统一日志管理。
|
|
"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from loguru import logger
|
|
|
|
from ..config.settings import settings
|
|
|
|
|
|
def setup_logger(
|
|
level: Optional[str] = None,
|
|
log_dir: Optional[str] = None,
|
|
enable_colors: Optional[bool] = None
|
|
):
|
|
"""
|
|
设置日志配置
|
|
|
|
Args:
|
|
level: 日志级别
|
|
log_dir: 日志目录
|
|
enable_colors: 是否启用颜色
|
|
"""
|
|
# 移除默认处理器
|
|
logger.remove()
|
|
|
|
# 使用配置中的值，如果没有传入参数
|
|
log_level = level or settings.log.level
|
|
log_dir_path = Path(log_dir or settings.logs_dir)
|
|
colors_enabled = enable_colors if enable_colors is not None else settings.log.enable_colors
|
|
|
|
# 确保日志目录存在
|
|
log_dir_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 控制台处理器
|
|
console_format = (
|
|
"<green>{time:HH:mm:ss}</green> | "
|
|
"<level>{level: <8}</level> | "
|
|
"<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
|
|
"<level>{message}</level>"
|
|
) if colors_enabled else (
|
|
"{time:HH:mm:ss} | "
|
|
"{level: <8} | "
|
|
"{name}:{function}:{line} - "
|
|
"{message}"
|
|
)
|
|
|
|
logger.add(
|
|
sys.stdout,
|
|
format=console_format,
|
|
level=log_level,
|
|
colorize=colors_enabled,
|
|
backtrace=True,
|
|
diagnose=True
|
|
)
|
|
|
|
# 文件处理器
|
|
file_format = (
|
|
"{time:YYYY-MM-DD HH:mm:ss.SSS} | "
|
|
"{level: <8} | "
|
|
"{name}:{function}:{line} | "
|
|
"{message}"
|
|
)
|
|
|
|
# 按日期分割的日志文件
|
|
logger.add(
|
|
log_dir_path / "{time:YYYY-MM-DD}.log",
|
|
format=file_format,
|
|
level=log_level,
|
|
rotation=settings.log.rotation,
|
|
retention=settings.log.retention,
|
|
compression=settings.log.compression,
|
|
backtrace=True,
|
|
diagnose=True,
|
|
encoding="utf-8"
|
|
)
|
|
|
|
# 错误日志单独文件
|
|
logger.add(
|
|
log_dir_path / "error_{time:YYYY-MM-DD}.log",
|
|
format=file_format,
|
|
level="ERROR",
|
|
rotation=settings.log.rotation,
|
|
retention=settings.log.retention,
|
|
compression=settings.log.compression,
|
|
backtrace=True,
|
|
diagnose=True,
|
|
encoding="utf-8"
|
|
)
|
|
|
|
# 添加平台过滤器标签
|
|
logger.configure(
|
|
extra={
|
|
"platform": None,
|
|
"account": None,
|
|
"task_id": None
|
|
}
|
|
)
|
|
|
|
return logger
|
|
|
|
|
|
def get_logger(name: Optional[str] = None):
|
|
"""
|
|
获取日志记录器
|
|
|
|
Args:
|
|
name: 日志记录器名称
|
|
|
|
Returns:
|
|
日志记录器实例
|
|
"""
|
|
if name:
|
|
return logger.bind(name=name)
|
|
return logger
|
|
|
|
|
|
class PlatformLogger:
|
|
"""平台特定的日志记录器"""
|
|
|
|
def __init__(self, platform: str, account: Optional[str] = None):
|
|
self.platform = platform
|
|
self.account = account
|
|
self.logger = logger.bind(platform=platform, account=account)
|
|
|
|
def info(self, message: str, **kwargs):
|
|
"""记录信息日志"""
|
|
self.logger.info(f"[{self.platform}] {message}", **kwargs)
|
|
|
|
def warning(self, message: str, **kwargs):
|
|
"""记录警告日志"""
|
|
self.logger.warning(f"[{self.platform}] {message}", **kwargs)
|
|
|
|
def error(self, message: str, **kwargs):
|
|
"""记录错误日志"""
|
|
self.logger.error(f"[{self.platform}] {message}", **kwargs)
|
|
|
|
def debug(self, message: str, **kwargs):
|
|
"""记录调试日志"""
|
|
self.logger.debug(f"[{self.platform}] {message}", **kwargs)
|
|
|
|
def success(self, message: str, **kwargs):
|
|
"""记录成功日志"""
|
|
self.logger.info(f"✅ [{self.platform}] {message}", **kwargs)
|
|
|
|
def failure(self, message: str, **kwargs):
|
|
"""记录失败日志"""
|
|
self.logger.error(f"❌ [{self.platform}] {message}", **kwargs)
|
|
|
|
def bind_task(self, task_id: str):
|
|
"""绑定任务ID"""
|
|
return self.logger.bind(task_id=task_id)
|
|
|
|
def log_operation(self, operation: str, status: str, details: Optional[dict] = None):
|
|
"""记录操作日志"""
|
|
message = f"操作: {operation} | 状态: {status}"
|
|
if details:
|
|
message += f" | 详情: {details}"
|
|
|
|
if status == "success":
|
|
self.success(message)
|
|
elif status == "failed":
|
|
self.failure(message)
|
|
elif status == "warning":
|
|
self.warning(message)
|
|
else:
|
|
self.info(message)
|
|
|
|
|
|
class TaskLogger:
|
|
"""任务特定的日志记录器"""
|
|
|
|
def __init__(self, task_id: str, platform: str, account: Optional[str] = None):
|
|
self.task_id = task_id
|
|
self.platform = platform
|
|
self.account = account
|
|
self.logger = logger.bind(
|
|
task_id=task_id,
|
|
platform=platform,
|
|
account=account
|
|
)
|
|
|
|
def start(self, message: str = "任务开始"):
|
|
"""记录任务开始"""
|
|
self.logger.info(f"🚀 [{self.task_id}] {message}")
|
|
|
|
def progress(self, message: str, progress: Optional[float] = None):
|
|
"""记录任务进度"""
|
|
if progress is not None:
|
|
message = f"⏳ [{self.task_id}] {message} ({progress:.1f}%)"
|
|
else:
|
|
message = f"⏳ [{self.task_id}] {message}"
|
|
self.logger.info(message)
|
|
|
|
def success(self, message: str = "任务完成"):
|
|
"""记录任务成功"""
|
|
self.logger.info(f"✅ [{self.task_id}] {message}")
|
|
|
|
def failure(self, message: str, error: Optional[Exception] = None):
|
|
"""记录任务失败"""
|
|
if error:
|
|
message = f"❌ [{self.task_id}] {message} - {str(error)}"
|
|
else:
|
|
message = f"❌ [{self.task_id}] {message}"
|
|
self.logger.error(message)
|
|
|
|
def warning(self, message: str):
|
|
"""记录任务警告"""
|
|
self.logger.warning(f"⚠️ [{self.task_id}] {message}")
|
|
|
|
def log_step(self, step: str, status: str, details: Optional[dict] = None):
|
|
"""记录任务步骤"""
|
|
message = f"步骤: {step} | 状态: {status}"
|
|
if details:
|
|
message += f" | 详情: {details}"
|
|
|
|
if status == "completed":
|
|
self.logger.info(f"✅ [{self.task_id}] {message}")
|
|
elif status == "failed":
|
|
self.logger.error(f"❌ [{self.task_id}] {message}")
|
|
elif status == "skipped":
|
|
self.logger.warning(f"⏭️ [{self.task_id}] {message}")
|
|
else:
|
|
self.logger.info(f"⏳ [{self.task_id}] {message}")
|
|
|
|
|
|
# 预定义的日志记录器
|
|
def get_platform_logger(platform: str, account: Optional[str] = None) -> PlatformLogger:
|
|
"""获取平台日志记录器"""
|
|
return PlatformLogger(platform, account)
|
|
|
|
|
|
def get_task_logger(task_id: str, platform: str, account: Optional[str] = None) -> TaskLogger:
|
|
"""获取任务日志记录器"""
|
|
return TaskLogger(task_id, platform, account)
|
|
|
|
|
|
# 初始化日志系统
|
|
if not logger._core.handlers:
|
|
setup_logger() |