| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485 |
- """日志装配。
- 两个格式档位：
- - ``text``（默认）：人类可读，本地开发用 ``tail`` 看
- - ``json``：一行一条 JSON，便于采集到 ELK/Loki
- trace_id 通过 ``obs.tracing.get_trace_id`` 动态注入——用 filter 而非
- formatter 参数，是为了让每条日志自动带上当前上下文的 trace_id，调用方
- 不需要在每条 log 里手动传。
- """
- from __future__ import annotations
- import json
- import logging
- import sys
- from datetime import datetime
- from obs.tracing import get_trace_id
- _TEXT_FMT = "%(asctime)s %(levelname)-5s [%(trace_id)s] %(name)s | %(message)s"
- _DATE_FMT = "%Y-%m-%d %H:%M:%S"
- # 这些第三方 logger 在 INFO 级别过于聒噪，统一压到 WARNING
- _NOISY_LOGGERS = (
- "httpx",
- "httpcore",
- "urllib3",
- "apscheduler.executors.default",
- "apscheduler.scheduler",
- "openai",
- )
- class _TraceIdFilter(logging.Filter):
- """把当前上下文的 trace_id 注入 LogRecord，缺省为 '-'。"""
- def filter(self, record: logging.LogRecord) -> bool:
- record.trace_id = get_trace_id() or "-"
- return True
- class _JsonFormatter(logging.Formatter):
- """一行一条 JSON 的格式化器。"""
- def format(self, record: logging.LogRecord) -> str:
- payload = {
- "ts": datetime.fromtimestamp(record.created).isoformat(timespec="milliseconds"),
- "level": record.levelname,
- "logger": record.name,
- "trace_id": getattr(record, "trace_id", "-"),
- "message": record.getMessage(),
- }
- if record.exc_info:
- payload["exc"] = self.formatException(record.exc_info)
- # 允许调用方通过 logger.info("...", extra={"kv": {...}}) 附加结构化字段
- kv = getattr(record, "kv", None)
- if isinstance(kv, dict):
- payload.update(kv)
- return json.dumps(payload, ensure_ascii=False)
- def setup_logging(level: str = "INFO", fmt: str = "text") -> None:
- """初始化根 logger。重复调用安全（先清空已有 handler）。"""
- root = logging.getLogger()
- root.setLevel(getattr(logging, str(level).upper(), logging.INFO))
- for handler in list(root.handlers):
- root.removeHandler(handler)
- handler = logging.StreamHandler(sys.stdout)
- handler.addFilter(_TraceIdFilter())
- if str(fmt).lower() == "json":
- handler.setFormatter(_JsonFormatter())
- else:
- handler.setFormatter(logging.Formatter(_TEXT_FMT, datefmt=_DATE_FMT))
- root.addHandler(handler)
- for name in _NOISY_LOGGERS:
- logging.getLogger(name).setLevel(logging.WARNING)
- def get_logger(name: str) -> logging.Logger:
- """取 logger 的统一入口（留一层薄封装，将来换实现只改这里）。"""
- return logging.getLogger(name)
|
