|
|
|
|
|
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测试群"], |
|
"nick_name_black_list": [], |
|
"group_welcome_msg": "", |
|
"trigger_by_self": False, |
|
"text_to_image": "dall-e-2", |
|
"image_proxy": True, |
|
"image_create_prefix": ["画", "看", "找"], |
|
"concurrency_in_session": 1, |
|
"image_create_size": "256x256", |
|
"group_chat_exit_group": False, |
|
|
|
"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": 180, |
|
"timeout": 120, |
|
|
|
"baidu_wenxin_model": "eb-instant", |
|
"baidu_wenxin_api_key": "", |
|
"baidu_wenxin_secret_key": "", |
|
|
|
"xunfei_app_id": "", |
|
"xunfei_api_key": "", |
|
"xunfei_api_secret": "", |
|
|
|
"claude_api_cookie": "", |
|
"claude_uuid": "", |
|
|
|
"qwen_access_key_id": "", |
|
"qwen_access_key_secret": "", |
|
"qwen_agent_key": "", |
|
"qwen_app_id": "", |
|
"qwen_node_id": "", |
|
|
|
"wework_smart": True, |
|
|
|
"speech_recognition": True, |
|
"group_speech_recognition": False, |
|
"voice_reply_voice": False, |
|
"always_reply_voice": False, |
|
"voice_to_text": "openai", |
|
"text_to_voice": "openai", |
|
"text_to_voice_model": "tts-1", |
|
"tts_voice_id": "alloy", |
|
|
|
"baidu_app_id": "", |
|
"baidu_api_key": "", |
|
"baidu_secret_key": "", |
|
|
|
"baidu_dev_pid": "1536", |
|
|
|
"azure_voice_api_key": "", |
|
"azure_voice_region": "japaneast", |
|
|
|
"xi_api_key": "", |
|
"xi_voice_id": "", |
|
|
|
"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": "", |
|
|
|
|
|
"feishu_port": 80, |
|
"feishu_app_id": "", |
|
"feishu_app_secret": "", |
|
"feishu_token": "", |
|
"feishu_bot_name": "", |
|
|
|
|
|
"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": [] |
|
} |
|
|