Duplicate from linhj07/chatgpt-on-wechat

Co-authored-by: Huijie Lin <[email protected]>

.gitattributes

@@ -0,0 +1,34 @@
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ckpt filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.mlmodel filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.npy filter=lfs diff=lfs merge=lfs -text
+*.npz filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pickle filter=lfs diff=lfs merge=lfs -text
+*.pkl filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+*.safetensors filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.wasm filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zst filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text

.github/ISSUE_TEMPLATE.md

@@ -0,0 +1,31 @@
+### 前置确认
+
+1. 网络能够访问openai接口
+2. python 已安装：版本在 3.7 ~ 3.10 之间
+3. `git pull` 拉取最新代码
+4. 执行`pip3 install -r requirements.txt`，检查依赖是否满足
+5. 拓展功能请执行`pip3 install -r requirements-optional.txt`，检查依赖是否满足
+6. 在已有 issue 中未搜索到类似问题
+7. [FAQS](https://github.com/zhayujie/chatgpt-on-wechat/wiki/FAQs) 中无类似问题
+
+
+### 问题描述
+
+> 简要说明、截图、复现步骤等，也可以是需求或想法
+
+
+
+
+### 终端日志 (如有报错)
+
+```
+[在此处粘贴终端日志, 可在主目录下`run.log`文件中找到]
+```
+
+
+
+### 环境
+
+ - 操作系统类型  (Mac/Windows/Linux)：
+ - Python版本  ( 执行 `python3 -V` )：                      
+ - pip版本  ( 依赖问题此项必填，执行 `pip3 -V`)：

.github/workflows/deploy-image.yml

@@ -0,0 +1,59 @@
+# This workflow uses actions that are not certified by GitHub.
+# They are provided by a third-party and are governed by
+# separate terms of service, privacy policy, and support
+# documentation.
+
+# GitHub recommends pinning actions to a commit SHA.
+# To get a newer version, you will need to update the SHA.
+# You can also reference a tag or branch, but the action may change without warning.
+
+name: Create and publish a Docker image
+
+on:
+  push:
+    branches: ['master']
+  create:
+env:
+  REGISTRY: ghcr.io
+  IMAGE_NAME: ${{ github.repository }}
+
+jobs:
+  build-and-push-image:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v3
+
+      - name: Log in to the Container registry
+        uses: docker/login-action@v2
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Extract metadata (tags, labels) for Docker
+        id: meta
+        uses: docker/metadata-action@v4
+        with:
+          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
+
+      - name: Build and push Docker image
+        uses: docker/build-push-action@v3
+        with:
+          context: .
+          push: true
+          file: ./docker/Dockerfile.latest
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+      
+      - uses: actions/delete-package-versions@v4
+        with: 
+          package-name: 'chatgpt-on-wechat'
+          package-type: 'container'
+          min-versions-to-keep: 10
+          delete-only-untagged-versions: 'true'
+          token: ${{ secrets.GITHUB_TOKEN }}

.gitignore

@@ -0,0 +1,14 @@
+.DS_Store
+.idea
+.wechaty/
+__pycache__/
+venv*
+*.pyc
+config.json
+QR.png
+nohup.out
+tmp
+plugins.json
+itchat.pkl
+*.log
+user_datas.pkl

Dockerfile

@@ -0,0 +1,3 @@
+FROM ghcr.io/zhayujie/chatgpt-on-wechat:latest
+
+ENTRYPOINT ["/entrypoint.sh"]

LICENSE

@@ -0,0 +1,19 @@
+Copyright (c) 2022 zhayujie
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,13 @@
+---
+title: wechat-bot
+emoji: 👀
+colorFrom: red
+colorTo: yellow
+sdk: gradio
+sdk_version: 3.19.1
+app_file: app.py
+pinned: false
+duplicated_from: linhj07/chatgpt-on-wechat
+---
+
+Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -0,0 +1,82 @@
+# encoding:utf-8
+
+import os
+from config import conf, load_config
+from channel import channel_factory
+from common.log import logger
+from plugins import *
+import signal
+import sys
+import config
+import gradio as gr
+from io import BytesIO
+from PIL import Image
+from concurrent.futures import ThreadPoolExecutor
+thread_pool = ThreadPoolExecutor(max_workers=8)
+
+def getImage(bytes):
+    bytes_stream = BytesIO(bytes)
+    image = Image.open(bytes_stream)
+    return image
+
+def getLoginUrl():
+    # load config
+    config.load_config()
+    # create channel
+    bot = channel_factory.create_channel("wx")
+    thread_pool.submit(bot.startup)
+    while (True):
+            if bot.getQrCode():
+                return getImage(bot.getQrCode())
+
+def sigterm_handler_wrap(_signo):
+    old_handler = signal.getsignal(_signo)
+    def func(_signo, _stack_frame):
+        logger.info("signal {} received, exiting...".format(_signo))
+        conf().save_user_datas()
+        return old_handler(_signo, _stack_frame)
+    signal.signal(_signo, func)
+
+def run():
+    try:
+        # load config
+        load_config()
+        # ctrl + c
+        sigterm_handler_wrap(signal.SIGINT)
+        # kill signal
+        sigterm_handler_wrap(signal.SIGTERM)
+
+        # create channel
+        channel_name=conf().get('channel_type', 'wx')
+        if channel_name == 'wxy':
+            os.environ['WECHATY_LOG']="warn"
+            # os.environ['WECHATY_PUPPET_SERVICE_ENDPOINT'] = '127.0.0.1:9001'
+
+        channel = channel_factory.create_channel(channel_name)
+        if channel_name in ['wx','wxy','wechatmp']:
+            PluginManager().load_plugins()
+
+        # startup channel
+        channel.startup()
+    except Exception as e:
+        logger.error("App startup failed!")
+        logger.exception(e)
+
+if __name__ == '__main__':
+    #run()
+    try:
+        
+        with gr.Blocks() as demo:
+            with gr.Row():
+                with gr.Column():
+                    btn = gr.Button(value="生成二维码")
+                with gr.Column():
+                    outputs=[gr.Pil()]
+                btn.click(getLoginUrl, outputs=outputs)
+
+        demo.launch()
+
+
+    except Exception as e:
+        logger.error("App startup failed!")
+        logger.exception(e)

bot/baidu/baidu_unit_bot.py

@@ -0,0 +1,28 @@
+# encoding:utf-8
+
+import requests
+from bot.bot import Bot
+from bridge.reply import Reply, ReplyType
+
+
+# Baidu Unit对话接口 (可用, 但能力较弱)
+class BaiduUnitBot(Bot):
+    def reply(self, query, context=None):
+        token = self.get_token()
+        url = 'https://aip.baidubce.com/rpc/2.0/unit/service/v3/chat?access_token=' + token
+        post_data = "{\"version\":\"3.0\",\"service_id\":\"S73177\",\"session_id\":\"\",\"log_id\":\"7758521\",\"skill_ids\":[\"1221886\"],\"request\":{\"terminal_id\":\"88888\",\"query\":\"" + query + "\", \"hyper_params\": {\"chat_custom_bot_profile\": 1}}}"
+        print(post_data)
+        headers = {'content-type': 'application/x-www-form-urlencoded'}
+        response = requests.post(url, data=post_data.encode(), headers=headers)
+        if response:
+            reply = Reply(ReplyType.TEXT, response.json()['result']['context']['SYS_PRESUMED_HIST'][1])
+            return reply
+
+    def get_token(self):
+        access_key = 'YOUR_ACCESS_KEY'
+        secret_key = 'YOUR_SECRET_KEY'
+        host = 'https://aip.baidubce.com/oauth/2.0/token?grant_type=client_credentials&client_id=' + access_key + '&client_secret=' + secret_key
+        response = requests.get(host)
+        if response:
+            print(response.json())
+            return response.json()['access_token']

bot/bot.py

@@ -0,0 +1,17 @@
+"""
+Auto-replay chat robot abstract class
+"""
+
+
+from bridge.context import Context
+from bridge.reply import Reply
+
+
+class Bot(object):
+    def reply(self, query, context : Context =None) -> Reply:
+        """
+        bot auto-reply content
+        :param req: received message
+        :return: reply content
+        """
+        raise NotImplementedError

bot/bot_factory.py

@@ -0,0 +1,32 @@
+"""
+channel factory
+"""
+from common import const
+
+
+def create_bot(bot_type):
+    """
+    create a bot_type instance
+    :param bot_type: bot type code
+    :return: bot instance
+    """
+    if bot_type == const.BAIDU:
+        # Baidu Unit对话接口
+        from bot.baidu.baidu_unit_bot import BaiduUnitBot
+        return BaiduUnitBot()
+
+    elif bot_type == const.CHATGPT:
+        # ChatGPT 网页端web接口
+        from bot.chatgpt.chat_gpt_bot import ChatGPTBot
+        return ChatGPTBot()
+
+    elif bot_type == const.OPEN_AI:
+        # OpenAI 官方对话模型API
+        from bot.openai.open_ai_bot import OpenAIBot
+        return OpenAIBot()
+
+    elif bot_type == const.CHATGPTONAZURE:
+        # Azure chatgpt service https://azure.microsoft.com/en-in/products/cognitive-services/openai-service/
+        from bot.chatgpt.chat_gpt_bot import AzureChatGPTBot
+        return AzureChatGPTBot()
+    raise RuntimeError

bot/chatgpt/chat_gpt_bot.py

@@ -0,0 +1,156 @@
+# encoding:utf-8
+
+from bot.bot import Bot
+from bot.chatgpt.chat_gpt_session import ChatGPTSession
+from bot.openai.open_ai_image import OpenAIImage
+from bot.session_manager import Session, SessionManager
+from bridge.context import ContextType
+from bridge.reply import Reply, ReplyType
+from config import conf, load_config
+from common.log import logger
+from common.token_bucket import TokenBucket
+from common.expired_dict import ExpiredDict
+import openai
+import openai.error
+import time
+
+# OpenAI对话模型API (可用)
+class ChatGPTBot(Bot,OpenAIImage):
+    def __init__(self):
+        super().__init__()
+        # set the default api_key
+        openai.api_key = conf().get('open_ai_api_key')
+        if conf().get('open_ai_api_base'):
+            openai.api_base = conf().get('open_ai_api_base')
+        proxy = conf().get('proxy')
+        if proxy:
+            openai.proxy = proxy
+        if conf().get('rate_limit_chatgpt'):
+            self.tb4chatgpt = TokenBucket(conf().get('rate_limit_chatgpt', 20))
+        
+        self.sessions = SessionManager(ChatGPTSession, model= conf().get("model") or "gpt-3.5-turbo")
+
+    def reply(self, query, context=None):
+        # acquire reply content
+        if context.type == ContextType.TEXT:
+            logger.info("[CHATGPT] query={}".format(query))
+
+
+            session_id = context['session_id']
+            reply = None
+            clear_memory_commands = conf().get('clear_memory_commands', ['#清除记忆'])
+            if query in clear_memory_commands:
+                self.sessions.clear_session(session_id)
+                reply = Reply(ReplyType.INFO, '记忆已清除')
+            elif query == '#清除所有':
+                self.sessions.clear_all_session()
+                reply = Reply(ReplyType.INFO, '所有人记忆已清除')
+            elif query == '#更新配置':
+                load_config()
+                reply = Reply(ReplyType.INFO, '配置已更新')
+            if reply:
+                return reply
+            session = self.sessions.session_query(query, session_id)
+            logger.debug("[CHATGPT] session query={}".format(session.messages))
+
+            api_key = context.get('openai_api_key')
+
+            # if context.get('stream'):
+            #     # reply in stream
+            #     return self.reply_text_stream(query, new_query, session_id)
+
+            reply_content = self.reply_text(session, session_id, api_key, 0)
+            logger.debug("[CHATGPT] new_query={}, session_id={}, reply_cont={}, completion_tokens={}".format(session.messages, session_id, reply_content["content"], reply_content["completion_tokens"]))
+            if reply_content['completion_tokens'] == 0 and len(reply_content['content']) > 0:
+                reply = Reply(ReplyType.ERROR, reply_content['content'])
+            elif reply_content["completion_tokens"] > 0:
+                self.sessions.session_reply(reply_content["content"], session_id, reply_content["total_tokens"])
+                reply = Reply(ReplyType.TEXT, reply_content["content"])
+            else:
+                reply = Reply(ReplyType.ERROR, reply_content['content'])
+                logger.debug("[CHATGPT] reply {} used 0 tokens.".format(reply_content))
+            return reply
+
+        elif context.type == ContextType.IMAGE_CREATE:
+            ok, retstring = self.create_img(query, 0)
+            reply = None
+            if ok:
+                reply = Reply(ReplyType.IMAGE_URL, retstring)
+            else:
+                reply = Reply(ReplyType.ERROR, retstring)
+            return reply
+        else:
+            reply = Reply(ReplyType.ERROR, 'Bot不支持处理{}类型的消息'.format(context.type))
+            return reply
+
+    def compose_args(self):
+        return {
+            "model": conf().get("model") or "gpt-3.5-turbo",  # 对话模型的名称
+            "temperature":conf().get('temperature', 0.9),  # 值在[0,1]之间，越大表示回复越具有不确定性
+            # "max_tokens":4096,  # 回复最大的字符数
+            "top_p":1,
+            "frequency_penalty":conf().get('frequency_penalty', 0.0),  # [-2,2]之间，该值越大则更倾向于产生不同的内容
+            "presence_penalty":conf().get('presence_penalty', 0.0),  # [-2,2]之间，该值越大则更倾向于产生不同的内容
+            "request_timeout": conf().get('request_timeout', 60),  # 请求超时时间，openai接口默认设置为600，对于难问题一般需要较长时间
+            "timeout": conf().get('request_timeout', 120), #重试超时时间，在这个时间内，将会自动重试
+        }
+
+    def reply_text(self, session:ChatGPTSession, session_id, api_key, retry_count=0) -> dict:
+        '''
+        call openai's ChatCompletion to get the answer
+        :param session: a conversation session
+        :param session_id: session id
+        :param retry_count: retry count
+        :return: {}
+        '''
+        try:
+            if conf().get('rate_limit_chatgpt') and not self.tb4chatgpt.get_token():
+                raise openai.error.RateLimitError("RateLimitError: rate limit exceeded")
+            # if api_key == None, the default openai.api_key will be used
+            response = openai.ChatCompletion.create(
+                api_key=api_key, messages=session.messages, **self.compose_args()
+            )
+            # logger.info("[ChatGPT] reply={}, total_tokens={}".format(response.choices[0]['message']['content'], response["usage"]["total_tokens"]))
+            return {"total_tokens": response["usage"]["total_tokens"],
+                    "completion_tokens": response["usage"]["completion_tokens"],
+                    "content": response.choices[0]['message']['content']}
+        except Exception as e:
+            need_retry = retry_count < 2
+            result = {"completion_tokens": 0, "content": "我现在有点累了，等会再来吧"}
+            if isinstance(e, openai.error.RateLimitError):
+                logger.warn("[CHATGPT] RateLimitError: {}".format(e))
+                result['content'] = "提问太快啦，请休息一下再问我吧"
+                if need_retry:
+                    time.sleep(5)
+            elif isinstance(e, openai.error.Timeout):
+                logger.warn("[CHATGPT] Timeout: {}".format(e))
+                result['content'] = "我没有收到你的消息"
+                if need_retry:
+                    time.sleep(5)
+            elif isinstance(e, openai.error.APIConnectionError):
+                logger.warn("[CHATGPT] APIConnectionError: {}".format(e))
+                need_retry = False
+                result['content'] = "我连接不到你的网络"
+            else:
+                logger.warn("[CHATGPT] Exception: {}".format(e))
+                need_retry = False
+                self.sessions.clear_session(session_id)
+
+            if need_retry:
+                logger.warn("[CHATGPT] 第{}次重试".format(retry_count+1))
+                return self.reply_text(session, session_id, api_key, retry_count+1)
+            else:
+                return result
+
+
+class AzureChatGPTBot(ChatGPTBot):
+    def __init__(self):
+        super().__init__()
+        openai.api_type = "azure"
+        openai.api_version = "2023-03-15-preview"
+
+    def compose_args(self):
+        args = super().compose_args()
+        args["engine"] = args["model"]
+        del(args["model"])
+        return args

bot/chatgpt/chat_gpt_session.py

@@ -0,0 +1,79 @@
+from bot.session_manager import Session
+from common.log import logger
+'''
+    e.g.  [
+        {"role": "system", "content": "You are a helpful assistant."},
+        {"role": "user", "content": "Who won the world series in 2020?"},
+        {"role": "assistant", "content": "The Los Angeles Dodgers won the World Series in 2020."},
+        {"role": "user", "content": "Where was it played?"}
+    ]
+'''
+class ChatGPTSession(Session):
+    def __init__(self, session_id, system_prompt=None, model= "gpt-3.5-turbo"):
+        super().__init__(session_id, system_prompt)
+        self.model = model
+        self.reset()
+    
+    def discard_exceeding(self, max_tokens, cur_tokens= None):
+        precise = True
+        try:
+            cur_tokens = num_tokens_from_messages(self.messages, self.model)
+        except Exception as e:
+            precise = False
+            if cur_tokens is None:
+                raise e
+            logger.debug("Exception when counting tokens precisely for query: {}".format(e))
+        while cur_tokens > max_tokens:
+            if len(self.messages) > 2:
+                self.messages.pop(1)
+            elif len(self.messages) == 2 and self.messages[1]["role"] == "assistant":
+                self.messages.pop(1)
+                if precise:
+                    cur_tokens = num_tokens_from_messages(self.messages, self.model)
+                else:
+                    cur_tokens = cur_tokens - max_tokens
+                break
+            elif len(self.messages) == 2 and self.messages[1]["role"] == "user":
+                logger.warn("user message exceed max_tokens. total_tokens={}".format(cur_tokens))
+                break
+            else:
+                logger.debug("max_tokens={}, total_tokens={}, len(messages)={}".format(max_tokens, cur_tokens, len(self.messages)))
+                break
+            if precise:
+                cur_tokens = num_tokens_from_messages(self.messages, self.model)
+            else:
+                cur_tokens = cur_tokens - max_tokens
+        return cur_tokens
+    
+
+# refer to https://github.com/openai/openai-cookbook/blob/main/examples/How_to_count_tokens_with_tiktoken.ipynb
+def num_tokens_from_messages(messages, model):
+    """Returns the number of tokens used by a list of messages."""
+    import tiktoken
+    try:
+        encoding = tiktoken.encoding_for_model(model)
+    except KeyError:
+        logger.debug("Warning: model not found. Using cl100k_base encoding.")
+        encoding = tiktoken.get_encoding("cl100k_base")
+    if model == "gpt-3.5-turbo":
+        return num_tokens_from_messages(messages, model="gpt-3.5-turbo-0301")
+    elif model == "gpt-4":
+        return num_tokens_from_messages(messages, model="gpt-4-0314")
+    elif model == "gpt-3.5-turbo-0301":
+        tokens_per_message = 4  # every message follows <|start|>{role/name}\n{content}<|end|>\n
+        tokens_per_name = -1  # if there's a name, the role is omitted
+    elif model == "gpt-4-0314":
+        tokens_per_message = 3
+        tokens_per_name = 1
+    else:
+        logger.warn(f"num_tokens_from_messages() is not implemented for model {model}. Returning num tokens assuming gpt-3.5-turbo-0301.")
+        return num_tokens_from_messages(messages, model="gpt-3.5-turbo-0301")
+    num_tokens = 0
+    for message in messages:
+        num_tokens += tokens_per_message
+        for key, value in message.items():
+            num_tokens += len(encoding.encode(value))
+            if key == "name":
+                num_tokens += tokens_per_name
+    num_tokens += 3  # every reply is primed with <|start|>assistant<|message|>
+    return num_tokens

bot/openai/open_ai_bot.py

@@ -0,0 +1,109 @@
+# encoding:utf-8
+
+from bot.bot import Bot
+from bot.openai.open_ai_image import OpenAIImage
+from bot.openai.open_ai_session import OpenAISession
+from bot.session_manager import SessionManager
+from bridge.context import ContextType
+from bridge.reply import Reply, ReplyType
+from config import conf
+from common.log import logger
+import openai
+import openai.error
+import time
+
+user_session = dict()
+
+# OpenAI对话模型API (可用)
+class OpenAIBot(Bot, OpenAIImage):
+    def __init__(self):
+        super().__init__()
+        openai.api_key = conf().get('open_ai_api_key')
+        if conf().get('open_ai_api_base'):
+            openai.api_base = conf().get('open_ai_api_base')
+        proxy = conf().get('proxy')
+        if proxy:
+            openai.proxy = proxy
+
+        self.sessions = SessionManager(OpenAISession, model= conf().get("model") or "text-davinci-003")
+
+    def reply(self, query, context=None):
+        # acquire reply content
+        if context and context.type:
+            if context.type == ContextType.TEXT:
+                logger.info("[OPEN_AI] query={}".format(query))
+                session_id = context['session_id']
+                reply = None
+                if query == '#清除记忆':
+                    self.sessions.clear_session(session_id)
+                    reply = Reply(ReplyType.INFO, '记忆已清除')
+                elif query == '#清除所有':
+                    self.sessions.clear_all_session()
+                    reply = Reply(ReplyType.INFO, '所有人记忆已清除')
+                else:
+                    session = self.sessions.session_query(query, session_id)
+                    new_query = str(session)
+                    logger.debug("[OPEN_AI] session query={}".format(new_query))
+
+                    total_tokens, completion_tokens, reply_content = self.reply_text(new_query, session_id, 0)
+                    logger.debug("[OPEN_AI] new_query={}, session_id={}, reply_cont={}, completion_tokens={}".format(new_query, session_id, reply_content, completion_tokens))
+
+                    if total_tokens == 0 :
+                        reply = Reply(ReplyType.ERROR, reply_content)
+                    else:
+                        self.sessions.session_reply(reply_content, session_id, total_tokens)
+                        reply = Reply(ReplyType.TEXT, reply_content)
+                return reply
+            elif context.type == ContextType.IMAGE_CREATE:
+                ok, retstring = self.create_img(query, 0)
+                reply = None
+                if ok:
+                    reply = Reply(ReplyType.IMAGE_URL, retstring)
+                else:
+                    reply = Reply(ReplyType.ERROR, retstring)
+                return reply
+
+    def reply_text(self, query, session_id, retry_count=0):
+        try:
+            response = openai.Completion.create(
+                model= conf().get("model") or "text-davinci-003",  # 对话模型的名称
+                prompt=query,
+                temperature=0.9,  # 值在[0,1]之间，越大表示回复越具有不确定性
+                max_tokens=1200,  # 回复最大的字符数
+                top_p=1,
+                frequency_penalty=0.0,  # [-2,2]之间，该值越大则更倾向于产生不同的内容
+                presence_penalty=0.0,  # [-2,2]之间，该值越大则更倾向于产生不同的内容
+                stop=["\n\n\n"]
+            )
+            res_content = response.choices[0]['text'].strip().replace('<|endoftext|>', '')
+            total_tokens = response["usage"]["total_tokens"]
+            completion_tokens = response["usage"]["completion_tokens"]
+            logger.info("[OPEN_AI] reply={}".format(res_content))
+            return total_tokens, completion_tokens, res_content
+        except Exception as e:
+            need_retry = retry_count < 2
+            result = [0,0,"我现在有点累了，等会再来吧"]
+            if isinstance(e, openai.error.RateLimitError):
+                logger.warn("[OPEN_AI] RateLimitError: {}".format(e))
+                result[2] = "提问太快啦，请休息一下再问我吧"
+                if need_retry:
+                    time.sleep(5)
+            elif isinstance(e, openai.error.Timeout):
+                logger.warn("[OPEN_AI] Timeout: {}".format(e))
+                result[2] = "我没有收到你的消息"
+                if need_retry:
+                    time.sleep(5)
+            elif isinstance(e, openai.error.APIConnectionError):
+                logger.warn("[OPEN_AI] APIConnectionError: {}".format(e))
+                need_retry = False
+                result[2] = "我连接不到你的网络"
+            else:
+                logger.warn("[OPEN_AI] Exception: {}".format(e))
+                need_retry = False
+                self.sessions.clear_session(session_id)
+
+            if need_retry:
+                logger.warn("[OPEN_AI] 第{}次重试".format(retry_count+1))
+                return self.reply_text(query, session_id, retry_count+1)
+            else:
+                return result

bot/openai/open_ai_image.py

@@ -0,0 +1,38 @@
+import time
+import openai
+import openai.error
+from common.token_bucket import TokenBucket
+from common.log import logger
+from config import conf
+
+# OPENAI提供的画图接口
+class OpenAIImage(object):
+    def __init__(self):
+        openai.api_key = conf().get('open_ai_api_key')
+        if conf().get('rate_limit_dalle'):
+            self.tb4dalle = TokenBucket(conf().get('rate_limit_dalle', 50))
+            
+    def create_img(self, query, retry_count=0):
+        try:
+            if conf().get('rate_limit_dalle') and not self.tb4dalle.get_token():
+                return False, "请求太快了，请休息一下再问我吧"
+            logger.info("[OPEN_AI] image_query={}".format(query))
+            response = openai.Image.create(
+                prompt=query,    #图片描述
+                n=1,             #每次生成图片的数量
+                size="256x256"   #图片大小,可选有 256x256, 512x512, 1024x1024
+            )
+            image_url = response['data'][0]['url']
+            logger.info("[OPEN_AI] image_url={}".format(image_url))
+            return True, image_url
+        except openai.error.RateLimitError as e:
+            logger.warn(e)
+            if retry_count < 1:
+                time.sleep(5)
+                logger.warn("[OPEN_AI] ImgCreate RateLimit exceed, 第{}次重试".format(retry_count+1))
+                return self.create_img(query, retry_count+1)
+            else:
+                return False, "提问太快啦，请休息一下再问我吧"
+        except Exception as e:
+            logger.exception(e)
+            return False, str(e)

bot/openai/open_ai_session.py

@@ -0,0 +1,67 @@
+from bot.session_manager import Session
+from common.log import logger
+class OpenAISession(Session):
+    def __init__(self, session_id, system_prompt=None, model= "text-davinci-003"):
+        super().__init__(session_id, system_prompt)
+        self.model = model
+        self.reset()
+
+    def __str__(self):
+        # 构造对话模型的输入
+        '''
+        e.g.  Q: xxx
+              A: xxx
+              Q: xxx
+        '''
+        prompt = ""
+        for item in self.messages:
+            if item['role'] == 'system':
+                prompt += item['content'] + "<|endoftext|>\n\n\n"
+            elif item['role'] == 'user':
+                prompt += "Q: " + item['content'] + "\n"
+            elif item['role'] == 'assistant':
+                prompt += "\n\nA: " + item['content'] + "<|endoftext|>\n"
+
+        if len(self.messages) > 0 and self.messages[-1]['role'] == 'user':
+            prompt += "A: "
+        return prompt
+
+    def discard_exceeding(self, max_tokens, cur_tokens= None):
+        precise = True
+        try:
+            cur_tokens = num_tokens_from_string(str(self), self.model)
+        except Exception as e:
+            precise = False
+            if cur_tokens is None:
+                raise e
+            logger.debug("Exception when counting tokens precisely for query: {}".format(e))
+        while cur_tokens > max_tokens:
+            if len(self.messages) > 1:
+                self.messages.pop(0)
+            elif len(self.messages) == 1 and self.messages[0]["role"] == "assistant":
+                self.messages.pop(0)
+                if precise:
+                    cur_tokens = num_tokens_from_string(str(self), self.model)
+                else:
+                    cur_tokens = len(str(self))
+                break
+            elif len(self.messages) == 1 and self.messages[0]["role"] == "user":
+                logger.warn("user question exceed max_tokens. total_tokens={}".format(cur_tokens))
+                break
+            else:
+                logger.debug("max_tokens={}, total_tokens={}, len(conversation)={}".format(max_tokens, cur_tokens, len(self.messages)))
+                break
+            if precise:
+                cur_tokens = num_tokens_from_string(str(self), self.model)
+            else:
+                cur_tokens = len(str(self))
+        return cur_tokens
+    
+
+# refer to https://github.com/openai/openai-cookbook/blob/main/examples/How_to_count_tokens_with_tiktoken.ipynb
+def num_tokens_from_string(string: str, model: str) -> int:
+    """Returns the number of tokens in a text string."""
+    import tiktoken
+    encoding = tiktoken.encoding_for_model(model)
+    num_tokens = len(encoding.encode(string,disallowed_special=()))
+    return num_tokens

bot/session_manager.py

@@ -0,0 +1,85 @@
+from common.expired_dict import ExpiredDict
+from common.log import logger
+from config import conf
+
+class Session(object):
+    def __init__(self, session_id, system_prompt=None):
+        self.session_id = session_id
+        self.messages = []
+        if system_prompt is None:
+            self.system_prompt = conf().get("character_desc", "")
+        else:
+            self.system_prompt = system_prompt
+
+    # 重置会话
+    def reset(self):
+        system_item = {'role': 'system', 'content': self.system_prompt}
+        self.messages = [system_item]
+
+    def set_system_prompt(self, system_prompt):
+        self.system_prompt = system_prompt
+        self.reset()
+
+    def add_query(self, query):
+        user_item = {'role': 'user', 'content': query}
+        self.messages.append(user_item)
+
+    def add_reply(self, reply):
+        assistant_item = {'role': 'assistant', 'content': reply}
+        self.messages.append(assistant_item)
+    
+    def discard_exceeding(self, max_tokens=None, cur_tokens=None):
+        raise NotImplementedError
+
+
+
+class SessionManager(object):
+    def __init__(self, sessioncls, **session_args):
+        if conf().get('expires_in_seconds'):
+            sessions = ExpiredDict(conf().get('expires_in_seconds'))
+        else:
+            sessions = dict()
+        self.sessions = sessions
+        self.sessioncls = sessioncls
+        self.session_args = session_args
+
+    def build_session(self, session_id, system_prompt=None):
+        '''
+            如果session_id不在sessions中，创建一个新的session并添加到sessions中
+            如果system_prompt不会空，会更新session的system_prompt并重置session
+        '''
+        if session_id not in self.sessions:
+            self.sessions[session_id] = self.sessioncls(session_id, system_prompt, **self.session_args)
+        elif system_prompt is not None:  # 如果有新的system_prompt，更新并重置session
+            self.sessions[session_id].set_system_prompt(system_prompt)
+        session = self.sessions[session_id]
+        return session
+    
+    def session_query(self, query, session_id):
+        session = self.build_session(session_id)
+        session.add_query(query)
+        try:
+            max_tokens = conf().get("conversation_max_tokens", 1000)
+            total_tokens = session.discard_exceeding(max_tokens, None)
+            logger.debug("prompt tokens used={}".format(total_tokens))
+        except Exception as e:
+            logger.debug("Exception when counting tokens precisely for prompt: {}".format(str(e)))
+        return session
+
+    def session_reply(self, reply, session_id, total_tokens = None):
+        session = self.build_session(session_id)
+        session.add_reply(reply)
+        try:
+            max_tokens = conf().get("conversation_max_tokens", 1000)
+            tokens_cnt = session.discard_exceeding(max_tokens, total_tokens)
+            logger.debug("raw total_tokens={}, savesession tokens={}".format(total_tokens, tokens_cnt))
+        except Exception as e:
+            logger.debug("Exception when counting tokens precisely for session: {}".format(str(e)))
+        return session
+
+    def clear_session(self, session_id):
+        if session_id in self.sessions:
+            del(self.sessions[session_id])
+
+    def clear_all_session(self):
+        self.sessions.clear()

bridge/bridge.py

@@ -0,0 +1,50 @@
+from bridge.context import Context
+from bridge.reply import Reply
+from common.log import logger
+from bot import bot_factory
+from common.singleton import singleton
+from voice import voice_factory
+from config import conf
+from common import const
+
+
+@singleton
+class Bridge(object):
+    def __init__(self):
+        self.btype={
+            "chat": const.CHATGPT,
+            "voice_to_text": conf().get("voice_to_text", "openai"),
+            "text_to_voice": conf().get("text_to_voice", "google")
+        }
+        model_type = conf().get("model")
+        if model_type in ["text-davinci-003"]:
+            self.btype['chat'] = const.OPEN_AI
+        if conf().get("use_azure_chatgpt"):
+            self.btype['chat'] = const.CHATGPTONAZURE
+        self.bots={}
+
+    def get_bot(self,typename):
+        if self.bots.get(typename) is None:
+            logger.info("create bot {} for {}".format(self.btype[typename],typename))
+            if typename == "text_to_voice":
+                self.bots[typename] = voice_factory.create_voice(self.btype[typename])
+            elif typename == "voice_to_text":
+                self.bots[typename] = voice_factory.create_voice(self.btype[typename])
+            elif typename == "chat":
+                self.bots[typename] = bot_factory.create_bot(self.btype[typename])
+        return self.bots[typename]
+    
+    def get_bot_type(self,typename):
+        return self.btype[typename]
+
+
+    def fetch_reply_content(self, query, context : Context) -> Reply:
+        return self.get_bot("chat").reply(query, context)
+
+
+    def fetch_voice_to_text(self, voiceFile) -> Reply:
+        return self.get_bot("voice_to_text").voiceToText(voiceFile)
+
+    def fetch_text_to_voice(self, text) -> Reply:
+        return self.get_bot("text_to_voice").textToVoice(text)
+

bridge/context.py

@@ -0,0 +1,57 @@
+# encoding:utf-8
+
+from enum import Enum
+
+class ContextType (Enum):
+    TEXT = 1         # 文本消息
+    VOICE = 2        # 音频消息
+    IMAGE_CREATE = 3 # 创建图片命令
+    
+    def __str__(self):
+        return self.name
+class Context:
+    def __init__(self, type : ContextType = None , content = None,  kwargs = dict()):
+        self.type = type
+        self.content = content
+        self.kwargs = kwargs
+
+    def __contains__(self, key):
+        if key == 'type':
+            return self.type is not None
+        elif key == 'content':
+            return self.content is not None
+        else:
+            return key in self.kwargs
+        
+    def __getitem__(self, key):
+        if key == 'type':
+            return self.type
+        elif key == 'content':
+            return self.content
+        else:
+            return self.kwargs[key]
+    
+    def get(self, key, default=None):
+        try:
+            return self[key]
+        except KeyError:
+            return default
+
+    def __setitem__(self, key, value):
+        if key == 'type':
+            self.type = value
+        elif key == 'content':
+            self.content = value
+        else:
+            self.kwargs[key] = value
+
+    def __delitem__(self, key):
+        if key == 'type':
+            self.type = None
+        elif key == 'content':
+            self.content = None
+        else:
+            del self.kwargs[key]
+    
+    def __str__(self):
+        return "Context(type={}, content={}, kwargs={})".format(self.type, self.content, self.kwargs)

bridge/reply.py

@@ -0,0 +1,22 @@
+
+# encoding:utf-8
+
+from enum import Enum
+
+class ReplyType(Enum):
+    TEXT = 1        # 文本
+    VOICE = 2       # 音频文件
+    IMAGE = 3       # 图片文件
+    IMAGE_URL = 4   # 图片URL
+    
+    INFO = 9
+    ERROR = 10
+    def __str__(self):
+        return self.name
+
+class Reply:
+    def __init__(self, type : ReplyType = None , content = None):
+        self.type = type
+        self.content = content
+    def __str__(self):
+        return "Reply(type={}, content={})".format(self.type, self.content)

channel/channel.py

@@ -0,0 +1,41 @@
+"""
+Message sending channel abstract class
+"""
+
+from bridge.bridge import Bridge
+from bridge.context import Context
+from bridge.reply import *
+
+class Channel(object):
+    NOT_SUPPORT_REPLYTYPE = [ReplyType.VOICE, ReplyType.IMAGE]
+    def startup(self):
+        """
+        init channel
+        """
+        raise NotImplementedError
+
+    def handle_text(self, msg):
+        """
+        process received msg
+        :param msg: message object
+        """
+        raise NotImplementedError
+
+    # 统一的发送函数，每个Channel自行实现，根据reply的type字段发送不同类型的消息
+    def send(self, reply: Reply, context: Context):
+        """
+        send message to user
+        :param msg: message content
+        :param receiver: receiver channel account
+        :return: 
+        """
+        raise NotImplementedError
+
+    def build_reply_content(self, query, context : Context=None) -> Reply:
+        return Bridge().fetch_reply_content(query, context)
+
+    def build_voice_to_text(self, voice_file) -> Reply:
+        return Bridge().fetch_voice_to_text(voice_file)
+    
+    def build_text_to_voice(self, text) -> Reply:
+        return Bridge().fetch_text_to_voice(text)

channel/channel_factory.py

@@ -0,0 +1,23 @@
+"""
+channel factory
+"""
+
+def create_channel(channel_type):
+    """
+    create a channel instance
+    :param channel_type: channel type code
+    :return: channel instance
+    """
+    if channel_type == 'wx':
+        from channel.wechat.wechat_channel import WechatChannel
+        return WechatChannel()
+    elif channel_type == 'wxy':
+        from channel.wechat.wechaty_channel import WechatyChannel
+        return WechatyChannel()
+    elif channel_type == 'terminal':
+        from channel.terminal.terminal_channel import TerminalChannel
+        return TerminalChannel()
+    elif channel_type == 'wechatmp':
+        from channel.wechatmp.wechatmp_channel import WechatMPChannel
+        return WechatMPChannel()
+    raise RuntimeError

channel/chat_channel.py

@@ -0,0 +1,316 @@
+
+
+from asyncio import CancelledError
+from concurrent.futures import Future, ThreadPoolExecutor
+import os
+import re
+import threading
+import time
+from common.dequeue import Dequeue
+from channel.channel import Channel
+from bridge.reply import *
+from bridge.context import *
+from config import conf
+from common.log import logger
+from plugins import *
+try:
+    from voice.audio_convert import any_to_wav
+except Exception as e:
+    pass
+
+# 抽象类, 它包含了与消息通道无关的通用处理逻辑
+class ChatChannel(Channel):
+    name = None # 登录的用户名
+    user_id = None # 登录的用户id
+    futures = {} # 记录每个session_id提交到线程池的future对象, 用于重置会话时把没执行的future取消掉，正在执行的不会被取消
+    sessions = {} # 用于控制并发，每个session_id同时只能有一个context在处理
+    lock = threading.Lock() # 用于控制对sessions的访问
+    handler_pool = ThreadPoolExecutor(max_workers=8)  # 处理消息的线程池
+
+    def __init__(self):
+        _thread = threading.Thread(target=self.consume)
+        _thread.setDaemon(True)
+        _thread.start()
+        
+
+    # 根据消息构造context，消息内容相关的触发项写在这里
+    def _compose_context(self, ctype: ContextType, content, **kwargs):
+        context = Context(ctype, content)
+        context.kwargs = kwargs
+        # context首次传入时，origin_ctype是None, 
+        # 引入的起因是：当输入语音时，会嵌套生成两个context，第一步语音转文本，第二步通过文本生成文字回复。
+        # origin_ctype用于第二步文本回复时，判断是否需要匹配前缀，如果是私聊的语音，就不需要匹配前缀
+        if 'origin_ctype' not in context:  
+            context['origin_ctype'] = ctype
+        # context首次传入时，receiver是None，根据类型设置receiver
+        first_in = 'receiver' not in context
+        # 群名匹配过程，设置session_id和receiver
+        if first_in: # context首次传入时，receiver是None，根据类型设置receiver
+            config = conf()
+            cmsg = context['msg']
+            if cmsg.from_user_id == self.user_id and not config.get('trigger_by_self', True):
+                logger.debug("[WX]self message skipped")
+                return None
+            if context["isgroup"]:
+                group_name = cmsg.other_user_nickname
+                group_id = cmsg.other_user_id
+
+                group_name_white_list = config.get('group_name_white_list', [])
+                group_name_keyword_white_list = config.get('group_name_keyword_white_list', [])
+                if any([group_name in group_name_white_list, 'ALL_GROUP' in group_name_white_list, check_contain(group_name, group_name_keyword_white_list)]):
+                    group_chat_in_one_session = conf().get('group_chat_in_one_session', [])
+                    session_id = cmsg.actual_user_id
+                    if any([group_name in group_chat_in_one_session, 'ALL_GROUP' in group_chat_in_one_session]):
+                        session_id = group_id
+                else:
+                    return None
+                context['session_id'] = session_id
+                context['receiver'] = group_id
+            else:
+                context['session_id'] = cmsg.other_user_id
+                context['receiver'] = cmsg.other_user_id
+
+        # 消息内容匹配过程，并处理content
+        if ctype == ContextType.TEXT:
+            if first_in and "」\n- - - - - - -" in content: # 初次匹配 过滤引用消息
+                logger.debug("[WX]reference query skipped")
+                return None
+            
+            if context["isgroup"]: # 群聊
+                # 校验关键字
+                match_prefix = check_prefix(content, conf().get('group_chat_prefix'))
+                match_contain = check_contain(content, conf().get('group_chat_keyword'))
+                flag = False
+                if match_prefix is not None or match_contain is not None:
+                    flag = True
+                    if match_prefix:
+                        content = content.replace(match_prefix, '', 1).strip()
+                if context['msg'].is_at:
+                    logger.info("[WX]receive group at")
+                    if not conf().get("group_at_off", False):
+                        flag = True
+                    pattern = f'@{self.name}(\u2005|\u0020)'
+                    content = re.sub(pattern, r'', content)
+                
+                if not flag:
+                    if context["origin_ctype"] == ContextType.VOICE:
+                        logger.info("[WX]receive group voice, but checkprefix didn't match")
+                    return None
+            else: # 单聊
+                match_prefix = check_prefix(content, conf().get('single_chat_prefix'))  
+                if match_prefix is not None: # 判断如果匹配到自定义前缀，则返回过滤掉前缀+空格后的内容
+                    content = content.replace(match_prefix, '', 1).strip()
+                elif context["origin_ctype"] == ContextType.VOICE: # 如果源消息是私聊的语音消息，允许不匹配前缀，放宽条件
+                    pass
+                else:
+                    return None     
+                                                  
+            img_match_prefix = check_prefix(content, conf().get('image_create_prefix'))
+            if img_match_prefix:
+                content = content.replace(img_match_prefix, '', 1).strip()
+                context.type = ContextType.IMAGE_CREATE
+            else:
+                context.type = ContextType.TEXT
+            context.content = content
+            if 'desire_rtype' not in context and conf().get('always_reply_voice') and ReplyType.VOICE not in self.NOT_SUPPORT_REPLYTYPE:
+                context['desire_rtype'] = ReplyType.VOICE
+        elif context.type == ContextType.VOICE: 
+            if 'desire_rtype' not in context and conf().get('voice_reply_voice') and ReplyType.VOICE not in self.NOT_SUPPORT_REPLYTYPE:
+                context['desire_rtype'] = ReplyType.VOICE
+
+        return context
+
+    def _handle(self, context: Context):
+        if context is None or not context.content:
+            return
+        logger.debug('[WX] ready to handle context: {}'.format(context))
+        # reply的构建步骤
+        reply = self._generate_reply(context)
+
+        logger.debug('[WX] ready to decorate reply: {}'.format(reply))
+        # reply的包装步骤
+        reply = self._decorate_reply(context, reply)
+
+        # reply的发送步骤
+        self._send_reply(context, reply)
+
+    def _generate_reply(self, context: Context, reply: Reply = Reply()) -> Reply:
+        e_context = PluginManager().emit_event(EventContext(Event.ON_HANDLE_CONTEXT, {
+            'channel': self, 'context': context, 'reply': reply}))
+        reply = e_context['reply']
+        if not e_context.is_pass():
+            logger.debug('[WX] ready to handle context: type={}, content={}'.format(context.type, context.content))
+            if context.type == ContextType.TEXT or context.type == ContextType.IMAGE_CREATE:  # 文字和图片消息
+                reply = super().build_reply_content(context.content, context)
+            elif context.type == ContextType.VOICE:  # 语音消息
+                cmsg = context['msg']
+                cmsg.prepare()
+                file_path = context.content
+                wav_path = os.path.splitext(file_path)[0] + '.wav'
+                try:
+                    any_to_wav(file_path, wav_path) 
+                except Exception as e:  # 转换失败，直接使用mp3，对于某些api，mp3也可以识别
+                    logger.warning("[WX]any to wav error, use raw path. " + str(e))
+                    wav_path = file_path
+                # 语音识别
+                reply = super().build_voice_to_text(wav_path)
+                # 删除临时文件
+                try:
+                    os.remove(file_path)
+                    if wav_path != file_path:
+                        os.remove(wav_path)
+                except Exception as e:
+                    pass
+                    # logger.warning("[WX]delete temp file error: " + str(e))
+
+                if reply.type == ReplyType.TEXT:
+                    new_context = self._compose_context(
+                        ContextType.TEXT, reply.content, **context.kwargs)
+                    if new_context:
+                        reply = self._generate_reply(new_context)
+                    else:
+                        return
+            else:
+                logger.error('[WX] unknown context type: {}'.format(context.type))
+                return
+        return reply
+
+    def _decorate_reply(self, context: Context, reply: Reply) -> Reply:
+        if reply and reply.type:
+            e_context = PluginManager().emit_event(EventContext(Event.ON_DECORATE_REPLY, {
+                'channel': self, 'context': context, 'reply': reply}))
+            reply = e_context['reply']
+            desire_rtype = context.get('desire_rtype')
+            if not e_context.is_pass() and reply and reply.type:
+                
+                if reply.type in self.NOT_SUPPORT_REPLYTYPE:
+                    logger.error("[WX]reply type not support: " + str(reply.type))
+                    reply.type = ReplyType.ERROR
+                    reply.content = "不支持发送的消息类型: " + str(reply.type)
+
+                if reply.type == ReplyType.TEXT:
+                    reply_text = reply.content
+                    if desire_rtype == ReplyType.VOICE and ReplyType.VOICE not in self.NOT_SUPPORT_REPLYTYPE:
+                        reply = super().build_text_to_voice(reply.content)
+                        return self._decorate_reply(context, reply)
+                    if context['isgroup']:
+                        reply_text = '@' +  context['msg'].actual_user_nickname + ' ' + reply_text.strip()
+                        reply_text = conf().get("group_chat_reply_prefix", "") + reply_text
+                    else:
+                        reply_text = conf().get("single_chat_reply_prefix", "") + reply_text
+                    reply.content = reply_text
+                elif reply.type == ReplyType.ERROR or reply.type == ReplyType.INFO:
+                    reply.content = "["+str(reply.type)+"]\n" + reply.content
+                elif reply.type == ReplyType.IMAGE_URL or reply.type == ReplyType.VOICE or reply.type == ReplyType.IMAGE:
+                    pass
+                else:
+                    logger.error('[WX] unknown reply type: {}'.format(reply.type))
+                    return
+            if desire_rtype and desire_rtype != reply.type and reply.type not in [ReplyType.ERROR, ReplyType.INFO]:
+                logger.warning('[WX] desire_rtype: {}, but reply type: {}'.format(context.get('desire_rtype'), reply.type))
+            return reply
+
+    def _send_reply(self, context: Context, reply: Reply):
+        if reply and reply.type:
+            e_context = PluginManager().emit_event(EventContext(Event.ON_SEND_REPLY, {
+                'channel': self, 'context': context, 'reply': reply}))
+            reply = e_context['reply']
+            if not e_context.is_pass() and reply and reply.type:
+                logger.debug('[WX] ready to send reply: {}, context: {}'.format(reply, context))
+                self._send(reply, context)
+
+    def _send(self, reply: Reply, context: Context, retry_cnt = 0):
+        try:
+            self.send(reply, context)
+        except Exception as e:
+            logger.error('[WX] sendMsg error: {}'.format(str(e)))
+            if isinstance(e, NotImplementedError):
+                return
+            logger.exception(e)
+            if retry_cnt < 2:
+                time.sleep(3+3*retry_cnt)
+                self._send(reply, context, retry_cnt+1)
+
+    def thread_pool_callback(self, session_id):
+        def func(worker:Future):
+            try:
+                worker_exception = worker.exception()
+                if worker_exception:
+                    logger.exception("Worker return exception: {}".format(worker_exception))
+            except CancelledError as e:
+                logger.info("Worker cancelled, session_id = {}".format(session_id))
+            except Exception as e:
+                logger.exception("Worker raise exception: {}".format(e))
+            with self.lock:
+                self.sessions[session_id][1].release()
+        return func
+
+    def produce(self, context: Context):
+        session_id = context['session_id']
+        with self.lock:
+            if session_id not in self.sessions:
+                self.sessions[session_id] = [Dequeue(), threading.BoundedSemaphore(conf().get("concurrency_in_session", 1))]
+            if context.type == ContextType.TEXT and context.content.startswith("#"): 
+                self.sessions[session_id][0].putleft(context) # 优先处理管理命令
+            else:
+                self.sessions[session_id][0].put(context)
+
+    # 消费者函数，单独线程，用于从消息队列中取出消息并处理
+    def consume(self):
+        while True:
+            with self.lock:
+                session_ids = list(self.sessions.keys())
+                for session_id in session_ids:
+                    context_queue, semaphore = self.sessions[session_id]
+                    if semaphore.acquire(blocking = False): # 等线程处理完毕才能删除
+                        if not context_queue.empty():
+                            context = context_queue.get()
+                            logger.debug("[WX] consume context: {}".format(context))
+                            future:Future = self.handler_pool.submit(self._handle, context)
+                            future.add_done_callback(self.thread_pool_callback(session_id))
+                            if session_id not in self.futures:
+                                self.futures[session_id] = []
+                            self.futures[session_id].append(future)
+                        elif semaphore._initial_value == semaphore._value+1: # 除了当前，没有任务再申请到信号量，说明所有任务都处理完毕
+                            self.futures[session_id] = [t for t in self.futures[session_id] if not t.done()]
+                            assert len(self.futures[session_id]) == 0, "thread pool error"
+                            del self.sessions[session_id]
+                        else:
+                            semaphore.release()
+            time.sleep(0.1)
+
+    # 取消session_id对应的所有任务，只能取消排队的消息和已提交线程池但未执行的任务
+    def cancel_session(self, session_id): 
+        with self.lock:
+            if session_id in self.sessions:
+                for future in self.futures[session_id]:
+                    future.cancel()
+                cnt = self.sessions[session_id][0].qsize()
+                if cnt>0:
+                    logger.info("Cancel {} messages in session {}".format(cnt, session_id))
+                self.sessions[session_id][0] = Dequeue()
+    
+    def cancel_all_session(self):
+        with self.lock:
+            for session_id in self.sessions:
+                for future in self.futures[session_id]:
+                    future.cancel()
+                cnt = self.sessions[session_id][0].qsize()
+                if cnt>0:
+                    logger.info("Cancel {} messages in session {}".format(cnt, session_id))
+                self.sessions[session_id][0] = Dequeue()
+    
+
+def check_prefix(content, prefix_list):
+    for prefix in prefix_list:
+        if content.startswith(prefix):
+            return prefix
+    return None
+
+def check_contain(content, keyword_list):
+    if not keyword_list:
+        return None
+    for ky in keyword_list:
+        if content.find(ky) != -1:
+            return True
+    return None

channel/chat_message.py

@@ -0,0 +1,83 @@
+
+""" 
+本类表示聊天消息，用于对itchat和wechaty的消息进行统一的封装
+
+ChatMessage
+msg_id: 消息id
+create_time: 消息创建时间
+
+ctype: 消息类型 : ContextType
+content: 消息内容, 如果是声音/图片，这里是文件路径
+
+from_user_id: 发送者id
+from_user_nickname: 发送者昵称
+to_user_id: 接收者id
+to_user_nickname: 接收者昵称
+
+other_user_id: 对方的id，如果你是发送者，那这个就是接收者id，如果你是接收者，那这个就是发送者id，如果是群消息，那这一直是群id
+other_user_nickname: 同上
+
+is_group: 是否是群消息
+is_at: 是否被at
+
+- (群消息时，一般会存在实际发送者，是群内某个成员的id和昵称，下列项仅在群消息时存在)
+actual_user_id: 实际发送者id
+actual_user_nickname：实际发送者昵称
+
+
+
+
+_prepare_fn: 准备函数，用于准备消息的内容，比如下载图片等,
+_prepared: 是否已经调用过准备函数
+_rawmsg: 原始消息对象
+
+"""
+class ChatMessage(object):
+    msg_id = None
+    create_time = None
+    
+    ctype = None
+    content = None
+    
+    from_user_id = None
+    from_user_nickname = None
+    to_user_id = None
+    to_user_nickname = None
+    other_user_id = None
+    other_user_nickname = None
+    
+    is_group = False
+    is_at = False
+    actual_user_id = None
+    actual_user_nickname = None
+
+    _prepare_fn = None
+    _prepared = False
+    _rawmsg = None
+
+
+    def __init__(self,_rawmsg):
+        self._rawmsg = _rawmsg
+
+    def prepare(self):
+        if self._prepare_fn and not self._prepared:
+            self._prepared = True
+            self._prepare_fn()
+
+    def __str__(self):
+        return 'ChatMessage: id={}, create_time={}, ctype={}, content={}, from_user_id={}, from_user_nickname={}, to_user_id={}, to_user_nickname={}, other_user_id={}, other_user_nickname={}, is_group={}, is_at={}, actual_user_id={}, actual_user_nickname={}'.format(
+            self.msg_id,
+            self.create_time,
+            self.ctype,
+            self.content,
+            self.from_user_id,
+            self.from_user_nickname,
+            self.to_user_id,
+            self.to_user_nickname,
+            self.other_user_id,
+            self.other_user_nickname,
+            self.is_group,
+            self.is_at,
+            self.actual_user_id,
+            self.actual_user_nickname,
+        )

channel/terminal/terminal_channel.py

@@ -0,0 +1,31 @@
+from bridge.context import *
+from channel.channel import Channel
+import sys
+
+class TerminalChannel(Channel):
+    def startup(self):
+        context = Context()
+        print("\nPlease input your question")
+        while True:
+            try:
+                prompt = self.get_input("User:\n")
+            except KeyboardInterrupt:
+                print("\nExiting...")
+                sys.exit()
+
+            context.type = ContextType.TEXT
+            context['session_id'] = "User"
+            context.content = prompt
+            print("Bot:")
+            sys.stdout.flush()
+            res = super().build_reply_content(prompt, context).content
+            print(res)
+
+
+    def get_input(self, prompt):
+        """
+        Multi-line input function
+        """
+        print(prompt, end="")
+        line = input()
+        return line

channel/wechat/wechat_channel.py

@@ -0,0 +1,194 @@
+# encoding:utf-8
+
+"""
+wechat channel
+"""
+
+import os
+import threading
+import requests
+import io
+import time
+import json
+from channel.chat_channel import ChatChannel
+from channel.wechat.wechat_message import *
+from common.singleton import singleton
+from common.log import logger
+from lib import itchat
+from lib.itchat.content import *
+from bridge.reply import *
+from bridge.context import *
+from config import conf
+from common.time_check import time_checker
+from common.expired_dict import ExpiredDict
+from plugins import *
+
+@itchat.msg_register(TEXT)
+def handler_single_msg(msg):
+    WechatChannel().handle_text(WeChatMessage(msg))
+    return None
+
+@itchat.msg_register(TEXT, isGroupChat=True)
+def handler_group_msg(msg):
+    WechatChannel().handle_group(WeChatMessage(msg,True))
+    return None
+
+@itchat.msg_register(VOICE)
+def handler_single_voice(msg):
+    WechatChannel().handle_voice(WeChatMessage(msg))
+    return None
+    
+@itchat.msg_register(VOICE, isGroupChat=True)
+def handler_group_voice(msg):
+    WechatChannel().handle_group_voice(WeChatMessage(msg,True))
+    return None
+
+def _check(func):
+    def wrapper(self, cmsg: ChatMessage):
+        msgId = cmsg.msg_id
+        if msgId in self.receivedMsgs:
+            logger.info("Wechat message {} already received, ignore".format(msgId))
+            return
+        self.receivedMsgs[msgId] = cmsg
+        create_time = cmsg.create_time            # 消息时间戳
+        if conf().get('hot_reload') == True and int(create_time) < int(time.time()) - 60:  # 跳过1分钟前的历史消息
+            logger.debug("[WX]history message {} skipped".format(msgId))
+            return
+        return func(self, cmsg)
+    return wrapper
+
+#可用的二维码生成接口
+#https://api.qrserver.com/v1/create-qr-code/?size=400×400&data=https://www.abc.com
+#https://api.isoyu.com/qr/?m=1&e=L&p=20&url=https://www.abc.com
+def qrCallback(uuid,status,qrcode):
+    # logger.debug("qrCallback: {} {}".format(uuid,status))
+    if status == '0':
+        try:
+            from PIL import Image
+            img = Image.open(io.BytesIO(qrcode))
+            _thread = threading.Thread(target=img.show, args=("QRCode",))
+            _thread.setDaemon(True)
+            _thread.start()
+        except Exception as e:
+            pass
+
+        import qrcode
+        url = f"https://login.weixin.qq.com/l/{uuid}"
+        
+        qr_api1="https://api.isoyu.com/qr/?m=1&e=L&p=20&url={}".format(url)
+        qr_api2="https://api.qrserver.com/v1/create-qr-code/?size=400×400&data={}".format(url)
+        qr_api3="https://api.pwmqr.com/qrcode/create/?url={}".format(url)
+        qr_api4="https://my.tv.sohu.com/user/a/wvideo/getQRCode.do?text={}".format(url)
+        print("You can also scan QRCode in any website below:")
+        print(qr_api3)
+        print(qr_api4)
+        print(qr_api2)
+        print(qr_api1)
+        
+        qr = qrcode.QRCode(border=1)
+        qr.add_data(url)
+        qr.make(fit=True)
+        qr.print_ascii(invert=True)
+
+@singleton
+class WechatChannel(ChatChannel):
+    NOT_SUPPORT_REPLYTYPE = []
+    def __init__(self):
+        super().__init__()
+        self.receivedMsgs = ExpiredDict(60*60*24) 
+
+    def startup(self):
+
+        itchat.instance.receivingRetryCount = 600  # 修改断线超时时间
+        # login by scan QRCode
+        hotReload = conf().get('hot_reload', False)
+        try:
+            itchat.auto_login(enableCmdQR=2, hotReload=hotReload, qrCallback=qrCallback)
+        except Exception as e:
+            if hotReload:
+                logger.error("Hot reload failed, try to login without hot reload")
+                itchat.logout()
+                os.remove("itchat.pkl")
+                itchat.auto_login(enableCmdQR=2, hotReload=hotReload, qrCallback=qrCallback)
+            else:
+                raise e
+        self.user_id = itchat.instance.storageClass.userName
+        self.name = itchat.instance.storageClass.nickName
+        logger.info("Wechat login success, user_id: {}, nickname: {}".format(self.user_id, self.name))
+        # start message listener
+        itchat.run()
+
+    # handle_* 系列函数处理收到的消息后构造Context，然后传入_handle函数中处理Context和发送回复
+    # Context包含了消息的所有信息，包括以下属性
+    #   type 消息类型, 包括TEXT、VOICE、IMAGE_CREATE
+    #   content 消息内容，如果是TEXT类型，content就是文本内容，如果是VOICE类型，content就是语音文件名，如果是IMAGE_CREATE类型，content就是图片生成命令
+    #   kwargs 附加参数字典，包含以下的key：
+    #        session_id: 会话id
+    #        isgroup: 是否是群聊
+    #        receiver: 需要回复的对象
+    #        msg: ChatMessage消息对象
+    #        origin_ctype: 原始消息类型，语音转文字后，私聊时如果匹配前缀失败，会根据初始消息是否是语音来放宽触发规则
+    #        desire_rtype: 希望回复类型，默认是文本回复��设置为ReplyType.VOICE是语音回复
+
+    @time_checker
+    @_check
+    def handle_voice(self, cmsg : ChatMessage):
+        if conf().get('speech_recognition') != True:
+            return
+        logger.debug("[WX]receive voice msg: {}".format(cmsg.content))
+        context = self._compose_context(ContextType.VOICE, cmsg.content, isgroup=False, msg=cmsg)
+        if context:
+            self.produce(context)
+
+    @time_checker
+    @_check
+    def handle_text(self, cmsg : ChatMessage):
+        logger.debug("[WX]receive text msg: {}, cmsg={}".format(json.dumps(cmsg._rawmsg, ensure_ascii=False), cmsg))
+        context = self._compose_context(ContextType.TEXT, cmsg.content, isgroup=False, msg=cmsg)
+        if context:
+            self.produce(context)
+
+    @time_checker
+    @_check
+    def handle_group(self, cmsg : ChatMessage):
+        logger.debug("[WX]receive group msg: {}, cmsg={}".format(json.dumps(cmsg._rawmsg, ensure_ascii=False), cmsg))
+        context = self._compose_context(ContextType.TEXT, cmsg.content, isgroup=True, msg=cmsg)
+        if context:
+            self.produce(context)
+    
+    @time_checker
+    @_check
+    def handle_group_voice(self, cmsg : ChatMessage):
+        if conf().get('group_speech_recognition', False) != True:
+            return
+        logger.debug("[WX]receive voice for group msg: {}".format(cmsg.content))
+        context = self._compose_context(ContextType.VOICE, cmsg.content, isgroup=True, msg=cmsg)
+        if context:
+            self.produce(context)
+    
+    # 统一的发送函数，每个Channel自行实现，根据reply的type字段发送不同类型的消息
+    def send(self, reply: Reply, context: Context):
+        receiver = context["receiver"]
+        if reply.type == ReplyType.TEXT:
+            itchat.send(reply.content, toUserName=receiver)
+            logger.info('[WX] sendMsg={}, receiver={}'.format(reply, receiver))
+        elif reply.type == ReplyType.ERROR or reply.type == ReplyType.INFO:
+            itchat.send(reply.content, toUserName=receiver)
+            logger.info('[WX] sendMsg={}, receiver={}'.format(reply, receiver))
+        elif reply.type == ReplyType.VOICE:
+            itchat.send_file(reply.content, toUserName=receiver)
+            logger.info('[WX] sendFile={}, receiver={}'.format(reply.content, receiver))
+        elif reply.type == ReplyType.IMAGE_URL: # 从网络下载图片
+            img_url = reply.content
+            pic_res = requests.get(img_url, stream=True)
+            image_storage = io.BytesIO()
+            for block in pic_res.iter_content(1024):
+                image_storage.write(block)
+            image_storage.seek(0)
+            itchat.send_image(image_storage, toUserName=receiver)
+            logger.info('[WX] sendImage url={}, receiver={}'.format(img_url,receiver))
+        elif reply.type == ReplyType.IMAGE: # 从文件读取图片
+            image_storage = reply.content
+            image_storage.seek(0)
+            itchat.send_image(image_storage, toUserName=receiver)
+            logger.info('[WX] sendImage, receiver={}'.format(receiver))

channel/wechat/wechat_message.py

@@ -0,0 +1,57 @@
+
+
+from bridge.context import ContextType
+from channel.chat_message import ChatMessage
+from common.tmp_dir import TmpDir
+from common.log import logger
+from lib.itchat.content import *
+from lib import itchat
+
+class WeChatMessage(ChatMessage):
+
+    def __init__(self, itchat_msg, is_group=False):
+        super().__init__( itchat_msg)
+        self.msg_id = itchat_msg['MsgId']
+        self.create_time = itchat_msg['CreateTime']
+        self.is_group = is_group
+        
+        if itchat_msg['Type'] == TEXT:
+            self.ctype = ContextType.TEXT
+            self.content = itchat_msg['Text']
+        elif itchat_msg['Type'] == VOICE:
+            self.ctype = ContextType.VOICE
+            self.content = TmpDir().path() + itchat_msg['FileName']  # content直接存临时目录路径
+            self._prepare_fn = lambda: itchat_msg.download(self.content)
+        else:
+            raise NotImplementedError("Unsupported message type: {}".format(itchat_msg['Type']))
+        
+        self.from_user_id = itchat_msg['FromUserName']
+        self.to_user_id = itchat_msg['ToUserName']
+        
+        user_id = itchat.instance.storageClass.userName
+        nickname = itchat.instance.storageClass.nickName
+        
+        # 虽然from_user_id和to_user_id用的少，但是为了保持一致性，还是要填充一下
+        # 以下很繁琐，一句话总结：能填的都填了。
+        if self.from_user_id == user_id:
+            self.from_user_nickname = nickname
+        if self.to_user_id == user_id:
+            self.to_user_nickname = nickname
+        try: # 陌生人时候, 'User'字段可能不存在
+            self.other_user_id = itchat_msg['User']['UserName']
+            self.other_user_nickname = itchat_msg['User']['NickName']
+            if self.other_user_id == self.from_user_id:
+                self.from_user_nickname = self.other_user_nickname
+            if self.other_user_id == self.to_user_id:
+                self.to_user_nickname = self.other_user_nickname
+        except KeyError as e: # 处理偶尔没有对方信息的情况
+            logger.warn("[WX]get other_user_id failed: " + str(e))
+            if self.from_user_id == user_id:
+                self.other_user_id = self.to_user_id
+            else:
+                self.other_user_id = self.from_user_id
+
+        if self.is_group:
+            self.is_at = itchat_msg['IsAt']
+            self.actual_user_id = itchat_msg['ActualUserName']
+            self.actual_user_nickname = itchat_msg['ActualNickName']

channel/wechat/wechaty_channel.py

@@ -0,0 +1,125 @@
+# encoding:utf-8
+
+"""
+wechaty channel
+Python Wechaty - https://github.com/wechaty/python-wechaty
+"""
+import base64
+import os
+import time
+import asyncio
+from bridge.context import Context
+from wechaty_puppet import FileBox
+from wechaty import Wechaty, Contact
+from wechaty.user import Message
+from bridge.reply import *
+from bridge.context import *
+from channel.chat_channel import ChatChannel
+from channel.wechat.wechaty_message import WechatyMessage
+from common.log import logger
+from common.singleton import singleton
+from config import conf
+try:
+    from voice.audio_convert import any_to_sil
+except Exception as e:
+    pass
+
+@singleton
+class WechatyChannel(ChatChannel):
+    NOT_SUPPORT_REPLYTYPE = []
+    def __init__(self):
+        super().__init__()
+
+    def startup(self):
+        config = conf()
+        token = config.get('wechaty_puppet_service_token')
+        os.environ['WECHATY_PUPPET_SERVICE_TOKEN'] = token
+        asyncio.run(self.main())
+
+    async def main(self):
+        
+        loop = asyncio.get_event_loop()
+        #将asyncio的loop传入处理线程
+        self.handler_pool._initializer= lambda: asyncio.set_event_loop(loop)
+        self.bot = Wechaty()
+        self.bot.on('login', self.on_login)
+        self.bot.on('message', self.on_message)
+        await self.bot.start()
+
+    async def on_login(self, contact: Contact):
+        self.user_id = contact.contact_id
+        self.name = contact.name
+        logger.info('[WX] login user={}'.format(contact))
+
+    # 统一的发送函数，每个Channel自行实现，根据reply的type字段发送不同类型的消息
+    def send(self, reply: Reply, context: Context):
+        receiver_id = context['receiver']
+        loop = asyncio.get_event_loop()
+        if context['isgroup']:
+            receiver = asyncio.run_coroutine_threadsafe(self.bot.Room.find(receiver_id),loop).result()
+        else:
+            receiver = asyncio.run_coroutine_threadsafe(self.bot.Contact.find(receiver_id),loop).result()
+        msg = None
+        if reply.type == ReplyType.TEXT:
+            msg = reply.content
+            asyncio.run_coroutine_threadsafe(receiver.say(msg),loop).result()
+            logger.info('[WX] sendMsg={}, receiver={}'.format(reply, receiver))
+        elif reply.type == ReplyType.ERROR or reply.type == ReplyType.INFO:
+            msg = reply.content
+            asyncio.run_coroutine_threadsafe(receiver.say(msg),loop).result()
+            logger.info('[WX] sendMsg={}, receiver={}'.format(reply, receiver))
+        elif reply.type == ReplyType.VOICE:
+            voiceLength = None
+            file_path = reply.content
+            sil_file = os.path.splitext(file_path)[0] + '.sil'
+            voiceLength = int(any_to_sil(file_path, sil_file))
+            if voiceLength >= 60000:
+                voiceLength = 60000
+                logger.info('[WX] voice too long, length={}, set to 60s'.format(voiceLength))
+            # 发送语音
+            t = int(time.time())
+            msg = FileBox.from_file(sil_file, name=str(t) + '.sil')
+            if voiceLength is not None:
+                msg.metadata['voiceLength'] = voiceLength
+            asyncio.run_coroutine_threadsafe(receiver.say(msg),loop).result()
+            try:
+                os.remove(file_path)
+                if sil_file != file_path:
+                    os.remove(sil_file)
+            except Exception as e:
+                pass
+            logger.info('[WX] sendVoice={}, receiver={}'.format(reply.content, receiver))
+        elif reply.type == ReplyType.IMAGE_URL: # 从网络下载图片
+            img_url = reply.content
+            t = int(time.time())
+            msg = FileBox.from_url(url=img_url, name=str(t) + '.png')
+            asyncio.run_coroutine_threadsafe(receiver.say(msg),loop).result()
+            logger.info('[WX] sendImage url={}, receiver={}'.format(img_url,receiver))
+        elif reply.type == ReplyType.IMAGE: # 从文件读取图片
+            image_storage = reply.content
+            image_storage.seek(0)
+            t = int(time.time())
+            msg = FileBox.from_base64(base64.b64encode(image_storage.read()), str(t) + '.png')
+            asyncio.run_coroutine_threadsafe(receiver.say(msg),loop).result()
+            logger.info('[WX] sendImage, receiver={}'.format(receiver))
+
+    async def on_message(self, msg: Message):
+        """
+        listen for message event
+        """
+        try:
+            cmsg = await WechatyMessage(msg)
+        except NotImplementedError as e:
+            logger.debug('[WX] {}'.format(e))
+            return
+        except Exception as e:
+            logger.exception('[WX] {}'.format(e))
+            return
+        logger.debug('[WX] message:{}'.format(cmsg))
+        room = msg.room()  # 获取消息来自的群聊. 如果消息不是来自群聊, 则返回None
+        isgroup = room is not None
+        ctype = cmsg.ctype
+        context = self._compose_context(ctype, cmsg.content, isgroup=isgroup, msg=cmsg)
+        if context:
+            logger.info('[WX] receiveMsg={}, context={}'.format(cmsg, context))
+            self.produce(context)

channel/wechat/wechaty_message.py

@@ -0,0 +1,85 @@
+import asyncio
+import re
+from wechaty import MessageType
+from bridge.context import ContextType
+from channel.chat_message import ChatMessage
+from common.tmp_dir import TmpDir
+from common.log import logger
+from wechaty.user import Message
+
+class aobject(object):
+    """Inheriting this class allows you to define an async __init__.
+
+    So you can create objects by doing something like `await MyClass(params)`
+    """
+    async def __new__(cls, *a, **kw):
+        instance = super().__new__(cls)
+        await instance.__init__(*a, **kw)
+        return instance
+
+    async def __init__(self):
+        pass
+class WechatyMessage(ChatMessage, aobject):
+
+    async def __init__(self, wechaty_msg: Message):
+        super().__init__(wechaty_msg)
+        
+        room = wechaty_msg.room()
+
+        self.msg_id = wechaty_msg.message_id
+        self.create_time = wechaty_msg.payload.timestamp
+        self.is_group = room is not None
+        
+        if wechaty_msg.type() == MessageType.MESSAGE_TYPE_TEXT:
+            self.ctype = ContextType.TEXT
+            self.content = wechaty_msg.text()
+        elif wechaty_msg.type() == MessageType.MESSAGE_TYPE_AUDIO:
+            self.ctype = ContextType.VOICE
+            voice_file = await wechaty_msg.to_file_box()
+            self.content = TmpDir().path() + voice_file.name  # content直接存临时目录路径
+
+            def func():
+                loop = asyncio.get_event_loop()
+                asyncio.run_coroutine_threadsafe(voice_file.to_file(self.content),loop).result()
+            self._prepare_fn = func
+            
+        else:
+            raise NotImplementedError("Unsupported message type: {}".format(wechaty_msg.type()))
+        
+        from_contact = wechaty_msg.talker()  # 获取消息的发送者
+        self.from_user_id = from_contact.contact_id
+        self.from_user_nickname = from_contact.name
+
+        # group中的from和to，wechaty跟itchat含义不一样
+        # wecahty: from是消息实际发送者, to:所在群
+        # itchat: 如果是你发送群消息，from和to是你自己和所在群，如果是别人发群消息，from和to是所在群和你自己
+        # 但这个差别不影响逻辑，group中只使用到：1.用from来判断是否是自己发的，2.actual_user_id来判断实际发送用户
+        
+        if self.is_group:
+            self.to_user_id = room.room_id
+            self.to_user_nickname = await room.topic()
+        else:
+            to_contact = wechaty_msg.to()
+            self.to_user_id = to_contact.contact_id
+            self.to_user_nickname = to_contact.name
+
+        if self.is_group or wechaty_msg.is_self(): # 如果是群消息，other_user设置为群，如果是私聊消息，而且自己发的，就设置成对方。
+            self.other_user_id = self.to_user_id
+            self.other_user_nickname = self.to_user_nickname
+        else:
+            self.other_user_id = self.from_user_id
+            self.other_user_nickname = self.from_user_nickname
+
+        
+
+        if self.is_group: # wechaty群聊中，实际发送用户就是from_user
+            self.is_at = await wechaty_msg.mention_self()
+            if not self.is_at: # 有时候复制粘贴的消息，不算做@，但是内容里面会有@xxx，这里做一下兼容
+                name = wechaty_msg.wechaty.user_self().name
+                pattern = f'@{name}(\u2005|\u0020)'
+                if re.search(pattern,self.content):
+                    logger.debug(f'wechaty message {self.msg_id} include at')
+                    self.is_at = True
+
+            self.actual_user_id = self.from_user_id
+            self.actual_user_nickname = self.from_user_nickname

channel/wechatmp/README.md

@@ -0,0 +1,46 @@
+# 个人微信公众号channel
+
+鉴于个人微信号在服务器上通过itchat登录有封号风险，这里新增了个人微信公众号channel，提供无风险的服务。
+但是由于个人微信公众号的众多接口限制，目前支持的功能有限，实现简陋，提供了一个最基本的文本对话服务，支持加载插件，优化了命令格式，支持私有api_key。暂未实现图片输入输出、语音输入输出等交互形式。
+如有公众号是企业主体且可以通过微信认证，即可获得更多接口，解除大多数限制。欢迎大家提供更多的支持。
+
+## 使用方法
+
+在开始部署前，你需要一个拥有公网IP的服务器，以提供微信服务器和我们自己服务器的连接。或者你需要进行内网穿透，否则微信服务器无法将消息发送给我们的服务器。
+
+此外，需要在我们的服务器上安装python的web框架web.py。
+以ubuntu为例(在ubuntu 22.04上测试):
+```
+pip3 install web.py
+```
+
+然后在[微信公众平台](https://mp.weixin.qq.com)注册一个自己的公众号，类型选择订阅号，主体为个人即可。
+
+然后根据[接入指南](https://developers.weixin.qq.com/doc/offiaccount/Basic_Information/Access_Overview.html)的说明，在[微信公众平台](https://mp.weixin.qq.com)的“设置与开发”-“基本配置”-“服务器配置”中填写服务器地址`URL`和令牌`Token`。这里的`URL`是`example.com/wx`的形式，不可以使用IP，`Token`是你自己编的一个特定的令牌。消息加解密方式目前选择的是明文模式。
+
+相关的服务器验证代码已经写好，你不需要再添加任何代码。你只需要在本项目根目录的`config.json`中添加
+```
+"channel_type": "wechatmp", 
+"wechatmp_token": "your Token", 
+"wechatmp_port": 8080,
+``` 
+然后运行`python3 app.py`启动web服务器。这里会默认监听8080端口，但是微信公众号的服务器配置只支持80/443端口，有两种方法来解决这个问题。第一个是推荐的方法，使用端口转发命令将80端口转发到8080端口（443同理，注意需要支持SSL，也就是https的访问，在`wechatmp_channel.py`需要修改相应的证书路径）：
+```
+sudo iptables -t nat -A PREROUTING -p tcp --dport 80 -j REDIRECT --to-port 8080
+sudo iptables-save > /etc/iptables/rules.v4
+```
+第二个方法是让python程序直接监听80端口。这样可能会导致权限问题，在linux上需要使用`sudo`。然而这会导致后续缓存文件的权限问题，因此不是推荐的方法。
+最后在刚才的“服务器配置”中点击`提交`即可验证你的服务器。
+
+随后在[微信公众平台](https://mp.weixin.qq.com)启用服务器，关闭手动填写规则的自动回复，即可实现ChatGPT的自动回复。
+
+## 个人微信公众号的限制
+由于目前测试的公众号不是企业主体，所以没有客服接口，因此公众号无法主动发出消息，只能被动回复。而微信官方对被动回复有5秒的时间限制，最多重试2次，因此最多只有15秒的自动回复时间窗口。因此如果问题比较复杂或者我们的服务器比较忙，ChatGPT的回答就没办法及时回复给用户。为了解决这个问题，这里做了回答缓存，它需要你在回复超时后，再次主动发送任意文字（例如1）来尝试拿到回答缓存。为了优化使用体验，目前设置了两分钟（120秒）的timeout，用户在至多两分钟后即可得到查询到回复或者错误原因。
+
+另外，由于微信官方的限制，自动回复有长度限制。因此这里将ChatGPT的回答拆分，分成每段600字回复（限制大约在700字）。
+
+## 私有api_key
+公共api有访问频率限制（免费账号每分钟最多20次ChatGPT的API调用），这在服务多人的时候会遇到问题。因此这里多加了一个设置私有api_key的功能。目前通过godcmd插件的命令来设置私有api_key。
+
+## 测试范围
+目前在`RoboStyle`这个公众号上进行了测试，感兴趣的可以关注并体验。开启了godcmd, Banwords, role, dungeon, finish这五个插件，其他的插件还没有测试。百度的接口暂未测试。语音对话没有测试。图片直接以链接形式回复（没有临时素材上传接口的权限）。

channel/wechatmp/receive.py

@@ -0,0 +1,42 @@
+# -*- coding: utf-8 -*-#
+# filename: receive.py
+import xml.etree.ElementTree as ET
+from bridge.context import ContextType
+from channel.chat_message import ChatMessage
+from common.log import logger
+
+
+def parse_xml(web_data):
+    if len(web_data) == 0:
+        return None
+    xmlData = ET.fromstring(web_data)
+    return WeChatMPMessage(xmlData)
+
+class WeChatMPMessage(ChatMessage):
+    def __init__(self, xmlData):
+        super().__init__(xmlData)
+        self.to_user_id = xmlData.find('ToUserName').text
+        self.from_user_id = xmlData.find('FromUserName').text
+        self.create_time = xmlData.find('CreateTime').text
+        self.msg_type = xmlData.find('MsgType').text
+        self.msg_id = xmlData.find('MsgId').text
+        self.is_group = False
+        
+        # reply to other_user_id
+        self.other_user_id = self.from_user_id
+
+        if self.msg_type == 'text':
+            self.ctype = ContextType.TEXT
+            self.content = xmlData.find('Content').text.encode("utf-8")
+        elif self.msg_type == 'voice':
+            self.ctype = ContextType.TEXT
+            self.content = xmlData.find('Recognition').text.encode("utf-8")  # 接收语音识别结果
+        elif self.msg_type == 'image':
+            # not implemented
+            self.pic_url = xmlData.find('PicUrl').text
+            self.media_id = xmlData.find('MediaId').text
+        elif self.msg_type == 'event':
+            self.event = xmlData.find('Event').text
+        else: # video, shortvideo, location, link
+            # not implemented
+            pass

channel/wechatmp/reply.py

@@ -0,0 +1,52 @@
+# -*- coding: utf-8 -*-#
+# filename: reply.py
+import time
+
+class Msg(object):
+    def __init__(self):
+        pass
+
+    def send(self):
+        return "success"
+
+class TextMsg(Msg):
+    def __init__(self, toUserName, fromUserName, content):
+        self.__dict = dict()
+        self.__dict['ToUserName'] = toUserName
+        self.__dict['FromUserName'] = fromUserName
+        self.__dict['CreateTime'] = int(time.time())
+        self.__dict['Content'] = content
+
+    def send(self):
+        XmlForm = """
+            <xml>
+                <ToUserName><![CDATA[{ToUserName}]]></ToUserName>
+                <FromUserName><![CDATA[{FromUserName}]]></FromUserName>
+                <CreateTime>{CreateTime}</CreateTime>
+                <MsgType><![CDATA[text]]></MsgType>
+                <Content><![CDATA[{Content}]]></Content>
+            </xml>
+            """
+        return XmlForm.format(**self.__dict)
+
+class ImageMsg(Msg):
+    def __init__(self, toUserName, fromUserName, mediaId):
+        self.__dict = dict()
+        self.__dict['ToUserName'] = toUserName
+        self.__dict['FromUserName'] = fromUserName
+        self.__dict['CreateTime'] = int(time.time())
+        self.__dict['MediaId'] = mediaId
+
+    def send(self):
+        XmlForm = """
+            <xml>
+                <ToUserName><![CDATA[{ToUserName}]]></ToUserName>
+                <FromUserName><![CDATA[{FromUserName}]]></FromUserName>
+                <CreateTime>{CreateTime}</CreateTime>
+                <MsgType><![CDATA[image]]></MsgType>
+                <Image>
+                <MediaId><![CDATA[{MediaId}]]></MediaId>
+                </Image>
+            </xml>
+            """
+        return XmlForm.format(**self.__dict)

channel/wechatmp/wechatmp_channel.py

@@ -0,0 +1,234 @@
+# -*- coding: utf-8 -*-
+import web
+import time
+import math
+import hashlib
+import textwrap
+from channel.chat_channel import ChatChannel
+import channel.wechatmp.reply as reply
+import channel.wechatmp.receive as receive
+from common.singleton import singleton
+from common.log import logger
+from config import conf
+from bridge.reply import *
+from bridge.context import *
+from plugins import *
+import traceback
+
+# If using SSL, uncomment the following lines, and modify the certificate path.
+# from cheroot.server import HTTPServer
+# from cheroot.ssl.builtin import BuiltinSSLAdapter
+# HTTPServer.ssl_adapter = BuiltinSSLAdapter(
+#         certificate='/ssl/cert.pem',
+#         private_key='/ssl/cert.key')
+
+
+# from concurrent.futures import ThreadPoolExecutor
+# thread_pool = ThreadPoolExecutor(max_workers=8)
+
+@singleton
+class WechatMPChannel(ChatChannel):
+    NOT_SUPPORT_REPLYTYPE = [ReplyType.IMAGE, ReplyType.VOICE]
+    def __init__(self):
+        super().__init__()
+        self.cache_dict = dict()
+        self.query1 = dict()
+        self.query2 = dict()
+        self.query3 = dict()
+
+
+    def startup(self):
+        urls = (
+            '/wx', 'SubsribeAccountQuery',
+        )
+        app = web.application(urls, globals())
+        port = conf().get('wechatmp_port', 8080)
+        web.httpserver.runsimple(app.wsgifunc(), ('0.0.0.0', port))
+
+
+    def send(self, reply: Reply, context: Context):
+        reply_cnt = math.ceil(len(reply.content) / 600)
+        receiver = context["receiver"]
+        self.cache_dict[receiver] = (reply_cnt, reply.content)
+        logger.debug("[send] reply to {} saved to cache: {}".format(receiver, reply))
+
+
+def verify_server():
+    try:
+        data = web.input()
+        if len(data) == 0:
+            return "None"
+        signature = data.signature
+        timestamp = data.timestamp
+        nonce = data.nonce
+        echostr = data.echostr
+        token = conf().get('wechatmp_token') #请按照公众平台官网\基本配置中信息填写
+
+        data_list = [token, timestamp, nonce]
+        data_list.sort()
+        sha1 = hashlib.sha1()
+        # map(sha1.update, data_list) #python2
+        sha1.update("".join(data_list).encode('utf-8'))
+        hashcode = sha1.hexdigest()
+        print("handle/GET func: hashcode, signature: ", hashcode, signature)
+        if hashcode == signature:
+            return echostr
+        else:
+            return ""
+    except Exception as Argument:
+        return Argument
+
+
+# This class is instantiated once per query
+class SubsribeAccountQuery():
+
+    def GET(self):
+        return verify_server()
+
+    def POST(self):
+        channel_instance = WechatMPChannel()
+        try:
+            query_time = time.time()
+            webData = web.data()
+            # logger.debug("[wechatmp] Receive request:\n" + webData.decode("utf-8"))
+            wechat_msg = receive.parse_xml(webData)
+            if wechat_msg.msg_type == 'text':
+                from_user = wechat_msg.from_user_id
+                to_user = wechat_msg.to_user_id
+                message = wechat_msg.content.decode("utf-8")
+                message_id = wechat_msg.msg_id
+
+                logger.info("[wechatmp] {}:{} Receive post query {} {}: {}".format(web.ctx.env.get('REMOTE_ADDR'), web.ctx.env.get('REMOTE_PORT'), from_user, message_id, message))
+
+                cache_key = from_user
+                cache = channel_instance.cache_dict.get(cache_key)
+
+                reply_text = ""
+                # New request
+                if cache == None:
+                    # The first query begin, reset the cache
+                    context = channel_instance._compose_context(ContextType.TEXT, message, isgroup=False, msg=wechat_msg)
+                    logger.debug("[wechatmp] context: {} {}".format(context, wechat_msg))
+                    if context:
+                        # set private openai_api_key
+                        # if from_user is not changed in itchat, this can be placed at chat_channel
+                        user_data = conf().get_user_data(from_user)
+                        context['openai_api_key'] = user_data.get('openai_api_key') # None or user openai_api_key
+                        channel_instance.cache_dict[cache_key] = (0, "")
+                        channel_instance.produce(context)
+                    else:
+                        trigger_prefix = conf().get('single_chat_prefix',[''])[0]
+                        if trigger_prefix:
+                            content = textwrap.dedent(f"""\
+                                请输入'{trigger_prefix}'接你想说的话跟我说话。
+                                例如:
+                                {trigger_prefix}你好，很高兴见到你。""")
+                        else:
+                            logger.error(f"[wechatmp] unknown error")
+                            content = textwrap.dedent("""\
+                                未知错误，请稍后再试""")
+                        replyMsg = reply.TextMsg(wechat_msg.from_user_id, wechat_msg.to_user_id, content)
+                        return replyMsg.send()
+                    channel_instance.query1[cache_key] = False
+                    channel_instance.query2[cache_key] = False
+                    channel_instance.query3[cache_key] = False
+                # Request again
+                elif cache[0] == 0 and channel_instance.query1.get(cache_key) == True and channel_instance.query2.get(cache_key) == True and channel_instance.query3.get(cache_key) == True:
+                    channel_instance.query1[cache_key] = False  #To improve waiting experience, this can be set to True.
+                    channel_instance.query2[cache_key] = False  #To improve waiting experience, this can be set to True.
+                    channel_instance.query3[cache_key] = False
+                elif cache[0] >= 1:
+                    # Skip the waiting phase
+                    channel_instance.query1[cache_key] = True
+                    channel_instance.query2[cache_key] = True
+                    channel_instance.query3[cache_key] = True
+
+
+                cache = channel_instance.cache_dict.get(cache_key)
+                if channel_instance.query1.get(cache_key) == False:
+                    # The first query from wechat official server
+                    logger.debug("[wechatmp] query1 {}".format(cache_key))
+                    channel_instance.query1[cache_key] = True
+                    cnt = 0
+                    while cache[0] == 0 and cnt < 45:
+                        cnt = cnt + 1
+                        time.sleep(0.1)
+                        cache = channel_instance.cache_dict.get(cache_key)
+                    if cnt == 45:
+                        # waiting for timeout (the POST query will be closed by wechat official server)
+                        time.sleep(5)
+                        # and do nothing
+                        return
+                    else:
+                        pass
+                elif channel_instance.query2.get(cache_key) == False:
+                    # The second query from wechat official server
+                    logger.debug("[wechatmp] query2 {}".format(cache_key))
+                    channel_instance.query2[cache_key] = True
+                    cnt = 0
+                    while cache[0] == 0 and cnt < 45:
+                        cnt = cnt + 1
+                        time.sleep(0.1)
+                        cache = channel_instance.cache_dict.get(cache_key)
+                    if cnt == 45:
+                        # waiting for timeout (the POST query will be closed by wechat official server)
+                        time.sleep(5)
+                        # and do nothing
+                        return
+                    else:
+                        pass
+                elif channel_instance.query3.get(cache_key) == False:
+                    # The third query from wechat official server
+                    logger.debug("[wechatmp] query3 {}".format(cache_key))
+                    channel_instance.query3[cache_key] = True
+                    cnt = 0
+                    while cache[0] == 0 and cnt < 40:
+                        cnt = cnt + 1
+                        time.sleep(0.1)
+                        cache = channel_instance.cache_dict.get(cache_key)
+                    if cnt == 40:
+                        # Have waiting for 3x5 seconds
+                        # return timeout message
+                        reply_text = "【正在思考中，回复任意文字尝试获取回复】"
+                        logger.info("[wechatmp] Three queries has finished For {}: {}".format(from_user, message_id))
+                        replyPost = reply.TextMsg(from_user, to_user, reply_text).send()
+                        return replyPost
+                    else:
+                        pass
+
+                if float(time.time()) - float(query_time) > 4.8:
+                    logger.info("[wechatmp] Timeout for {} {}".format(from_user, message_id))
+                    return
+
+
+                if cache[0] > 1:
+                    reply_text = cache[1][:600] + "\n【未完待续，回复任意文字以继续】" #wechatmp auto_reply length limit
+                    channel_instance.cache_dict[cache_key] = (cache[0] - 1, cache[1][600:])
+                elif cache[0] == 1:
+                    reply_text = cache[1]
+                    channel_instance.cache_dict.pop(cache_key)
+                logger.info("[wechatmp] {}:{} Do send {}".format(web.ctx.env.get('REMOTE_ADDR'), web.ctx.env.get('REMOTE_PORT'), reply_text))
+                replyPost = reply.TextMsg(from_user, to_user, reply_text).send()
+                return replyPost
+
+            elif wechat_msg.msg_type == 'event':
+                logger.info("[wechatmp] Event {} from {}".format(wechat_msg.Event, wechat_msg.from_user_id))
+                trigger_prefix = conf().get('single_chat_prefix',[''])[0]
+                content = textwrap.dedent(f"""\
+                    感谢您的关注！
+                    ��里是ChatGPT，可以自由对话。
+                    资源有限，回复较慢，请勿着急。
+                    支持通用表情输入。
+                    暂时不支持图片输入。
+                    支持图片输出，画字开头的问题将回复图片链接。
+                    支持角色扮演和文字冒险两种定制模式对话。
+                    输入'{trigger_prefix}#帮助' 查看详细指令。""")
+                replyMsg = reply.TextMsg(wechat_msg.from_user_id, wechat_msg.to_user_id, content)
+                return replyMsg.send()
+            else:
+                logger.info("暂且不处理")
+                return "success"
+        except Exception as exc:
+            logger.exception(exc)
+            return exc
+

common/const.py

@@ -0,0 +1,5 @@
+# bot_type
+OPEN_AI = "openAI"
+CHATGPT = "chatGPT"
+BAIDU = "baidu"
+CHATGPTONAZURE = "chatGPTOnAzure"

common/dequeue.py

@@ -0,0 +1,33 @@
+
+from queue import Full, Queue
+from time import monotonic as time
+
+# add implementation of putleft to Queue
+class Dequeue(Queue):
+    def putleft(self, item, block=True, timeout=None):
+        with self.not_full:
+            if self.maxsize > 0:
+                if not block:
+                    if self._qsize() >= self.maxsize:
+                        raise Full
+                elif timeout is None:
+                    while self._qsize() >= self.maxsize:
+                        self.not_full.wait()
+                elif timeout < 0:
+                    raise ValueError("'timeout' must be a non-negative number")
+                else:
+                    endtime = time() + timeout
+                    while self._qsize() >= self.maxsize:
+                        remaining = endtime - time()
+                        if remaining <= 0.0:
+                            raise Full
+                        self.not_full.wait(remaining)
+            self._putleft(item)
+            self.unfinished_tasks += 1
+            self.not_empty.notify()
+
+    def putleft_nowait(self, item):
+        return self.putleft(item, block=False)
+
+    def _putleft(self, item):
+        self.queue.appendleft(item)

common/expired_dict.py

@@ -0,0 +1,42 @@
+from datetime import datetime, timedelta
+
+
+class ExpiredDict(dict):
+    def __init__(self, expires_in_seconds):
+        super().__init__()
+        self.expires_in_seconds = expires_in_seconds
+
+    def __getitem__(self, key):
+        value, expiry_time = super().__getitem__(key)
+        if datetime.now() > expiry_time:
+            del self[key]
+            raise KeyError("expired {}".format(key))
+        self.__setitem__(key, value)
+        return value
+
+    def __setitem__(self, key, value):
+        expiry_time = datetime.now() + timedelta(seconds=self.expires_in_seconds)
+        super().__setitem__(key, (value, expiry_time))
+
+    def get(self, key, default=None):
+        try:
+            return self[key]
+        except KeyError:
+            return default
+
+    def __contains__(self, key):
+        try:
+            self[key]
+            return True
+        except KeyError:
+            return False
+
+    def keys(self):
+        keys = list(super().keys())
+        return [key for key in keys if key in self]
+
+    def items(self):
+        return [(key, self[key]) for key in self.keys()]
+
+    def __iter__(self):
+        return self.keys().__iter__()

common/log.py

@@ -0,0 +1,20 @@
+import logging
+import sys
+
+
+def _get_logger():
+    log = logging.getLogger('log')
+    log.setLevel(logging.INFO)
+    console_handle = logging.StreamHandler(sys.stdout)
+    console_handle.setFormatter(logging.Formatter('[%(levelname)s][%(asctime)s][%(filename)s:%(lineno)d] - %(message)s',
+                                                  datefmt='%Y-%m-%d %H:%M:%S'))
+    file_handle = logging.FileHandler('run.log', encoding='utf-8')
+    file_handle.setFormatter(logging.Formatter('[%(levelname)s][%(asctime)s][%(filename)s:%(lineno)d] - %(message)s',
+                                                  datefmt='%Y-%m-%d %H:%M:%S'))
+    log.addHandler(file_handle)
+    log.addHandler(console_handle)
+    return log
+
+
+# 日志句柄
+logger = _get_logger()

common/singleton.py

@@ -0,0 +1,9 @@
+def singleton(cls):
+    instances = {}
+
+    def get_instance(*args, **kwargs):
+        if cls not in instances:
+            instances[cls] = cls(*args, **kwargs)
+        return instances[cls]
+
+    return get_instance

common/sorted_dict.py

@@ -0,0 +1,65 @@
+import heapq
+
+
+class SortedDict(dict):
+    def __init__(self, sort_func=lambda k, v: k, init_dict=None, reverse=False):
+        if init_dict is None:
+            init_dict = []
+        if isinstance(init_dict, dict):
+            init_dict = init_dict.items()
+        self.sort_func = sort_func
+        self.sorted_keys = None
+        self.reverse = reverse
+        self.heap = []
+        for k, v in init_dict:
+            self[k] = v
+
+    def __setitem__(self, key, value):
+        if key in self:
+            super().__setitem__(key, value)
+            for i, (priority, k) in enumerate(self.heap):
+                if k == key:
+                    self.heap[i] = (self.sort_func(key, value), key)
+                    heapq.heapify(self.heap)
+                    break
+            self.sorted_keys = None
+        else:
+            super().__setitem__(key, value)
+            heapq.heappush(self.heap, (self.sort_func(key, value), key))
+            self.sorted_keys = None
+
+    def __delitem__(self, key):
+        super().__delitem__(key)
+        for i, (priority, k) in enumerate(self.heap):
+            if k == key:
+                del self.heap[i]
+                heapq.heapify(self.heap)
+                break
+        self.sorted_keys = None
+
+    def keys(self):
+        if self.sorted_keys is None:
+            self.sorted_keys = [k for _, k in sorted(self.heap, reverse=self.reverse)]
+        return self.sorted_keys
+
+    def items(self):
+        if self.sorted_keys is None:
+            self.sorted_keys = [k for _, k in sorted(self.heap, reverse=self.reverse)]
+        sorted_items = [(k, self[k]) for k in self.sorted_keys]
+        return sorted_items
+
+    def _update_heap(self, key):
+        for i, (priority, k) in enumerate(self.heap):
+            if k == key:
+                new_priority = self.sort_func(key, self[key])
+                if new_priority != priority:
+                    self.heap[i] = (new_priority, key)
+                    heapq.heapify(self.heap)
+                    self.sorted_keys = None
+                break
+
+    def __iter__(self):
+        return iter(self.keys())
+
+    def __repr__(self):
+        return f'{type(self).__name__}({dict(self)}, sort_func={self.sort_func.__name__}, reverse={self.reverse})'

common/time_check.py

@@ -0,0 +1,38 @@
+import time,re,hashlib
+import config
+from common.log import logger
+
+def time_checker(f):
+    def _time_checker(self, *args, **kwargs):
+        _config = config.conf()
+        chat_time_module = _config.get("chat_time_module", False)
+        if chat_time_module:
+            chat_start_time = _config.get("chat_start_time", "00:00")
+            chat_stopt_time = _config.get("chat_stop_time", "24:00")
+            time_regex = re.compile(r'^([01]?[0-9]|2[0-4])(:)([0-5][0-9])$')  #时间匹配，包含24:00
+
+            starttime_format_check = time_regex.match(chat_start_time)  # 检查停止时间格式
+            stoptime_format_check = time_regex.match(chat_stopt_time)  # 检查停止时间格式
+            chat_time_check = chat_start_time < chat_stopt_time # 确定启动时间<停止时间
+
+            # 时间格式检查
+            if not (starttime_format_check and stoptime_format_check and chat_time_check):
+                logger.warn('时间格式不正确,请在config.json中修改您的CHAT_START_TIME/CHAT_STOP_TIME,否则可能会影响您正常使用,开始({})-结束({})'.format(starttime_format_check,stoptime_format_check))
+            if chat_start_time>"23:59":
+                logger.error('启动时间可能存在问题，请修改!')
+
+            # 服务时间检查
+            now_time = time.strftime("%H:%M", time.localtime())
+            if chat_start_time <= now_time <= chat_stopt_time:  # 服务时间内，正常返回回答
+                f(self, *args, **kwargs)
+                return None
+            else:
+                if args[0]['Content'] == "#更新配置":  # 不在服务时间内也可以更新配置
+                    f(self, *args, **kwargs)
+                else:
+                    logger.info('非服务时间内,不接受访问')
+                    return None
+        else:
+            f(self, *args, **kwargs)  # 未开启时间模块则直接回答
+    return _time_checker
+

common/tmp_dir.py

@@ -0,0 +1,20 @@
+
+import os
+import pathlib
+from config import conf
+
+
+class TmpDir(object):
+    """A temporary directory that is deleted when the object is destroyed.
+    """
+
+    tmpFilePath = pathlib.Path('./tmp/')
+    
+    def __init__(self):
+        pathExists = os.path.exists(self.tmpFilePath)
+        if not pathExists and conf().get('speech_recognition') == True:
+            os.makedirs(self.tmpFilePath)
+
+    def path(self):
+        return str(self.tmpFilePath) + '/'
+    

common/token_bucket.py

@@ -0,0 +1,45 @@
+import threading
+import time
+
+
+class TokenBucket:
+    def __init__(self, tpm, timeout=None):
+        self.capacity = int(tpm)  # 令牌桶容量
+        self.tokens = 0  # 初始令牌数为0
+        self.rate = int(tpm) / 60  # 令牌每秒生成速率
+        self.timeout = timeout  # 等待令牌超时时间
+        self.cond = threading.Condition()  # 条件变量
+        self.is_running = True
+        # 开启令牌生成线程
+        threading.Thread(target=self._generate_tokens).start()
+
+    def _generate_tokens(self):
+        """生成令牌"""
+        while self.is_running:
+            with self.cond:
+                if self.tokens < self.capacity:
+                    self.tokens += 1
+                self.cond.notify()  # 通知获取令牌的线程
+            time.sleep(1 / self.rate)
+
+    def get_token(self):
+        """获取令牌"""
+        with self.cond:
+            while self.tokens <= 0:
+                flag = self.cond.wait(self.timeout)
+                if not flag:  # 超时
+                    return False
+            self.tokens -= 1
+        return True
+
+    def close(self):
+        self.is_running = False
+
+
+if __name__ == "__main__":
+    token_bucket = TokenBucket(20, None)  # 创建一个每分钟生产20个tokens的令牌桶
+    # token_bucket = TokenBucket(20, 0.1)
+    for i in range(3):
+        if token_bucket.get_token():
+            print(f"第{i+1}次请求成功")
+    token_bucket.close()

config-template.json

@@ -0,0 +1,18 @@
+{
+  "open_ai_api_key": "YOUR API KEY",
+  "model": "gpt-3.5-turbo",
+  "proxy": "",
+  "use_azure_chatgpt": false,
+  "single_chat_prefix": ["bot", "@bot"],
+  "single_chat_reply_prefix": "[bot] ",
+  "group_chat_prefix": ["@bot"],
+  "group_name_white_list": ["ChatGPT测试群", "ChatGPT测试群2"],
+  "group_chat_in_one_session": ["ChatGPT测试群"],
+  "image_create_prefix": ["画", "看", "找"],
+  "speech_recognition": false,
+  "group_speech_recognition": false,
+  "voice_reply_voice": false,
+  "conversation_max_tokens": 1000,
+  "expires_in_seconds": 3600,
+  "character_desc": "你是ChatGPT, 一个由OpenAI训练的大型语言模型, 你旨在回答并解决人们的任何问题，并且可以使用多种语言与人交流。"
+}

config.json

@@ -0,0 +1,18 @@
+{
+  "open_ai_api_key": "YOUR API KEY",
+  "model": "gpt-3.5-turbo",
+  "proxy": "",
+  "use_azure_chatgpt": false,
+  "single_chat_prefix": [""],
+  "single_chat_reply_prefix": "[bot] ",
+  "group_chat_prefix": ["@bot"],
+  "group_name_white_list": [""],
+  "group_chat_in_one_session": [""],
+  "image_create_prefix": ["画", "看", "找"],
+  "speech_recognition": true,
+  "group_speech_recognition": false,
+  "voice_reply_voice": false,
+  "conversation_max_tokens": 1000,
+  "expires_in_seconds": 3600,
+  "character_desc": "你是接入微信的ChatGPT, 一个由OpenAI训练的大型语言模型, 你旨在回答并解决人们的任何问题，并且可以使用多种语言与人交流。"
+}

config.py

@@ -0,0 +1,198 @@
+# encoding:utf-8
+
+import json
+import logging
+import os
+from common.log import logger
+import pickle
+
+# 将所有可用的配置项写在字典里, 请使用小写字母
+available_setting = {
+    # openai api配置
+    "open_ai_api_key": "",  # openai api key
+    # openai apibase，当use_azure_chatgpt为true时，需要设置对应的api base
+    "open_ai_api_base": "https://api.openai.com/v1",
+    "proxy": "",  # openai使用的代理
+    # chatgpt模型， 当use_azure_chatgpt为true时，其名称为Azure上model deployment名称
+    "model": "gpt-3.5-turbo",
+    "use_azure_chatgpt": False,  # 是否使用azure的chatgpt
+
+    # Bot触发配置
+    "single_chat_prefix": ["bot", "@bot"],  # 私聊时文本需要包含该前缀才能触发机器人回复
+    "single_chat_reply_prefix": "[bot] ",  # 私聊时自动回复的前缀，用于区分真人
+    "group_chat_prefix": ["@bot"],  # 群聊时包含该前缀则会触发机器人回复
+    "group_chat_reply_prefix": "",  # 群聊时自动回复的前缀
+    "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测试群"],  # 支持会话上下文共享的群名称
+    "trigger_by_self": False,  # 是否允许机器人触发
+    "image_create_prefix": ["画", "看", "找"],  # 开启图片回复的前缀
+    "concurrency_in_session": 1, # 同一会话最多有多少条消息在处理中，大于1可能乱序
+
+    # 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": 60, # chatgpt请求超时时间，openai接口默认设置为600，对于难问题一般需要较长时间
+    "timeout": 120,         # chatgpt重试超时时间，在这个时间内，将会自动重试
+
+    # 语音设置
+    "speech_recognition": False,  # 是否开启语音识别
+    "group_speech_recognition": False,  # 是否开启群组语音识别
+    "voice_reply_voice": False,  # 是否使用语音回复语音，需要设置对应语音合成引擎的api key
+    "always_reply_voice": False,  # 是否一直使用语音回复
+    "voice_to_text": "openai",  # 语音识别引擎，支持openai,google,azure
+    "text_to_voice": "baidu",  # 语音合成引擎，支持baidu,google,pytts(offline),azure
+
+    # 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",
+
+    # 服务时间限制，目前支持itchat
+    "chat_time_module": False,  # 是否开启服务时间限制
+    "chat_start_time": "00:00",  # 服务开始时间
+    "chat_stop_time": "24:00",  # 服务结束时间
+
+    # itchat的配置
+    "hot_reload": False,  # 是否开启热重载
+
+    # wechaty的配置
+    "wechaty_puppet_service_token": "",  # wechaty的token
+
+    # wechatmp的配置
+    "wechatmp_token": "",  # 微信公众平台的Token
+    "wechatmp_port": 8080, # 微信公众平台的端口,需要端口转发到80或443
+
+    # chatgpt指令自定义触发词
+    "clear_memory_commands": ['#清除记忆'],  # 重置会话指令，必须以#开头
+
+    # channel配置
+    "channel_type": "wx", # 通道类型，支持：{wx,wxy,terminal,wechatmp}
+
+    "debug": False,  # 是否开启debug模式，开启后会打印更多日志
+
+    # 插件配置
+    "plugin_trigger_prefix": "$",  # 规范插件提供聊天相关指令的前缀，建议不要和管理员指令前缀"#"冲突
+}
+
+
+class Config(dict):
+    def __init__(self, d:dict={}):
+        super().__init__(d)
+        # 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('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('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))
+
+    # 将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(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

docker/Dockerfile.alpine

@@ -0,0 +1,39 @@
+FROM python:3.10-alpine
+
+LABEL maintainer="[email protected]"
+ARG TZ='Asia/Shanghai'
+
+ARG CHATGPT_ON_WECHAT_VER
+
+ENV BUILD_PREFIX=/app
+
+RUN apk add --no-cache \
+        bash \
+        curl \
+        wget \
+    && export BUILD_GITHUB_TAG=${CHATGPT_ON_WECHAT_VER:-`curl -sL "https://api.github.com/repos/zhayujie/chatgpt-on-wechat/releases/latest" | \
+        grep '"tag_name":' | \
+        sed -E 's/.*"([^"]+)".*/\1/'`} \
+    && wget -t 3 -T 30 -nv -O chatgpt-on-wechat-${BUILD_GITHUB_TAG}.tar.gz \
+            https://github.com/zhayujie/chatgpt-on-wechat/archive/refs/tags/${BUILD_GITHUB_TAG}.tar.gz \
+    && tar -xzf chatgpt-on-wechat-${BUILD_GITHUB_TAG}.tar.gz \
+    && mv chatgpt-on-wechat-${BUILD_GITHUB_TAG} ${BUILD_PREFIX} \
+    && rm chatgpt-on-wechat-${BUILD_GITHUB_TAG}.tar.gz \
+    && cd ${BUILD_PREFIX} \
+    && cp config-template.json ${BUILD_PREFIX}/config.json \
+    && /usr/local/bin/python -m pip install --no-cache --upgrade pip \
+    && pip install --no-cache -r requirements.txt \
+    && pip install --no-cache -r requirements-optional.txt \
+    && apk del curl wget
+
+WORKDIR ${BUILD_PREFIX}
+
+ADD ./entrypoint.sh /entrypoint.sh
+
+RUN chmod +x /entrypoint.sh \
+    && adduser -D -h /home/noroot -u 1000 -s /bin/bash noroot \
+    && chown -R noroot:noroot ${BUILD_PREFIX}
+
+USER noroot
+
+ENTRYPOINT ["/entrypoint.sh"]

docker/Dockerfile.debian

@@ -0,0 +1,40 @@
+FROM python:3.10
+
+LABEL maintainer="[email protected]"
+ARG TZ='Asia/Shanghai'
+
+ARG CHATGPT_ON_WECHAT_VER
+
+ENV BUILD_PREFIX=/app
+
+RUN apt-get update  \
+    && apt-get install -y --no-install-recommends \
+        wget \
+        curl  \
+    && rm -rf /var/lib/apt/lists/* \
+    && export BUILD_GITHUB_TAG=${CHATGPT_ON_WECHAT_VER:-`curl -sL "https://api.github.com/repos/zhayujie/chatgpt-on-wechat/releases/latest" | \
+        grep '"tag_name":' | \
+        sed -E 's/.*"([^"]+)".*/\1/'`} \
+    && wget -t 3 -T 30 -nv -O chatgpt-on-wechat-${BUILD_GITHUB_TAG}.tar.gz \
+            https://github.com/zhayujie/chatgpt-on-wechat/archive/refs/tags/${BUILD_GITHUB_TAG}.tar.gz \
+    && tar -xzf chatgpt-on-wechat-${BUILD_GITHUB_TAG}.tar.gz \
+    && mv chatgpt-on-wechat-${BUILD_GITHUB_TAG} ${BUILD_PREFIX} \
+    && rm chatgpt-on-wechat-${BUILD_GITHUB_TAG}.tar.gz \
+    && cd ${BUILD_PREFIX} \
+    && cp config-template.json ${BUILD_PREFIX}/config.json \
+    && /usr/local/bin/python -m pip install --no-cache --upgrade pip \
+    && pip install --no-cache -r requirements.txt \
+    && pip install --no-cache -r requirements-optional.txt
+
+WORKDIR ${BUILD_PREFIX}
+
+ADD ./entrypoint.sh /entrypoint.sh
+
+RUN chmod +x /entrypoint.sh \
+    && groupadd -r noroot \
+    && useradd -r -g noroot -s /bin/bash -d /home/noroot noroot \
+    && chown -R noroot:noroot ${BUILD_PREFIX} 
+
+USER noroot
+
+ENTRYPOINT ["/entrypoint.sh"]

docker/Dockerfile.debian.latest

@@ -0,0 +1,33 @@
+FROM python:3.10-slim
+
+LABEL maintainer="[email protected]"
+ARG TZ='Asia/Shanghai'
+
+ARG CHATGPT_ON_WECHAT_VER
+
+ENV BUILD_PREFIX=/app
+
+ADD . ${BUILD_PREFIX}
+
+RUN apt-get update \
+    &&apt-get install -y --no-install-recommends  bash \
+    ffmpeg espeak \
+    && cd ${BUILD_PREFIX} \
+    && cp config-template.json config.json \
+    && /usr/local/bin/python -m pip install --no-cache --upgrade pip \
+    && pip install --no-cache -r requirements.txt \
+    && pip install --no-cache -r requirements-optional.txt \
+    && pip install azure-cognitiveservices-speech
+    
+WORKDIR ${BUILD_PREFIX}
+
+ADD docker/entrypoint.sh /entrypoint.sh
+
+RUN chmod +x /entrypoint.sh \
+    && groupadd -r noroot \
+    && useradd -r -g noroot -s /bin/bash -d /home/noroot noroot \
+    && chown -R noroot:noroot ${BUILD_PREFIX}
+
+USER noroot
+
+ENTRYPOINT ["docker/entrypoint.sh"]

docker/Dockerfile.latest

@@ -0,0 +1,29 @@
+FROM python:3.10-alpine
+
+LABEL maintainer="[email protected]"
+ARG TZ='Asia/Shanghai'
+
+ARG CHATGPT_ON_WECHAT_VER
+
+ENV BUILD_PREFIX=/app
+
+ADD . ${BUILD_PREFIX}
+
+RUN apk add --no-cache bash ffmpeg espeak \
+    && cd ${BUILD_PREFIX} \
+    && cp config-template.json config.json \
+    && /usr/local/bin/python -m pip install --no-cache --upgrade pip \
+    && pip install --no-cache -r requirements.txt \
+    && pip install --no-cache -r requirements-optional.txt
+
+WORKDIR ${BUILD_PREFIX}
+
+ADD docker/entrypoint.sh /entrypoint.sh
+
+RUN chmod +x /entrypoint.sh \
+    && adduser -D -h /home/noroot -u 1000 -s /bin/bash noroot \
+    && chown -R noroot:noroot ${BUILD_PREFIX}
+
+USER noroot
+
+ENTRYPOINT ["docker/entrypoint.sh"]

docker/build.alpine.sh

@@ -0,0 +1,16 @@
+#!/bin/bash
+
+# fetch latest release tag
+CHATGPT_ON_WECHAT_TAG=`curl -sL "https://api.github.com/repos/zhayujie/chatgpt-on-wechat/releases/latest" | \
+     grep '"tag_name":' | \
+     sed -E 's/.*"([^"]+)".*/\1/'`
+
+# build image
+docker build -f Dockerfile.alpine \
+             --build-arg CHATGPT_ON_WECHAT_VER=$CHATGPT_ON_WECHAT_TAG \
+             -t zhayujie/chatgpt-on-wechat .
+
+# tag image
+docker tag zhayujie/chatgpt-on-wechat zhayujie/chatgpt-on-wechat:alpine 
+docker tag zhayujie/chatgpt-on-wechat zhayujie/chatgpt-on-wechat:$CHATGPT_ON_WECHAT_TAG-alpine
+