126 lines
3.8 KiB
Python
126 lines
3.8 KiB
Python
"""统一日志模块"""
|
||
import os
|
||
import json
|
||
import logging
|
||
from datetime import datetime
|
||
from typing import Optional, Any
|
||
from functools import lru_cache
|
||
|
||
from .trace import get_trace_id, get_tenant_id, get_user_id
|
||
|
||
|
||
class PlatformLogger:
|
||
"""平台日志器
|
||
|
||
支持本地输出和远程上报两种模式
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
app_code: str,
|
||
platform_url: Optional[str] = None,
|
||
api_key: Optional[str] = None,
|
||
local_only: bool = True
|
||
):
|
||
self.app_code = app_code
|
||
self.platform_url = platform_url or os.getenv("PLATFORM_URL", "")
|
||
self.api_key = api_key or os.getenv("PLATFORM_API_KEY", "")
|
||
self.local_only = local_only or not self.platform_url
|
||
|
||
# 设置本地日志
|
||
self._logger = logging.getLogger(app_code)
|
||
self._logger.setLevel(logging.DEBUG)
|
||
|
||
if not self._logger.handlers:
|
||
handler = logging.StreamHandler()
|
||
handler.setFormatter(logging.Formatter(
|
||
'%(asctime)s | %(levelname)s | %(name)s | %(message)s'
|
||
))
|
||
self._logger.addHandler(handler)
|
||
|
||
def _format_message(self, message: str, **kwargs) -> str:
|
||
"""格式化日志消息，包含trace_id"""
|
||
trace_id = get_trace_id()
|
||
prefix = f"[{trace_id[:8]}] " if trace_id else ""
|
||
|
||
if kwargs:
|
||
extra = " | " + " ".join(f"{k}={v}" for k, v in kwargs.items())
|
||
else:
|
||
extra = ""
|
||
|
||
return f"{prefix}{message}{extra}"
|
||
|
||
def _log(
|
||
self,
|
||
level: str,
|
||
message: str,
|
||
log_type: str = "app",
|
||
category: Optional[str] = None,
|
||
context: Optional[dict] = None,
|
||
**kwargs
|
||
):
|
||
"""内部日志方法"""
|
||
formatted = self._format_message(message, **kwargs)
|
||
|
||
# 本地日志
|
||
log_method = getattr(self._logger, level.lower(), self._logger.info)
|
||
log_method(formatted)
|
||
|
||
# TODO: 远程上报（异步）
|
||
if not self.local_only:
|
||
self._send_to_platform(level, message, log_type, category, context, kwargs)
|
||
|
||
def _send_to_platform(
|
||
self,
|
||
level: str,
|
||
message: str,
|
||
log_type: str,
|
||
category: Optional[str],
|
||
context: Optional[dict],
|
||
extra: dict
|
||
):
|
||
"""发送日志到平台（异步，后续实现）"""
|
||
# TODO: 使用httpx异步发送
|
||
pass
|
||
|
||
def debug(self, message: str, **kwargs):
|
||
"""调试日志"""
|
||
self._log("debug", message, **kwargs)
|
||
|
||
def info(self, message: str, **kwargs):
|
||
"""信息日志"""
|
||
self._log("info", message, **kwargs)
|
||
|
||
def warn(self, message: str, **kwargs):
|
||
"""警告日志"""
|
||
self._log("warn", message, **kwargs)
|
||
|
||
def warning(self, message: str, **kwargs):
|
||
"""警告日志（别名）"""
|
||
self.warn(message, **kwargs)
|
||
|
||
def error(self, message: str, error: Optional[Exception] = None, **kwargs):
|
||
"""错误日志"""
|
||
if error:
|
||
kwargs["error_type"] = type(error).__name__
|
||
kwargs["error_msg"] = str(error)
|
||
self._log("error", message, log_type="error", **kwargs)
|
||
|
||
def audit(self, action: str, target_type: str, target_id: str, **kwargs):
|
||
"""审计日志"""
|
||
self._log(
|
||
"info",
|
||
f"AUDIT: {action} {target_type}:{target_id}",
|
||
log_type="audit",
|
||
action=action,
|
||
target_type=target_type,
|
||
target_id=target_id,
|
||
**kwargs
|
||
)
|
||
|
||
|
||
@lru_cache()
|
||
def get_logger(app_code: str) -> PlatformLogger:
|
||
"""获取日志器（单例）"""
|
||
return PlatformLogger(app_code)
|