From 4f3d12129c35787d1c54d1175d4f5e5c9e2dd390 Mon Sep 17 00:00:00 2001 From: lanvent Date: Fri, 31 Mar 2023 01:29:04 +0800 Subject: [PATCH 1/5] refactor: wrap wechat msg to make logic general --- channel/chat_message.py | 64 ++++++++++++++++++++++ channel/wechat/wechat_channel.py | 93 ++++++++++++++------------------ channel/wechat/wechat_message.py | 57 ++++++++++++++++++++ 3 files changed, 160 insertions(+), 54 deletions(-) create mode 100644 channel/chat_message.py create mode 100644 channel/wechat/wechat_message.py diff --git a/channel/chat_message.py b/channel/chat_message.py new file mode 100644 index 0000000..28ee054 --- /dev/null +++ b/channel/chat_message.py @@ -0,0 +1,64 @@ + +""" +本类表示聊天消息，用于对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 +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() \ No newline at end of file diff --git a/channel/wechat/wechat_channel.py b/channel/wechat/wechat_channel.py index 2e7b075..cdbb333 100644 --- a/channel/wechat/wechat_channel.py +++ b/channel/wechat/wechat_channel.py @@ -9,6 +9,7 @@ import re import requests import io import time +from channel.wechat.wechat_message import * from common.singleton import singleton from lib import itchat import json @@ -38,36 +39,36 @@ def thread_pool_callback(worker): @itchat.msg_register(TEXT) def handler_single_msg(msg): - WechatChannel().handle_text(msg) + WechatChannel().handle_text(WeChatMessage(msg)) return None @itchat.msg_register(TEXT, isGroupChat=True) def handler_group_msg(msg): - WechatChannel().handle_group(msg) + WechatChannel().handle_group(WeChatMessage(msg,True)) return None @itchat.msg_register(VOICE) def handler_single_voice(msg): - WechatChannel().handle_voice(msg) + WechatChannel().handle_voice(WeChatMessage(msg)) return None @itchat.msg_register(VOICE, isGroupChat=True) def handler_group_voice(msg): - WechatChannel().handle_group_voice(msg) + WechatChannel().handle_group_voice(WeChatMessage(msg,True)) return None def _check(func): - def wrapper(self, msg): - msgId = msg['MsgId'] + 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] = msg - create_time = msg['CreateTime'] # 消息时间 + 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, msg) + return func(self, cmsg) return wrapper @singleton @@ -112,57 +113,40 @@ class WechatChannel(Channel): @time_checker @_check - def handle_voice(self, msg): + def handle_voice(self, cmsg : ChatMessage): if conf().get('speech_recognition') != True: return - logger.debug("[WX]receive voice msg: " + msg['FileName']) - to_user_id = msg['ToUserName'] - from_user_id = msg['FromUserName'] - try: - other_user_id = msg['User']['UserName'] # 对手方id - except Exception as e: - logger.warn("[WX]get other_user_id failed: " + str(e)) - if from_user_id == self.userName: - other_user_id = to_user_id - else: - other_user_id = from_user_id + logger.debug("[WX]receive voice msg: {}".format(cmsg.content)) + from_user_id = cmsg.from_user_id + other_user_id = cmsg.other_user_id if from_user_id == other_user_id: - context = self._compose_context(ContextType.VOICE, msg['FileName'], isgroup=False, msg=msg, receiver=other_user_id, session_id=other_user_id) + context = self._compose_context(ContextType.VOICE, cmsg.content, isgroup=False, msg=cmsg, receiver=other_user_id, session_id=other_user_id) if context: thread_pool.submit(self.handle, context).add_done_callback(thread_pool_callback) @time_checker @_check - def handle_text(self, msg): - logger.debug("[WX]receive text msg: " + json.dumps(msg, ensure_ascii=False)) - content = msg['Text'] - from_user_id = msg['FromUserName'] - to_user_id = msg['ToUserName'] # 接收人id - try: - other_user_id = msg['User']['UserName'] # 对手方id - except Exception as e: - logger.warn("[WX]get other_user_id failed: " + str(e)) - if from_user_id == self.userName: - other_user_id = to_user_id - else: - other_user_id = from_user_id - if "」\n- - - - - - - - - - - - - - -" in content: + def handle_text(self, cmsg : ChatMessage): + logger.debug("[WX]receive text msg: " + json.dumps(cmsg._rawmsg, ensure_ascii=False)) + content = cmsg.content + other_user_id = cmsg.other_user_id + if "」\n- - - - - - - - - - - - - - -" in cmsg.content: logger.debug("[WX]reference query skipped") return - context = self._compose_context(ContextType.TEXT, content, isgroup=False, msg=msg, receiver=other_user_id, session_id=other_user_id) + context = self._compose_context(ContextType.TEXT, content, isgroup=False, msg=cmsg, receiver=other_user_id, session_id=other_user_id) if context: thread_pool.submit(self.handle, context).add_done_callback(thread_pool_callback) @time_checker @_check - def handle_group(self, msg): - logger.debug("[WX]receive group msg: " + json.dumps(msg, ensure_ascii=False)) - group_name = msg['User'].get('NickName', None) - group_id = msg['User'].get('UserName', None) + def handle_group(self, cmsg : ChatMessage): + logger.debug("[WX]receive group msg: " + json.dumps(cmsg._rawmsg, ensure_ascii=False)) + group_name = cmsg.other_user_nickname + group_id = cmsg.other_user_id if not group_name: return "" - content = msg.content + content = cmsg.content if "」\n- - - - - - - - - - - - - - -" in content: logger.debug("[WX]reference query skipped") return "" @@ -175,21 +159,22 @@ class WechatChannel(Channel): 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 = msg['ActualUserName'] + 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 - context = self._compose_context(ContextType.TEXT, content, isgroup=True, msg=msg, receiver=group_id, session_id=session_id) + + context = self._compose_context(ContextType.TEXT, content, isgroup=True, msg=cmsg, receiver=group_id, session_id=session_id) if context: thread_pool.submit(self.handle, context).add_done_callback(thread_pool_callback) @time_checker @_check - def handle_group_voice(self, msg): + def handle_group_voice(self, cmsg : ChatMessage): if conf().get('group_speech_recognition', False) != True: return - logger.debug("[WX]receive voice for group msg: " + msg['FileName']) - group_name = msg['User'].get('NickName', None) - group_id = msg['User'].get('UserName', None) + logger.debug("[WX]receive voice for group msg: {}".format(cmsg.content)) + group_name = cmsg.other_user_nickname + group_id = cmsg.other_user_id # 验证群名 if not group_name: return "" @@ -199,10 +184,10 @@ class WechatChannel(Channel): 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 =msg['ActualUserName'] + 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 - context = self._compose_context(ContextType.VOICE, msg['FileName'], isgroup=True, msg=msg, receiver=group_id, session_id=session_id) + context = self._compose_context(ContextType.VOICE, cmsg.content, isgroup=True, msg=cmsg, receiver=group_id, session_id=session_id) if context: thread_pool.submit(self.handle, context).add_done_callback(thread_pool_callback) @@ -222,7 +207,7 @@ class WechatChannel(Channel): # 判断如果匹配到自定义前缀，则返回过滤掉前缀+空格后的内容，用于实现类似自定义+前缀触发生成AI图片的功能 if match_prefix: content = content.replace(match_prefix, '', 1).strip() - elif context['msg']['IsAt'] and not conf().get("group_at_off", False): + elif context['msg'].is_at and not conf().get("group_at_off", False): logger.info("[WX]receive group at, continue") elif context["origin_ctype"] == ContextType.VOICE: logger.info("[WX]receive group voice, checkprefix didn't match") @@ -306,8 +291,8 @@ class WechatChannel(Channel): reply = super().build_reply_content(context.content, context) elif context.type == ContextType.VOICE: # 语音消息 msg = context['msg'] - mp3_path = TmpDir().path() + context.content - msg.download(mp3_path) + msg.prepare() + mp3_path = context.content # mp3转wav wav_path = os.path.splitext(mp3_path)[0] + '.wav' try: @@ -349,7 +334,7 @@ class WechatChannel(Channel): reply = super().build_text_to_voice(reply.content) return self._decorate_reply(context, reply) if context['isgroup']: - reply_text = '@' + context['msg']['ActualNickName'] + ' ' + reply_text.strip() + 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 diff --git a/channel/wechat/wechat_message.py b/channel/wechat/wechat_message.py new file mode 100644 index 0000000..48a5867 --- /dev/null +++ b/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'] From 50a03e7c15bc77007d993428aab8268d108e42a8 Mon Sep 17 00:00:00 2001 From: lanvent Date: Fri, 31 Mar 2023 04:11:25 +0800 Subject: [PATCH 2/5] refactor: wrap wechaty message --- channel/chat_message.py | 20 +++++++++- channel/wechat/wechaty_message.py | 65 +++++++++++++++++++++++++++++++ 2 files changed, 84 insertions(+), 1 deletion(-) create mode 100644 channel/wechat/wechaty_message.py diff --git a/channel/chat_message.py b/channel/chat_message.py index 28ee054..9fc2075 100644 --- a/channel/chat_message.py +++ b/channel/chat_message.py @@ -61,4 +61,22 @@ class ChatMessage(object): def prepare(self): if self._prepare_fn and not self._prepared: self._prepared = True - self._prepare_fn() \ No newline at end of file + 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, + ) \ No newline at end of file diff --git a/channel/wechat/wechaty_message.py b/channel/wechat/wechaty_message.py new file mode 100644 index 0000000..f5ef0ef --- /dev/null +++ b/channel/wechat/wechaty_message.py @@ -0,0 +1,65 @@ +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直接存临时目录路径 + self._prepare_fn = lambda: voice_file.to_file(self.content) + 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 + + 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 wechaty_msg.is_self(): + 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: + self.is_at = await wechaty_msg.mention_self() + self.actual_user_id = self.other_user_id + self.actual_user_nickname = self.other_user_nickname From 87df588c8065749fc00ab569bffa7907d07a4ff4 Mon Sep 17 00:00:00 2001 From: lanvent Date: Fri, 31 Mar 2023 22:31:10 +0800 Subject: [PATCH 3/5] refactor: stripp processing unrelated to channel --- channel/channel.py | 2 +- channel/chat_channel.py | 223 ++++++++++++++++++++++++ channel/chat_message.py | 3 +- channel/wechat/wechat_channel.py | 284 +++++-------------------------- 4 files changed, 267 insertions(+), 245 deletions(-) create mode 100644 channel/chat_channel.py diff --git a/channel/channel.py b/channel/channel.py index 62e1ad3..f65a031 100644 --- a/channel/channel.py +++ b/channel/channel.py @@ -20,7 +20,7 @@ class Channel(object): """ raise NotImplementedError - def send(self, msg, receiver): + def send(self, reply: Reply, context: Context): """ send message to user :param msg: message content diff --git a/channel/chat_channel.py b/channel/chat_channel.py new file mode 100644 index 0000000..42470f9 --- /dev/null +++ b/channel/chat_channel.py @@ -0,0 +1,223 @@ + + + +import re +import time +from common.expired_dict import ExpiredDict +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 * + + +# 抽象类, 它包含了与消息通道无关的通用处理逻辑 +class ChatChannel(Channel): + name = None # 登录的用户名 + user_id = None # 登录的用户id + def __init__(self): + pass + + # 根据消息构造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 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 + 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')) + if match_prefix is not None or match_contain is not None: + if match_prefix: + content = content.replace(match_prefix, '', 1).strip() + elif context['msg'].is_at and not conf().get("group_at_off", False): + logger.info("[WX]receive group at, continue") + pattern = f'@{self.name}(\u2005|\u0020)' + content = re.sub(pattern, r'', content) + elif context["origin_ctype"] == ContextType.VOICE: + logger.info("[WX]receive group voice, checkprefix didn't match") + return None + else: + 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 + elif context.type == ContextType.VOICE: + if 'desire_rtype' not in context and conf().get('voice_reply_voice'): + context['desire_rtype'] = ReplyType.VOICE + + + return context + + # 统一的发送函数，每个Channel自行实现，根据reply的type字段发送不同类型的消息 + def send(self, reply: Reply, context: Context, retry_cnt = 0): + raise NotImplementedError + + # 处理消息 TODO: 如果wechaty解耦，此处逻辑可以放置到父类 + 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: # 语音消息 + msg = context['msg'] + msg.prepare() + mp3_path = context.content + # mp3转wav + wav_path = os.path.splitext(mp3_path)[0] + '.wav' + try: + mp3_to_wav(mp3_path=mp3_path, wav_path=wav_path) + except Exception as e: # 转换失败，直接使用mp3，对于某些api，mp3也可以识别 + logger.warning("[WX]mp3 to wav error, use mp3 path. " + str(e)) + wav_path = mp3_path + # 语音识别 + reply = super().build_voice_to_text(wav_path) + # 删除临时文件 + try: + os.remove(wav_path) + os.remove(mp3_path) + except Exception as e: + 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 == ReplyType.TEXT: + reply_text = reply.content + if desire_rtype == ReplyType.VOICE: + 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: {} to {}'.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(e)) + if retry_cnt < 2: + time.sleep(3+3*retry_cnt) + self._send(reply, context, retry_cnt+1) + + + +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 diff --git a/channel/chat_message.py b/channel/chat_message.py index 9fc2075..4fdb1dd 100644 --- a/channel/chat_message.py +++ b/channel/chat_message.py @@ -1,6 +1,7 @@ """ 本类表示聊天消息，用于对itchat和wechaty的消息进行统一的封装 + ChatMessage msg_id: 消息id create_time: 消息创建时间 @@ -13,7 +14,7 @@ from_user_nickname: 发送者昵称 to_user_id: 接收者id to_user_nickname: 接收者昵称 -other_user_id: 对方的id，如果你是发送者，那这个就是接收者id，如果你是接收者，那这个就是发送者id +other_user_id: 对方的id，如果你是发送者，那这个就是接收者id，如果你是接收者，那这个就是发送者id，如果是群消息，那这一直是群id other_user_nickname: 同上 is_group: 是否是群消息 diff --git a/channel/wechat/wechat_channel.py b/channel/wechat/wechat_channel.py index cdbb333..6e9abaa 100644 --- a/channel/wechat/wechat_channel.py +++ b/channel/wechat/wechat_channel.py @@ -5,10 +5,10 @@ wechat channel """ import os -import re import requests import io import time +from channel.chat_channel import ChatChannel from channel.wechat.wechat_message import * from common.singleton import singleton from lib import itchat @@ -19,7 +19,6 @@ 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 common.expired_dict import ExpiredDict @@ -72,10 +71,9 @@ def _check(func): return wrapper @singleton -class WechatChannel(Channel): +class WechatChannel(ChatChannel): def __init__(self): - self.user_id = None - self.name = None + super().__init__() self.receivedMsgs = ExpiredDict(60*60*24) def startup(self): @@ -99,7 +97,7 @@ class WechatChannel(Channel): # start message listener itchat.run() - # handle_* 系列函数处理收到的消息后构造Context，然后传入handle函数中处理Context和发送回复 + # handle_* 系列函数处理收到的消息后构造Context，然后传入_handle函数中处理Context和发送回复 # Context包含了消息的所有信息，包括以下属性 # type 消息类型, 包括TEXT、VOICE、IMAGE_CREATE # content 消息内容，如果是TEXT类型，content就是文本内容，如果是VOICE类型，content就是语音文件名，如果是IMAGE_CREATE类型，content就是图片生成命令 @@ -107,7 +105,7 @@ class WechatChannel(Channel): # session_id: 会话id # isgroup: 是否是群聊 # receiver: 需要回复的对象 - # msg: itchat的原始消息对象 + # msg: ChatMessage消息对象 # origin_ctype: 原始消息类型，语音转文字后，私聊时如果匹配前缀失败，会根据初始消息是否是语音来放宽触发规则 # desire_rtype: 希望回复类型，默认是文本回复，设置为ReplyType.VOICE是语音回复 @@ -117,55 +115,25 @@ class WechatChannel(Channel): if conf().get('speech_recognition') != True: return logger.debug("[WX]receive voice msg: {}".format(cmsg.content)) - from_user_id = cmsg.from_user_id - other_user_id = cmsg.other_user_id - if from_user_id == other_user_id: - context = self._compose_context(ContextType.VOICE, cmsg.content, isgroup=False, msg=cmsg, receiver=other_user_id, session_id=other_user_id) - if context: - thread_pool.submit(self.handle, context).add_done_callback(thread_pool_callback) + context = self._compose_context(ContextType.VOICE, cmsg.content, isgroup=False, msg=cmsg) + if context: + thread_pool.submit(self._handle, context).add_done_callback(thread_pool_callback) @time_checker @_check def handle_text(self, cmsg : ChatMessage): logger.debug("[WX]receive text msg: " + json.dumps(cmsg._rawmsg, ensure_ascii=False)) - content = cmsg.content - other_user_id = cmsg.other_user_id - if "」\n- - - - - - - - - - - - - - -" in cmsg.content: - logger.debug("[WX]reference query skipped") - return - - context = self._compose_context(ContextType.TEXT, content, isgroup=False, msg=cmsg, receiver=other_user_id, session_id=other_user_id) + context = self._compose_context(ContextType.TEXT, cmsg.content, isgroup=False, msg=cmsg) if context: - thread_pool.submit(self.handle, context).add_done_callback(thread_pool_callback) + thread_pool.submit(self._handle, context).add_done_callback(thread_pool_callback) @time_checker @_check def handle_group(self, cmsg : ChatMessage): logger.debug("[WX]receive group msg: " + json.dumps(cmsg._rawmsg, ensure_ascii=False)) - group_name = cmsg.other_user_nickname - group_id = cmsg.other_user_id - if not group_name: - return "" - content = cmsg.content - if "」\n- - - - - - - - - - - - - - -" in content: - logger.debug("[WX]reference query skipped") - return "" - pattern = f'@{self.name}(\u2005|\u0020)' - content = re.sub(pattern, r'', content) - - config = conf() - 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 - - context = self._compose_context(ContextType.TEXT, content, isgroup=True, msg=cmsg, receiver=group_id, session_id=session_id) - if context: - thread_pool.submit(self.handle, context).add_done_callback(thread_pool_callback) + context = self._compose_context(ContextType.TEXT, cmsg.content, isgroup=True, msg=cmsg) + if context: + thread_pool.submit(self._handle, context).add_done_callback(thread_pool_callback) @time_checker @_check @@ -173,203 +141,33 @@ class WechatChannel(Channel): if conf().get('group_speech_recognition', False) != True: return logger.debug("[WX]receive voice for group msg: {}".format(cmsg.content)) - group_name = cmsg.other_user_nickname - group_id = cmsg.other_user_id - # 验证群名 - if not group_name: - return "" - - config = conf() - 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 - context = self._compose_context(ContextType.VOICE, cmsg.content, isgroup=True, msg=cmsg, receiver=group_id, session_id=session_id) - if context: - thread_pool.submit(self.handle, context).add_done_callback(thread_pool_callback) - - # 根据消息构造context，消息内容相关的触发项写在这里 - def _compose_context(self, ctype: ContextType, content, **kwargs): - context = Context(ctype, content) - context.kwargs = kwargs - if 'origin_ctype' not in context: - context['origin_ctype'] = ctype - - if ctype == ContextType.TEXT: - if context["isgroup"]: # 群聊 - # 校验关键字 - match_prefix = check_prefix(content, conf().get('group_chat_prefix')) - match_contain = check_contain(content, conf().get('group_chat_keyword')) - if match_prefix is not None or match_contain is not None: - # 判断如果匹配到自定义前缀，则返回过滤掉前缀+空格后的内容，用于实现类似自定义+前缀触发生成AI图片的功能 - if match_prefix: - content = content.replace(match_prefix, '', 1).strip() - elif context['msg'].is_at and not conf().get("group_at_off", False): - logger.info("[WX]receive group at, continue") - elif context["origin_ctype"] == ContextType.VOICE: - logger.info("[WX]receive group voice, checkprefix didn't match") - return None - else: - 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 - elif context.type == ContextType.VOICE: - if 'desire_rtype' not in context and conf().get('voice_reply_voice'): - context['desire_rtype'] = ReplyType.VOICE - return context + context = self._compose_context(ContextType.VOICE, cmsg.content, isgroup=True, msg=cmsg) + if context: + thread_pool.submit(self._handle, context).add_done_callback(thread_pool_callback) # 统一的发送函数，每个Channel自行实现，根据reply的type字段发送不同类型的消息 - def send(self, reply: Reply, receiver, retry_cnt = 0): - try: - 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)) - except Exception as e: - logger.error('[WX] sendMsg error: {}, receiver={}'.format(e, receiver)) - if retry_cnt < 2: - time.sleep(3+3*retry_cnt) - self.send(reply, receiver, retry_cnt + 1) - - # 处理消息 TODO: 如果wechaty解耦，此处逻辑可以放置到父类 - 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: # 语音消息 - msg = context['msg'] - msg.prepare() - mp3_path = context.content - # mp3转wav - wav_path = os.path.splitext(mp3_path)[0] + '.wav' - try: - mp3_to_wav(mp3_path=mp3_path, wav_path=wav_path) - except Exception as e: # 转换失败，直接使用mp3，对于某些api，mp3也可以识别 - logger.warning("[WX]mp3 to wav error, use mp3 path. " + str(e)) - wav_path = mp3_path - # 语音识别 - reply = super().build_voice_to_text(wav_path) - # 删除临时文件 - try: - os.remove(wav_path) - os.remove(mp3_path) - except Exception as e: - 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 == ReplyType.TEXT: - reply_text = reply.content - if desire_rtype == ReplyType.VOICE: - 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: {} to {}'.format(reply, context['receiver'])) - self.send(reply, context['receiver']) - - -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 + 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)) From 02cd553990f73b7b9b18fd806e85cd508b3aa50f Mon Sep 17 00:00:00 2001 From: lanvent Date: Sat, 1 Apr 2023 04:23:19 +0800 Subject: [PATCH 4/5] refactor: using one processing logic in chat channel --- channel/channel.py | 1 + channel/chat_channel.py | 36 +-- channel/wechat/wechat_channel.py | 14 +- channel/wechat/wechat_message.py | 2 +- channel/wechat/wechaty_channel.py | 403 ++++++++---------------------- channel/wechat/wechaty_message.py | 32 ++- voice/audio_convert.py | 10 + 7 files changed, 159 insertions(+), 339 deletions(-) diff --git a/channel/channel.py b/channel/channel.py index f65a031..8aae3a6 100644 --- a/channel/channel.py +++ b/channel/channel.py @@ -20,6 +20,7 @@ class Channel(object): """ raise NotImplementedError + # 统一的发送函数，每个Channel自行实现，根据reply的type字段发送不同类型的消息 def send(self, reply: Reply, context: Context): """ send message to user diff --git a/channel/chat_channel.py b/channel/chat_channel.py index 42470f9..1efe391 100644 --- a/channel/chat_channel.py +++ b/channel/chat_channel.py @@ -1,6 +1,7 @@ +import os import re import time from common.expired_dict import ExpiredDict @@ -10,7 +11,10 @@ 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): @@ -30,11 +34,13 @@ class ChatChannel(Channel): 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: + logger.debug("[WX]self message skipped") + return None if context["isgroup"]: group_name = cmsg.other_user_nickname group_id = cmsg.other_user_id @@ -47,14 +53,13 @@ class ChatChannel(Channel): if any([group_name in group_chat_in_one_session, 'ALL_GROUP' in group_chat_in_one_session]): session_id = group_id else: - return + 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: # 初次匹配 过滤引用消息 @@ -99,10 +104,6 @@ class ChatChannel(Channel): return context - - # 统一的发送函数，每个Channel自行实现，根据reply的type字段发送不同类型的消息 - def send(self, reply: Reply, context: Context, retry_cnt = 0): - raise NotImplementedError # 处理消息 TODO: 如果wechaty解耦，此处逻辑可以放置到父类 def _handle(self, context: Context): @@ -128,22 +129,21 @@ class ChatChannel(Channel): 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'] - msg.prepare() - mp3_path = context.content - # mp3转wav - wav_path = os.path.splitext(mp3_path)[0] + '.wav' + cmsg = context['msg'] + cmsg.prepare() + file_path = context.content + wav_path = os.path.splitext(file_path)[0] + '.wav' try: - mp3_to_wav(mp3_path=mp3_path, wav_path=wav_path) + any_to_wav(file_path, wav_path) except Exception as e: # 转换失败，直接使用mp3，对于某些api，mp3也可以识别 - logger.warning("[WX]mp3 to wav error, use mp3 path. " + str(e)) - wav_path = mp3_path + 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) os.remove(wav_path) - os.remove(mp3_path) except Exception as e: logger.warning("[WX]delete temp file error: " + str(e)) @@ -204,7 +204,7 @@ class ChatChannel(Channel): logger.error('[WX] sendMsg error: {}'.format(e)) if retry_cnt < 2: time.sleep(3+3*retry_cnt) - self._send(reply, context, retry_cnt+1) + self._send(reply, context, retry_cnt+1) diff --git a/channel/wechat/wechat_channel.py b/channel/wechat/wechat_channel.py index 6e9abaa..df9c0ac 100644 --- a/channel/wechat/wechat_channel.py +++ b/channel/wechat/wechat_channel.py @@ -8,28 +8,22 @@ import os 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 -import json from lib.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 config import conf from common.time_check import time_checker from common.expired_dict import ExpiredDict from plugins import * -try: - from voice.audio_convert import mp3_to_wav -except Exception as e: - pass thread_pool = ThreadPoolExecutor(max_workers=8) - def thread_pool_callback(worker): worker_exception = worker.exception() if worker_exception: @@ -122,7 +116,7 @@ class WechatChannel(ChatChannel): @time_checker @_check def handle_text(self, cmsg : ChatMessage): - logger.debug("[WX]receive text msg: " + json.dumps(cmsg._rawmsg, ensure_ascii=False)) + 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: thread_pool.submit(self._handle, context).add_done_callback(thread_pool_callback) @@ -130,7 +124,7 @@ class WechatChannel(ChatChannel): @time_checker @_check def handle_group(self, cmsg : ChatMessage): - logger.debug("[WX]receive group msg: " + json.dumps(cmsg._rawmsg, ensure_ascii=False)) + 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: thread_pool.submit(self._handle, context).add_done_callback(thread_pool_callback) diff --git a/channel/wechat/wechat_message.py b/channel/wechat/wechat_message.py index 48a5867..43050f8 100644 --- a/channel/wechat/wechat_message.py +++ b/channel/wechat/wechat_message.py @@ -44,7 +44,7 @@ class WeChatMessage(ChatMessage): 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: + 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 diff --git a/channel/wechat/wechaty_channel.py b/channel/wechat/wechaty_channel.py index ac6717c..cb39313 100644 --- a/channel/wechat/wechaty_channel.py +++ b/channel/wechat/wechaty_channel.py @@ -4,21 +4,32 @@ wechaty channel Python Wechaty - https://github.com/wechaty/python-wechaty """ +import base64 +from concurrent.futures import ThreadPoolExecutor import os import time import asyncio -from typing import Optional, Union -from bridge.context import Context, ContextType -from wechaty_puppet import MessageType, FileBox, ScanStatus # type: ignore +from bridge.context import Context +from wechaty_puppet import FileBox from wechaty import Wechaty, Contact -from wechaty.user import Message, MiniProgram, UrlLink -from channel.channel import Channel +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.tmp_dir import TmpDir from config import conf -from voice.audio_convert import sil_to_wav, mp3_to_sil - -class WechatyChannel(Channel): +try: + from voice.audio_convert import mp3_to_sil +except Exception as e: + pass + +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)) +class WechatyChannel(ChatChannel): def __init__(self): pass @@ -28,312 +39,96 @@ class WechatyChannel(Channel): async def main(self): config = conf() - # 使用PadLocal协议 比较稳定(免费web协议 os.environ['WECHATY_PUPPET_SERVICE_ENDPOINT'] = '127.0.0.1:8080') token = config.get('wechaty_puppet_service_token') os.environ['WECHATY_PUPPET_SERVICE_TOKEN'] = token - global bot - bot = Wechaty() - - bot.on('scan', self.on_scan) - bot.on('login', self.on_login) - bot.on('message', self.on_message) - await bot.start() + 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)) - async def on_scan(self, status: ScanStatus, qr_code: Optional[str] = None, - data: Optional[str] = None): - pass - # contact = self.Contact.load(self.contact_id) - # logger.info('[WX] scan user={}, scan status={}, scan qr_code={}'.format(contact, status.name, qr_code)) - # print(f'user <{contact}> scan status: {status.name} , 'f'qr_code: {qr_code}') + # 统一的发送函数，每个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 + if reply.content.endswith('.mp3'): + mp3_file = reply.content + sil_file = os.path.splitext(mp3_file)[0] + '.sil' + voiceLength = mp3_to_sil(mp3_file, sil_file) + try: + os.remove(mp3_file) + except Exception as e: + pass + elif reply.content.endswith('.sil'): + sil_file = reply.content + else: + raise Exception('voice file must be mp3 or sil format') + # 发送语音 + 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(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 """ - from_contact = msg.talker() # 获取消息的发送者 - to_contact = msg.to() # 接收人 - room = msg.room() # 获取消息来自的群聊. 如果消息不是来自群聊, 则返回None - from_user_id = from_contact.contact_id - to_user_id = to_contact.contact_id # 接收人id - # other_user_id = msg['User']['UserName'] # 对手方id - content = msg.text() - mention_content = await msg.mention_text() # 返回过滤掉@name后的消息 - match_prefix = self.check_prefix(content, conf().get('single_chat_prefix')) - # conversation: Union[Room, Contact] = from_contact if room is None else room - - if room is None and msg.type() == MessageType.MESSAGE_TYPE_TEXT: - if not msg.is_self() 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() - - 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() - await self._do_send_img(content, from_user_id) - else: - await self._do_send(content, from_user_id) - elif msg.is_self() 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() - await self._do_send_img(content, to_user_id) - else: - await self._do_send(content, to_user_id) - elif room is None and msg.type() == MessageType.MESSAGE_TYPE_AUDIO: - if not msg.is_self(): # 接收语音消息 - # 下载语音文件 - voice_file = await msg.to_file_box() - silk_file = TmpDir().path() + voice_file.name - await voice_file.to_file(silk_file) - logger.info("[WX]receive voice file: " + silk_file) - # 将文件转成wav格式音频 - wav_file = os.path.splitext(silk_file)[0] + '.wav' - sil_to_wav(silk_file, wav_file) - # 语音识别为文本 - query = super().build_voice_to_text(wav_file).content - # 交验关键字 - match_prefix = self.check_prefix(query, conf().get('single_chat_prefix')) - if match_prefix is not None: - if match_prefix != '': - str_list = query.split(match_prefix, 1) - if len(str_list) == 2: - query = str_list[1].strip() - # 返回消息 - if conf().get('voice_reply_voice'): - await self._do_send_voice(query, from_user_id) - else: - await self._do_send(query, from_user_id) - else: - logger.info("[WX]receive voice check prefix: " + 'False') - # 清除缓存文件 - os.remove(wav_file) - os.remove(silk_file) - elif room and msg.type() == MessageType.MESSAGE_TYPE_TEXT: - # 群组&文本消息 - room_id = room.room_id - room_name = await room.topic() - from_user_id = from_contact.contact_id - from_user_name = from_contact.name - is_at = await msg.mention_self() - content = mention_content - config = conf() - match_prefix = (is_at and not config.get("group_at_off", False)) \ - or self.check_prefix(content, config.get('group_chat_prefix')) \ - or self.check_contain(content, config.get('group_chat_keyword')) - # Wechaty判断is_at为True，返回的内容是过滤掉@之后的内容；而is_at为False，则会返回完整的内容 - # 故判断如果匹配到自定义前缀，则返回过滤掉前缀+空格后的内容，用于实现类似自定义+前缀触发生成AI图片的功能 - prefixes = config.get('group_chat_prefix') - for prefix in prefixes: - if content.startswith(prefix): - content = content.replace(prefix, '', 1).strip() - break - if ('ALL_GROUP' in config.get('group_name_white_list') or room_name in config.get( - 'group_name_white_list') or self.check_contain(room_name, config.get( - 'group_name_keyword_white_list'))) and match_prefix: - 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() - await self._do_send_group_img(content, room_id) - else: - await self._do_send_group(content, room_id, room_name, from_user_id, from_user_name) - elif room and msg.type() == MessageType.MESSAGE_TYPE_AUDIO: - # 群组&语音消息 - room_id = room.room_id - room_name = await room.topic() - from_user_id = from_contact.contact_id - from_user_name = from_contact.name - is_at = await msg.mention_self() - config = conf() - # 是否开启语音识别、群消息响应功能、群名白名单符合等条件 - if config.get('group_speech_recognition') and ( - 'ALL_GROUP' in config.get('group_name_white_list') or room_name in config.get( - 'group_name_white_list') or self.check_contain(room_name, config.get( - 'group_name_keyword_white_list'))): - # 下载语音文件 - voice_file = await msg.to_file_box() - silk_file = TmpDir().path() + voice_file.name - await voice_file.to_file(silk_file) - logger.info("[WX]receive voice file: " + silk_file) - # 将文件转成wav格式音频 - wav_file = os.path.splitext(silk_file)[0] + '.wav' - sil_to_wav(silk_file, wav_file) - # 语音识别为文本 - query = super().build_voice_to_text(wav_file).content - # 校验关键字 - match_prefix = self.check_prefix(query, config.get('group_chat_prefix')) \ - or self.check_contain(query, config.get('group_chat_keyword')) - # Wechaty判断is_at为True，返回的内容是过滤掉@之后的内容；而is_at为False，则会返回完整的内容 - if match_prefix is not None: - # 故判断如果匹配到自定义前缀，则返回过滤掉前缀+空格后的内容，用于实现类似自定义+前缀触发生成AI图片的功能 - prefixes = config.get('group_chat_prefix') - for prefix in prefixes: - if query.startswith(prefix): - query = query.replace(prefix, '', 1).strip() - break - # 返回消息 - img_match_prefix = self.check_prefix(query, conf().get('image_create_prefix')) - if img_match_prefix: - query = query.split(img_match_prefix, 1)[1].strip() - await self._do_send_group_img(query, room_id) - elif config.get('voice_reply_voice'): - await self._do_send_group_voice(query, room_id, room_name, from_user_id, from_user_name) - else: - await self._do_send_group(query, room_id, room_name, from_user_id, from_user_name) - else: - logger.info("[WX]receive voice check prefix: " + 'False') - # 清除缓存文件 - os.remove(wav_file) - os.remove(silk_file) - - async def send(self, message: Union[str, Message, FileBox, Contact, UrlLink, MiniProgram], receiver): - logger.info('[WX] sendMsg={}, receiver={}'.format(message, receiver)) - if receiver: - contact = await bot.Contact.find(receiver) - await contact.say(message) - - async def send_group(self, message: Union[str, Message, FileBox, Contact, UrlLink, MiniProgram], receiver): - logger.info('[WX] sendMsg={}, receiver={}'.format(message, receiver)) - if receiver: - room = await bot.Room.find(receiver) - await room.say(message) - - async def _do_send(self, query, reply_user_id): - try: - if not query: - return - context = Context(ContextType.TEXT, query) - context['session_id'] = reply_user_id - 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: - logger.exception(e) - - async def _do_send_voice(self, query, reply_user_id): - try: - if not query: - return - context = Context(ContextType.TEXT, query) - context['session_id'] = reply_user_id - reply_text = super().build_reply_content(query, context).content - if reply_text: - # 转换 mp3 文件为 silk 格式 - mp3_file = super().build_text_to_voice(reply_text).content - silk_file = os.path.splitext(mp3_file)[0] + '.sil' - voiceLength = mp3_to_sil(mp3_file, silk_file) - # 发送语音 - t = int(time.time()) - file_box = FileBox.from_file(silk_file, name=str(t) + '.sil') - file_box.metadata = {'voiceLength': voiceLength} - await self.send(file_box, reply_user_id) - # 清除缓存文件 - os.remove(mp3_file) - os.remove(silk_file) - except Exception as e: - logger.exception(e) - - async def _do_send_img(self, query, reply_user_id): try: - if not query: - return - context = Context(ContextType.IMAGE_CREATE, query) - img_url = super().build_reply_content(query, context).content - if not img_url: - return - # 图片下载 - # 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) - - # 图片发送 - logger.info('[WX] sendImage, receiver={}'.format(reply_user_id)) - t = int(time.time()) - file_box = FileBox.from_url(url=img_url, name=str(t) + '.png') - await self.send(file_box, reply_user_id) - except Exception as e: - logger.exception(e) - - async def _do_send_group(self, query, group_id, group_name, group_user_id, group_user_name): - if not query: + cmsg = await WechatyMessage(msg) + except NotImplementedError as e: + logger.debug('[WX] {}'.format(e)) return - 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 \ - self.check_contain(group_name, group_chat_in_one_session)): - 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).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) - - async def _do_send_group_voice(self, query, group_id, group_name, group_user_id, group_user_name): - if not query: - return - 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 \ - self.check_contain(group_name, group_chat_in_one_session)): - 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).content - if reply_text: - reply_text = '@' + group_user_name + ' ' + reply_text.strip() - # 转换 mp3 文件为 silk 格式 - mp3_file = super().build_text_to_voice(reply_text).content - silk_file = os.path.splitext(mp3_file)[0] + '.sil' - voiceLength = mp3_to_sil(mp3_file, silk_file) - # 发送语音 - t = int(time.time()) - file_box = FileBox.from_file(silk_file, name=str(t) + '.silk') - file_box.metadata = {'voiceLength': voiceLength} - await self.send_group(file_box, group_id) - # 清除缓存文件 - os.remove(mp3_file) - os.remove(silk_file) - - async def _do_send_group_img(self, query, reply_room_id): - try: - if not query: - return - context = Context(ContextType.IMAGE_CREATE, query) - img_url = super().build_reply_content(query, context).content - if not img_url: - return - # 图片发送 - logger.info('[WX] sendImage, receiver={}'.format(reply_room_id)) - t = int(time.time()) - file_box = FileBox.from_url(url=img_url, name=str(t) + '.png') - await self.send_group(file_box, reply_room_id) except Exception as e: - logger.exception(e) - - def check_prefix(self, content, prefix_list): - for prefix in prefix_list: - if content.startswith(prefix): - return prefix - 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 + 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)) + thread_pool.submit(self._handle_loop, context, asyncio.get_event_loop()).add_done_callback(thread_pool_callback) + + def _handle_loop(self,context,loop): + asyncio.set_event_loop(loop) + self._handle(context) \ No newline at end of file diff --git a/channel/wechat/wechaty_message.py b/channel/wechat/wechaty_message.py index f5ef0ef..be2fe5d 100644 --- a/channel/wechat/wechaty_message.py +++ b/channel/wechat/wechaty_message.py @@ -1,3 +1,5 @@ +import asyncio +import re from wechaty import MessageType from bridge.context import ContextType from channel.chat_message import ChatMessage @@ -17,7 +19,6 @@ class aobject(object): async def __init__(self): pass - class WechatyMessage(ChatMessage, aobject): async def __init__(self, wechaty_msg: Message): @@ -36,13 +37,23 @@ class WechatyMessage(ChatMessage, aobject): self.ctype = ContextType.VOICE voice_file = await wechaty_msg.to_file_box() self.content = TmpDir().path() + voice_file.name # content直接存临时目录路径 - self._prepare_fn = lambda: voice_file.to_file(self.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 @@ -52,14 +63,23 @@ class WechatyMessage(ChatMessage, aobject): self.to_user_id = to_contact.contact_id self.to_user_nickname = to_contact.name - if wechaty_msg.is_self(): + 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: + + + if self.is_group: # wechaty群聊中，实际发送用户就是from_user self.is_at = await wechaty_msg.mention_self() - self.actual_user_id = self.other_user_id - self.actual_user_nickname = self.other_user_nickname + 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 diff --git a/voice/audio_convert.py b/voice/audio_convert.py index 73d18c8..2551bcc 100644 --- a/voice/audio_convert.py +++ b/voice/audio_convert.py @@ -21,6 +21,16 @@ def mp3_to_wav(mp3_path, wav_path): audio = AudioSegment.from_mp3(mp3_path) audio.export(wav_path, format="wav") +def any_to_wav(any_path, wav_path): + """ + 把任意格式转成wav文件 + """ + if any_path.endswith('.wav'): + return + if any_path.endswith('.sil') or any_path.endswith('.silk') or any_path.endswith('.slk'): + return sil_to_wav(any_path, wav_path) + audio = AudioSegment.from_file(any_path) + audio.export(wav_path, format="wav") def pcm_to_silk(pcm_path, silk_path): """ From cda9d5873d60d054a4ffd5ceebb94bfc4c6c68ae Mon Sep 17 00:00:00 2001 From: lanvent Date: Sat, 1 Apr 2023 04:26:34 +0800 Subject: [PATCH 5/5] plugins: support wechaty channel --- app.py | 2 +- channel/wechat/wechaty_channel.py | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/app.py b/app.py index c78a72c..7338f56 100644 --- a/app.py +++ b/app.py @@ -14,7 +14,7 @@ def run(): # create channel channel_name=conf().get('channel_type', 'wx') channel = channel_factory.create_channel(channel_name) - if channel_name=='wx': + if channel_name in ['wx','wxy']: PluginManager().load_plugins() # startup channel diff --git a/channel/wechat/wechaty_channel.py b/channel/wechat/wechaty_channel.py index cb39313..5f2f858 100644 --- a/channel/wechat/wechaty_channel.py +++ b/channel/wechat/wechaty_channel.py @@ -41,6 +41,8 @@ class WechatyChannel(ChatChannel): config = conf() token = config.get('wechaty_puppet_service_token') os.environ['WECHATY_PUPPET_SERVICE_TOKEN'] = token + os.environ['WECHATY_LOG']="warn" + # os.environ['WECHATY_PUPPET_SERVICE_ENDPOINT'] = '127.0.0.1:9001' self.bot = Wechaty() self.bot.on('login', self.on_login) self.bot.on('message', self.on_message)