DataMiddleware/SourceCode/DataMiddleware/tools/utils/config_helper.py

import os

import yaml


class ConfigHelper:
    _instance = None

    # 默认配置文件路径
    default_config_path = os.path.join(os.path.dirname(__file__), "..", "config.yml")

    # 类变量存储加载的配置
    _config = None
    _path = None

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            cls._instance = super(ConfigHelper, cls).__new__(cls)
        return cls._instance

    def load_config(self, path=None):
        if self._config is None:
            if not path:
                # print(f"使用默认配置文件：{self.default_config_path}")
                self._path = self.default_config_path
            else:
                self._path = path
            if not os.path.exists(self._path):
                raise FileNotFoundError(f"没有找到文件或目录：'{self._path}'")
        with open(self._path, "r", encoding="utf-8") as file:
            self._config = yaml.safe_load(file)
        # 合并环境变量配置
        self._merge_env_vars()
        # print(f"加载的配置文件内容：{self._config}")
        return self._config

    def _merge_env_vars(self, env_prefix="APP_"):  # 环境变量前缀为 APP_
        for key, value in os.environ.items():
            if key.startswith(env_prefix):
                config_key = key[len(env_prefix) :].lower()
                self._set_nested_key(self._config, config_key.split("__"), value)

    def _set_nested_key(self, config, keys, value):
        if len(keys) > 1:
            if keys[0] not in config or not isinstance(config[keys[0]], dict):
                config[keys[0]] = {}
            self._set_nested_key(config[keys[0]], keys[1:], value)
        else:
            config[keys[0]] = value

    def get(self, key: str, default: str = None):
        if self._config is None:
            self.load_config(self._path)
        keys = key.split(".")
        config = self._config
        for k in keys:
            if isinstance(config, dict) and k in config:
                config = config[k]
            else:
                return default
        return config

    def get_bool(self, key: str) -> bool:
        val = str(self.get(key, "0"))
        return True if val.lower() == "true" or val == "1" else False

    def get_int(self, key: str, default: int = 0) -> int:
        val = self.get(key)
        if not val:
            return default
        try:
            return int(val)
        except ValueError:
            return default

    def get_object(self, key: str, default: dict = None):
        val = self.get(key)
        if not val:
            return default
        if isinstance(val, dict):
            return val
        try:
            return yaml.safe_load(val)
        except yaml.YAMLError as e:
            print(f"Error loading YAML object: {e}")
            return default

    def get_all(self):
        if self._config is None:
            self.load_config(self._path)
        return self._config