diff --git a/src/plugins/group_nickname/nickname_manager.py b/src/plugins/group_nickname/nickname_manager.py index a5a3872f..b54c92b3 100644 --- a/src/plugins/group_nickname/nickname_manager.py +++ b/src/plugins/group_nickname/nickname_manager.py @@ -1,12 +1,9 @@ import asyncio import threading -import queue -import traceback import time import json import re from typing import Dict, Optional, List, Any - from pymongo.errors import OperationFailure, DuplicateKeyError from src.common.logger_manager import get_logger from src.common.database import db @@ -15,39 +12,76 @@ from src.plugins.models.utils_model import LLMRequest from .nickname_db import NicknameDB from .nickname_mapper import _build_mapping_prompt from .nickname_utils import select_nicknames_for_prompt, format_nickname_prompt_injection - -# 依赖于 person_info_manager 来生成 person_id from ..person_info.person_info import person_info_manager - -# 依赖于 relationship_manager 来获取用户名称和现有绰号 from ..person_info.relationship_manager import relationship_manager - -# 导入消息和聊天流相关的类型和工具 from src.plugins.chat.chat_stream import ChatStream from src.plugins.chat.message import MessageRecv from src.plugins.utils.chat_message_builder import build_readable_messages, get_raw_msg_before_timestamp_with_chat logger = get_logger("NicknameManager") +logger_helper = get_logger("AsyncLoopHelper") # 为辅助函数创建单独的 logger +def run_async_loop(loop: asyncio.AbstractEventLoop, coro): + """ + 运行给定的协程直到完成，并确保循环最终关闭。 + + Args: + loop: 要使用的 asyncio 事件循环。 + coro: 要在循环中运行的主协程。 + """ + try: + logger_helper.debug(f"Running coroutine in loop {id(loop)}...") + result = loop.run_until_complete(coro) + logger_helper.debug(f"Coroutine completed in loop {id(loop)}.") + return result + except asyncio.CancelledError: + logger_helper.info(f"Coroutine in loop {id(loop)} was cancelled.") + # 取消是预期行为，不视为错误 + except Exception as e: + logger_helper.error(f"Error in async loop {id(loop)}: {e}", exc_info=True) + finally: + try: + # 1. 取消所有剩余任务 + all_tasks = asyncio.all_tasks(loop) + current_task = asyncio.current_task(loop) + tasks_to_cancel = [task for task in all_tasks if task is not current_task] # 避免取消 run_until_complete 本身 + if tasks_to_cancel: + logger_helper.info(f"Cancelling {len(tasks_to_cancel)} outstanding tasks in loop {id(loop)}...") + for task in tasks_to_cancel: + task.cancel() + # 等待取消完成 + loop.run_until_complete(asyncio.gather(*tasks_to_cancel, return_exceptions=True)) + logger_helper.info(f"Outstanding tasks cancelled in loop {id(loop)}.") + + # 2. 停止循环 (如果仍在运行) + if loop.is_running(): + loop.stop() + logger_helper.info(f"Asyncio loop {id(loop)} stopped.") + + # 3. 关闭循环 (如果未关闭) + if not loop.is_closed(): + # 在关闭前再运行一次以处理挂起的关闭回调 + loop.run_until_complete(loop.shutdown_asyncgens()) # 关闭异步生成器 + loop.close() + logger_helper.info(f"Asyncio loop {id(loop)} closed.") + except Exception as close_err: + logger_helper.error(f"Error during asyncio loop cleanup for loop {id(loop)}: {close_err}", exc_info=True) class NicknameManager: """ 管理群组绰号分析、处理、存储和使用的单例类。 封装了 LLM 调用、后台处理线程和数据库交互。 """ - _instance = None _lock = threading.Lock() - # Singleton Implementation def __new__(cls, *args, **kwargs): if not cls._instance: with cls._lock: - # 再次检查，防止多线程并发创建实例 if not cls._instance: logger.info("正在创建 NicknameManager 单例实例...") cls._instance = super(NicknameManager, cls).__new__(cls) - cls._instance._initialized = False # 添加初始化标志 + cls._instance._initialized = False return cls._instance def __init__(self): @@ -55,12 +89,11 @@ class NicknameManager: 初始化 NicknameManager。 使用锁和标志确保实际初始化只执行一次。 """ - if self._initialized: # 如果已初始化，直接返回 + if hasattr(self, '_initialized') and self._initialized: return with self._lock: - # 再次检查初始化标志，防止重复初始化 - if self._initialized: + if hasattr(self, '_initialized') and self._initialized: return logger.info("正在初始化 NicknameManager 组件...") @@ -68,11 +101,11 @@ class NicknameManager: self.is_enabled = self.config.ENABLE_NICKNAME_MAPPING # 数据库处理器 - person_info_collection = getattr(db, "person_info", None) + person_info_collection = getattr(db, 'person_info', None) self.db_handler = NicknameDB(person_info_collection) if not self.db_handler.is_available(): logger.error("数据库处理器初始化失败，NicknameManager 功能受限。") - self.is_enabled = False # 如果数据库不可用，禁用功能 + self.is_enabled = False # LLM 映射器 self.llm_mapper: Optional[LLMRequest] = None @@ -82,8 +115,8 @@ class NicknameManager: if model_config and model_config.get("name"): self.llm_mapper = LLMRequest( model=model_config, - temperature=model_config.get("temp", 0.5), # 使用 get 获取并提供默认值 - max_tokens=model_config.get("max_tokens", 256), # 使用 get 获取并提供默认值 + temperature=model_config.get("temp", 0.5), + max_tokens=model_config.get("max_tokens", 256), request_type="nickname_mapping", ) logger.info("绰号映射 LLM 映射器初始化成功。") @@ -101,16 +134,15 @@ class NicknameManager: # 队列和线程 self.queue_max_size = getattr(self.config, "NICKNAME_QUEUE_MAX_SIZE", 100) - self.nickname_queue: queue.Queue = queue.Queue(maxsize=self.queue_max_size) - self._stop_event = threading.Event() + # 使用 asyncio.Queue + self.nickname_queue: asyncio.Queue = asyncio.Queue(maxsize=self.queue_max_size) + self._stop_event = threading.Event() # stop_event 仍然使用 threading.Event，因为它是由另一个线程设置的 self._nickname_thread: Optional[threading.Thread] = None - self.sleep_interval = getattr(self.config, "NICKNAME_PROCESS_SLEEP_INTERVAL", 0.5) + self.sleep_interval = getattr(self.config, "NICKNAME_PROCESS_SLEEP_INTERVAL", 0.5) # 超时时间 - self._initialized = True # 标记为已初始化 + self._initialized = True logger.info("NicknameManager 初始化完成。") - # 公共方法 - def start_processor(self): """启动后台处理线程（如果已启用且未运行）。""" if not self.is_enabled: @@ -118,10 +150,10 @@ class NicknameManager: return if self._nickname_thread is None or not self._nickname_thread.is_alive(): logger.info("正在启动绰号处理器线程...") - self._stop_event.clear() # 清除停止事件标志 + self._stop_event.clear() self._nickname_thread = threading.Thread( - target=self._run_processor_in_thread, # 线程执行的入口函数 - daemon=True, # 设置为守护线程，主程序退出时自动结束 + target=self._run_processor_in_thread, # 线程目标函数不变 + daemon=True ) self._nickname_thread.start() logger.info(f"绰号处理器线程已启动 (ID: {self._nickname_thread.ident})") @@ -132,18 +164,10 @@ class NicknameManager: """停止后台处理线程。""" if self._nickname_thread and self._nickname_thread.is_alive(): logger.info("正在停止绰号处理器线程...") - self._stop_event.set() # 设置停止事件标志 + self._stop_event.set() # 设置停止事件，_processing_loop 会检测到 try: - # 可选：尝试清空队列，避免丢失未处理的任务 - # while not self.nickname_queue.empty(): - # try: - # self.nickname_queue.get_nowait() - # self.nickname_queue.task_done() - # except queue.Empty: - # break - # logger.info("绰号处理队列已清空。") - - self._nickname_thread.join(timeout=10) # 等待线程结束，设置超时 + # 不需要清空 asyncio.Queue，让循环自然结束或被取消 + self._nickname_thread.join(timeout=10) # 等待线程结束 if self._nickname_thread.is_alive(): logger.warning("绰号处理器线程在超时后仍未停止。") except Exception as e: @@ -151,7 +175,7 @@ class NicknameManager: finally: if self._nickname_thread and not self._nickname_thread.is_alive(): logger.info("绰号处理器线程已成功停止。") - self._nickname_thread = None # 清理线程对象引用 + self._nickname_thread = None else: logger.info("绰号处理器线程未在运行或已被清理。") @@ -163,10 +187,10 @@ class NicknameManager: ): """ 准备数据并将其排队等待绰号分析（如果满足条件）。 - 取代了旧的 trigger_nickname_analysis_if_needed 函数。 + (现在调用异步的 _add_to_queue) """ if not self.is_enabled: - return # 功能禁用则直接返回 + return current_chat_stream = chat_stream or anchor_message.chat_stream if not current_chat_stream or not current_chat_stream.group_info: @@ -182,56 +206,41 @@ class NicknameManager: timestamp=time.time(), limit=history_limit, ) - # 格式化历史记录 chat_history_str = await build_readable_messages( messages=history_messages, - replace_bot_name=True, - merge_messages=False, - timestamp_mode="relative", - read_mark=0.0, - truncate=False, + replace_bot_name=True, merge_messages=False, timestamp_mode="relative", + read_mark=0.0, truncate=False, ) - # 2. 获取 Bot 回复 bot_reply_str = " ".join(bot_reply) if bot_reply else "" - # 3. 获取群组和平台信息 group_id = str(current_chat_stream.group_info.group_id) platform = current_chat_stream.platform - # 4. 构建用户 ID 到名称的映射 (user_name_map) - user_ids_in_history = { - str(msg["user_info"]["user_id"]) for msg in history_messages if msg.get("user_info", {}).get("user_id") - } + user_ids_in_history = {str(msg["user_info"]["user_id"]) for msg in history_messages if msg.get("user_info", {}).get("user_id")} user_name_map = {} if user_ids_in_history: try: - # 使用 relationship_manager 批量获取名称 names_data = await relationship_manager.get_person_names_batch(platform, list(user_ids_in_history)) except Exception as e: logger.error(f"{log_prefix} 批量获取 person_name 时出错: {e}", exc_info=True) names_data = {} - - # 填充 user_name_map for user_id in user_ids_in_history: if user_id in names_data: user_name_map[user_id] = names_data[user_id] else: - # 回退查找历史记录中的 nickname latest_nickname = next( - ( - m["user_info"].get("user_nickname") - for m in reversed(history_messages) - if str(m["user_info"].get("user_id")) == user_id and m["user_info"].get("user_nickname") - ), + (m["user_info"].get("user_nickname") + for m in reversed(history_messages) + if str(m["user_info"].get("user_id")) == user_id and m["user_info"].get("user_nickname")), None, ) user_name_map[user_id] = latest_nickname or f"未知({user_id})" - # 5. 添加到内部处理队列 + item = (chat_history_str, bot_reply_str, platform, group_id, user_name_map) - self._add_to_queue(item, platform, group_id) # 调用私有方法入队 + await self._add_to_queue(item, platform, group_id) except Exception as e: logger.error(f"{log_prefix} 触发绰号分析时出错: {e}", exc_info=True) @@ -239,24 +248,16 @@ class NicknameManager: async def get_nickname_prompt_injection(self, chat_stream: ChatStream, message_list_before_now: List[Dict]) -> str: """ 获取并格式化用于 Prompt 注入的绰号信息字符串。 - 取代了旧的 get_nickname_injection_for_prompt 函数。 """ if not self.is_enabled or not chat_stream or not chat_stream.group_info: - return "" # 功能禁用或非群聊则返回空 + return "" log_prefix = f"[{chat_stream.stream_id}]" try: group_id = str(chat_stream.group_info.group_id) platform = chat_stream.platform + user_ids_in_context = {str(msg["user_info"]["user_id"]) for msg in message_list_before_now if msg.get("user_info", {}).get("user_id")} - # 确定上下文中的用户 ID - user_ids_in_context = { - str(msg["user_info"]["user_id"]) - for msg in message_list_before_now - if msg.get("user_info", {}).get("user_id") - } - - # 如果消息列表为空，尝试获取最近发言者 if not user_ids_in_context: recent_speakers = chat_stream.get_recent_speakers(limit=5) user_ids_in_context.update(str(speaker["user_id"]) for speaker in recent_speakers) @@ -265,41 +266,40 @@ class NicknameManager: logger.warning(f"{log_prefix} 未找到上下文用户用于绰号注入。") return "" - # 使用 relationship_manager 批量获取这些用户的群组绰号 all_nicknames_data = await relationship_manager.get_users_group_nicknames( platform, list(user_ids_in_context), group_id ) if all_nicknames_data: - # 使用 nickname_utils 中的工具函数进行选择和格式化 selected_nicknames = select_nicknames_for_prompt(all_nicknames_data) injection_str = format_nickname_prompt_injection(selected_nicknames) if injection_str: logger.debug(f"{log_prefix} 生成的绰号 Prompt 注入:\n{injection_str}") return injection_str else: - return "" # 没有获取到绰号数据 + return "" except Exception as e: logger.error(f"{log_prefix} 获取绰号注入时出错: {e}", exc_info=True) - return "" # 出错时返回空 + return "" + # 私有/内部方法 - def _add_to_queue(self, item: tuple, platform: str, group_id: str): - """将项目添加到内部处理队列。""" + async def _add_to_queue(self, item: tuple, platform: str, group_id: str): + """将项目异步添加到内部处理队列 (asyncio.Queue)。""" try: - self.nickname_queue.put_nowait(item) - logger.debug( - f"已将项目添加到平台 '{platform}' 群组 '{group_id}' 的绰号队列。当前大小: {self.nickname_queue.qsize()}" - ) - except queue.Full: - logger.warning( - f"绰号队列已满 (最大={self.queue_max_size})。平台 '{platform}' 群组 '{group_id}' 的项目被丢弃。" - ) + # 使用 await put()，如果队列满则异步等待 + await self.nickname_queue.put(item) + logger.debug(f"已将项目添加到平台 '{platform}' 群组 '{group_id}' 的绰号队列。当前大小: {self.nickname_queue.qsize()}") + except asyncio.QueueFull: + # 理论上 await put() 不会直接抛 QueueFull，除非 maxsize=0 + # 但保留以防万一或未来修改 + logger.warning(f"绰号队列已满 (最大={self.queue_max_size})。平台 '{platform}' 群组 '{group_id}' 的项目被丢弃。") except Exception as e: logger.error(f"将项目添加到绰号队列时出错: {e}", exc_info=True) + async def _analyze_and_update_nicknames(self, item: tuple): """处理单个队列项目：调用 LLM 分析并更新数据库。""" if not isinstance(item, tuple) or len(item) != 5: @@ -307,8 +307,9 @@ class NicknameManager: return chat_history_str, bot_reply, platform, group_id, user_name_map = item - thread_id = threading.get_ident() - log_prefix = f"[线程 {thread_id}][{platform}:{group_id}]" + # 使用 asyncio.get_running_loop().call_soon(threading.get_ident) 可能不准确，线程ID是同步概念 + # 可以考虑移除线程ID日志或寻找异步安全的获取标识符的方式 + log_prefix = f"[{platform}:{group_id}]" # 简化日志前缀 logger.debug(f"{log_prefix} 开始处理绰号分析任务...") if not self.llm_mapper: @@ -318,16 +319,16 @@ class NicknameManager: logger.error(f"{log_prefix} 数据库处理器不可用，无法更新计数。") return - # 1. 调用 LLM 分析 (逻辑从 nickname_mapper 移入) + # 1. 调用 LLM 分析 (内部逻辑不变) analysis_result = await self._call_llm_for_analysis(chat_history_str, bot_reply, user_name_map) - # 2. 如果分析成功且找到映射，则更新数据库 + # 2. 如果分析成功且找到映射，则更新数据库 (内部逻辑不变) if analysis_result.get("is_exist") and analysis_result.get("data"): nickname_map_to_update = analysis_result["data"] logger.info(f"{log_prefix} LLM 找到绰号映射，准备更新数据库: {nickname_map_to_update}") for user_id_str, nickname in nickname_map_to_update.items(): - # 基本验证 + # ... (验证和数据库更新逻辑保持不变) ... if not user_id_str or not nickname: logger.warning(f"{log_prefix} 跳过无效条目: user_id='{user_id_str}', nickname='{nickname}'") continue @@ -335,24 +336,15 @@ class NicknameManager: logger.warning(f"{log_prefix} 无效的用户ID格式 (非纯数字): '{user_id_str}'，跳过。") continue user_id_int = int(user_id_str) - # 结束验证 try: - # 步骤 1: 生成 person_id person_id = person_info_manager.get_person_id(platform, user_id_str) if not person_id: - logger.error( - f"{log_prefix} 无法为 platform='{platform}', user_id='{user_id_str}' 生成 person_id，跳过此用户。" - ) + logger.error(f"{log_prefix} 无法为 platform='{platform}', user_id='{user_id_str}' 生成 person_id，跳过此用户。") continue - - # 步骤 2: 确保 Person 文档存在 (调用 DB Handler) self.db_handler.upsert_person(person_id, user_id_int, platform) - - # 步骤 3: 更新群组绰号 (调用 DB Handler) self.db_handler.update_group_nickname_count(person_id, group_id, nickname) - - except (OperationFailure, DuplicateKeyError) as db_err: # 捕获特定的数据库错误 + except (OperationFailure, DuplicateKeyError) as db_err: logger.exception( f"{log_prefix} 数据库操作失败 ({type(db_err).__name__}): 用户 {user_id_str}, 绰号 {nickname}. 错误: {db_err}" ) @@ -361,6 +353,7 @@ class NicknameManager: else: logger.debug(f"{log_prefix} LLM 未找到可靠的绰号映射或分析失败。") + async def _call_llm_for_analysis( self, chat_history_str: str, @@ -369,17 +362,16 @@ class NicknameManager: ) -> Dict[str, Any]: """ 内部方法：调用 LLM 分析聊天记录和 Bot 回复，提取可靠的 用户ID-绰号 映射。 - (逻辑从 analyze_chat_for_nicknames 移入) """ - if not self.llm_mapper: # 再次检查 LLM 映射器 + # ... (此方法内部逻辑保持不变) ... + if not self.llm_mapper: logger.error("LLM 映射器未初始化，无法执行分析。") return {"is_exist": False} prompt = _build_mapping_prompt(chat_history_str, bot_reply, user_name_map) - logger.debug(f"构建的绰号映射 Prompt:\n{prompt[:500]}...") # 截断日志输出 + logger.debug(f"构建的绰号映射 Prompt:\n{prompt[:500]}...") try: - # 调用 LLM response_content, _, _ = await self.llm_mapper.generate_response(prompt) logger.debug(f"LLM 原始响应 (绰号映射): {response_content}") @@ -387,28 +379,23 @@ class NicknameManager: logger.warning("LLM 返回了空的绰号映射内容。") return {"is_exist": False} - # 清理可能的 Markdown 代码块标记 response_content = response_content.strip() markdown_code_regex = re.compile(r"^```(?:\w+)?\s*\n(.*?)\n\s*```$", re.DOTALL | re.IGNORECASE) match = markdown_code_regex.match(response_content) if match: response_content = match.group(1).strip() - # 尝试直接解析 JSON，即使没有代码块标记 elif response_content.startswith("{") and response_content.endswith("}"): pass # 可能是纯 JSON else: - # 尝试在文本中查找 JSON 对象 - json_match = re.search(r"\{.*\}", response_content, re.DOTALL) + json_match = re.search(r'\{.*\}', response_content, re.DOTALL) if json_match: response_content = json_match.group(0) else: logger.warning(f"LLM 响应似乎不包含有效的 JSON 对象。响应: {response_content}") return {"is_exist": False} - # 解析 JSON result = json.loads(response_content) - # 结果验证和过滤 if not isinstance(result, dict): logger.warning(f"LLM 响应不是一个有效的 JSON 对象 (字典类型)。响应内容: {response_content}") return {"is_exist": False} @@ -419,7 +406,7 @@ class NicknameManager: original_data = result.get("data") if isinstance(original_data, dict) and original_data: logger.info(f"LLM 找到的原始绰号映射: {original_data}") - filtered_data = self._filter_llm_results(original_data, user_name_map) # 调用过滤函数 + filtered_data = self._filter_llm_results(original_data, user_name_map) if not filtered_data: logger.info("所有找到的绰号映射都被过滤掉了。") return {"is_exist": False} @@ -427,13 +414,12 @@ class NicknameManager: logger.info(f"过滤后的绰号映射: {filtered_data}") return {"is_exist": True, "data": filtered_data} else: - # is_exist 为 True 但 data 缺失、不是字典或为空 logger.warning(f"LLM 响应格式错误: is_exist=True 但 data 无效。原始 data: {original_data}") return {"is_exist": False} elif is_exist is False: logger.info("LLM 明确指示未找到可靠的绰号映射 (is_exist=False)。") return {"is_exist": False} - else: # is_exist 不是 True 或 False (包括 None) + else: logger.warning(f"LLM 响应格式错误: 'is_exist' 的值 '{is_exist}' 无效。") return {"is_exist": False} @@ -444,104 +430,85 @@ class NicknameManager: logger.error(f"绰号映射 LLM 调用或处理过程中发生意外错误: {e}", exc_info=True) return {"is_exist": False} + def _filter_llm_results(self, original_data: Dict[str, str], user_name_map: Dict[str, str]) -> Dict[str, str]: """过滤 LLM 返回的绰号映射结果。""" filtered_data = {} - bot_qq_str = str(self.config.BOT_QQ) if hasattr(self.config, "BOT_QQ") else None + bot_qq_str = str(self.config.BOT_QQ) if hasattr(self.config, 'BOT_QQ') else None for user_id, nickname in original_data.items(): - # 过滤条件 1: user_id 必须是字符串 if not isinstance(user_id, str): logger.warning(f"过滤掉非字符串 user_id: {user_id}") continue - # 过滤条件 2: 排除机器人自身 if bot_qq_str and user_id == bot_qq_str: logger.debug(f"过滤掉机器人自身的映射: ID {user_id}") continue - # 过滤条件 3: 排除 nickname 为空或仅包含空白的情况 if not nickname or nickname.isspace(): logger.debug(f"过滤掉用户 {user_id} 的空绰号。") continue - - # 过滤条件 4 (可选，根据 Prompt 效果决定是否保留): 排除 nickname 与已知名称相同的情况 # person_name = user_name_map.get(user_id) # if person_name and person_name == nickname: # logger.debug(f"过滤掉用户 {user_id} 的映射: 绰号 '{nickname}' 与其名称 '{person_name}' 相同。") # continue - - # 如果通过所有过滤条件，则保留 - filtered_data[user_id] = nickname.strip() # 保留时去除首尾空白 + filtered_data[user_id] = nickname.strip() return filtered_data + # 线程相关 + # 修改：使用 run_async_loop 辅助函数 def _run_processor_in_thread(self): - """后台线程的入口函数，负责创建和运行 asyncio 事件循环。""" - loop = None - thread_id = threading.get_ident() + """后台线程入口函数，使用辅助函数管理 asyncio 事件循环。""" + thread_id = threading.get_ident() # 获取线程ID用于日志 logger.info(f"绰号处理器线程启动 (线程 ID: {thread_id})...") - try: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - logger.info(f"(线程 ID: {thread_id}) Asyncio 事件循环已创建并设置。") - # 运行主处理循环直到停止事件被设置 - loop.run_until_complete(self._processing_loop()) - except Exception as e: - logger.error(f"(线程 ID: {thread_id}) 运行绰号处理器线程时出错: {e}", exc_info=True) - finally: - # 确保循环被正确关闭 - if loop: - try: - if loop.is_running(): - logger.info(f"(线程 ID: {thread_id}) 正在停止 asyncio 循环...") - all_tasks = asyncio.all_tasks(loop) - if all_tasks: - logger.info(f"(线程 ID: {thread_id}) 正在取消 {len(all_tasks)} 个运行中的任务...") - for task in all_tasks: - task.cancel() - # 等待任务取消完成 - loop.run_until_complete(asyncio.gather(*all_tasks, return_exceptions=True)) - logger.info(f"(线程 ID: {thread_id}) 所有任务已取消。") - loop.stop() - logger.info(f"(线程 ID: {thread_id}) 循环已停止。") - if not loop.is_closed(): - loop.close() - logger.info(f"(线程 ID: {thread_id}) Asyncio 循环已关闭。") - except Exception as loop_close_err: - logger.error(f"(线程 ID: {thread_id}) 关闭循环时出错: {loop_close_err}", exc_info=True) - logger.info(f"绰号处理器线程结束 (线程 ID: {thread_id}).") + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) # 为当前线程设置事件循环 + logger.info(f"(线程 ID: {thread_id}) Asyncio 事件循环已创建并设置。") + # 调用辅助函数来运行主处理协程并管理循环生命周期 + run_async_loop(loop, self._processing_loop()) + + logger.info(f"绰号处理器线程结束 (线程 ID: {thread_id}).") + # 结束修改 + + + # 修改：使用 asyncio.Queue 和 wait_for async def _processing_loop(self): - """后台线程中运行的异步处理循环。""" - thread_id = threading.get_ident() - logger.info(f"绰号处理循环已启动 (线程 ID: {thread_id})。") + """后台线程中运行的异步处理循环 (使用 asyncio.Queue)。""" + # 移除线程ID日志，因为它在异步上下文中不一定准确 + logger.info("绰号异步处理循环已启动。") - while not self._stop_event.is_set(): + while not self._stop_event.is_set(): # 仍然检查同步的停止事件 try: - # 从队列中获取项目，设置超时以允许检查停止事件 - item = self.nickname_queue.get(block=True, timeout=self.sleep_interval) + # 使用 asyncio.wait_for 从异步队列获取项目，并设置超时 + item = await asyncio.wait_for( + self.nickname_queue.get(), + timeout=self.sleep_interval + ) - # 处理获取到的项目 + # 处理获取到的项目 (调用异步方法) await self._analyze_and_update_nicknames(item) - self.nickname_queue.task_done() # 标记任务完成 + self.nickname_queue.task_done() # 标记任务完成 - except queue.Empty: - # 超时，队列为空，继续循环检查停止事件 + except asyncio.TimeoutError: + # 等待超时，相当于之前 queue.Empty，继续循环检查停止事件 continue except asyncio.CancelledError: - logger.info(f"绰号处理循环被取消 (线程 ID: {thread_id})。") - break # 任务被取消，退出循环 + # 协程被取消 (通常在 stop_processor 中发生) + logger.info("绰号处理循环被取消。") + break # 退出循环 except Exception as e: # 捕获处理单个项目时可能发生的其他异常 - logger.error(f"(线程 ID: {thread_id}) 绰号处理循环出错: {e}\n{traceback.format_exc()}") - # 可以在这里添加错误处理逻辑，例如将失败的任务放回队列或记录到错误日志 - # 短暂休眠避免快速连续失败 + logger.error(f"绰号处理循环出错: {e}", exc_info=True) + # 短暂异步休眠避免快速连续失败 await asyncio.sleep(5) - logger.info(f"绰号处理循环已结束 (线程 ID: {thread_id})。") + logger.info("绰号异步处理循环已结束。") + # 可以在这里添加清理逻辑，比如确保队列为空或处理剩余项目 + # 例如：await self.nickname_queue.join() # 等待所有任务完成 (如果需要) + # 结束修改 # 在模块级别创建单例实例 -# 这使得其他模块可以通过 `from .nickname_manager import nickname_manager` 来导入和使用 nickname_manager = NicknameManager()