From 66a0f18e694a055e8f4b49137288305c8784900f Mon Sep 17 00:00:00 2001 From: Pliosauroidea Date: Tue, 11 Mar 2025 20:38:14 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E4=BA=86=E7=A7=81=E8=81=8A?= =?UTF-8?q?=E6=97=B6=E4=BA=A7=E7=94=9Freply=E6=B6=88=E6=81=AF=E7=9A=84bug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/plugins/chat/message.py | 223 ++++++++++++++++------------- src/plugins/chat/message_sender.py | 93 +++++++----- 2 files changed, 181 insertions(+), 135 deletions(-) diff --git a/src/plugins/chat/message.py b/src/plugins/chat/message.py index 9301a20a4..e502e357a 100644 --- a/src/plugins/chat/message.py +++ b/src/plugins/chat/message.py @@ -8,112 +8,122 @@ from loguru import logger from .utils_image import image_manager from .message_base import Seg, GroupInfo, UserInfo, BaseMessageInfo, MessageBase from .chat_stream import ChatStream, chat_manager + # 禁用SSL警告 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) -#这个类是消息数据类，用于存储和管理消息数据。 -#它定义了消息的属性，包括群组ID、用户ID、消息ID、原始消息内容、纯文本内容和时间戳。 -#它还定义了两个辅助属性：keywords用于提取消息的关键词，is_plain_text用于判断消息是否为纯文本。 - +# 这个类是消息数据类，用于存储和管理消息数据。 +# 它定义了消息的属性，包括群组ID、用户ID、消息ID、原始消息内容、纯文本内容和时间戳。 +# 它还定义了两个辅助属性：keywords用于提取消息的关键词，is_plain_text用于判断消息是否为纯文本。 + + @dataclass class MessageRecv(MessageBase): """接收消息类，用于处理从MessageCQ序列化的消息""" - + def __init__(self, message_dict: Dict): """从MessageCQ的字典初始化 - + Args: message_dict: MessageCQ序列化后的字典 """ - message_info = BaseMessageInfo.from_dict(message_dict.get('message_info', {})) - message_segment = Seg.from_dict(message_dict.get('message_segment', {})) - raw_message = message_dict.get('raw_message') - + message_info = BaseMessageInfo.from_dict(message_dict.get("message_info", {})) + message_segment = Seg.from_dict(message_dict.get("message_segment", {})) + raw_message = message_dict.get("raw_message") + super().__init__( message_info=message_info, message_segment=message_segment, - raw_message=raw_message + raw_message=raw_message, ) - + # 处理消息内容 self.processed_plain_text = "" # 初始化为空字符串 - self.detailed_plain_text = "" # 初始化为空字符串 - self.is_emoji=False - def update_chat_stream(self,chat_stream:ChatStream): - self.chat_stream=chat_stream - + self.detailed_plain_text = "" # 初始化为空字符串 + self.is_emoji = False + + def update_chat_stream(self, chat_stream: ChatStream): + self.chat_stream = chat_stream + async def process(self) -> None: """处理消息内容，生成纯文本和详细文本 - + 这个方法必须在创建实例后显式调用，因为它包含异步操作。 """ - self.processed_plain_text = await self._process_message_segments(self.message_segment) + self.processed_plain_text = await self._process_message_segments( + self.message_segment + ) self.detailed_plain_text = self._generate_detailed_text() async def _process_message_segments(self, segment: Seg) -> str: """递归处理消息段，转换为文字描述 - + Args: segment: 要处理的消息段 - + Returns: str: 处理后的文本 """ - if segment.type == 'seglist': + if segment.type == "seglist": # 处理消息段列表 segments_text = [] for seg in segment.data: processed = await self._process_message_segments(seg) if processed: segments_text.append(processed) - return ' '.join(segments_text) + return " ".join(segments_text) else: # 处理单个消息段 return await self._process_single_segment(segment) async def _process_single_segment(self, seg: Seg) -> str: """处理单个消息段 - + Args: seg: 要处理的消息段 - + Returns: str: 处理后的文本 """ try: - if seg.type == 'text': + if seg.type == "text": return seg.data - elif seg.type == 'image': + elif seg.type == "image": # 如果是base64图片数据 if isinstance(seg.data, str): return await image_manager.get_image_description(seg.data) - return '[图片]' - elif seg.type == 'emoji': - self.is_emoji=True + return "[图片]" + elif seg.type == "emoji": + self.is_emoji = True if isinstance(seg.data, str): return await image_manager.get_emoji_description(seg.data) - return '[表情]' + return "[表情]" else: return f"[{seg.type}:{str(seg.data)}]" except Exception as e: - logger.error(f"处理消息段失败: {str(e)}, 类型: {seg.type}, 数据: {seg.data}") + logger.error( + f"处理消息段失败: {str(e)}, 类型: {seg.type}, 数据: {seg.data}" + ) return f"[处理失败的{seg.type}消息]" def _generate_detailed_text(self) -> str: """生成详细文本，包含时间和用户信息""" - time_str = time.strftime("%m-%d %H:%M:%S", time.localtime(self.message_info.time)) + time_str = time.strftime( + "%m-%d %H:%M:%S", time.localtime(self.message_info.time) + ) user_info = self.message_info.user_info name = ( f"{user_info.user_nickname}(ta的昵称:{user_info.user_cardname},ta的id:{user_info.user_id})" - if user_info.user_cardname!='' + if user_info.user_cardname != "" else f"{user_info.user_nickname}(ta的id:{user_info.user_id})" ) return f"[{time_str}] {name}: {self.processed_plain_text}\n" - + + @dataclass class Message(MessageBase): - chat_stream: ChatStream=None - reply: Optional['Message'] = None + chat_stream: ChatStream = None + reply: Optional["Message"] = None detailed_plain_text: str = "" processed_plain_text: str = "" @@ -124,7 +134,7 @@ class Message(MessageBase): chat_stream: ChatStream, user_info: UserInfo, message_segment: Optional[Seg] = None, - reply: Optional['MessageRecv'] = None, + reply: Optional["MessageRecv"] = None, detailed_plain_text: str = "", processed_plain_text: str = "", ): @@ -134,21 +144,19 @@ class Message(MessageBase): message_id=message_id, time=time, group_info=chat_stream.group_info, - user_info=user_info + user_info=user_info, ) # 调用父类初始化 super().__init__( - message_info=message_info, - message_segment=message_segment, - raw_message=None + message_info=message_info, message_segment=message_segment, raw_message=None ) self.chat_stream = chat_stream # 文本处理相关属性 self.processed_plain_text = detailed_plain_text self.detailed_plain_text = processed_plain_text - + # 回复消息 self.reply = reply @@ -156,14 +164,14 @@ class Message(MessageBase): @dataclass class MessageProcessBase(Message): """消息处理基类，用于处理中和发送中的消息""" - + def __init__( self, message_id: str, chat_stream: ChatStream, bot_user_info: UserInfo, message_segment: Optional[Seg] = None, - reply: Optional['MessageRecv'] = None + reply: Optional["MessageRecv"] = None, ): # 调用父类初始化 super().__init__( @@ -172,7 +180,7 @@ class MessageProcessBase(Message): chat_stream=chat_stream, user_info=bot_user_info, message_segment=message_segment, - reply=reply + reply=reply, ) # 处理状态相关属性 @@ -186,78 +194,83 @@ class MessageProcessBase(Message): async def _process_message_segments(self, segment: Seg) -> str: """递归处理消息段，转换为文字描述 - + Args: segment: 要处理的消息段 - + Returns: str: 处理后的文本 """ - if segment.type == 'seglist': + if segment.type == "seglist": # 处理消息段列表 segments_text = [] for seg in segment.data: processed = await self._process_message_segments(seg) if processed: segments_text.append(processed) - return ' '.join(segments_text) + return " ".join(segments_text) else: # 处理单个消息段 return await self._process_single_segment(segment) async def _process_single_segment(self, seg: Seg) -> str: """处理单个消息段 - + Args: seg: 要处理的消息段 - + Returns: str: 处理后的文本 """ try: - if seg.type == 'text': + if seg.type == "text": return seg.data - elif seg.type == 'image': + elif seg.type == "image": # 如果是base64图片数据 if isinstance(seg.data, str): return await image_manager.get_image_description(seg.data) - return '[图片]' - elif seg.type == 'emoji': + return "[图片]" + elif seg.type == "emoji": if isinstance(seg.data, str): return await image_manager.get_emoji_description(seg.data) - return '[表情]' - elif seg.type == 'at': + return "[表情]" + elif seg.type == "at": return f"[@{seg.data}]" - elif seg.type == 'reply': - if self.reply and hasattr(self.reply, 'processed_plain_text'): + elif seg.type == "reply": + if self.reply and hasattr(self.reply, "processed_plain_text"): return f"[回复：{self.reply.processed_plain_text}]" else: return f"[{seg.type}:{str(seg.data)}]" except Exception as e: - logger.error(f"处理消息段失败: {str(e)}, 类型: {seg.type}, 数据: {seg.data}") + logger.error( + f"处理消息段失败: {str(e)}, 类型: {seg.type}, 数据: {seg.data}" + ) return f"[处理失败的{seg.type}消息]" def _generate_detailed_text(self) -> str: """生成详细文本，包含时间和用户信息""" - time_str = time.strftime("%m-%d %H:%M:%S", time.localtime(self.message_info.time)) + time_str = time.strftime( + "%m-%d %H:%M:%S", time.localtime(self.message_info.time) + ) user_info = self.message_info.user_info name = ( f"{user_info.user_nickname}(ta的昵称:{user_info.user_cardname},ta的id:{user_info.user_id})" - if user_info.user_cardname != '' + if user_info.user_cardname != "" else f"{user_info.user_nickname}(ta的id:{user_info.user_id})" ) return f"[{time_str}] {name}: {self.processed_plain_text}\n" + @dataclass class MessageThinking(MessageProcessBase): """思考状态的消息类""" - + def __init__( self, message_id: str, chat_stream: ChatStream, bot_user_info: UserInfo, - reply: Optional['MessageRecv'] = None + reply: Optional["MessageRecv"] = None, ): # 调用父类初始化 super().__init__( @@ -265,26 +278,27 @@ class MessageThinking(MessageProcessBase): chat_stream=chat_stream, bot_user_info=bot_user_info, message_segment=None, # 思考状态不需要消息段 - reply=reply + reply=reply, ) - + # 思考状态特有属性 self.interrupt = False + @dataclass class MessageSending(MessageProcessBase): """发送状态的消息类""" - + def __init__( self, message_id: str, chat_stream: ChatStream, bot_user_info: UserInfo, - sender_info:UserInfo, # 用来记录发送者信息,用于私聊回复 + sender_info: UserInfo, # 用来记录发送者信息,用于私聊回复 message_segment: Seg, - reply: Optional['MessageRecv'] = None, + reply: Optional["MessageRecv"] = None, is_head: bool = False, - is_emoji: bool = False + is_emoji: bool = False, ): # 调用父类初始化 super().__init__( @@ -292,29 +306,34 @@ class MessageSending(MessageProcessBase): chat_stream=chat_stream, bot_user_info=bot_user_info, message_segment=message_segment, - reply=reply + reply=reply, ) - + # 发送状态特有属性 - self.sender_info=sender_info + self.sender_info = sender_info self.reply_to_message_id = reply.message_info.message_id if reply else None self.is_head = is_head self.is_emoji = is_emoji - - def set_reply(self, reply: Optional['MessageRecv']) -> None: + + def set_reply(self, reply: Optional["MessageRecv"]) -> None: """设置回复消息""" if reply: self.reply = reply self.reply_to_message_id = self.reply.message_info.message_id - self.message_segment = Seg(type='seglist', data=[ - Seg(type='reply', data=reply.message_info.message_id), - self.message_segment - ]) + self.message_segment = Seg( + type="seglist", + data=[ + Seg(type="reply", data=reply.message_info.message_id), + self.message_segment, + ], + ) async def process(self) -> None: """处理消息内容，生成纯文本和详细文本""" if self.message_segment: - self.processed_plain_text = await self._process_message_segments(self.message_segment) + self.processed_plain_text = await self._process_message_segments( + self.message_segment + ) self.detailed_plain_text = self._generate_detailed_text() @classmethod @@ -323,8 +342,8 @@ class MessageSending(MessageProcessBase): thinking: MessageThinking, message_segment: Seg, is_head: bool = False, - is_emoji: bool = False - ) -> 'MessageSending': + is_emoji: bool = False, + ) -> "MessageSending": """从思考状态消息创建发送状态消息""" return cls( message_id=thinking.message_info.message_id, @@ -333,41 +352,50 @@ class MessageSending(MessageProcessBase): bot_user_info=thinking.message_info.user_info, reply=thinking.reply, is_head=is_head, - is_emoji=is_emoji + is_emoji=is_emoji, ) - + def to_dict(self): - ret= super().to_dict() - ret['message_info']['user_info']=self.chat_stream.user_info.to_dict() + ret = super().to_dict() + ret["message_info"]["user_info"] = self.chat_stream.user_info.to_dict() return ret + def is_private_message(self) -> bool: + """判断是否为私聊消息""" + return ( + self.message_info.group_info is None + or self.message_info.group_info.group_id is None + ) + + @dataclass class MessageSet: """消息集合类，可以存储多个发送消息""" + def __init__(self, chat_stream: ChatStream, message_id: str): self.chat_stream = chat_stream self.message_id = message_id self.messages: List[MessageSending] = [] self.time = round(time.time(), 2) - + def add_message(self, message: MessageSending) -> None: """添加消息到集合""" if not isinstance(message, MessageSending): raise TypeError("MessageSet只能添加MessageSending类型的消息") self.messages.append(message) self.messages.sort(key=lambda x: x.message_info.time) - + def get_message_by_index(self, index: int) -> Optional[MessageSending]: """通过索引获取消息""" if 0 <= index < len(self.messages): return self.messages[index] return None - + def get_message_by_time(self, target_time: float) -> Optional[MessageSending]: """获取最接近指定时间的消息""" if not self.messages: return None - + left, right = 0, len(self.messages) - 1 while left < right: mid = (left + right) // 2 @@ -375,25 +403,22 @@ class MessageSet: left = mid + 1 else: right = mid - + return self.messages[left] - + def clear_messages(self) -> None: """清空所有消息""" self.messages.clear() - + def remove_message(self, message: MessageSending) -> bool: """移除指定消息""" if message in self.messages: self.messages.remove(message) return True return False - + def __str__(self) -> str: return f"MessageSet(id={self.message_id}, count={len(self.messages)})" - + def __len__(self) -> int: return len(self.messages) - - - diff --git a/src/plugins/chat/message_sender.py b/src/plugins/chat/message_sender.py index f987cf999..55272953c 100644 --- a/src/plugins/chat/message_sender.py +++ b/src/plugins/chat/message_sender.py @@ -7,7 +7,7 @@ from nonebot.adapters.onebot.v11 import Bot from .cq_code import cq_code_tool from .message_cq import MessageSendCQ -from .message import MessageSending, MessageThinking, MessageRecv,MessageSet +from .message import MessageSending, MessageThinking, MessageRecv, MessageSet from .storage import MessageStorage from .config import global_config from .chat_stream import chat_manager @@ -26,23 +26,24 @@ class Message_Sender: self._current_bot = bot async def send_message( - self, - message: MessageSending, + self, + message: MessageSending, ) -> None: """发送消息""" - + if isinstance(message, MessageSending): message_json = message.to_dict() - message_send=MessageSendCQ( - data=message_json - ) + message_send = MessageSendCQ(data=message_json) # logger.debug(message_send.message_info,message_send.raw_message) - if message_send.message_info.group_info.group_id: + if ( + message_send.message_info.group_info + and message_send.message_info.group_info.group_id + ): try: await self._current_bot.send_group_msg( group_id=message.message_info.group_info.group_id, message=message_send.raw_message, - auto_escape=False + auto_escape=False, ) logger.success(f"[调试] 发送消息{message.processed_plain_text}成功") except Exception as e: @@ -54,7 +55,7 @@ class Message_Sender: await self._current_bot.send_private_msg( user_id=message.sender_info.user_id, message=message_send.raw_message, - auto_escape=False + auto_escape=False, ) logger.success(f"[调试] 发送消息{message.processed_plain_text}成功") except Exception as e: @@ -64,13 +65,14 @@ class Message_Sender: class MessageContainer: """单个聊天流的发送/思考消息容器""" + def __init__(self, chat_id: str, max_size: int = 100): self.chat_id = chat_id self.max_size = max_size self.messages = [] self.last_send_time = 0 self.thinking_timeout = 20 # 思考超时时间（秒） - + def get_timeout_messages(self) -> List[MessageSending]: """获取所有超时的Message_Sending对象（思考时间超过30秒），按thinking_start_time排序""" current_time = time.time() @@ -85,12 +87,12 @@ class MessageContainer: timeout_messages.sort(key=lambda x: x.thinking_start_time) return timeout_messages - + def get_earliest_message(self) -> Optional[Union[MessageThinking, MessageSending]]: """获取thinking_start_time最早的消息对象""" if not self.messages: return None - earliest_time = float('inf') + earliest_time = float("inf") earliest_message = None for msg in self.messages: msg_time = msg.thinking_start_time @@ -98,7 +100,7 @@ class MessageContainer: earliest_time = msg_time earliest_message = msg return earliest_message - + def add_message(self, message: Union[MessageThinking, MessageSending]) -> None: """添加消息到队列""" if isinstance(message, MessageSet): @@ -106,7 +108,7 @@ class MessageContainer: self.messages.append(single_message) else: self.messages.append(message) - + def remove_message(self, message: Union[MessageThinking, MessageSending]) -> bool: """移除消息，如果消息存在则返回True，否则返回False""" try: @@ -121,7 +123,7 @@ class MessageContainer: def has_messages(self) -> bool: """检查是否有待发送的消息""" return bool(self.messages) - + def get_all_messages(self) -> List[Union[MessageSending, MessageThinking]]: """获取所有消息""" return list(self.messages) @@ -129,72 +131,91 @@ class MessageContainer: class MessageManager: """管理所有聊天流的消息容器""" + def __init__(self): self.containers: Dict[str, MessageContainer] = {} # chat_id -> MessageContainer self.storage = MessageStorage() self._running = True - + def get_container(self, chat_id: str) -> MessageContainer: """获取或创建聊天流的消息容器""" if chat_id not in self.containers: self.containers[chat_id] = MessageContainer(chat_id) return self.containers[chat_id] - - def add_message(self, message: Union[MessageThinking, MessageSending, MessageSet]) -> None: + + def add_message( + self, message: Union[MessageThinking, MessageSending, MessageSet] + ) -> None: chat_stream = message.chat_stream if not chat_stream: raise ValueError("无法找到对应的聊天流") container = self.get_container(chat_stream.stream_id) container.add_message(message) - + async def process_chat_messages(self, chat_id: str): """处理聊天流消息""" container = self.get_container(chat_id) if container.has_messages(): # print(f"处理有message的容器chat_id: {chat_id}") message_earliest = container.get_earliest_message() - + if isinstance(message_earliest, MessageThinking): message_earliest.update_thinking_time() thinking_time = message_earliest.thinking_time - print(f"消息正在思考中，已思考{int(thinking_time)}秒\r", end='', flush=True) + print( + f"消息正在思考中，已思考{int(thinking_time)}秒\r", + end="", + flush=True, + ) # 检查是否超时 if thinking_time > global_config.thinking_timeout: logger.warning(f"消息思考超时({thinking_time}秒)，移除该消息") container.remove_message(message_earliest) else: - - if message_earliest.is_head and message_earliest.update_thinking_time() > 30: + + if ( + message_earliest.is_head + and message_earliest.update_thinking_time() > 30 + and not message_earliest.is_private_message() # 避免在私聊时插入reply + ): await message_sender.send_message(message_earliest.set_reply()) else: await message_sender.send_message(message_earliest) await message_earliest.process() - - print(f"\033[1;34m[调试]\033[0m 消息'{message_earliest.processed_plain_text}'正在发送中") - - await self.storage.store_message(message_earliest, message_earliest.chat_stream,None) - + + print( + f"\033[1;34m[调试]\033[0m 消息'{message_earliest.processed_plain_text}'正在发送中" + ) + + await self.storage.store_message( + message_earliest, message_earliest.chat_stream, None + ) + container.remove_message(message_earliest) - + message_timeout = container.get_timeout_messages() if message_timeout: logger.warning(f"发现{len(message_timeout)}条超时消息") for msg in message_timeout: if msg == message_earliest: continue - + try: - if msg.is_head and msg.update_thinking_time() > 30: + if ( + msg.is_head + and msg.update_thinking_time() > 30 + and not message_earliest.is_private_message() # 避免在私聊时插入reply + ): await message_sender.send_message(msg.set_reply()) else: await message_sender.send_message(msg) - + # if msg.is_emoji: # msg.processed_plain_text = "[表情包]" await msg.process() - await self.storage.store_message(msg,msg.chat_stream, None) - + await self.storage.store_message(msg, msg.chat_stream, None) + if not container.remove_message(msg): logger.warning("尝试删除不存在的消息") except Exception: @@ -208,7 +229,7 @@ class MessageManager: tasks = [] for chat_id in self.containers.keys(): tasks.append(self.process_chat_messages(chat_id)) - + await asyncio.gather(*tasks)