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

Merge branch 'dev' of https://github.com/MaiM-with-u/MaiBot into dev

Browse Source
This commit is contained in:
SengokuCola
2025-04-08 23:12:00 +08:00
parent 687c3f6710 cf3e64c753
commit a76437f710
45 changed files with 1177 additions and 1224 deletions
Download Patch File Download Diff File Expand all files Collapse all files

106
src/plugins/PFC/chat_observer.py
View File

@@ -1,6 +1,6 @@
import time
import asyncio
from typing import Optional, Dict, Any, List, Tuple
from typing import Optional, Dict, Any, List, Tuple
from src.common.logger import get_module_logger
from ..message.message_base import UserInfo
from ..config.config import global_config
@@ -9,16 +9,17 @@ from .message_storage import MessageStorage, MongoDBMessageStorage
logger = get_module_logger("chat_observer")
class ChatObserver:
"""聊天状态观察器"""
# 类级别的实例管理
_instances: Dict[str, 'ChatObserver'] = {}
_instances: Dict[str, "ChatObserver"] = {}
@classmethod
def get_instance(cls, stream_id: str, message_storage: Optional[MessageStorage] = None) -> 'ChatObserver':
"""获取或创建观察器实例
Args:
stream_id: 聊天流ID
message_storage: 消息存储实现，如果为None则使用MongoDB实现
@@ -32,14 +33,14 @@ class ChatObserver:
def __init__(self, stream_id: str, message_storage: Optional[MessageStorage] = None):
"""初始化观察器
Args:
stream_id: 聊天流ID
message_storage: 消息存储实现，如果为None则使用MongoDB实现
"""
if stream_id in self._instances:
raise RuntimeError(f"ChatObserver for {stream_id} already exists. Use get_instance() instead.")
self.stream_id = stream_id
self.message_storage = message_storage or MongoDBMessageStorage()
@@ -53,9 +54,9 @@ class ChatObserver:
# 消息历史记录
self.message_history: List[Dict[str, Any]] = [] # 所有消息历史
self.last_message_id: Optional[str] = None # 最后一条消息的ID
self.message_count: int = 0 # 消息计数
self.last_message_id: Optional[str] = None # 最后一条消息的ID
self.message_count: int = 0 # 消息计数
# 运行状态
self._running: bool = False
self._task: Optional[asyncio.Task] = None
@@ -77,7 +78,7 @@ class ChatObserver:
async def check(self) -> bool:
"""检查距离上一次观察之后是否有了新消息
Returns:
bool: 是否有新消息
"""
@@ -91,7 +92,7 @@ class ChatObserver:
if new_message_exists:
logger.debug("发现新消息")
self.last_check_time = time.time()
return new_message_exists
async def _add_message_to_history(self, message: Dict[str, Any]):
@@ -104,7 +105,7 @@ class ChatObserver:
self.last_message_id = message["message_id"]
self.last_message_time = message["time"] # 更新最后消息时间
self.message_count += 1
# 更新说话时间
user_info = UserInfo.from_dict(message.get("user_info", {}))
if user_info.user_id == global_config.BOT_QQ:
@@ -186,41 +187,40 @@ class ChatObserver:
start_time: Optional[float] = None,
end_time: Optional[float] = None,
limit: Optional[int] = None,
user_id: Optional[str] = None
user_id: Optional[str] = None,
) -> List[Dict[str, Any]]:
"""获取消息历史
Args:
start_time: 开始时间戳
end_time: 结束时间戳
limit: 限制返回消息数量
user_id: 指定用户ID
Returns:
List[Dict[str, Any]]: 消息列表
"""
filtered_messages = self.message_history
if start_time is not None:
filtered_messages = [m for m in filtered_messages if m["time"] >= start_time]
if end_time is not None:
filtered_messages = [m for m in filtered_messages if m["time"] <= end_time]
if user_id is not None:
filtered_messages = [
m for m in filtered_messages
if UserInfo.from_dict(m.get("user_info", {})).user_id == user_id
m for m in filtered_messages if UserInfo.from_dict(m.get("user_info", {})).user_id == user_id
]
if limit is not None:
filtered_messages = filtered_messages[-limit:]
return filtered_messages
async def _fetch_new_messages(self) -> List[Dict[str, Any]]:
"""获取新消息
Returns:
List[Dict[str, Any]]: 新消息列表
"""
@@ -231,15 +231,15 @@ class ChatObserver:
if new_messages:
self.last_message_read = new_messages[-1]["message_id"]
return new_messages
async def _fetch_new_messages_before(self, time_point: float) -> List[Dict[str, Any]]:
"""获取指定时间点之前的消息
Args:
time_point: 时间戳
Returns:
List[Dict[str, Any]]: 最多5条消息
"""
@@ -250,7 +250,7 @@ class ChatObserver:
if new_messages:
self.last_message_read = new_messages[-1]["message_id"]
return new_messages
'''主要观察循环'''
@@ -263,7 +263,7 @@ class ChatObserver:
await self._add_message_to_history(message)
except Exception as e:
logger.error(f"缓冲消息出错: {e}")
while self._running:
try:
# 等待事件或超时（1秒）
@@ -271,13 +271,13 @@ class ChatObserver:
await asyncio.wait_for(self._update_event.wait(), timeout=1)
except asyncio.TimeoutError:
pass # 超时后也执行一次检查
self._update_event.clear() # 重置触发事件
self._update_complete.clear() # 重置完成事件
# 获取新消息
new_messages = await self._fetch_new_messages()
if new_messages:
# 处理新消息
for message in new_messages:
@@ -285,21 +285,21 @@ class ChatObserver:
# 设置完成事件
self._update_complete.set()
except Exception as e:
logger.error(f"更新循环出错: {e}")
self._update_complete.set() # 即使出错也要设置完成事件
def trigger_update(self):
"""触发一次立即更新"""
self._update_event.set()
async def wait_for_update(self, timeout: float = 5.0) -> bool:
"""等待更新完成
Args:
timeout: 超时时间（秒）
Returns:
bool: 是否成功完成更新（False表示超时）
"""
@@ -309,16 +309,16 @@ class ChatObserver:
except asyncio.TimeoutError:
logger.warning(f"等待更新完成超时（{timeout}秒）")
return False
def start(self):
"""启动观察器"""
if self._running:
return
self._running = True
self._task = asyncio.create_task(self._update_loop())
logger.info(f"ChatObserver for {self.stream_id} started")
def stop(self):
"""停止观察器"""
self._running = False
@@ -327,15 +327,15 @@ class ChatObserver:
if self._task:
self._task.cancel()
logger.info(f"ChatObserver for {self.stream_id} stopped")
async def process_chat_history(self, messages: list):
"""处理聊天历史
Args:
messages: 消息列表
"""
self.update_check_time()
for msg in messages:
try:
user_info = UserInfo.from_dict(msg.get("user_info", {}))
@@ -345,33 +345,33 @@ class ChatObserver:
self.update_user_speak_time(msg["time"])
except Exception as e:
logger.warning(f"处理消息时间时出错: {e}")
continue
continue
def update_check_time(self):
"""更新查看时间"""
self.last_check_time = time.time()
def update_bot_speak_time(self, speak_time: Optional[float] = None):
"""更新机器人说话时间"""
self.last_bot_speak_time = speak_time or time.time()
def update_user_speak_time(self, speak_time: Optional[float] = None):
"""更新用户说话时间"""
self.last_user_speak_time = speak_time or time.time()
def get_time_info(self) -> str:
"""获取时间信息文本"""
current_time = time.time()
time_info = ""
if self.last_bot_speak_time:
bot_speak_ago = current_time - self.last_bot_speak_time
time_info += f"\n距离你上次发言已经过去了{int(bot_speak_ago)}秒"
if self.last_user_speak_time:
user_speak_ago = current_time - self.last_user_speak_time
time_info += f"\n距离对方上次发言已经过去了{int(user_speak_ago)}秒"
return time_info
def start_periodic_update(self):