Merge branch 'master' into master

channel/channel.py

@@ -3,6 +3,8 @@ Message sending channel abstract class
 """
 
 from bridge.bridge import Bridge
+from bridge.context import Context
+from bridge.reply import Reply
 
 class Channel(object):
     def startup(self):
@@ -27,11 +29,11 @@ class Channel(object):
         """
         raise NotImplementedError
 
-    def build_reply_content(self, query, context=None):
+    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):
+    def build_voice_to_text(self, voice_file) -> Reply:
         return Bridge().fetch_voice_to_text(voice_file)
     
-    def build_text_to_voice(self, text):
+    def build_text_to_voice(self, text) -> Reply:
         return Bridge().fetch_text_to_voice(text)

channel/terminal/terminal_channel.py

@@ -1,9 +1,10 @@
+from bridge.context import *
 from channel.channel import Channel
 import sys
 
 class TerminalChannel(Channel):
     def startup(self):
-        context = {"from_user_id": "User"}
+        context = Context()
         print("\nPlease input your question")
         while True:
             try:
@@ -12,12 +13,13 @@ class TerminalChannel(Channel):
                 print("\nExiting...")
                 sys.exit()
 
+            context.type = ContextType.TEXT
+            context['session_id'] = "User"
+            context.content = prompt
             print("Bot:")
             sys.stdout.flush()
-            for res in super().build_reply_content(prompt, context):
-                print(res, end="")
-                sys.stdout.flush()
-            print("\n")
+            res = super().build_reply_content(prompt, context).content
+            print(res)
 
 
     def get_input(self, prompt):

channel/wechat/wechat_channel.py

@@ -7,18 +7,25 @@ wechat channel
 import itchat
 import json
 from itchat.content import *
+from bridge.reply import *
+from bridge.context import *
 from channel.channel import Channel
 from concurrent.futures import ThreadPoolExecutor
 from common.log import logger
 from common.tmp_dir import TmpDir
 from config import conf
 from common.time_check import time_checker
+from plugins import *
 import requests
 import io
 import time
 
-thread_pool = ThreadPoolExecutor(max_workers=8)
 
+thread_pool = ThreadPoolExecutor(max_workers=8)
+def thread_pool_callback(worker):
+    worker_exception = worker.exception()
+    if worker_exception:
+        logger.exception("Worker return exception: {}".format(worker_exception))
 
 @itchat.msg_register(TEXT)
 def handler_single_msg(msg):
@@ -49,68 +56,59 @@ class WechatChannel(Channel):
         # 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: itchat的原始消息对象
+
     def handle_voice(self, msg):
-        if conf().get('speech_recognition') != True :
+        if conf().get('speech_recognition') != True:
             return
         logger.debug("[WX]receive voice msg: " + msg['FileName'])
-        thread_pool.submit(self._do_handle_voice, msg)
-
-    def _do_handle_voice(self, msg):
         from_user_id = msg['FromUserName']
         other_user_id = msg['User']['UserName']
         if from_user_id == other_user_id:
-            file_name = TmpDir().path() + msg['FileName']
-            msg.download(file_name)
-            query = super().build_voice_to_text(file_name)
-            if conf().get('voice_reply_voice'):
-                self._do_send_voice(query, from_user_id)
-            else:
-                self._do_send_text(query, from_user_id)
+            context = Context(ContextType.VOICE,msg['FileName'])
+            context.kwargs = {'isgroup': False, 'msg': msg, 'receiver': other_user_id, 'session_id': other_user_id}
+            thread_pool.submit(self.handle, context).add_done_callback(thread_pool_callback)
 
 
     @time_checker
     def handle_text(self, msg):
         logger.debug("[WX]receive text msg: " + json.dumps(msg, ensure_ascii=False))
         content = msg['Text']
-        self._handle_single_msg(msg, content)
-
-    def _handle_single_msg(self, msg, content):
         from_user_id = msg['FromUserName']
         to_user_id = msg['ToUserName']              # 接收人id
         other_user_id = msg['User']['UserName']     # 对手方id
         create_time = msg['CreateTime']             # 消息时间
-        match_prefix = self.check_prefix(content, conf().get('single_chat_prefix'))
+        match_prefix = check_prefix(content, conf().get('single_chat_prefix'))
         if conf().get('hot_reload') == True and int(create_time) < int(time.time()) - 60:    #跳过1分钟前的历史消息
             logger.debug("[WX]history message skipped")
             return
         if "」\n- - - - - - - - - - - - - - -" in content:
             logger.debug("[WX]reference query skipped")
             return
-        if from_user_id == other_user_id and match_prefix is not None:
-            # 好友向自己发送消息
-            if match_prefix != '':
-                str_list = content.split(match_prefix, 1)
-                if len(str_list) == 2:
-                    content = str_list[1].strip()
+        if match_prefix:
+            content = content.replace(match_prefix, '', 1).strip()
+        else:
+            return
+        context = Context()
+        context.kwargs = {'isgroup': False, 'msg': msg, 'receiver': other_user_id, 'session_id': other_user_id}
 
-            img_match_prefix = self.check_prefix(content, conf().get('image_create_prefix'))
-            if img_match_prefix:
-                content = content.split(img_match_prefix, 1)[1].strip()
-                thread_pool.submit(self._do_send_img, content, from_user_id)
-            else :
-                thread_pool.submit(self._do_send_text, content, from_user_id)
-        elif to_user_id == other_user_id and match_prefix:
-            # 自己给好友发送消息
-            str_list = content.split(match_prefix, 1)
-            if len(str_list) == 2:
-                content = str_list[1].strip()
-            img_match_prefix = self.check_prefix(content, conf().get('image_create_prefix'))
-            if img_match_prefix:
-                content = content.split(img_match_prefix, 1)[1].strip()
-                thread_pool.submit(self._do_send_img, content, to_user_id)
-            else:
-                thread_pool.submit(self._do_send_text, content, to_user_id)
+        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
+        thread_pool.submit(self.handle, context).add_done_callback(thread_pool_callback)
 
     @time_checker
     def handle_group(self, msg):
@@ -135,100 +133,128 @@ class WechatChannel(Channel):
             logger.debug("[WX]reference query skipped")
             return ""
         config = conf()
-        match_prefix = (msg['IsAt'] and not config.get("group_at_off", False)) or self.check_prefix(origin_content, config.get('group_chat_prefix')) \
-                       or self.check_contain(origin_content, config.get('group_chat_keyword'))
-        if ('ALL_GROUP' in config.get('group_name_white_list') or group_name in config.get('group_name_white_list') or self.check_contain(group_name, config.get('group_name_keyword_white_list'))) and match_prefix:
-            img_match_prefix = self.check_prefix(content, conf().get('image_create_prefix'))
+        match_prefix = (msg['IsAt'] and not config.get("group_at_off", False)) or check_prefix(origin_content, config.get('group_chat_prefix')) \
+                       or check_contain(origin_content, config.get('group_chat_keyword'))
+        if ('ALL_GROUP' in config.get('group_name_white_list') or group_name in config.get('group_name_white_list') or check_contain(group_name, config.get('group_name_keyword_white_list'))) and match_prefix:
+            context = Context()
+            context.kwargs = { 'isgroup': True, 'msg': msg, 'receiver': group_id}
+            
+            img_match_prefix = check_prefix(content, conf().get('image_create_prefix'))
             if img_match_prefix:
-                content = content.split(img_match_prefix, 1)[1].strip()
-                thread_pool.submit(self._do_send_img, content, group_id)
+                content = content.replace(img_match_prefix, '', 1).strip()
+                context.type = ContextType.IMAGE_CREATE
             else:
-                thread_pool.submit(self._do_send_group, content, msg)
+                context.type = ContextType.TEXT
+            context.content = content
 
-    def send(self, msg, receiver):
-        itchat.send(msg, toUserName=receiver)
-        logger.info('[WX] sendMsg={}, receiver={}'.format(msg, receiver))
+            group_chat_in_one_session = conf().get('group_chat_in_one_session', [])
+            if ('ALL_GROUP' in group_chat_in_one_session or
+                    group_name in group_chat_in_one_session or
+                    check_contain(group_name, group_chat_in_one_session)):
+                context['session_id'] = group_id
+            else:
+                context['session_id'] = msg['ActualUserName']
 
-    def _do_send_voice(self, query, reply_user_id):
-        try:
-            if not query:
-                return
-            context = dict()
-            context['from_user_id'] = reply_user_id
-            reply_text = super().build_reply_content(query, context)
-            if reply_text:
-                replyFile = super().build_text_to_voice(reply_text)
-                itchat.send_file(replyFile, toUserName=reply_user_id)
-                logger.info('[WX] sendFile={}, receiver={}'.format(replyFile, reply_user_id))
-        except Exception as e:
-            logger.exception(e)
+            thread_pool.submit(self.handle, context).add_done_callback(thread_pool_callback)
 
-    def _do_send_text(self, query, reply_user_id):
-        try:
-            if not query:
-                return
-            context = dict()
-            context['session_id'] = reply_user_id
-            reply_text = super().build_reply_content(query, context)
-            if reply_text:
-                self.send(conf().get("single_chat_reply_prefix") + reply_text, reply_user_id)
-        except Exception as e:
-            logger.exception(e)
-
-    def _do_send_img(self, query, reply_user_id):
-        try:
-            if not query:
-                return
-            context = dict()
-            context['type'] = 'IMAGE_CREATE'
-            img_url = super().build_reply_content(query, context)
-            if not img_url:
-                return
-
-            # 图片下载
+    # 统一的发送函数，每个Channel自行实现，根据reply的type字段发送不同类型的消息
+    def send(self, reply : Reply, 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))
 
-            # 图片发送
-            itchat.send_image(image_storage, reply_user_id)
-            logger.info('[WX] sendImage, receiver={}'.format(reply_user_id))
-        except Exception as e:
-            logger.exception(e)
+    # 处理消息 TODO: 如果wechaty解耦，此处逻辑可以放置到父类
+    def handle(self, context):
+        reply = Reply()
 
-    def _do_send_group(self, query, msg):
-        if not query:
-            return
-        context = dict()
-        group_name = msg['User']['NickName']
-        group_id = msg['User']['UserName']
-        group_chat_in_one_session = conf().get('group_chat_in_one_session', [])
-        if ('ALL_GROUP' in group_chat_in_one_session or \
-                group_name in group_chat_in_one_session or \
-                self.check_contain(group_name, group_chat_in_one_session)):
-            context['session_id'] = group_id
-        else:
-            context['session_id'] = msg['ActualUserName']
-        reply_text = super().build_reply_content(query, context)
-        if reply_text:
-            reply_text = '@' + msg['ActualNickName'] + ' ' + reply_text.strip()
-            self.send(conf().get("group_chat_reply_prefix", "") + reply_text, group_id)
+        logger.debug('[WX] ready to handle context: {}'.format(context))
+        
+        # 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:
+                msg = context['msg']
+                file_name = TmpDir().path() + context.content
+                msg.download(file_name)
+                reply = super().build_voice_to_text(file_name)
+                if reply.type != ReplyType.ERROR and reply.type != ReplyType.INFO:
+                    context.content = reply.content # 语音转文字后，将文字内容作为新的context
+                    context.type = ContextType.TEXT
+                    reply = super().build_reply_content(context.content, context)
+                    if reply.type == ReplyType.TEXT:
+                        if conf().get('voice_reply_voice'):
+                            reply = super().build_text_to_voice(reply.content)
+            else:
+                logger.error('[WX] unknown context type: {}'.format(context.type))
+                return
+
+        logger.debug('[WX] ready to decorate reply: {}'.format(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']
+            if not e_context.is_pass() and reply and reply.type:
+                if reply.type == ReplyType.TEXT:
+                    reply_text = reply.content
+                    if context['isgroup']:
+                        reply_text = '@' +  context['msg']['ActualNickName'] + ' ' + 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
+
+        # 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: {} to {}'.format(reply, context['receiver']))
+                self.send(reply, context['receiver'])
 
 
-    def check_prefix(self, content, prefix_list):
-        for prefix in prefix_list:
-            if content.startswith(prefix):
-                return prefix
+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
-
-
-    def check_contain(self, content, keyword_list):
-        if not keyword_list:
-            return None
-        for ky in keyword_list:
-            if content.find(ky) != -1:
-                return True
-        return None
-
+    for ky in keyword_list:
+        if content.find(ky) != -1:
+            return True
+    return None

channel/wechat/wechaty_channel.py

@@ -14,6 +14,7 @@ import pysilk
 import wave
 from pydub import AudioSegment
 from typing import Optional, Union
+from bridge.context import Context, ContextType
 from wechaty_puppet import MessageType, FileBox, ScanStatus  # type: ignore
 from wechaty import Wechaty, Contact
 from wechaty.user import Message, Room, MiniProgram, UrlLink
@@ -180,9 +181,9 @@ class WechatyChannel(Channel):
         try:
             if not query:
                 return
-            context = dict()
+            context = Context(ContextType.TEXT, query)
             context['session_id'] = reply_user_id
-            reply_text = super().build_reply_content(query, context)
+            reply_text = super().build_reply_content(query, context).content
             if reply_text:
                 await self.send(conf().get("single_chat_reply_prefix") + reply_text, reply_user_id)
         except Exception as e:
@@ -193,12 +194,12 @@ class WechatyChannel(Channel):
         try:
             if not query:
                 return
-            context = dict()
+            context = Context(ContextType.TEXT, query)
             context['session_id'] = reply_user_id
-            reply_text = super().build_reply_content(query, context)
+            reply_text = super().build_reply_content(query, context).content
             if reply_text:
                 # 转换 mp3 文件为 silk 格式
-                mp3_file = super().build_text_to_voice(reply_text)
+                mp3_file = super().build_text_to_voice(reply_text).content
                 silk_file = mp3_file.replace(".mp3", ".silk")
                 # Load the MP3 file
                 audio = AudioSegment.from_file(mp3_file, format="mp3")
@@ -225,9 +226,8 @@ class WechatyChannel(Channel):
         try:
             if not query:
                 return
-            context = dict()
-            context['type'] = 'IMAGE_CREATE'
-            img_url = super().build_reply_content(query, context)
+            context = Context(ContextType.IMAGE_CREATE, query)
+            img_url = super().build_reply_content(query, context).content
             if not img_url:
                 return
             # 图片下载
@@ -248,7 +248,7 @@ class WechatyChannel(Channel):
     async def _do_send_group(self, query, group_id, group_name, group_user_id, group_user_name):
         if not query:
             return
-        context = dict()
+        context = Context(ContextType.TEXT, query)
         group_chat_in_one_session = conf().get('group_chat_in_one_session', [])
         if ('ALL_GROUP' in group_chat_in_one_session or \
                 group_name in group_chat_in_one_session or \
@@ -256,7 +256,7 @@ class WechatyChannel(Channel):
             context['session_id'] = str(group_id)
         else:
             context['session_id'] = str(group_id) + '-' + str(group_user_id)
-        reply_text = super().build_reply_content(query, context)
+        reply_text = super().build_reply_content(query, context).content
         if reply_text:
             reply_text = '@' + group_user_name + ' ' + reply_text.strip()
             await self.send_group(conf().get("group_chat_reply_prefix", "") + reply_text, group_id)
@@ -265,9 +265,8 @@ class WechatyChannel(Channel):
         try:
             if not query:
                 return
-            context = dict()
-            context['type'] = 'IMAGE_CREATE'
-            img_url = super().build_reply_content(query, context)
+            context = Context(ContextType.IMAGE_CREATE, query)
+            img_url = super().build_reply_content(query, context).content
             if not img_url:
                 return
             # 图片发送