Simon
127 lines
4.5 KiB
Python
127 lines
4.5 KiB
Python
# =============================================================================
|
||
# IT智能服务台 — 日志配置
|
||
# =============================================================================
|
||
# 说明：统一日志格式，支持 JSON 输出便于日志收集
|
||
# =============================================================================
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import sys
|
||
from datetime import datetime
|
||
from logging.handlers import RotatingFileHandler
|
||
from typing import Any, Optional
|
||
|
||
|
||
class JSONFormatter(logging.Formatter):
|
||
"""JSON 格式日志 formatter"""
|
||
|
||
def format(self, record: logging.LogRecord) -> str:
|
||
"""将日志记录格式化为 JSON"""
|
||
log_data: dict[str, Any] = {
|
||
"timestamp": datetime.utcnow().isoformat() + "Z",
|
||
"level": record.levelname,
|
||
"logger": record.name,
|
||
"message": record.getMessage(),
|
||
"module": record.module,
|
||
"function": record.funcName,
|
||
"line": record.lineno,
|
||
}
|
||
|
||
# 添加异常信息
|
||
if record.exc_info:
|
||
log_data["exception"] = self.formatException(record.exc_info)
|
||
|
||
# 添加额外字段
|
||
if hasattr(record, "request_id"):
|
||
log_data["request_id"] = record.request_id
|
||
if hasattr(record, "user_id"):
|
||
log_data["user_id"] = record.user_id
|
||
if hasattr(record, "extra"):
|
||
log_data.update(record.extra)
|
||
|
||
return json.dumps(log_data, ensure_ascii=False)
|
||
|
||
|
||
class PlainFormatter(logging.Formatter):
|
||
"""普通格式日志 formatter（开发环境使用）"""
|
||
|
||
def __init__(self):
|
||
super().__init__(
|
||
fmt="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||
datefmt="%Y-%m-%d %H:%M:%S",
|
||
)
|
||
|
||
|
||
def setup_logging(level: str = "INFO", json_format: bool = False, log_dir: Optional[str] = None) -> None:
|
||
"""配置日志系统
|
||
|
||
Args:
|
||
level: 日志级别 (DEBUG, INFO, WARNING, ERROR, CRITICAL)
|
||
json_format: 是否使用 JSON 格式输出
|
||
log_dir: 运行期日志文件目录。若目录存在或可被创建，则在 stdout handler
|
||
之外额外写入 <log_dir>/wecom-it-desk.log
|
||
（RotatingFileHandler，单文件上限 20MB，保留 5 个备份，复用 JSONFormatter）。
|
||
为空 / 不可创建时仅保留 stdout handler，不影响启动。
|
||
"""
|
||
log_level = getattr(logging, level.upper(), logging.INFO)
|
||
|
||
# 获取 root logger
|
||
root_logger = logging.getLogger()
|
||
root_logger.setLevel(log_level)
|
||
|
||
# 清除现有 handlers
|
||
for handler in root_logger.handlers[:]:
|
||
root_logger.removeHandler(handler)
|
||
|
||
# 创建 console handler（保持原有 stdout 行为不变）
|
||
console_handler = logging.StreamHandler(sys.stdout)
|
||
console_handler.setLevel(log_level)
|
||
|
||
# 设置 formatter
|
||
if json_format:
|
||
formatter = JSONFormatter()
|
||
else:
|
||
formatter = PlainFormatter()
|
||
|
||
console_handler.setFormatter(formatter)
|
||
root_logger.addHandler(console_handler)
|
||
|
||
# 运行期日志文件 handler（独立于 stdout，便于"运行期日志"管理页面读取）
|
||
# 始终使用 JSONFormatter，保证日志文件为结构化 JSON 行，便于解析与下载。
|
||
if log_dir:
|
||
try:
|
||
os.makedirs(log_dir, exist_ok=True)
|
||
log_file = os.path.join(log_dir, "wecom-it-desk.log")
|
||
file_handler = RotatingFileHandler(
|
||
log_file,
|
||
maxBytes=20 * 1024 * 1024, # 单文件上限 20MB
|
||
backupCount=5, # 保留 5 个轮转备份
|
||
encoding="utf-8",
|
||
)
|
||
file_handler.setLevel(log_level)
|
||
file_handler.setFormatter(JSONFormatter())
|
||
root_logger.addHandler(file_handler)
|
||
except (OSError, PermissionError) as e:
|
||
# 目录不可创建 / 不可写：不阻塞启动，仅记录告警，降级为仅 stdout 日志
|
||
logging.getLogger(__name__).warning(
|
||
f"无法为运行期日志创建文件 handler（log_dir={log_dir}）: {e}；仅保留 stdout 日志。"
|
||
)
|
||
|
||
# 设置第三方库日志级别
|
||
logging.getLogger("uvicorn").setLevel(logging.WARNING)
|
||
logging.getLogger("fastapi").setLevel(logging.WARNING)
|
||
logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
|
||
|
||
|
||
def get_logger(name: str) -> logging.Logger:
|
||
"""获取 logger 实例
|
||
|
||
Args:
|
||
name: logger 名称，通常使用 __name__
|
||
|
||
Returns:
|
||
Logger 实例
|
||
"""
|
||
return logging.getLogger(name)
|