ChatGPT-on-WeChat配置教程

# encoding:utf-8

import json
import logging
import os
import pickle
import copy

from common.log import logger

# 将所有可用的配置项写在字典里, 请使用小写字母
# 此处的配置值无实际意义，程序不会读取此处的配置，仅用于提示格式，请将配置加入到 config.json 中
available_setting = {
    # OpenAI API 配置
    "open_ai_api_key": "sk-xxxxxx",  # OpenAI API Key
    # 当 use_azure_chatgpt 为 True 时，需要设置对应的 API Base URL
    "open_ai_api_base": "https://api.n1n.ai/v1",
    "proxy": "",  # OpenAI 使用的代理
    # ChatGPT 模型， 当 use_azure_chatgpt 为True 时，其名称为 Azure 上 model deployment 名称
    "model": "gpt-3.5-turbo",  # 可选择: gpt-4o, pt-4o-mini, gpt-4-turbo, claude-3-sonnet, wenxin, moonshot, qwen-turbo, xunfei, glm-4, minimax, gemini 等模型，全部可选模型详见 common/const.py 文件
    "bot_type": "",  # 可选配置，使用兼容 OpenAI 格式的三方服务时候，需填"ChatGPT"。Bot 具体名称详见 common/const.py 文件列出的 bot_type，如不填根据 model 名称判断，
    "use_azure_chatgpt": False,  # 是否使用 azure 的 ChatGPT
    "azure_deployment_id": "",  # Azure 模型部署名称
    "azure_api_version": "",  # Azure API 版本
    # Bot触发配置
    "single_chat_prefix": ["bot", "@bot"],  # 私聊时文本需要包含该前缀才能触发机器人回复
    "single_chat_reply_prefix": "[bot] ",  # 私聊时自动回复的前缀，用于区分真人
    "single_chat_reply_suffix": "",  # 私聊时自动回复的后缀，\n 可以换行
    "group_chat_prefix": ["@bot"],  # 群聊时包含该前缀则会触发机器人回复
    "no_need_at": False,  # 群聊回复时是否不需要艾特
    "group_chat_reply_prefix": "",  # 群聊时自动回复的前缀
    "group_chat_reply_suffix": "",  # 群聊时自动回复的后缀，\n 可以换行
    "group_chat_keyword": [],  # 群聊时包含该关键词则会触发机器人回复
    "group_at_off": False,  # 是否关闭群聊时@bot的触发
    "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",  # 图片生成模型，可选 dall-e-2, dall-e-3
    # Azure OpenAI dall-e-3 配置
    "dalle3_image_style": "vivid", # 图片生成 dalle3 的风格，可选有 vivid, natural
    "dalle3_image_quality": "hd", # 图片生成 dalle3 的质量，可选有 standard, hd
    # Azure OpenAI DALL-E API 配置, 当 use_azure_chatgpt 为 True 时,用于将文字回复的资源和 Dall-E 的资源分开.
    "azure_openai_dalle_api_base": "", # [可选] Azure OpenAI 用于回复图片的资源  endpoint，默认使用 open_ai_api_base
    "azure_openai_dalle_api_key": "", # [可选] Azure OpenAI 用于回复图片的资源 key，默认使用 open_ai_api_key
    "azure_openai_dalle_deployment_id":"", # [可选] Azure OpenAI 用于回复图片的资源  Deployment ID，默认使用 text_to_image
    "image_proxy": True,  # 是否需要图片代理，国内访问 LinkAI 时需要
    "image_create_prefix": ["画", "看", "找"],  # 开启图片回复的前缀
    "concurrency_in_session": 1,  # 同一会话最多有多少条消息在处理中，大于 1 可能乱序
    "image_create_size": "256x256",  # 图片大小,可选有 256x256, 512x512, 1024x1024 (dall-e-3 默认为 1024x1024)
    "group_chat_exit_group": False,
    # ChatGPT 会话参数
    "expires_in_seconds": 3600,  # 无操作会话的过期时间
    # 人格描述
    "character_desc": "你是 ChatGPT, 一个由 OpenAI 训练的大型语言模型, 你旨在回答并解决人们的任何问题，并且可以使用多种语言与人交流。",
    "conversation_max_tokens": 1000,  # 支持上下文记忆的最多字符数
    # chatgpt限流配置
    "rate_limit_chatgpt": 20,  # ChatGPT 的调用频率限制
    "rate_limit_dalle": 50,  # OpenAI Dalle 的调用频率限制
    # ChatGPT API 参数参考https://platform.openai.com/docs/api-reference/chat/create
    "temperature": 0.9,
    "top_p": 1,
    "frequency_penalty": 0,
    "presence_penalty": 0,
    "request_timeout": 180,  # ChatGPT 请求超时时间，OpenAI 接口默认设置为 600，对于难问题一般需要较长时间
    "timeout": 120,  # ChatGPT 重试超时时间，在这个时间内，将会自动重试
    # Baidu 文心一言参数
    "baidu_wenxin_model": "eb-instant",  # 默认使用 ERNIE-Bot-turbo 模型
    "baidu_wenxin_api_key": "",  # Baidu API Key
    "baidu_wenxin_secret_key": "",  # Baidu Secret Key
    "baidu_wenxin_prompt_enabled": False,  # Enable prompt if you are using ernie character model
    # 讯飞星火 API
    "xunfei_app_id": "",  # 讯飞应用 ID
    "xunfei_api_key": "",  # 讯飞 API Key
    "xunfei_api_secret": "",  # 讯飞 API Secret
    "xunfei_domain": "",  # 讯飞模型对应的 Domain 参数，Spark4.0 Ultra 为 4.0Ultra，其他模型详见: https://www.xfyun.cn/doc/spark/Web.html
    "xunfei_spark_url": "",  # 讯飞模型对应的请求地址，Spark4.0 Ultra 为 wss://spark-api.xf-yun.com/v4.0/chat，其他模型参考详见: https://www.xfyun.cn/doc/spark/Web.html
    # Claude 配置
    "claude_api_cookie": "",
    "claude_uuid": "",
    # Claude API Key
    "claude_api_key": "",
    # 通义千问 API, 获取方式查看文档 https://help.aliyun.com/document_detail/2587494.html
    "qwen_access_key_id": "",
    "qwen_access_key_secret": "",
    "qwen_agent_key": "",
    "qwen_app_id": "",
    "qwen_node_id": "",  # 流程编排模型用到的 ID，如果没有用到 qwen_node_id，请务必保持为空字符串
    # 阿里灵积（通义新版 sdk）模型 API Key
    "dashscope_api_key": "",
    # Google Gemini API Key
    "gemini_api_key": "",
    # wework 的通用配置
    "wework_smart": True,  # 配置 wework 是否使用已登录的企业微信，False 为多开
    # 语音设置
    "speech_recognition": True,  # 是否开启语音识别
    "group_speech_recognition": False,  # 是否开启群组语音识别
    "voice_reply_voice": False,  # 是否使用语音回复语音，需要设置对应语音合成引擎的 API Key
    "always_reply_voice": False,  # 是否一直使用语音回复
    "voice_to_text": "openai",  # 语音识别引擎，支持 OpenAI, Baidu, Google, Azure, Xunfei, Ali
    "text_to_voice": "openai",  # 语音合成引擎，支持 OpenAI, Baidu, Google, Azure, Xunfei, Ali, Pytts (offline), Elevenlabs, Edge (online)
    "text_to_voice_model": "tts-1",
    "tts_voice_id": "alloy",
    # Baidu 语音 API 配置， 使用百度语音识别和语音合成时需要
    "baidu_app_id": "",
    "baidu_api_key": "",
    "baidu_secret_key": "",
    # 1536 普通话（支持简单的英文识别）1737 英语 1637 粤语 1837 四川话 1936 普通话远场
    "baidu_dev_pid": 1536,
    # azure 语音 API 配置， 使用 Azure 语音识别和语音合成时需要
    "azure_voice_api_key": "",
    "azure_voice_region": "japaneast",
    # elevenlabs 语音 API 配置
    "xi_api_key": "",  # 获取 API 的方法可以参考https://docs.elevenlabs.io/api-reference/quick-start/authentication
    "xi_voice_id": "",  # ElevenLabs 提供了 9 种英式、美式等英语发音 ID，分别是“Adam/Antoni/Arnold/Bella/Domi/Elli/Josh/Rachel/Sam”
    # 服务时间限制，目前支持 itchat
    "chat_time_module": False,  # 是否开启服务时间限制
    "chat_start_time": "00:00",  # 服务开始时间
    "chat_stop_time": "24:00",  # 服务结束时间
    # 翻译 API
    "translate": "baidu",  # 翻译 API，支持Baidu
    # Baidu 翻译 API 的配置
    "baidu_translate_app_id": "",  # 百度翻译 API 的 Appid
    "baidu_translate_app_key": "",  # 百度翻译 API 的秘钥
    # itchat的配置
    "hot_reload": False,  # 是否开启热重载
    # wechaty 的配置
    "wechaty_puppet_service_token": "",  # wechaty 的 token
    # wechatmp的配置
    "wechatmp_token": "",  # 微信公众平台的 Token
    "wechatmp_port": 8080,  # 微信公众平台的端口，需要端口转发到 80 或 443
    "wechatmp_app_id": "",  # 微信公众平台 AppID
    "wechatmp_app_secret": "",  # 微信公众平台的 AppSecret
    "wechatmp_aes_key": "",  # 微信公众平台的EncodingAESKey，加密模式需要
    # wechatcom 的通用配置
    "wechatcom_corp_id": "",  # 企业微信公司的 CorpID
    # wechatcomapp 的配置
    "wechatcomapp_token": "",  # 企业微信 App 的 token
    "wechatcomapp_port": 9898,  # 企业微信 App 的服务端口，不需要端口转发
    "wechatcomapp_secret": "",  # 企业微信 App 的 Secret
    "wechatcomapp_agent_id": "",  # 企业微信 App 的 agent_id
    "wechatcomapp_aes_key": "",  # 企业微信 App的 aes_key
    # 飞书配置
    "feishu_port": 80,  # 飞书 Bot 监听端口
    "feishu_app_id": "",  # 飞书机器人应用 APP ID
    "feishu_app_secret": "",  # 飞书机器人 APP Secret
    "feishu_token": "",  # 飞书 Verification Token
    "feishu_bot_name": "",  # 飞书机器人的名字
    # 钉钉配置
    "dingtalk_client_id": "",  # 钉钉机器人 Client ID 
    "dingtalk_client_secret": "",  # 钉钉机器人 Client Secret
    "dingtalk_card_enabled": False,
    
    # ChatGPT 指令自定义触发词
    "clear_memory_commands": ["#清除记忆"],  # 重置会话指令，必须以#开头
    # Channel 配置
    "channel_type": "",  # 通道类型，支持：{wx, wxy, terminal, wechatmp, wechatmp_service, wechatcom_app, dingtalk}
    "subscribe_msg": "",  # 订阅消息, 支持: wechatmp, wechatmp_service, wechatcom_app
    "debug": False,  # 是否开启 Debug 模式，开启后会打印更多日志
    "appdata_dir": "",  # 数据目录
    # 插件配置
    "plugin_trigger_prefix": "$",  # 规范插件提供聊天相关指令的前缀，建议不要和管理员指令前缀"#"冲突
    # 是否使用全局插件配置
    "use_global_plugin_config": False,
    "max_media_send_count": 3,  # 单次最大发送媒体资源的个数
    "media_send_interval": 1,  # 发送图片的事件间隔，单位秒
    # 智谱 AI 平台配置
    "zhipu_ai_api_key": "",
    "zhipu_ai_api_base": "https://open.bigmodel.cn/api/paas/v4",
    "moonshot_api_key": "",
    "moonshot_base_url": "https://api.moonshot.cn/v1/chat/completions",
    # LinkAI 平台配置
    "use_linkai": False,
    "linkai_api_key": "",
    "linkai_app_code": "",
    "linkai_api_base": "https://api.link-ai.tech",  # linkAI 服务地址
    "Minimax_api_key": "",
    "Minimax_group_id": "",
    "Minimax_base_url": "",
}


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

    # Make sure to return a dictionary to ensure atomic
    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 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_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)))

    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)


# 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": []}