選択できるのは25トピックまでです。 トピックは、先頭が英数字で、英数字とダッシュ('-')を使用した35文字以内のものにしてください。

3ヶ月前
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179
  1. import json
  2. import logging
  3. import os
  4. import pickle
  5. import copy
  6. from common.log import logger
  7. # from common.log import logger
  8. # 示例配置文件
  9. DEBUG = True
  10. available_setting = {
  11. "qwen_access_key_id": "",
  12. "qwen_access_key_secret": "",
  13. "debug": False,
  14. #redis 配置
  15. "redis_host":"",
  16. "redis_port":0,
  17. "redis_password":"",
  18. "redis_db":0,
  19. # kafka配置
  20. "kafka_bootstrap_servers":"",
  21. # aiops平台
  22. "aiops_api":""
  23. }
  24. class Config(dict):
  25. def __init__(self, d=None):
  26. super().__init__()
  27. if d is None:
  28. d = {}
  29. for k, v in d.items():
  30. self[k] = v
  31. # user_datas: 用户数据,key为用户名,value为用户数据,也是dict
  32. self.user_datas = {}
  33. def __getitem__(self, key):
  34. if key not in available_setting:
  35. raise Exception("key {} not in available_setting".format(key))
  36. return super().__getitem__(key)
  37. def __setitem__(self, key, value):
  38. if key not in available_setting:
  39. raise Exception("key {} not in available_setting".format(key))
  40. return super().__setitem__(key, value)
  41. def get(self, key, default=None):
  42. try:
  43. return self[key]
  44. except KeyError as e:
  45. return default
  46. except Exception as e:
  47. raise e
  48. config = Config()
  49. def drag_sensitive(config):
  50. try:
  51. if isinstance(config, str):
  52. conf_dict: dict = json.loads(config)
  53. conf_dict_copy = copy.deepcopy(conf_dict)
  54. for key in conf_dict_copy:
  55. if "key" in key or "secret" in key:
  56. if isinstance(conf_dict_copy[key], str):
  57. conf_dict_copy[key] = conf_dict_copy[key][0:3] + "*" * 5 + conf_dict_copy[key][-3:]
  58. return json.dumps(conf_dict_copy, indent=4)
  59. elif isinstance(config, dict):
  60. config_copy = copy.deepcopy(config)
  61. for key in config:
  62. if "key" in key or "secret" in key:
  63. if isinstance(config_copy[key], str):
  64. config_copy[key] = config_copy[key][0:3] + "*" * 5 + config_copy[key][-3:]
  65. return config_copy
  66. except Exception as e:
  67. logger.exception(e)
  68. return config
  69. return config
  70. def load_config():
  71. global config
  72. # config_path = "./config.json"
  73. # if not os.path.exists(config_path):
  74. # logger.info("配置文件不存在,将使用config-template.json模板")
  75. # config_path = "./config-template.json"
  76. # 默认加载 config.json 或者 config-template.json
  77. environment = os.environ.get('environment', 'default') # 默认是生产环境
  78. logger.info(f"当前环境: {environment}")
  79. if environment == "test":
  80. config_path = "./config-test.json"
  81. elif environment == "production":
  82. config_path = "./config-production.json"
  83. elif environment == "dev":
  84. config_path = "./config-dev.json"
  85. elif environment == "default":
  86. config_path = "./config.json"
  87. else:
  88. logger.error("无效的环境配置,使用默认的 config-template.json")
  89. config_path = "./config-template.json"
  90. # 加载配置文件
  91. if not os.path.exists(config_path):
  92. logger.info(f"配置文件 {config_path} 不存在,将使用 config-template.json 模板")
  93. config_path = "./config-template.json"
  94. config_str = read_file(config_path)
  95. logger.debug("[INIT] config str: {}".format(drag_sensitive(config_str)))
  96. # 将json字符串反序列化为dict类型
  97. config = Config(json.loads(config_str))
  98. # override config with environment variables.
  99. # 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.
  100. for name, value in os.environ.items():
  101. name = name.lower()
  102. if name in available_setting:
  103. logger.info("[INIT] override config by environ args: {}={}".format(name, value))
  104. try:
  105. config[name] = eval(value)
  106. except:
  107. if value == "false":
  108. config[name] = False
  109. elif value == "true":
  110. config[name] = True
  111. else:
  112. config[name] = value
  113. if config.get("debug", False):
  114. logger.setLevel(logging.DEBUG)
  115. logger.debug("[INIT] set log level to DEBUG")
  116. logger.info("[INIT] load config: {}".format(drag_sensitive(config)))
  117. def get_root():
  118. return os.path.dirname(os.path.abspath(__file__))
  119. def read_file(path):
  120. with open(path, mode="r", encoding="utf-8") as f:
  121. return f.read()
  122. def conf():
  123. return config
  124. # # global plugin config
  125. # plugin_config = {}
  126. # def write_plugin_config(pconf: dict):
  127. # """
  128. # 写入插件全局配置
  129. # :param pconf: 全量插件配置
  130. # """
  131. # global plugin_config
  132. # for k in pconf:
  133. # plugin_config[k.lower()] = pconf[k]
  134. # def pconf(plugin_name: str) -> dict:
  135. # """
  136. # 根据插件名称获取配置
  137. # :param plugin_name: 插件名称
  138. # :return: 该插件的配置项
  139. # """
  140. # return plugin_config.get(plugin_name.lower())
  141. # # 全局配置,用于存放全局生效的状态
  142. # global_config = {"admin_users": []}