Merge branch 'dev' into plugin

src/chat/message_receive/__init__.py

@@ -1,12 +1,10 @@
 from src.chat.emoji_system.emoji_manager import get_emoji_manager
 from src.chat.message_receive.chat_stream import get_chat_manager
-from src.chat.message_receive.normal_message_sender import message_manager
 from src.chat.message_receive.storage import MessageStorage
 
 
 __all__ = [
     "get_emoji_manager",
     "get_chat_manager",
-    "message_manager",
     "MessageStorage",
 ]

src/chat/message_receive/bot.py

@@ -84,7 +84,6 @@ class ChatBot:
 
         # 创建初始化PFC管理器的任务，会在_ensure_started时执行
         self.only_process_chat = MessageProcessor()
-        self.pfc_manager = PFCManager.get_instance()
         self.s4u_message_processor = S4UMessageProcessor()
 
     async def _ensure_started(self):

src/chat/message_receive/message.py

@@ -438,3 +438,52 @@ class MessageSet:
 
     def __len__(self) -> int:
         return len(self.messages)
+
+
+def message_recv_from_dict(message_dict: dict) -> MessageRecv:
+    return MessageRecv(message_dict)
+
+
+def message_from_db_dict(db_dict: dict) -> MessageRecv:
+    """从数据库字典创建MessageRecv实例"""
+    # 转换扁平的数据库字典为嵌套结构
+    message_info_dict = {
+        "platform": db_dict.get("chat_info_platform"),
+        "message_id": db_dict.get("message_id"),
+        "time": db_dict.get("time"),
+        "group_info": {
+            "platform": db_dict.get("chat_info_group_platform"),
+            "group_id": db_dict.get("chat_info_group_id"),
+            "group_name": db_dict.get("chat_info_group_name"),
+        },
+        "user_info": {
+            "platform": db_dict.get("user_platform"),
+            "user_id": db_dict.get("user_id"),
+            "user_nickname": db_dict.get("user_nickname"),
+            "user_cardname": db_dict.get("user_cardname"),
+        },
+    }
+
+    processed_text = db_dict.get("processed_plain_text", "")
+
+    # 构建 MessageRecv 需要的字典
+    recv_dict = {
+        "message_info": message_info_dict,
+        "message_segment": {"type": "text", "data": processed_text},  # 从纯文本重建消息段
+        "raw_message": None,  # 数据库中未存储原始消息
+        "processed_plain_text": processed_text,
+        "detailed_plain_text": db_dict.get("detailed_plain_text", ""),
+    }
+
+    # 创建 MessageRecv 实例
+    msg = MessageRecv(recv_dict)
+
+    # 从数据库字典中填充其他可选字段
+    msg.interest_value = db_dict.get("interest_value")
+    msg.is_mentioned = db_dict.get("is_mentioned")
+    msg.priority_mode = db_dict.get("priority_mode", "interest")
+    msg.priority_info = db_dict.get("priority_info")
+    msg.is_emoji = db_dict.get("is_emoji", False)
+    msg.is_picid = db_dict.get("is_picid", False)
+
+    return msg

src/chat/message_receive/normal_message_sender.py

@@ -1,297 +0,0 @@
-import asyncio
-import time
-from asyncio import Task
-from typing import Union
-from rich.traceback import install
-
-from src.common.logger import get_logger
-from src.common.message.api import get_global_api
-from src.chat.message_receive.storage import MessageStorage
-from src.chat.utils.utils import truncate_message, calculate_typing_time, count_messages_between
-from .message import MessageSending, MessageThinking, MessageSet
-
-install(extra_lines=3)
-
-logger = get_logger("sender")
-
-
-async def send_via_ws(message: MessageSending) -> None:
-    """通过 WebSocket 发送消息"""
-    try:
-        await get_global_api().send_message(message)
-    except Exception as e:
-        logger.error(f"WS发送失败: {e}")
-        raise ValueError(f"未找到平台：{message.message_info.platform} 的url配置，请检查配置文件") from e
-
-
-async def send_message(
-    message: MessageSending,
-) -> None:
-    """发送消息（核心发送逻辑）"""
-
-    # --- 添加计算打字和延迟的逻辑 (从 heartflow_message_sender 移动并调整) ---
-    typing_time = calculate_typing_time(
-        input_string=message.processed_plain_text,
-        thinking_start_time=message.thinking_start_time,
-        is_emoji=message.is_emoji,
-    )
-    # logger.debug(f"{message.processed_plain_text},{typing_time},计算输入时间结束") # 减少日志
-    await asyncio.sleep(typing_time)
-    # logger.debug(f"{message.processed_plain_text},{typing_time},等待输入时间结束") # 减少日志
-    # --- 结束打字延迟 ---
-
-    message_preview = truncate_message(message.processed_plain_text)
-
-    try:
-        await send_via_ws(message)
-        logger.info(f"发送消息   '{message_preview}'   成功")  # 调整日志格式
-    except Exception as e:
-        logger.error(f"发送消息   '{message_preview}'   失败: {str(e)}")
-
-
-class MessageSender:
-    """发送器 (不再是单例)"""
-
-    def __init__(self):
-        self.message_interval = (0.5, 1)  # 消息间隔时间范围(秒)
-        self.last_send_time = 0
-        self._current_bot = None
-
-    def set_bot(self, bot):
-        """设置当前bot实例"""
-        pass
-
-
-class MessageContainer:
-    """单个聊天流的发送/思考消息容器"""
-
-    def __init__(self, chat_id: str, max_size: int = 100):
-        self.chat_id = chat_id
-        self.max_size = max_size
-        self.messages: list[MessageThinking | MessageSending] = []  # 明确类型
-        self.last_send_time = 0
-        self.thinking_wait_timeout = 20  # 思考等待超时时间（秒） - 从旧 sender 合并
-
-    def count_thinking_messages(self) -> int:
-        """计算当前容器中思考消息的数量"""
-        return sum(isinstance(msg, MessageThinking) for msg in self.messages)
-
-    def get_timeout_sending_messages(self) -> list[MessageSending]:
-        # sourcery skip: merge-nested-ifs
-        """获取所有超时的MessageSending对象（思考时间超过20秒），按thinking_start_time排序 - 从旧 sender 合并"""
-        current_time = time.time()
-        timeout_messages = []
-
-        for msg in self.messages:
-            # 只检查 MessageSending 类型
-            if isinstance(msg, MessageSending):
-                # 确保 thinking_start_time 有效
-                if msg.thinking_start_time and current_time - msg.thinking_start_time > self.thinking_wait_timeout:
-                    timeout_messages.append(msg)
-
-        # 按thinking_start_time排序，时间早的在前面
-        timeout_messages.sort(key=lambda x: x.thinking_start_time)
-        return timeout_messages
-
-    def get_earliest_message(self):
-        """获取thinking_start_time最早的消息对象"""
-        if not self.messages:
-            return None
-        earliest_time = float("inf")
-        earliest_message = None
-        for msg in self.messages:
-            # 确保消息有 thinking_start_time 属性
-            msg_time = getattr(msg, "thinking_start_time", float("inf"))
-            if msg_time < earliest_time:
-                earliest_time = msg_time
-                earliest_message = msg
-        return earliest_message
-
-    def add_message(self, message: Union[MessageThinking, MessageSending, MessageSet]):
-        """添加消息到队列"""
-        if isinstance(message, MessageSet):
-            for single_message in message.messages:
-                self.messages.append(single_message)
-        else:
-            self.messages.append(message)
-
-    def remove_message(self, message_to_remove: Union[MessageThinking, MessageSending]):
-        """移除指定的消息对象，如果消息存在则返回True，否则返回False"""
-        try:
-            _initial_len = len(self.messages)
-            # 使用列表推导式或 message_filter 创建新列表，排除要删除的元素
-            # self.messages = [msg for msg in self.messages if msg is not message_to_remove]
-            # 或者直接 remove (如果确定对象唯一性)
-            if message_to_remove in self.messages:
-                self.messages.remove(message_to_remove)
-                return True
-            # logger.debug(f"Removed message {getattr(message_to_remove, 'message_info', {}).get('message_id', 'UNKNOWN')}. Old len: {initial_len}, New len: {len(self.messages)}")
-            # return len(self.messages) < initial_len
-            return False
-
-        except Exception as e:
-            logger.exception(f"移除消息时发生错误: {e}")
-            return False
-
-    def has_messages(self) -> bool:
-        """检查是否有待发送的消息"""
-        return bool(self.messages)
-
-    def get_all_messages(self) -> list[MessageThinking | MessageSending]:
-        """获取所有消息"""
-        return list(self.messages)  # 返回副本
-
-
-class MessageManager:
-    """管理所有聊天流的消息容器 (不再是单例)"""
-
-    def __init__(self):
-        self._processor_task: Task | None = None
-        self.containers: dict[str, MessageContainer] = {}
-        self.storage = MessageStorage()  # 添加 storage 实例
-        self._running = True  # 处理器运行状态
-        self._container_lock = asyncio.Lock()  # 保护 containers 字典的锁
-        # self.message_sender = MessageSender() # 创建发送器实例 (改为全局实例)
-
-    async def start(self):
-        """启动后台处理器任务。"""
-        # 检查是否已有任务在运行，避免重复启动
-        if self._processor_task is not None and not self._processor_task.done():
-            logger.warning("Processor task already running.")
-            return
-        self._processor_task = asyncio.create_task(self._start_processor_loop())
-        logger.debug("MessageManager processor task started.")
-
-    def stop(self):
-        """停止后台处理器任务。"""
-        self._running = False
-        if self._processor_task is not None and not self._processor_task.done():
-            self._processor_task.cancel()
-            logger.debug("MessageManager processor task stopping.")
-        else:
-            logger.debug("MessageManager processor task not running or already stopped.")
-
-    async def get_container(self, chat_id: str) -> MessageContainer:
-        """获取或创建聊天流的消息容器 (异步，使用锁)"""
-        async with self._container_lock:
-            if chat_id not in self.containers:
-                self.containers[chat_id] = MessageContainer(chat_id)
-            return self.containers[chat_id]
-
-    async def add_message(self, message: Union[MessageThinking, MessageSending, MessageSet]) -> None:
-        """添加消息到对应容器"""
-        chat_stream = message.chat_stream
-        if not chat_stream:
-            logger.error("消息缺少 chat_stream，无法添加到容器")
-            return  # 或者抛出异常
-        container = await self.get_container(chat_stream.stream_id)
-        container.add_message(message)
-
-    async def _handle_sending_message(self, container: MessageContainer, message: MessageSending):
-        """处理单个 MessageSending 消息 (包含 set_reply 逻辑)"""
-        try:
-            _ = message.update_thinking_time()  # 更新思考时间
-            thinking_start_time = message.thinking_start_time
-            now_time = time.time()
-            # logger.debug(f"thinking_start_time:{thinking_start_time},now_time:{now_time}")
-            thinking_messages_count, thinking_messages_length = count_messages_between(
-                start_time=thinking_start_time, end_time=now_time, stream_id=message.chat_stream.stream_id
-            )
-
-            if (
-                message.is_head
-                and (thinking_messages_count > 3 or thinking_messages_length > 200)
-                and not message.is_private_message()
-            ):
-                logger.debug(
-                    f"[{message.chat_stream.stream_id}] 应用 set_reply 逻辑: {message.processed_plain_text[:20]}..."
-                )
-                message.build_reply()
-            # --- 结束条件 set_reply ---
-
-            await message.process()  # 预处理消息内容
-
-            # logger.debug(f"{message}")
-
-            # 使用全局 message_sender 实例
-            await send_message(message)
-            await self.storage.store_message(message, message.chat_stream)
-
-            # 移除消息要在发送 *之后*
-            container.remove_message(message)
-            # logger.debug(f"[{message.chat_stream.stream_id}] Sent and removed message: {message.message_info.message_id}")
-
-        except Exception as e:
-            logger.error(
-                f"[{message.chat_stream.stream_id}] 处理发送消息 {getattr(message.message_info, 'message_id', 'N/A')} 时出错: {e}"
-            )
-            logger.exception("详细错误信息:")
-            if container.remove_message(message):
-                logger.warning(f"[{message.chat_stream.stream_id}] 已移除处理出错的消息。")
-
-    async def _process_chat_messages(self, chat_id: str):
-        """处理单个聊天流消息 (合并后的逻辑)"""
-        container = await self.get_container(chat_id)  # 获取容器是异步的了
-
-        if container.has_messages():
-            message_earliest = container.get_earliest_message()
-
-            if not message_earliest:  # 如果最早消息为空，则退出
-                return
-
-            if isinstance(message_earliest, MessageThinking):
-                # --- 处理思考消息 (来自旧 sender) ---
-                message_earliest.update_thinking_time()
-                thinking_time = message_earliest.thinking_time
-                # 减少控制台刷新频率或只在时间显著变化时打印
-                if int(thinking_time) % 5 == 0:  # 每5秒打印一次
-                    print(
-                        f"消息 {message_earliest.message_info.message_id} 正在思考中，已思考 {int(thinking_time)} 秒\r",
-                        end="",
-                        flush=True,
-                    )
-
-            elif isinstance(message_earliest, MessageSending):
-                # --- 处理发送消息 ---
-                await self._handle_sending_message(container, message_earliest)
-
-            if timeout_sending_messages := container.get_timeout_sending_messages():
-                logger.debug(f"[{chat_id}] 发现 {len(timeout_sending_messages)} 条超时的发送消息")
-                for msg in timeout_sending_messages:
-                    # 确保不是刚刚处理过的最早消息 (虽然理论上应该已被移除，但以防万一)
-                    if msg is message_earliest:
-                        continue
-                    logger.info(f"[{chat_id}] 处理超时发送消息: {msg.message_info.message_id}")
-                    await self._handle_sending_message(container, msg)  # 复用处理逻辑
-
-    async def _start_processor_loop(self):
-        # sourcery skip: list-comprehension, move-assign-in-block, use-named-expression
-        """消息处理器主循环"""
-        while self._running:
-            tasks = []
-            # 使用异步锁保护迭代器创建过程
-            async with self._container_lock:
-                # 创建 keys 的快照以安全迭代
-                chat_ids = list(self.containers.keys())
-
-            tasks.extend(asyncio.create_task(self._process_chat_messages(chat_id)) for chat_id in chat_ids)
-            if tasks:
-                try:
-                    # 等待当前批次的所有任务完成
-                    await asyncio.gather(*tasks)
-                except Exception as e:
-                    logger.error(f"消息处理循环 gather 出错: {e}")
-
-            # 等待一小段时间，避免CPU空转
-            try:
-                await asyncio.sleep(0.1)  # 稍微降低轮询频率
-            except asyncio.CancelledError:
-                logger.info("Processor loop sleep cancelled.")
-                break  # 退出循环
-        logger.info("MessageManager processor loop finished.")
-
-
-# --- 创建全局实例 ---
-message_manager = MessageManager()
-message_sender = MessageSender()
-# --- 结束全局实例 ---

src/chat/message_receive/storage.py

@@ -37,11 +37,21 @@ class MessageStorage:
                 else:
                     filtered_display_message = ""
                 interest_value = 0
+                is_mentioned = False
                 reply_to = message.reply_to
+                priority_mode = ""
+                priority_info = {}
+                is_emoji = False
+                is_picid = False
             else:
                 filtered_display_message = ""
                 interest_value = message.interest_value
+                is_mentioned = message.is_mentioned
                 reply_to = ""
+                priority_mode = message.priority_mode
+                priority_info = message.priority_info
+                is_emoji = message.is_emoji
+                is_picid = message.is_picid
 
             chat_info_dict = chat_stream.to_dict()
             user_info_dict = message.message_info.user_info.to_dict()  # type: ignore
@@ -60,6 +70,7 @@ class MessageStorage:
                 chat_id=chat_stream.stream_id,
                 # Flattened chat_info
                 reply_to=reply_to,
+                is_mentioned=is_mentioned,
                 chat_info_stream_id=chat_info_dict.get("stream_id"),
                 chat_info_platform=chat_info_dict.get("platform"),
                 chat_info_user_platform=user_info_from_chat.get("platform"),
@@ -81,6 +92,10 @@ class MessageStorage:
                 display_message=filtered_display_message,
                 memorized_times=message.memorized_times,
                 interest_value=interest_value,
+                priority_mode=priority_mode,
+                priority_info=priority_info,
+                is_emoji=is_emoji,
+                is_picid=is_picid,
             )
         except Exception:
             logger.exception("存储消息失败")

src/chat/message_receive/uni_message_sender.py

@@ -1,12 +1,11 @@
 import asyncio
 import traceback
 
-from typing import Dict, Optional
 from rich.traceback import install
 
 from src.common.message.api import get_global_api
 from src.common.logger import get_logger
-from src.chat.message_receive.message import MessageSending, MessageThinking
+from src.chat.message_receive.message import MessageSending
 from src.chat.message_receive.storage import MessageStorage
 from src.chat.utils.utils import truncate_message
 from src.chat.utils.utils import calculate_typing_time
@@ -37,42 +36,6 @@ class HeartFCSender:
 
     def __init__(self):
         self.storage = MessageStorage()
-        # 用于存储活跃的思考消息
-        self.thinking_messages: Dict[str, Dict[str, MessageThinking]] = {}
-        self._thinking_lock = asyncio.Lock()  # 保护 thinking_messages 的锁
-
-    async def register_thinking(self, thinking_message: MessageThinking):
-        """注册一个思考中的消息。"""
-        if not thinking_message.chat_stream or not thinking_message.message_info.message_id:
-            logger.error("无法注册缺少 chat_stream 或 message_id 的思考消息")
-            return
-
-        chat_id = thinking_message.chat_stream.stream_id
-        message_id = thinking_message.message_info.message_id
-
-        async with self._thinking_lock:
-            if chat_id not in self.thinking_messages:
-                self.thinking_messages[chat_id] = {}
-            if message_id in self.thinking_messages[chat_id]:
-                logger.warning(f"[{chat_id}] 尝试注册已存在的思考消息 ID: {message_id}")
-            self.thinking_messages[chat_id][message_id] = thinking_message
-            logger.debug(f"[{chat_id}] Registered thinking message: {message_id}")
-
-    async def complete_thinking(self, chat_id: str, message_id: str):
-        """完成并移除一个思考中的消息记录。"""
-        async with self._thinking_lock:
-            if chat_id in self.thinking_messages and message_id in self.thinking_messages[chat_id]:
-                del self.thinking_messages[chat_id][message_id]
-                logger.debug(f"[{chat_id}] Completed thinking message: {message_id}")
-                if not self.thinking_messages[chat_id]:
-                    del self.thinking_messages[chat_id]
-                    logger.debug(f"[{chat_id}] Removed empty thinking message container.")
-
-    async def get_thinking_start_time(self, chat_id: str, message_id: str) -> Optional[float]:
-        """获取已注册思考消息的开始时间。"""
-        async with self._thinking_lock:
-            thinking_message = self.thinking_messages.get(chat_id, {}).get(message_id)
-            return thinking_message.thinking_start_time if thinking_message else None
 
     async def send_message(self, message: MessageSending, typing=False, set_reply=False, storage_message=True):
         """
@@ -122,5 +85,3 @@ class HeartFCSender:
         except Exception as e:
             logger.error(f"[{chat_id}] 处理或存储消息 {message_id} 时出错: {e}")
             raise e
-        finally:
-            await self.complete_thinking(chat_id, message_id)