|
-
-
- import json
- import logging
- import os
- import pickle
-
- from common.log import logger
-
-
-
- available_setting = {
-
- "open_ai_api_key": "",
-
- "open_ai_api_base": "https://api.openai.com/v1",
- "proxy": "",
-
- "model": "gpt-3.5-turbo",
- "use_azure_chatgpt": False,
- "azure_deployment_id": "",
- "azure_api_version": "",
-
- "single_chat_prefix": ["bot", "@bot"],
- "single_chat_reply_prefix": "[bot] ",
- "single_chat_reply_suffix": "",
- "group_chat_prefix": ["@bot"],
- "group_chat_reply_prefix": "",
- "group_chat_reply_suffix": "",
- "group_chat_keyword": [],
- "group_at_off": False,
- "group_name_white_list": ["ChatGPT测试群", "ChatGPT测试群2"],
- "group_name_keyword_white_list": [],
- "group_chat_in_one_session": ["ChatGPT测试群"],
- "trigger_by_self": False,
- "image_create_prefix": ["画", "看", "找"],
- "concurrency_in_session": 1,
- "image_create_size": "256x256",
-
- "expires_in_seconds": 3600,
-
- "character_desc": "你是ChatGPT, 一个由OpenAI训练的大型语言模型, 你旨在回答并解决人们的任何问题,并且可以使用多种语言与人交流。",
- "conversation_max_tokens": 1000,
-
- "rate_limit_chatgpt": 20,
- "rate_limit_dalle": 50,
-
- "temperature": 0.9,
- "top_p": 1,
- "frequency_penalty": 0,
- "presence_penalty": 0,
- "request_timeout": 60,
- "timeout": 120,
-
- "baidu_wenxin_model": "eb-instant",
- "baidu_wenxin_api_key": "",
- "baidu_wenxin_secret_key": "",
-
- "speech_recognition": False,
- "group_speech_recognition": False,
- "voice_reply_voice": False,
- "always_reply_voice": False,
- "voice_to_text": "openai",
- "text_to_voice": "baidu",
-
- "baidu_app_id": "",
- "baidu_api_key": "",
- "baidu_secret_key": "",
-
- "baidu_dev_pid": "1536",
-
- "azure_voice_api_key": "",
- "azure_voice_region": "japaneast",
-
- "chat_time_module": False,
- "chat_start_time": "00:00",
- "chat_stop_time": "24:00",
-
- "translate": "baidu",
-
- "baidu_translate_app_id": "",
- "baidu_translate_app_key": "",
-
- "hot_reload": False,
-
- "wechaty_puppet_service_token": "",
-
- "wechatmp_token": "",
- "wechatmp_port": 8080,
- "wechatmp_app_id": "",
- "wechatmp_app_secret": "",
- "wechatmp_aes_key": "",
-
- "wechatcom_corp_id": "",
-
- "wechatcomapp_token": "",
- "wechatcomapp_port": 9898,
- "wechatcomapp_secret": "",
- "wechatcomapp_agent_id": "",
- "wechatcomapp_aes_key": "",
-
- "clear_memory_commands": ["#清除记忆"],
-
- "channel_type": "wx",
- "subscribe_msg": "",
- "debug": False,
- "appdata_dir": "",
-
- "plugin_trigger_prefix": "$",
-
- "use_global_plugin_config": False,
-
- "use_linkai": False,
- "linkai_api_key": "",
- "linkai_app_code": "",
- "linkai_api_base": "https://api.link-ai.chat"
- }
-
-
- class Config(dict):
- def __init__(self, d=None):
- super().__init__()
- if d is None:
- d = {}
- for k, v in d.items():
- self[k] = v
-
- 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
-
-
- def get_user_data(self, user) -> dict:
- if self.user_datas.get(user) is None:
- self.user_datas[user] = {}
- return self.user_datas[user]
-
- def load_user_datas(self):
- try:
- with open(os.path.join(get_appdata_dir(), "user_datas.pkl"), "rb") as f:
- self.user_datas = pickle.load(f)
- logger.info("[Config] User datas loaded.")
- except FileNotFoundError as e:
- logger.info("[Config] User datas file not found, ignore.")
- except Exception as e:
- logger.info("[Config] User datas error: {}".format(e))
- self.user_datas = {}
-
- def save_user_datas(self):
- try:
- with open(os.path.join(get_appdata_dir(), "user_datas.pkl"), "wb") as f:
- pickle.dump(self.user_datas, f)
- logger.info("[Config] User datas saved.")
- except Exception as e:
- logger.info("[Config] User datas error: {}".format(e))
-
-
- config = 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_str = read_file(config_path)
- logger.debug("[INIT] config str: {}".format(config_str))
-
-
- config = Config(json.loads(config_str))
-
-
-
- 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(config))
-
- config.load_user_datas()
-
-
- 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
-
-
- def get_appdata_dir():
- data_path = os.path.join(get_root(), conf().get("appdata_dir", ""))
- if not os.path.exists(data_path):
- logger.info("[INIT] data path not exists, create it: {}".format(data_path))
- os.makedirs(data_path)
- return data_path
-
-
- def subscribe_msg():
- trigger_prefix = conf().get("single_chat_prefix", [""])[0]
- msg = conf().get("subscribe_msg", "")
- return msg.format(trigger_prefix=trigger_prefix)
-
-
-
- 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": []
- }
|