# -*- coding: utf-8 -*- """ Dev Trace Log 配置模块 从环境变量读取 trace 相关配置,支持运行时动态修改(内存状态)。 重启后回退到 .env 配置值。使用单例模式,全局唯一实例。 环境变量缺失时使用合理默认值: - DEV_TRACE_ENABLED: false(默认关闭,显式开启才采集) - DEV_TRACE_LOG_DIR: export/dev-trace-logs - DEV_TRACE_LOG_RETENTION_DAYS: 7 - DEV_TRACE_LOG_SQL: true - DEV_TRACE_LOG_PARAMS: true """ from __future__ import annotations import os import threading from typing import Any def _parse_bool(value: str, default: bool) -> bool: """解析布尔型环境变量,支持 true/1/yes(不区分大小写)。""" if not value: return default return value.strip().lower() in ("true", "1", "yes") class TraceConfig: """全链路日志配置(单例)。 属性从环境变量初始化,运行时可通过 update_settings() 动态修改。 重启后自动回退到 .env 值(因为 __init__ 重新读取环境变量)。 """ _instance: TraceConfig | None = None _lock: threading.Lock = threading.Lock() def __new__(cls) -> TraceConfig: # 双重检查锁定,线程安全的单例 if cls._instance is None: with cls._lock: if cls._instance is None: instance = super().__new__(cls) instance._init_from_env() cls._instance = instance return cls._instance def _init_from_env(self) -> None: """从环境变量读取配置,作为基线值和当前运行时值。""" # 基线值(.env 原始值,reset_to_env 时回退到这里) self._env_enabled: bool = _parse_bool( os.environ.get("DEV_TRACE_ENABLED", ""), default=False ) raw_log_dir: str = os.environ.get( "DEV_TRACE_LOG_DIR", "export/dev-trace-logs" ) # CHANGE 2026-03-22 | trace日志路径修复 # 相对路径基于项目根目录解析,避免受后端 cwd(apps/backend/)影响 if not os.path.isabs(raw_log_dir): project_root = os.environ.get("NEOZQYY_ROOT", "") if project_root: raw_log_dir = os.path.join(project_root, raw_log_dir) self._env_log_dir: str = raw_log_dir self._env_retention_days: int = int( os.environ.get("DEV_TRACE_LOG_RETENTION_DAYS", "7") ) self._env_log_sql: bool = _parse_bool( os.environ.get("DEV_TRACE_LOG_SQL", ""), default=True ) self._env_log_params: bool = _parse_bool( os.environ.get("DEV_TRACE_LOG_PARAMS", ""), default=True ) # 运行时值(初始等于基线值,API 可动态修改) self.enabled: bool = self._env_enabled self.log_dir: str = self._env_log_dir self.retention_days: int = self._env_retention_days self.log_sql: bool = self._env_log_sql self.log_params: bool = self._env_log_params # ------------------------------------------------------------------ # 运行时开关方法 # ------------------------------------------------------------------ def update_settings( self, *, enabled: bool | None = None, retention_days: int | None = None, log_sql: bool | None = None, log_params: bool | None = None, ) -> None: """更新内存中的运行时设置(不写 .env,重启后回退)。""" if enabled is not None: self.enabled = enabled if retention_days is not None: self.retention_days = retention_days if log_sql is not None: self.log_sql = log_sql if log_params is not None: self.log_params = log_params def get_settings(self) -> dict[str, Any]: """返回当前运行时设置字典。""" return { "enabled": self.enabled, "log_dir": self.log_dir, "retention_days": self.retention_days, "log_sql": self.log_sql, "log_params": self.log_params, } def reset_to_env(self) -> None: """重置运行时值为 .env 基线值。""" self.enabled = self._env_enabled self.log_dir = self._env_log_dir self.retention_days = self._env_retention_days self.log_sql = self._env_log_sql self.log_params = self._env_log_params # ------------------------------------------------------------------ # 测试辅助:重置单例(仅测试使用) # ------------------------------------------------------------------ @classmethod def _reset_singleton(cls) -> None: """销毁单例实例,下次访问时重新从环境变量初始化。仅供测试使用。""" with cls._lock: cls._instance = None def get_trace_config() -> TraceConfig: """获取全局 TraceConfig 单例。""" return TraceConfig()