"""统一日志配置。 业务代码用 `logger = logging.getLogger("shagua.xxx")` 即可, 本模块在 main.py 启动时调一次。 - 控制台(stdout): 人类可读文本, 给 systemd / 本地看; 有 trace 时行尾附 `trace=xxx`。 - 文件 `logs/app-server.log`: 单行 JSON, 供**阿里云 SLS / Logtail** 采集(JSON 模式零正则); 异常栈内嵌为字段不换行 → 每条日志一行。 - **结构化字段(SLS 可直接查/聚合)**: - `trace_id`: 请求级贯穿——在入口 `trace_id_ctx.set(...)` 后, 本请求内**每一行日志**(含 run_in_threadpool 里的 harvest, contextvars 自动拷进线程)都自动带上, 无需手写。 SLS 里 `trace_id: "xxx"` 一查即得整条比价链路, 按 time 升序即请求顺序。 - 任意 `logger.info(msg, extra={"phase": ..., "step": ..., "command": ...})` 的 extra 键都会平铺进 JSON 顶层 → SLS 可按 phase/step/command/cost_ms 等过滤聚合。 - 环境变量: - LOG_JSON_CONSOLE=1 控制台也输出 JSON - LOG_DIR / LOG_FILE 改落盘路径(默认 logs/app-server.log) ⚠️ admin 子进程(app.admin.main, 独立进程)应设不同 LOG_FILE, 避免与主进程争抢同一轮转文件 - LOG_SERVICE_NAME JSON 里的 service 字段(默认 app-server) """ from __future__ import annotations import json import logging import os import shutil import sys from contextvars import ContextVar from datetime import datetime from logging.handlers import RotatingFileHandler from pathlib import Path # 请求级 trace_id:入口(如 compare.py 透传壳)set 之后, 本请求上下文(含 run_in_threadpool # 拷贝出去的线程)内所有日志自动带上。默认空串 = 非请求上下文(启动/后台 worker)。 trace_id_ctx: ContextVar[str] = ContextVar("trace_id", default="") # 标准 LogRecord 属性 + 格式化期附加项:凡不在此集合的 record 属性都视为业务 extra, 平铺进 JSON。 _RESERVED = set( logging.LogRecord("", 0, "", 0, "", (), None).__dict__ ) | {"message", "asctime", "trace_id", "taskName"} class _ContextFilter(logging.Filter): """把 trace_id_ctx 注入每条 record(供两个 formatter 取用)。挂在 handler 上, 命中每条(含 propagate 上来的)记录, 在 format 之前置好 record.trace_id。""" def filter(self, record: logging.LogRecord) -> bool: if not hasattr(record, "trace_id"): record.trace_id = trace_id_ctx.get() return True class JsonFormatter(logging.Formatter): """LogRecord → 单行 JSON(SLS/Logtail 友好)。trace_id 提到顶层、extra 键平铺, 异常栈内嵌。""" def __init__(self, service: str = "app-server"): super().__init__() self.service = service def format(self, record: logging.LogRecord) -> str: dt = datetime.fromtimestamp(record.created) data = { "time": dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{int(record.msecs):03d}", "level": record.levelname, "service": self.service, "logger": record.name, } tid = getattr(record, "trace_id", "") or trace_id_ctx.get() if tid: data["trace_id"] = tid # 业务 extra 字段(phase / step / endpoint / command / cost_ms / status ...)平铺进顶层 for k, v in record.__dict__.items(): if k not in _RESERVED and not k.startswith("_"): data[k] = v data["func"] = record.funcName data["line"] = record.lineno data["message"] = record.getMessage() if record.exc_info: data["exception"] = self.formatException(record.exc_info) if record.stack_info: data["stack"] = self.formatStack(record.stack_info) return json.dumps(data, ensure_ascii=False, default=str) class TextFormatter(logging.Formatter): """控制台文本:标准行 + 有 trace_id 时行尾附 `trace=xxx`(无 trace 的启动/后台日志不加噪)。""" def format(self, record: logging.LogRecord) -> str: base = super().format(record) tid = getattr(record, "trace_id", "") or trace_id_ctx.get() return f"{base} trace={tid}" if tid else base class SafeRotatingFileHandler(RotatingFileHandler): """Windows 下不会被外部句柄卡死的 RotatingFileHandler。 stdlib 轮转靠 rename 活动文件(app-server.log → .1);Windows 只要有别的句柄(IDE 索引、 app.admin.main 第二进程、残留 --reload worker、杀软扫描)开着它, rename 就 WinError 32, 轮转永久卡死——文件停在 maxBytes、之后每条日志被丢。这里 Windows 改用 copytruncate:把活动 文件拷进备份、再通过自己的句柄原地清空, 从不 rename 活动文件, 故外部句柄开着也能转。 POSIX(生产 Linux)rename 打开中的文件本就合法, 保留 stdlib 的原子轮转不变。 代价:copytruncate 在“拷贝→清空”极窄窗口内并发写可能丢几行(仅跨进程;同进程 emit 有 handler 锁串行, 无此问题)。对本地开发日志可接受。 """ def doRollover(self) -> None: if os.name != "nt": super().doRollover() return if self.stream is None: self.stream = self._open() else: self.stream.flush() try: self._copytruncate_backups() except OSError: # 备份腾挪是尽力而为:任一备份被占用也绝不能挡住下面的清空, 否则活动文件继续涨、 # 轮转又卡死——那就白改了。 pass # 通过自己独占的句柄原地清空:不涉及 rename, 外部只读句柄不受影响。 self.stream.seek(0) self.stream.truncate() self.stream.flush() def _copytruncate_backups(self) -> None: """把 .N-1→.N 逐级腾挪, 再把活动文件拷到 .1(不动活动文件本身)。""" if self.backupCount <= 0: return for i in range(self.backupCount - 1, 0, -1): sfn = self.rotation_filename(f"{self.baseFilename}.{i}") dfn = self.rotation_filename(f"{self.baseFilename}.{i + 1}") if os.path.exists(sfn): if os.path.exists(dfn): os.remove(dfn) os.replace(sfn, dfn) dfn = self.rotation_filename(f"{self.baseFilename}.1") if os.path.exists(dfn): os.remove(dfn) shutil.copyfile(self.baseFilename, dfn) _CONFIGURED = False def setup_logging(debug: bool = False) -> None: global _CONFIGURED if _CONFIGURED: return level = logging.DEBUG if debug else logging.INFO service = os.getenv("LOG_SERVICE_NAME", "app-server") root = logging.getLogger() root.setLevel(level) # 清掉 basicConfig/uvicorn 可能预置的 root handler, 避免重复输出 for h in list(root.handlers): root.removeHandler(h) ctx_filter = _ContextFilter() # 控制台: 默认文本(systemd/本地看); LOG_JSON_CONSOLE=1 时输出 JSON console = logging.StreamHandler(sys.stdout) if os.getenv("LOG_JSON_CONSOLE") == "1": console.setFormatter(JsonFormatter(service)) else: console.setFormatter( TextFormatter("%(asctime)s %(levelname)s %(name)s: %(message)s") ) console.addFilter(ctx_filter) root.addHandler(console) # 文件: 单行 JSON, 供 Logtail 采集(自动轮转, 单文件 10MB, 保留 5 个) log_file = os.getenv("LOG_FILE") or str( Path(os.getenv("LOG_DIR", "logs")) / "app-server.log" ) Path(log_file).parent.mkdir(parents=True, exist_ok=True) file_handler = SafeRotatingFileHandler( log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8", ) file_handler.setFormatter(JsonFormatter(service)) file_handler.addFilter(ctx_filter) root.addHandler(file_handler) # 第三方库降噪 logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("httpcore").setLevel(logging.WARNING) # uvicorn --reload 的文件监视器: DEBUG 下会把每次"检测到变更"打出来。我们的 root 文件 # handler 又把这行写回 logs/app-server.log → watchfiles 再次检测 → 自我喂食死循环。 # 降到 WARNING 同时消除噪音和这个回环(真正的 .py 热重载不受影响)。 logging.getLogger("watchfiles").setLevel(logging.WARNING) _CONFIGURED = True