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

refactor(logging): 将多个info日志级别的记录更改为debug级别，以减少日志输出

Browse Source
This commit is contained in:
Windpicker-owo
2025-11-01 00:44:50 +08:00
parent 49f376dc1c
commit 69ee822ef7
9 changed files with 105 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/chat/message_manager/context_manager.py
View File

@@ -39,7 +39,7 @@ class SingleStreamContextManager:
# 标记是否已初始化历史消息
self._history_initialized = False
logger.info(f"[新建] 单流上下文管理器初始化: {stream_id} (id={id(self)})")
logger.debug(f"单流上下文管理器初始化: {stream_id}")
# 异步初始化历史消息（不阻塞构造函数）
asyncio.create_task(self._initialize_history_from_db())
@@ -237,7 +237,7 @@ class SingleStreamContextManager:
else:
setattr(self.context, attr, time.time())
await self._update_stream_energy()
logger.info(f"清空单流上下文: {self.stream_id}")
logger.debug(f"清空单流上下文: {self.stream_id}")
return True
except Exception as e:
logger.error(f"清空单流上下文失败 {self.stream_id}: {e}", exc_info=True)
@@ -303,15 +303,15 @@ class SingleStreamContextManager:
async def _initialize_history_from_db(self):
"""从数据库初始化历史消息到context中"""
if self._history_initialized:
logger.info(f"历史消息已初始化，跳过: {self.stream_id}")
logger.debug(f"历史消息已初始化，跳过: {self.stream_id}")
return
# 立即设置标志，防止并发重复加载
logger.info(f"设置历史初始化标志: {self.stream_id}")
logger.debug(f"设置历史初始化标志: {self.stream_id}")
self._history_initialized = True
try:
logger.info(f"开始从数据库加载历史消息: {self.stream_id}")
logger.debug(f"开始从数据库加载历史消息: {self.stream_id}")
from src.chat.utils.chat_message_builder import get_raw_msg_before_timestamp_with_chat
@@ -339,7 +339,7 @@ class SingleStreamContextManager:
logger.warning(f"转换历史消息失败 (message_id={msg_dict.get('message_id', 'unknown')}): {e}")
continue
logger.info(f"成功从数据库加载 {len(self.context.history_messages)} 条历史消息到内存: {self.stream_id}")
logger.debug(f"成功从数据库加载 {len(self.context.history_messages)} 条历史消息到内存: {self.stream_id}")
else:
logger.debug(f"没有历史消息需要加载: {self.stream_id}")

26
src/chat/message_manager/distribution_manager.py
View File

@@ -118,12 +118,12 @@ class StreamLoopManager:
# 如果是强制启动且任务仍在运行，先取消旧任务
if force and context.stream_loop_task and not context.stream_loop_task.done():
logger.info(f"强制启动模式：先取消现有流循环任务: {stream_id}")
logger.debug(f"强制启动模式：先取消现有流循环任务: {stream_id}")
old_task = context.stream_loop_task
old_task.cancel()
try:
await asyncio.wait_for(old_task, timeout=2.0)
logger.info(f"旧流循环任务已结束: {stream_id}")
logger.debug(f"旧流循环任务已结束: {stream_id}")
except (asyncio.TimeoutError, asyncio.CancelledError):
logger.debug(f"旧流循环任务已取消或超时: {stream_id}")
except Exception as e:
@@ -140,7 +140,7 @@ class StreamLoopManager:
self.stats["active_streams"] += 1
self.stats["total_loops"] += 1
logger.info(f"启动流循环任务: {stream_id}")
logger.debug(f"启动流循环任务: {stream_id}")
return True
except Exception as e:
@@ -183,7 +183,7 @@ class StreamLoopManager:
# 清空 StreamContext 中的任务记录
context.stream_loop_task = None
logger.info(f"停止流循环: {stream_id}")
logger.debug(f"停止流循环: {stream_id}")
return True
async def _stream_loop_worker(self, stream_id: str) -> None:
@@ -192,7 +192,7 @@ class StreamLoopManager:
Args:
stream_id: 流ID
"""
logger.info(f"流循环工作器启动: {stream_id}")
logger.debug(f"流循环工作器启动: {stream_id}")
try:
while self.is_running:
@@ -243,7 +243,7 @@ class StreamLoopManager:
await asyncio.sleep(interval)
except asyncio.CancelledError:
logger.info(f"流循环被取消: {stream_id}")
logger.debug(f"流循环被取消: {stream_id}")
break
except Exception as e:
logger.error(f"流循环出错 {stream_id}: {e}", exc_info=True)
@@ -263,7 +263,7 @@ class StreamLoopManager:
# 清理间隔记录
self._last_intervals.pop(stream_id, None)
logger.info(f"流循环结束: {stream_id}")
logger.debug(f"流循环结束: {stream_id}")
async def _get_stream_context(self, stream_id: str) -> Any | None:
"""获取流上下文
@@ -333,7 +333,7 @@ class StreamLoopManager:
# 在处理开始前，先刷新缓存到未读消息
cached_messages = await self._flush_cached_messages_to_unread(stream_id)
if cached_messages:
logger.info(f"处理开始前刷新缓存消息: stream={stream_id}, 数量={len(cached_messages)}")
logger.debug(f"处理开始前刷新缓存消息: stream={stream_id}, 数量={len(cached_messages)}")
# 设置触发用户ID，以实现回复保护
last_message = context.get_last_message()
@@ -357,7 +357,7 @@ class StreamLoopManager:
# 处理成功后，再次刷新缓存中可能的新消息
additional_messages = await self._flush_cached_messages_to_unread(stream_id)
if additional_messages:
logger.info(f"处理完成后刷新新消息: stream={stream_id}, 数量={len(additional_messages)}")
logger.debug(f"处理完成后刷新新消息: stream={stream_id}, 数量={len(additional_messages)}")
process_time = time.time() - start_time
logger.debug(f"流处理成功: {stream_id} (耗时: {process_time:.2f}s)")
@@ -367,7 +367,7 @@ class StreamLoopManager:
return success
except asyncio.CancelledError:
logger.info(f"流处理被取消: {stream_id}")
logger.debug(f"流处理被取消: {stream_id}")
# 取消所有子任务
for child_task in child_tasks:
if not child_task.done():
@@ -552,7 +552,7 @@ class StreamLoopManager:
chatter_manager: chatter管理器实例
"""
self.chatter_manager = chatter_manager
logger.info(f"设置chatter管理器: {chatter_manager.__class__.__name__}")
logger.debug(f"设置chatter管理器: {chatter_manager.__class__.__name__}")
async def _should_force_dispatch_for_stream(self, stream_id: str) -> bool:
if not self.force_dispatch_unread_threshold or self.force_dispatch_unread_threshold <= 0:
@@ -652,7 +652,7 @@ class StreamLoopManager:
Args:
stream_id: 流ID
"""
logger.info(f"强制分发流处理: {stream_id}")
logger.debug(f"强制分发流处理: {stream_id}")
try:
# 获取流上下文
@@ -663,7 +663,7 @@ class StreamLoopManager:
# 检查是否有现有的 stream_loop_task
if context.stream_loop_task and not context.stream_loop_task.done():
logger.info(f"发现现有流循环 {stream_id}，将先取消再重新创建")
logger.debug(f"发现现有流循环 {stream_id}，将先取消再重新创建")
existing_task = context.stream_loop_task
existing_task.cancel()
# 创建异步任务来等待取消完成，并添加异常处理

7
src/chat/message_manager/global_notice_manager.py
View File

@@ -74,7 +74,7 @@ class GlobalNoticeManager:
"last_cleanup_time": 0,
}
logger.info("全局Notice管理器初始化完成")
logger.debug("全局Notice管理器初始化完成")
def add_notice(
self,
@@ -135,7 +135,7 @@ class GlobalNoticeManager:
# 定期清理过期消息
self._cleanup_expired_notices()
logger.info(f"✅ Notice已添加: id={message.message_id}, type={self._get_notice_type(message)}, scope={scope.value}, target={target_stream_id}, storage_key={storage_key}, ttl={ttl}s")
logger.debug(f"Notice已添加: id={message.message_id}, type={self._get_notice_type(message)}, scope={scope.value}")
return True
except Exception as e:
@@ -282,7 +282,8 @@ class GlobalNoticeManager:
for key in keys_to_remove:
del self._notices[key]
logger.info(f"清理notice消息: {removed_count} 条")
if removed_count > 0:
logger.debug(f"清理notice消息: {removed_count} 条")
return removed_count
except Exception as e:

33
src/chat/message_manager/message_manager.py
View File

@@ -72,13 +72,13 @@ class MessageManager:
logger.error(f"启动批量数据库写入器失败: {e}")
# 启动消息缓存系统（内置）
logger.info("📦 消息缓存系统已启动")
logger.debug("消息缓存系统已启动")
# 启动流循环管理器并设置chatter_manager
await stream_loop_manager.start()
stream_loop_manager.set_chatter_manager(self.chatter_manager)
logger.info("🚀 消息管理器已启动 | 流循环管理器已启动")
logger.info("消息管理器已启动")
async def stop(self):
"""停止消息管理器"""
@@ -92,19 +92,19 @@ class MessageManager:
from src.chat.message_manager.batch_database_writer import shutdown_batch_writer
await shutdown_batch_writer()
logger.info("📦 批量数据库写入器已停止")
logger.debug("批量数据库写入器已停止")
except Exception as e:
logger.error(f"停止批量数据库写入器失败: {e}")
# 停止消息缓存系统（内置）
self.message_caches.clear()
self.stream_processing_status.clear()
logger.info("📦 消息缓存系统已停止")
logger.debug("消息缓存系统已停止")
# 停止流循环管理器
await stream_loop_manager.stop()
logger.info("🛑 消息管理器已停止 | 流循环管理器已停止")
logger.info("消息管理器已停止")
async def add_message(self, stream_id: str, message: DatabaseMessages):
"""添加消息到指定聊天流"""
@@ -113,15 +113,15 @@ class MessageManager:
# 检查是否为notice消息
if self._is_notice_message(message):
# Notice消息处理 - 添加到全局管理器
logger.info(f"📢 检测到notice消息: notice_type={getattr(message, 'notice_type', None)}")
logger.debug(f"检测到notice消息: notice_type={getattr(message, 'notice_type', None)}")
await self._handle_notice_message(stream_id, message)
# 根据配置决定是否继续处理（触发聊天流程）
if not global_config.notice.enable_notice_trigger_chat:
logger.info(f"根据配置，流 {stream_id} 的Notice消息将被忽略，不触发聊天流程。")
logger.debug(f"Notice消息将被忽略，不触发聊天流程: {stream_id}")
return # 停止处理，不进入未读消息队列
else:
logger.info(f"根据配置，流 {stream_id} 的Notice消息将触发聊天流程。")
logger.debug(f"Notice消息将触发聊天流程: {stream_id}")
# 继续执行，将消息添加到未读队列
# 普通消息处理
@@ -201,7 +201,7 @@ class MessageManager:
if hasattr(context, "processing_task") and context.processing_task and not context.processing_task.done():
context.processing_task.cancel()
logger.info(f"停用聊天流: {stream_id}")
logger.debug(f"停用聊天流: {stream_id}")
except Exception as e:
logger.error(f"停用聊天流 {stream_id} 时发生错误: {e}")
@@ -218,7 +218,7 @@ class MessageManager:
context = chat_stream.context_manager.context
context.is_active = True
logger.info(f"激活聊天流: {stream_id}")
logger.debug(f"激活聊天流: {stream_id}")
except Exception as e:
logger.error(f"激活聊天流 {stream_id} 时发生错误: {e}")
@@ -354,8 +354,7 @@ class MessageManager:
# 取消 stream_loop_task，子任务会通过 try-catch 自动取消
try:
stream_loop_task.cancel()
logger.info(f"已发送取消信号到流循环任务: {chat_stream.stream_id}")
# 等待任务真正结束（设置超时避免死锁）
try:
await asyncio.wait_for(stream_loop_task, timeout=2.0)
@@ -401,21 +400,21 @@ class MessageManager:
# 确保有未读消息需要处理
unread_messages = context.get_unread_messages()
if not unread_messages:
logger.debug(f"💭 聊天流 {stream_id} 没有未读消息，跳过重新处理")
logger.debug(f"聊天流 {stream_id} 没有未读消息，跳过重新处理")
return
logger.info(f"💬 准备重新处理 {len(unread_messages)} 条未读消息: {stream_id}")
logger.debug(f"准备重新处理 {len(unread_messages)} 条未读消息: {stream_id}")
# 重新创建 stream_loop 任务
success = await stream_loop_manager.start_stream_loop(stream_id, force=True)
if success:
logger.info(f"✅ 成功重新创建流循环任务: {stream_id}")
logger.debug(f"成功重新创建流循环任务: {stream_id}")
else:
logger.warning(f"⚠️ 重新创建流循环任务失败: {stream_id}")
logger.warning(f"重新创建流循环任务失败: {stream_id}")
except Exception as e:
logger.error(f"🚨 触发重新处理时出错: {e}")
logger.error(f"触发重新处理时出错: {e}")
async def clear_all_unread_messages(self, stream_id: str):
"""清除指定上下文中的所有未读消息，在消息处理完成后调用"""