100 lines
3.4 KiB
Python
100 lines
3.4 KiB
Python
"""日志配置模块"""
|
|
|
|
import sys
|
|
import logging
|
|
import re
|
|
from logging.handlers import TimedRotatingFileHandler
|
|
from pathlib import Path
|
|
|
|
from loguru import logger
|
|
|
|
|
|
_LOG_LEVELS = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
|
|
|
|
|
|
def _normalize_log_level(level: str | None) -> str:
|
|
"""规范化环境变量日志等级，非法值回退到 INFO。"""
|
|
normalized = str(level or "INFO").strip().upper()
|
|
return normalized if normalized in _LOG_LEVELS else "INFO"
|
|
|
|
|
|
class _RelevantAccessFilter(logging.Filter):
|
|
"""仅保留异常 HTTP 请求，避免前端轮询淹没业务日志。"""
|
|
|
|
_STATUS_RE = re.compile(r'"\s+(\d{3})\b')
|
|
|
|
def filter(self, record: logging.LogRecord) -> bool:
|
|
if record.levelno >= logging.WARNING:
|
|
return True
|
|
matched = self._STATUS_RE.search(record.getMessage())
|
|
return bool(matched and int(matched.group(1)) >= 400)
|
|
|
|
|
|
def _configure_standard_logging(level: str, log_dir: Path) -> logging.Logger:
|
|
"""将业务日志和关键框架日志收敛到单一轮转文件。"""
|
|
log_level = getattr(logging, level.upper(), logging.INFO)
|
|
formatter = logging.Formatter(
|
|
"%(asctime)s | %(levelname)-8s | %(name)s | %(message)s"
|
|
)
|
|
|
|
console_handler = logging.StreamHandler(sys.stdout)
|
|
console_handler.setLevel(log_level)
|
|
console_handler.setFormatter(formatter)
|
|
|
|
file_handler = TimedRotatingFileHandler(
|
|
log_dir / "app.log", when="midnight", backupCount=7, encoding="utf-8"
|
|
)
|
|
file_handler.setLevel(log_level)
|
|
file_handler.setFormatter(formatter)
|
|
|
|
app_logger = logging.getLogger("app")
|
|
app_logger.handlers.clear()
|
|
app_logger.handlers = [console_handler, file_handler]
|
|
app_logger.setLevel(log_level)
|
|
app_logger.propagate = False
|
|
|
|
for name in ("uvicorn", "uvicorn.error", "fastapi", "starlette"):
|
|
standard_logger = logging.getLogger(name)
|
|
standard_logger.handlers = [console_handler, file_handler]
|
|
standard_logger.setLevel(log_level)
|
|
standard_logger.propagate = False
|
|
|
|
access_logger = logging.getLogger("uvicorn.access")
|
|
access_logger.handlers = [console_handler, file_handler]
|
|
access_logger.filters = [_RelevantAccessFilter()]
|
|
access_logger.setLevel(logging.INFO)
|
|
access_logger.propagate = False
|
|
return app_logger
|
|
|
|
|
|
def setup_logger(level: str = "INFO", log_dir: str = None, log_file: str = None) -> None:
|
|
"""
|
|
配置日志
|
|
|
|
Args:
|
|
level: 日志级别
|
|
log_dir: 日志目录路径（写入 app.log，按日轮转）
|
|
log_file: 日志文件路径（兼容旧接口，优先级低于 log_dir）
|
|
"""
|
|
level = _normalize_log_level(level)
|
|
# 移除默认 handler，由标准 logging 统一处理控制台和文件输出。
|
|
logger.remove()
|
|
|
|
# 确定日志文件路径
|
|
if log_dir:
|
|
dir_path = Path(log_dir)
|
|
dir_path.mkdir(parents=True, exist_ok=True)
|
|
app_logger = _configure_standard_logging(level, dir_path)
|
|
elif log_file:
|
|
file_path = Path(log_file)
|
|
file_path.parent.mkdir(parents=True, exist_ok=True)
|
|
app_logger = _configure_standard_logging(level, file_path.parent)
|
|
else:
|
|
return
|
|
|
|
def forward_to_standard(message) -> None:
|
|
record = message.record
|
|
app_logger.log(record["level"].no, f"{record['name']} | {record['message']}")
|
|
|
|
logger.add(forward_to_standard, level=level, format="{message}")
|