gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复了私聊时产生reply消息的bug

Browse Source
This commit is contained in:
Pliosauroidea
2025-03-11 20:38:14 +08:00
parent 33cd83b895
commit 66a0f18e69
2 changed files with 181 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

223
src/plugins/chat/message.py
View File

@@ -8,112 +8,122 @@ from loguru import logger
from .utils_image import image_manager from .utils_image import image_manager
from .message_base import Seg, GroupInfo, UserInfo, BaseMessageInfo, MessageBase from .message_base import Seg, GroupInfo, UserInfo, BaseMessageInfo, MessageBase
from .chat_stream import ChatStream, chat_manager from .chat_stream import ChatStream, chat_manager
# 禁用SSL警告 # 禁用SSL警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
#这个类是消息数据类，用于存储和管理消息数据。 # 这个类是消息数据类，用于存储和管理消息数据。
#它定义了消息的属性，包括群组ID、用户ID、消息ID、原始消息内容、纯文本内容和时间戳。 # 它定义了消息的属性，包括群组ID、用户ID、消息ID、原始消息内容、纯文本内容和时间戳。
#它还定义了两个辅助属性：keywords用于提取消息的关键词，is_plain_text用于判断消息是否为纯文本。 # 它还定义了两个辅助属性：keywords用于提取消息的关键词，is_plain_text用于判断消息是否为纯文本。
@dataclass @dataclass
class MessageRecv(MessageBase): class MessageRecv(MessageBase):
"""接收消息类，用于处理从MessageCQ序列化的消息""" """接收消息类，用于处理从MessageCQ序列化的消息"""
def __init__(self, message_dict: Dict): def __init__(self, message_dict: Dict):
"""从MessageCQ的字典初始化 """从MessageCQ的字典初始化
Args: Args:
message_dict: MessageCQ序列化后的字典 message_dict: MessageCQ序列化后的字典
""" """
message_info = BaseMessageInfo.from_dict(message_dict.get('message_info', {})) message_info = BaseMessageInfo.from_dict(message_dict.get("message_info", {}))
message_segment = Seg.from_dict(message_dict.get('message_segment', {})) message_segment = Seg.from_dict(message_dict.get("message_segment", {}))
raw_message = message_dict.get('raw_message') raw_message = message_dict.get("raw_message")
super().__init__( super().__init__(
message_info=message_info, message_info=message_info,
message_segment=message_segment, message_segment=message_segment,
raw_message=raw_message raw_message=raw_message,
) )
# 处理消息内容 # 处理消息内容
self.processed_plain_text = "" # 初始化为空字符串 self.processed_plain_text = "" # 初始化为空字符串
self.detailed_plain_text = "" # 初始化为空字符串 self.detailed_plain_text = "" # 初始化为空字符串
self.is_emoji=False self.is_emoji = False
def update_chat_stream(self,chat_stream:ChatStream):
self.chat_stream=chat_stream def update_chat_stream(self, chat_stream: ChatStream):
self.chat_stream = chat_stream
async def process(self) -> None: 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() self.detailed_plain_text = self._generate_detailed_text()
async def _process_message_segments(self, segment: Seg) -> str: async def _process_message_segments(self, segment: Seg) -> str:
"""递归处理消息段，转换为文字描述 """递归处理消息段，转换为文字描述
Args: Args:
segment: 要处理的消息段 segment: 要处理的消息段
Returns: Returns:
str: 处理后的文本 str: 处理后的文本
""" """
if segment.type == 'seglist': if segment.type == "seglist":
# 处理消息段列表 # 处理消息段列表
segments_text = [] segments_text = []
for seg in segment.data: for seg in segment.data:
processed = await self._process_message_segments(seg) processed = await self._process_message_segments(seg)
if processed: if processed:
segments_text.append(processed) segments_text.append(processed)
return ' '.join(segments_text) return " ".join(segments_text)
else: else:
# 处理单个消息段 # 处理单个消息段
return await self._process_single_segment(segment) return await self._process_single_segment(segment)
async def _process_single_segment(self, seg: Seg) -> str: async def _process_single_segment(self, seg: Seg) -> str:
"""处理单个消息段 """处理单个消息段
Args: Args:
seg: 要处理的消息段 seg: 要处理的消息段
Returns: Returns:
str: 处理后的文本 str: 处理后的文本
""" """
try: try:
if seg.type == 'text': if seg.type == "text":
return seg.data return seg.data
elif seg.type == 'image': elif seg.type == "image":
# 如果是base64图片数据 # 如果是base64图片数据
if isinstance(seg.data, str): if isinstance(seg.data, str):
return await image_manager.get_image_description(seg.data) return await image_manager.get_image_description(seg.data)
return '[图片]' return "[图片]"
elif seg.type == 'emoji': elif seg.type == "emoji":
self.is_emoji=True self.is_emoji = True
if isinstance(seg.data, str): if isinstance(seg.data, str):
return await image_manager.get_emoji_description(seg.data) return await image_manager.get_emoji_description(seg.data)
return '[表情]' return "[表情]"
else: else:
return f"[{seg.type}:{str(seg.data)}]" return f"[{seg.type}:{str(seg.data)}]"
except Exception as e: 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}消息]" return f"[处理失败的{seg.type}消息]"
def _generate_detailed_text(self) -> str: 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 user_info = self.message_info.user_info
name = ( name = (
f"{user_info.user_nickname}(ta的昵称:{user_info.user_cardname},ta的id:{user_info.user_id})" 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})" else f"{user_info.user_nickname}(ta的id:{user_info.user_id})"
) )
return f"[{time_str}] {name}: {self.processed_plain_text}\n" return f"[{time_str}] {name}: {self.processed_plain_text}\n"
@dataclass @dataclass
class Message(MessageBase): class Message(MessageBase):
chat_stream: ChatStream=None chat_stream: ChatStream = None
reply: Optional['Message'] = None reply: Optional["Message"] = None
detailed_plain_text: str = "" detailed_plain_text: str = ""
processed_plain_text: str = "" processed_plain_text: str = ""
@@ -124,7 +134,7 @@ class Message(MessageBase):
chat_stream: ChatStream, chat_stream: ChatStream,
user_info: UserInfo, user_info: UserInfo,
message_segment: Optional[Seg] = None, message_segment: Optional[Seg] = None,
reply: Optional['MessageRecv'] = None, reply: Optional["MessageRecv"] = None,
detailed_plain_text: str = "", detailed_plain_text: str = "",
processed_plain_text: str = "", processed_plain_text: str = "",
): ):
@@ -134,21 +144,19 @@ class Message(MessageBase):
message_id=message_id, message_id=message_id,
time=time, time=time,
group_info=chat_stream.group_info, group_info=chat_stream.group_info,
user_info=user_info user_info=user_info,
) )
# 调用父类初始化 # 调用父类初始化
super().__init__( super().__init__(
message_info=message_info, message_info=message_info, message_segment=message_segment, raw_message=None
message_segment=message_segment,
raw_message=None
) )
self.chat_stream = chat_stream self.chat_stream = chat_stream
# 文本处理相关属性 # 文本处理相关属性
self.processed_plain_text = detailed_plain_text self.processed_plain_text = detailed_plain_text
self.detailed_plain_text = processed_plain_text self.detailed_plain_text = processed_plain_text
# 回复消息 # 回复消息
self.reply = reply self.reply = reply
@@ -156,14 +164,14 @@ class Message(MessageBase):
@dataclass @dataclass
class MessageProcessBase(Message): class MessageProcessBase(Message):
"""消息处理基类，用于处理中和发送中的消息""" """消息处理基类，用于处理中和发送中的消息"""
def __init__( def __init__(
self, self,
message_id: str, message_id: str,
chat_stream: ChatStream, chat_stream: ChatStream,
bot_user_info: UserInfo, bot_user_info: UserInfo,
message_segment: Optional[Seg] = None, message_segment: Optional[Seg] = None,
reply: Optional['MessageRecv'] = None reply: Optional["MessageRecv"] = None,
): ):
# 调用父类初始化 # 调用父类初始化
super().__init__( super().__init__(
@@ -172,7 +180,7 @@ class MessageProcessBase(Message):
chat_stream=chat_stream, chat_stream=chat_stream,
user_info=bot_user_info, user_info=bot_user_info,
message_segment=message_segment, message_segment=message_segment,
reply=reply reply=reply,
) )
# 处理状态相关属性 # 处理状态相关属性
@@ -186,78 +194,83 @@ class MessageProcessBase(Message):
async def _process_message_segments(self, segment: Seg) -> str: async def _process_message_segments(self, segment: Seg) -> str:
"""递归处理消息段，转换为文字描述 """递归处理消息段，转换为文字描述
Args: Args:
segment: 要处理的消息段 segment: 要处理的消息段
Returns: Returns:
str: 处理后的文本 str: 处理后的文本
""" """
if segment.type == 'seglist': if segment.type == "seglist":
# 处理消息段列表 # 处理消息段列表
segments_text = [] segments_text = []
for seg in segment.data: for seg in segment.data:
processed = await self._process_message_segments(seg) processed = await self._process_message_segments(seg)
if processed: if processed:
segments_text.append(processed) segments_text.append(processed)
return ' '.join(segments_text) return " ".join(segments_text)
else: else:
# 处理单个消息段 # 处理单个消息段
return await self._process_single_segment(segment) return await self._process_single_segment(segment)
async def _process_single_segment(self, seg: Seg) -> str: async def _process_single_segment(self, seg: Seg) -> str:
"""处理单个消息段 """处理单个消息段
Args: Args:
seg: 要处理的消息段 seg: 要处理的消息段
Returns: Returns:
str: 处理后的文本 str: 处理后的文本
""" """
try: try:
if seg.type == 'text': if seg.type == "text":
return seg.data return seg.data
elif seg.type == 'image': elif seg.type == "image":
# 如果是base64图片数据 # 如果是base64图片数据
if isinstance(seg.data, str): if isinstance(seg.data, str):
return await image_manager.get_image_description(seg.data) return await image_manager.get_image_description(seg.data)
return '[图片]' return "[图片]"
elif seg.type == 'emoji': elif seg.type == "emoji":
if isinstance(seg.data, str): if isinstance(seg.data, str):
return await image_manager.get_emoji_description(seg.data) return await image_manager.get_emoji_description(seg.data)
return '[表情]' return "[表情]"
elif seg.type == 'at': elif seg.type == "at":
return f"[@{seg.data}]" return f"[@{seg.data}]"
elif seg.type == 'reply': elif seg.type == "reply":
if self.reply and hasattr(self.reply, 'processed_plain_text'): if self.reply and hasattr(self.reply, "processed_plain_text"):
return f"[回复：{self.reply.processed_plain_text}]" return f"[回复：{self.reply.processed_plain_text}]"
else: else:
return f"[{seg.type}:{str(seg.data)}]" return f"[{seg.type}:{str(seg.data)}]"
except Exception as e: 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}消息]" return f"[处理失败的{seg.type}消息]"
def _generate_detailed_text(self) -> str: 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 user_info = self.message_info.user_info
name = ( name = (
f"{user_info.user_nickname}(ta的昵称:{user_info.user_cardname},ta的id:{user_info.user_id})" 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})" else f"{user_info.user_nickname}(ta的id:{user_info.user_id})"
) )
return f"[{time_str}] {name}: {self.processed_plain_text}\n" return f"[{time_str}] {name}: {self.processed_plain_text}\n"
@dataclass @dataclass
class MessageThinking(MessageProcessBase): class MessageThinking(MessageProcessBase):
"""思考状态的消息类""" """思考状态的消息类"""
def __init__( def __init__(
self, self,
message_id: str, message_id: str,
chat_stream: ChatStream, chat_stream: ChatStream,
bot_user_info: UserInfo, bot_user_info: UserInfo,
reply: Optional['MessageRecv'] = None reply: Optional["MessageRecv"] = None,
): ):
# 调用父类初始化 # 调用父类初始化
super().__init__( super().__init__(
@@ -265,26 +278,27 @@ class MessageThinking(MessageProcessBase):
chat_stream=chat_stream, chat_stream=chat_stream,
bot_user_info=bot_user_info, bot_user_info=bot_user_info,
message_segment=None, # 思考状态不需要消息段 message_segment=None, # 思考状态不需要消息段
reply=reply reply=reply,
) )
# 思考状态特有属性 # 思考状态特有属性
self.interrupt = False self.interrupt = False
@dataclass @dataclass
class MessageSending(MessageProcessBase): class MessageSending(MessageProcessBase):
"""发送状态的消息类""" """发送状态的消息类"""
def __init__( def __init__(
self, self,
message_id: str, message_id: str,
chat_stream: ChatStream, chat_stream: ChatStream,
bot_user_info: UserInfo, bot_user_info: UserInfo,
sender_info:UserInfo, # 用来记录发送者信息,用于私聊回复 sender_info: UserInfo, # 用来记录发送者信息,用于私聊回复
message_segment: Seg, message_segment: Seg,
reply: Optional['MessageRecv'] = None, reply: Optional["MessageRecv"] = None,
is_head: bool = False, is_head: bool = False,
is_emoji: bool = False is_emoji: bool = False,
): ):
# 调用父类初始化 # 调用父类初始化
super().__init__( super().__init__(
@@ -292,29 +306,34 @@ class MessageSending(MessageProcessBase):
chat_stream=chat_stream, chat_stream=chat_stream,
bot_user_info=bot_user_info, bot_user_info=bot_user_info,
message_segment=message_segment, 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.reply_to_message_id = reply.message_info.message_id if reply else None
self.is_head = is_head self.is_head = is_head
self.is_emoji = is_emoji self.is_emoji = is_emoji
def set_reply(self, reply: Optional['MessageRecv']) -> None: def set_reply(self, reply: Optional["MessageRecv"]) -> None:
"""设置回复消息""" """设置回复消息"""
if reply: if reply:
self.reply = reply self.reply = reply
self.reply_to_message_id = self.reply.message_info.message_id self.reply_to_message_id = self.reply.message_info.message_id
self.message_segment = Seg(type='seglist', data=[ self.message_segment = Seg(
Seg(type='reply', data=reply.message_info.message_id), type="seglist",
self.message_segment data=[
]) Seg(type="reply", data=reply.message_info.message_id),
self.message_segment,
],
)
async def process(self) -> None: async def process(self) -> None:
"""处理消息内容，生成纯文本和详细文本""" """处理消息内容，生成纯文本和详细文本"""
if self.message_segment: 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() self.detailed_plain_text = self._generate_detailed_text()
@classmethod @classmethod
@@ -323,8 +342,8 @@ class MessageSending(MessageProcessBase):
thinking: MessageThinking, thinking: MessageThinking,
message_segment: Seg, message_segment: Seg,
is_head: bool = False, is_head: bool = False,
is_emoji: bool = False is_emoji: bool = False,
) -> 'MessageSending': ) -> "MessageSending":
"""从思考状态消息创建发送状态消息""" """从思考状态消息创建发送状态消息"""
return cls( return cls(
message_id=thinking.message_info.message_id, message_id=thinking.message_info.message_id,
@@ -333,41 +352,50 @@ class MessageSending(MessageProcessBase):
bot_user_info=thinking.message_info.user_info, bot_user_info=thinking.message_info.user_info,
reply=thinking.reply, reply=thinking.reply,
is_head=is_head, is_head=is_head,
is_emoji=is_emoji is_emoji=is_emoji,
) )
def to_dict(self): def to_dict(self):
ret= super().to_dict() ret = super().to_dict()
ret['message_info']['user_info']=self.chat_stream.user_info.to_dict() ret["message_info"]["user_info"] = self.chat_stream.user_info.to_dict()
return ret 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 @dataclass
class MessageSet: class MessageSet:
"""消息集合类，可以存储多个发送消息""" """消息集合类，可以存储多个发送消息"""
def __init__(self, chat_stream: ChatStream, message_id: str): def __init__(self, chat_stream: ChatStream, message_id: str):
self.chat_stream = chat_stream self.chat_stream = chat_stream
self.message_id = message_id self.message_id = message_id
self.messages: List[MessageSending] = [] self.messages: List[MessageSending] = []
self.time = round(time.time(), 2) self.time = round(time.time(), 2)
def add_message(self, message: MessageSending) -> None: def add_message(self, message: MessageSending) -> None:
"""添加消息到集合""" """添加消息到集合"""
if not isinstance(message, MessageSending): if not isinstance(message, MessageSending):
raise TypeError("MessageSet只能添加MessageSending类型的消息") raise TypeError("MessageSet只能添加MessageSending类型的消息")
self.messages.append(message) self.messages.append(message)
self.messages.sort(key=lambda x: x.message_info.time) self.messages.sort(key=lambda x: x.message_info.time)
def get_message_by_index(self, index: int) -> Optional[MessageSending]: def get_message_by_index(self, index: int) -> Optional[MessageSending]:
"""通过索引获取消息""" """通过索引获取消息"""
if 0 <= index < len(self.messages): if 0 <= index < len(self.messages):
return self.messages[index] return self.messages[index]
return None return None
def get_message_by_time(self, target_time: float) -> Optional[MessageSending]: def get_message_by_time(self, target_time: float) -> Optional[MessageSending]:
"""获取最接近指定时间的消息""" """获取最接近指定时间的消息"""
if not self.messages: if not self.messages:
return None return None
left, right = 0, len(self.messages) - 1 left, right = 0, len(self.messages) - 1
while left < right: while left < right:
mid = (left + right) // 2 mid = (left + right) // 2
@@ -375,25 +403,22 @@ class MessageSet:
left = mid + 1 left = mid + 1
else: else:
right = mid right = mid
return self.messages[left] return self.messages[left]
def clear_messages(self) -> None: def clear_messages(self) -> None:
"""清空所有消息""" """清空所有消息"""
self.messages.clear() self.messages.clear()
def remove_message(self, message: MessageSending) -> bool: def remove_message(self, message: MessageSending) -> bool:
"""移除指定消息""" """移除指定消息"""
if message in self.messages: if message in self.messages:
self.messages.remove(message) self.messages.remove(message)
return True return True
return False return False
def __str__(self) -> str: def __str__(self) -> str:
return f"MessageSet(id={self.message_id}, count={len(self.messages)})" return f"MessageSet(id={self.message_id}, count={len(self.messages)})"
def __len__(self) -> int: def __len__(self) -> int:
return len(self.messages) return len(self.messages)

93
src/plugins/chat/message_sender.py
View File

@@ -7,7 +7,7 @@ from nonebot.adapters.onebot.v11 import Bot
from .cq_code import cq_code_tool from .cq_code import cq_code_tool
from .message_cq import MessageSendCQ from .message_cq import MessageSendCQ
from .message import MessageSending, MessageThinking, MessageRecv,MessageSet from .message import MessageSending, MessageThinking, MessageRecv, MessageSet
from .storage import MessageStorage from .storage import MessageStorage
from .config import global_config from .config import global_config
from .chat_stream import chat_manager from .chat_stream import chat_manager
@@ -26,23 +26,24 @@ class Message_Sender:
self._current_bot = bot self._current_bot = bot
async def send_message( async def send_message(
self, self,
message: MessageSending, message: MessageSending,
) -> None: ) -> None:
"""发送消息""" """发送消息"""
if isinstance(message, MessageSending): if isinstance(message, MessageSending):
message_json = message.to_dict() message_json = message.to_dict()
message_send=MessageSendCQ( message_send = MessageSendCQ(data=message_json)
data=message_json
)
# logger.debug(message_send.message_info,message_send.raw_message) # 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: try:
await self._current_bot.send_group_msg( await self._current_bot.send_group_msg(
group_id=message.message_info.group_info.group_id, group_id=message.message_info.group_info.group_id,
message=message_send.raw_message, message=message_send.raw_message,
auto_escape=False auto_escape=False,
) )
logger.success(f"[调试] 发送消息{message.processed_plain_text}成功") logger.success(f"[调试] 发送消息{message.processed_plain_text}成功")
except Exception as e: except Exception as e:
@@ -54,7 +55,7 @@ class Message_Sender:
await self._current_bot.send_private_msg( await self._current_bot.send_private_msg(
user_id=message.sender_info.user_id, user_id=message.sender_info.user_id,
message=message_send.raw_message, message=message_send.raw_message,
auto_escape=False auto_escape=False,
) )
logger.success(f"[调试] 发送消息{message.processed_plain_text}成功") logger.success(f"[调试] 发送消息{message.processed_plain_text}成功")
except Exception as e: except Exception as e:
@@ -64,13 +65,14 @@ class Message_Sender:
class MessageContainer: class MessageContainer:
"""单个聊天流的发送/思考消息容器""" """单个聊天流的发送/思考消息容器"""
def __init__(self, chat_id: str, max_size: int = 100): def __init__(self, chat_id: str, max_size: int = 100):
self.chat_id = chat_id self.chat_id = chat_id
self.max_size = max_size self.max_size = max_size
self.messages = [] self.messages = []
self.last_send_time = 0 self.last_send_time = 0
self.thinking_timeout = 20 # 思考超时时间（秒） self.thinking_timeout = 20 # 思考超时时间（秒）
def get_timeout_messages(self) -> List[MessageSending]: def get_timeout_messages(self) -> List[MessageSending]:
"""获取所有超时的Message_Sending对象（思考时间超过30秒），按thinking_start_time排序""" """获取所有超时的Message_Sending对象（思考时间超过30秒），按thinking_start_time排序"""
current_time = time.time() current_time = time.time()
@@ -85,12 +87,12 @@ class MessageContainer:
timeout_messages.sort(key=lambda x: x.thinking_start_time) timeout_messages.sort(key=lambda x: x.thinking_start_time)
return timeout_messages return timeout_messages
def get_earliest_message(self) -> Optional[Union[MessageThinking, MessageSending]]: def get_earliest_message(self) -> Optional[Union[MessageThinking, MessageSending]]:
"""获取thinking_start_time最早的消息对象""" """获取thinking_start_time最早的消息对象"""
if not self.messages: if not self.messages:
return None return None
earliest_time = float('inf') earliest_time = float("inf")
earliest_message = None earliest_message = None
for msg in self.messages: for msg in self.messages:
msg_time = msg.thinking_start_time msg_time = msg.thinking_start_time
@@ -98,7 +100,7 @@ class MessageContainer:
earliest_time = msg_time earliest_time = msg_time
earliest_message = msg earliest_message = msg
return earliest_message return earliest_message
def add_message(self, message: Union[MessageThinking, MessageSending]) -> None: def add_message(self, message: Union[MessageThinking, MessageSending]) -> None:
"""添加消息到队列""" """添加消息到队列"""
if isinstance(message, MessageSet): if isinstance(message, MessageSet):
@@ -106,7 +108,7 @@ class MessageContainer:
self.messages.append(single_message) self.messages.append(single_message)
else: else:
self.messages.append(message) self.messages.append(message)
def remove_message(self, message: Union[MessageThinking, MessageSending]) -> bool: def remove_message(self, message: Union[MessageThinking, MessageSending]) -> bool:
"""移除消息，如果消息存在则返回True，否则返回False""" """移除消息，如果消息存在则返回True，否则返回False"""
try: try:
@@ -121,7 +123,7 @@ class MessageContainer:
def has_messages(self) -> bool: def has_messages(self) -> bool:
"""检查是否有待发送的消息""" """检查是否有待发送的消息"""
return bool(self.messages) return bool(self.messages)
def get_all_messages(self) -> List[Union[MessageSending, MessageThinking]]: def get_all_messages(self) -> List[Union[MessageSending, MessageThinking]]:
"""获取所有消息""" """获取所有消息"""
return list(self.messages) return list(self.messages)
@@ -129,72 +131,91 @@ class MessageContainer:
class MessageManager: class MessageManager:
"""管理所有聊天流的消息容器""" """管理所有聊天流的消息容器"""
def __init__(self): def __init__(self):
self.containers: Dict[str, MessageContainer] = {} # chat_id -> MessageContainer self.containers: Dict[str, MessageContainer] = {} # chat_id -> MessageContainer
self.storage = MessageStorage() self.storage = MessageStorage()
self._running = True self._running = True
def get_container(self, chat_id: str) -> MessageContainer: def get_container(self, chat_id: str) -> MessageContainer:
"""获取或创建聊天流的消息容器""" """获取或创建聊天流的消息容器"""
if chat_id not in self.containers: if chat_id not in self.containers:
self.containers[chat_id] = MessageContainer(chat_id) self.containers[chat_id] = MessageContainer(chat_id)
return self.containers[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 chat_stream = message.chat_stream
if not chat_stream: if not chat_stream:
raise ValueError("无法找到对应的聊天流") raise ValueError("无法找到对应的聊天流")
container = self.get_container(chat_stream.stream_id) container = self.get_container(chat_stream.stream_id)
container.add_message(message) container.add_message(message)
async def process_chat_messages(self, chat_id: str): async def process_chat_messages(self, chat_id: str):
"""处理聊天流消息""" """处理聊天流消息"""
container = self.get_container(chat_id) container = self.get_container(chat_id)
if container.has_messages(): if container.has_messages():
# print(f"处理有message的容器chat_id: {chat_id}") # print(f"处理有message的容器chat_id: {chat_id}")
message_earliest = container.get_earliest_message() message_earliest = container.get_earliest_message()
if isinstance(message_earliest, MessageThinking): if isinstance(message_earliest, MessageThinking):
message_earliest.update_thinking_time() message_earliest.update_thinking_time()
thinking_time = message_earliest.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: if thinking_time > global_config.thinking_timeout:
logger.warning(f"消息思考超时({thinking_time}秒)，移除该消息") logger.warning(f"消息思考超时({thinking_time}秒)，移除该消息")
container.remove_message(message_earliest) container.remove_message(message_earliest)
else: 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()) await message_sender.send_message(message_earliest.set_reply())
else: else:
await message_sender.send_message(message_earliest) await message_sender.send_message(message_earliest)
await message_earliest.process() await message_earliest.process()
print(f"\033[1;34m[调试]\033[0m 消息'{message_earliest.processed_plain_text}'正在发送中") print(
f"\033[1;34m[调试]\033[0m 消息'{message_earliest.processed_plain_text}'正在发送中"
await self.storage.store_message(message_earliest, message_earliest.chat_stream,None) )
await self.storage.store_message(
message_earliest, message_earliest.chat_stream, None
)
container.remove_message(message_earliest) container.remove_message(message_earliest)
message_timeout = container.get_timeout_messages() message_timeout = container.get_timeout_messages()
if message_timeout: if message_timeout:
logger.warning(f"发现{len(message_timeout)}条超时消息") logger.warning(f"发现{len(message_timeout)}条超时消息")
for msg in message_timeout: for msg in message_timeout:
if msg == message_earliest: if msg == message_earliest:
continue continue
try: 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()) await message_sender.send_message(msg.set_reply())
else: else:
await message_sender.send_message(msg) await message_sender.send_message(msg)
# if msg.is_emoji: # if msg.is_emoji:
# msg.processed_plain_text = "[表情包]" # msg.processed_plain_text = "[表情包]"
await msg.process() 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): if not container.remove_message(msg):
logger.warning("尝试删除不存在的消息") logger.warning("尝试删除不存在的消息")
except Exception: except Exception:
@@ -208,7 +229,7 @@ class MessageManager:
tasks = [] tasks = []
for chat_id in self.containers.keys(): for chat_id in self.containers.keys():
tasks.append(self.process_chat_messages(chat_id)) tasks.append(self.process_chat_messages(chat_id))
await asyncio.gather(*tasks) await asyncio.gather(*tasks)