fix: chat when single_chat_prefix is None

channel/chat_channel.py

@@ -97,7 +97,7 @@ class ChatChannel(Channel):
                         logger.info("[WX]receive group voice, but checkprefix didn't match")
                     return None
             else: # 单聊
-                match_prefix = check_prefix(content, conf().get('single_chat_prefix'))  
+                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: # 如果源消息是私聊的语音消息，允许不匹配前缀，放宽条件
@@ -231,12 +231,15 @@ class ChatChannel(Channel):
                 time.sleep(3+3*retry_cnt)
                 self._send(reply, context, retry_cnt+1)
 
-    def thread_pool_callback(self, session_id):
+    def _fail_callback(self, session_id, exception, **kwargs): # 线程异常结束时的回调函数
+        logger.exception("Worker return exception: {}".format(exception))
+
+    def _thread_pool_callback(self, session_id, **kwargs):
         def func(worker:Future):
             try:
                 worker_exception = worker.exception()
                 if worker_exception:
-                    logger.exception("Worker return exception: {}".format(worker_exception))
+                    self._fail_callback(session_id, exception = worker_exception, **kwargs)
             except CancelledError as e:
                 logger.info("Worker cancelled, session_id = {}".format(session_id))
             except Exception as e:
@@ -249,7 +252,7 @@ class ChatChannel(Channel):
         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))]
+                self.sessions[session_id] = [Dequeue(), threading.BoundedSemaphore(conf().get("concurrency_in_session", 4))]
             if context.type == ContextType.TEXT and context.content.startswith("#"): 
                 self.sessions[session_id][0].putleft(context) # 优先处理管理命令
             else:
@@ -267,7 +270,7 @@ class ChatChannel(Channel):
                             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))
+                            future.add_done_callback(self._thread_pool_callback(session_id, context = context))
                             if session_id not in self.futures:
                                 self.futures[session_id] = []
                             self.futures[session_id].append(future)
@@ -302,6 +305,8 @@ class ChatChannel(Channel):
     
 
 def check_prefix(content, prefix_list):
+    if not prefix_list:
+        return None
     for prefix in prefix_list:
         if content.startswith(prefix):
             return prefix

channel/chat_message.py

@@ -1,27 +1,29 @@
 
 """ 
-本类表示聊天消息，用于对itchat和wechaty的消息进行统一的封装
+本类表示聊天消息，用于对itchat和wechaty的消息进行统一的封装。
+
+填好必填项(群聊6个，非群聊8个)，即可接入ChatChannel，并支持插件，参考TerminalChannel
 
 ChatMessage
-msg_id: 消息id
+msg_id: 消息id (必填)
 create_time: 消息创建时间
 
-ctype: 消息类型 : ContextType
-content: 消息内容, 如果是声音/图片，这里是文件路径
+ctype: 消息类型 : ContextType (必填)
+content: 消息内容, 如果是声音/图片，这里是文件路径 (必填)
 
-from_user_id: 发送者id
+from_user_id: 发送者id (必填)
 from_user_nickname: 发送者昵称
-to_user_id: 接收者id
+to_user_id: 接收者id (必填)
 to_user_nickname: 接收者昵称
 
-other_user_id: 对方的id，如果你是发送者，那这个就是接收者id，如果你是接收者，那这个就是发送者id，如果是群消息，那这一直是群id
+other_user_id: 对方的id，如果你是发送者，那这个就是接收者id，如果你是接收者，那这个就是发送者id，如果是群消息，那这一直是群id (必填)
 other_user_nickname: 同上
 
-is_group: 是否是群消息
-is_at: 是否被at
+is_group: 是否是群消息 (群聊必填)
+is_at: 是否被at 
 
 - (群消息时，一般会存在实际发送者，是群内某个成员的id和昵称，下列项仅在群消息时存在)
-actual_user_id: 实际发送者id
+actual_user_id: 实际发送者id (群聊必填)
 actual_user_nickname：实际发送者昵称
 
 

channel/wechatmp/wechatmp_channel.py

@@ -7,6 +7,7 @@ import textwrap
 from channel.chat_channel import ChatChannel
 import channel.wechatmp.reply as reply
 import channel.wechatmp.receive as receive
+from common.expired_dict import ExpiredDict
 from common.singleton import singleton
 from common.log import logger
 from config import conf
@@ -26,16 +27,18 @@ import traceback
 # from concurrent.futures import ThreadPoolExecutor
 # thread_pool = ThreadPoolExecutor(max_workers=8)
 
+MAX_UTF8_LEN = 2048
 @singleton
 class WechatMPChannel(ChatChannel):
     NOT_SUPPORT_REPLYTYPE = [ReplyType.IMAGE, ReplyType.VOICE]
     def __init__(self):
         super().__init__()
         self.cache_dict = dict()
+        self.running = set()
         self.query1 = dict()
         self.query2 = dict()
         self.query3 = dict()
-
+        self.received_msgs = ExpiredDict(60*60*24) 
 
     def startup(self):
         urls = (
@@ -47,11 +50,16 @@ class WechatMPChannel(ChatChannel):
 
 
     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)
+        self.cache_dict[receiver] = reply.content
+        self.running.remove(receiver)
         logger.debug("[send] reply to {} saved to cache: {}".format(receiver, reply))
 
+    def _fail_callback(self, session_id, exception, context, **kwargs):
+        logger.exception("[wechatmp] Fail to generation message to user, msgId={}, exception={}".format(context['msg'].msg_id, exception))
+        assert session_id not in self.cache_dict
+        self.running.remove(session_id)
+
 
 def verify_server():
     try:
@@ -86,11 +94,11 @@ class SubsribeAccountQuery():
         return verify_server()
 
     def POST(self):
-        channel_instance = WechatMPChannel()
+        channel = WechatMPChannel()
         try:
             query_time = time.time()
             webData = web.data()
-            # logger.debug("[wechatmp] Receive request:\n" + webData.decode("utf-8"))
+            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
@@ -99,93 +107,99 @@ class SubsribeAccountQuery():
                 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))
-
+                supported = True
+                if "【收到不支持的消息类型，暂无法显示】" in message:
+                    supported = False # not supported, used to refresh
                 cache_key = from_user
-                cache = channel_instance.cache_dict.get(cache_key)
 
                 reply_text = ""
                 # New request
-                if cache == None:
+                if cache_key not in channel.cache_dict and cache_key not in channel.running:
                     # The first query begin, reset the cache
-                    context = channel_instance._compose_context(ContextType.TEXT, message, isgroup=False, msg=wechat_msg)
+                    context = channel._compose_context(ContextType.TEXT, message, isgroup=False, msg=wechat_msg)
                     logger.debug("[wechatmp] context: {} {}".format(context, wechat_msg))
-                    if context:
+                    if message_id in channel.received_msgs: # received and finished
+                        return 
+                    if supported and 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)
+                        channel.received_msgs[message_id] = wechat_msg
+                        channel.running.add(cache_key)
+                        channel.produce(context)
                     else:
                         trigger_prefix = conf().get('single_chat_prefix',[''])[0]
-                        if trigger_prefix:
-                            content = textwrap.dedent(f"""\
-                                请输入'{trigger_prefix}'接你想说的话跟我说话。
-                                例如:
-                                {trigger_prefix}你好，很高兴见到你。""")
+                        if trigger_prefix or not supported:
+                            if trigger_prefix:
+                                content = textwrap.dedent(f"""\
+                                    请输入'{trigger_prefix}'接你想说的话跟我说话。
+                                    例如:
+                                    {trigger_prefix}你好，很高兴见到你。""")
+                            else:
+                                content = textwrap.dedent("""\
+                                    你好，很高兴见到你。
+                                    请跟我说话吧。""")
                         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
+                    channel.query1[cache_key] = False
+                    channel.query2[cache_key] = False
+                    channel.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:
+                elif cache_key in channel.running:
+                    channel.query1[cache_key] = False  #To improve waiting experience, this can be set to True.
+                    channel.query2[cache_key] = False  #To improve waiting experience, this can be set to True.
+                    channel.query3[cache_key] = False
+                elif cache_key in channel.cache_dict:
                     # Skip the waiting phase
-                    channel_instance.query1[cache_key] = True
-                    channel_instance.query2[cache_key] = True
-                    channel_instance.query3[cache_key] = True
+                    channel.query1[cache_key] = True
+                    channel.query2[cache_key] = True
+                    channel.query3[cache_key] = True
 
+                assert not (cache_key in channel.cache_dict and cache_key in channel.running)
 
-                cache = channel_instance.cache_dict.get(cache_key)
-                if channel_instance.query1.get(cache_key) == False:
+                if channel.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
+                    channel.query1[cache_key] = True
                     cnt = 0
-                    while cache[0] == 0 and cnt < 45:
+                    while cache_key not in channel.cache_dict 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)
+                        time.sleep(1)
                         # and do nothing
                         return
                     else:
                         pass
-                elif channel_instance.query2.get(cache_key) == False:
+                elif channel.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
+                    channel.query2[cache_key] = True
                     cnt = 0
-                    while cache[0] == 0 and cnt < 45:
+                    while cache_key not in channel.cache_dict 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)
+                        time.sleep(1)
                         # and do nothing
                         return
                     else:
                         pass
-                elif channel_instance.query3.get(cache_key) == False:
+                elif channel.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
+                    channel.query3[cache_key] = True
                     cnt = 0
-                    while cache[0] == 0 and cnt < 40:
+                    while cache_key not in channel.cache_dict 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
@@ -197,16 +211,21 @@ class SubsribeAccountQuery():
                         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)
+                    reply_text = "【正在思考中，回复任意文字尝试获取回复】"
+                    logger.info("[wechatmp] Timeout for {} {}, return".format(from_user, message_id))
+                    replyPost = reply.TextMsg(from_user, to_user, reply_text).send()
+                    return replyPost
+                
+                if cache_key in channel.cache_dict:
+                    content = channel.cache_dict[cache_key]
+                    if len(content.encode('utf8'))<=MAX_UTF8_LEN:
+                        reply_text = channel.cache_dict[cache_key]
+                        channel.cache_dict.pop(cache_key)
+                    else:
+                        continue_text = "\n【未完待续，回复任意文字以继续】"
+                        splits = split_string_by_utf8_length(content, MAX_UTF8_LEN - len(continue_text.encode('utf-8')), max_split= 1)
+                        reply_text = splits[0] + continue_text
+                        channel.cache_dict[cache_key] = splits[1]
                 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
@@ -232,3 +251,18 @@ class SubsribeAccountQuery():
             logger.exception(exc)
             return exc
 
+def split_string_by_utf8_length(string, max_length, max_split=0):
+    encoded = string.encode('utf-8')
+    start, end = 0, 0
+    result = []
+    while end < len(encoded):
+        if max_split > 0 and len(result) >= max_split:
+            result.append(encoded[start:].decode('utf-8'))
+            break
+        end = start + max_length
+        # 如果当前字节不是 UTF-8 编码的开始字节，则向前查找直到找到开始字节为止
+        while end < len(encoded) and (encoded[end] & 0b11000000) == 0b10000000:
+            end -= 1
+        result.append(encoded[start:end].decode('utf-8'))
+        start = end
+    return result

plugins/plugin_manager.py

@@ -238,7 +238,7 @@ class PluginManager:
             if os.path.exists(os.path.join(dirname,"requirements.txt")):
                 logger.info("detect requirements.txt，installing...")
             pkgmgr.install_requirements(os.path.join(dirname,"requirements.txt"))
-            return True, "安装插件成功，请使用 #scanp 命令扫描插件或重启程序"
+            return True, "安装插件成功，请使用 #scanp 命令扫描插件或重启程序，开启前请检查插件是否需要配置"
         except Exception as e:
             logger.error("Failed to install plugin, {}".format(e))
             return False, "安装插件失败，"+str(e)

plugins/source.json

@@ -3,6 +3,10 @@
         "sdwebui": {
             "url": "https://github.com/lanvent/plugin_sdwebui.git",
             "desc": "利用stable-diffusion画图的插件"
+        },
+        "replicate": {
+            "url": "https://github.com/lanvent/plugin_replicate.git",
+            "desc": "利用replicate api画图的插件"
         }
     }
 }