AiOpsWechat/config.py

import json
import logging
import os
import pickle
import copy

from common.log import logger

# from common.log import logger

# 示例配置文件
DEBUG = True

available_setting = {

    "qwen_access_key_id": "",
    "qwen_access_key_secret": "",

    "debug": False, 
    #redis 配置
    "redis_host":"",
    "redis_port":0,
    "redis_password":"",
    "redis_db":0,
    # kafka配置
    "kafka_bootstrap_servers":"",
    # aiops平台
    "aiops_api":"",
    "wx_chat_api":""
}

class Config(dict):
    def __init__(self, d=None):
        super().__init__()
        if d is None:
            d = {}
        for k, v in d.items():
            self[k] = v
        # user_datas: 用户数据，key为用户名，value为用户数据，也是dict
        self.user_datas = {}

    def __getitem__(self, key):
        if key not in available_setting:
            raise Exception("key {} not in available_setting".format(key))
        return super().__getitem__(key)

    def __setitem__(self, key, value):
        if key not in available_setting:
            raise Exception("key {} not in available_setting".format(key))
        return super().__setitem__(key, value)

    def get(self, key, default=None):
        try:
            return self[key]
        except KeyError as e:
            return default
        except Exception as e:
            raise e


config = Config()


def drag_sensitive(config):
    try:
        if isinstance(config, str):
            conf_dict: dict = json.loads(config)
            conf_dict_copy = copy.deepcopy(conf_dict)
            for key in conf_dict_copy:
                if "key" in key or "secret" in key:
                    if isinstance(conf_dict_copy[key], str):
                        conf_dict_copy[key] = conf_dict_copy[key][0:3] + "*" * 5 + conf_dict_copy[key][-3:]
            return json.dumps(conf_dict_copy, indent=4)

        elif isinstance(config, dict):
            config_copy = copy.deepcopy(config)
            for key in config:
                if "key" in key or "secret" in key:
                    if isinstance(config_copy[key], str):
                        config_copy[key] = config_copy[key][0:3] + "*" * 5 + config_copy[key][-3:]
            return config_copy
    except Exception as e:
        logger.exception(e)
        return config
    return config


def load_config():
    global config
    # config_path = "./config.json"
    # if not os.path.exists(config_path):
    #     logger.info("配置文件不存在，将使用config-template.json模板")
    #     config_path = "./config-template.json"

    # 默认加载 config.json 或者 config-template.json
    environment = os.environ.get('environment', 'default')  # 默认是生产环境
    logger.info(f"当前环境: {environment}")

    if environment == "test":
        config_path = "./config-test.json"
    elif environment == "production":
        config_path = "./config-production.json"
    elif environment == "dev":
        config_path = "./config-dev.json"
    elif environment == "default":
        config_path = "./config.json"
    else:
        logger.error("无效的环境配置，使用默认的 config-template.json")
        config_path = "./config-template.json"

    # 加载配置文件
    if not os.path.exists(config_path):
        logger.info(f"配置文件 {config_path} 不存在，将使用 config-template.json 模板")
        config_path = "./config-template.json"

    config_str = read_file(config_path)
    logger.debug("[INIT] config str: {}".format(drag_sensitive(config_str)))

    # 将json字符串反序列化为dict类型
    config = Config(json.loads(config_str))

    # override config with environment variables.
    # Some online deployment platforms (e.g. Railway) deploy project from github directly. So you shouldn't put your secrets like api key in a config file, instead use environment variables to override the default config.
    for name, value in os.environ.items():
        name = name.lower()
        if name in available_setting:
            logger.info("[INIT] override config by environ args: {}={}".format(name, value))
            try:
                config[name] = eval(value)
            except:
                if value == "false":
                    config[name] = False
                elif value == "true":
                    config[name] = True
                else:
                    config[name] = value

    if config.get("debug", False):
        logger.setLevel(logging.DEBUG)
        logger.debug("[INIT] set log level to DEBUG")

    logger.info("[INIT] load config: {}".format(drag_sensitive(config)))

def get_root():
    return os.path.dirname(os.path.abspath(__file__))


def read_file(path):
    with open(path, mode="r", encoding="utf-8") as f:
        return f.read()


def conf():
    return config

# # global plugin config
# plugin_config = {}


# def write_plugin_config(pconf: dict):
#     """
#     写入插件全局配置
#     :param pconf: 全量插件配置
#     """
#     global plugin_config
#     for k in pconf:
#         plugin_config[k.lower()] = pconf[k]


# def pconf(plugin_name: str) -> dict:
#     """
#     根据插件名称获取配置
#     :param plugin_name: 插件名称
#     :return: 该插件的配置项
#     """
#     return plugin_config.get(plugin_name.lower())


# # 全局配置，用于存放全局生效的状态
# global_config = {"admin_users": []}