diff --git a/main.py b/main.py index b7efee4..c00d14e 100644 --- a/main.py +++ b/main.py @@ -109,8 +109,6 @@ class QQGroupDailyAnalysis(Star): logger.info( f" - 平台 {platform_id}: {type(bot_instance).__name__}" ) - # 预先创建编排器 - self._get_orchestrator(platform_id, bot_instance=bot_instance) # 启动调度器 self.auto_scheduler.schedule_jobs(self.context) @@ -142,10 +140,8 @@ class QQGroupDailyAnalysis(Star): # 重置实例属性 self.auto_scheduler = None self.bot_manager = None - self.message_analyzer = None self.report_generator = None self.config_manager = None - self.orchestrators = {} logger.info("QQ群日常分析插件资源清理完成") @@ -598,3 +594,25 @@ class QQGroupDailyAnalysis(Star): 💡 可用命令: enable, disable, status, reload, test 💡 支持的输出格式: image, text, pdf (图片和PDF包含活跃度可视化) 💡 其他命令: /设置格式, /安装PDF""") + + def _get_group_id_from_event(self, event: AstrMessageEvent) -> str | None: + """从消息事件中安全获取群组 ID""" + try: + group_id = event.get_group_id() + return group_id if group_id else None + except Exception: + return None + + def _get_platform_id_from_event(self, event: AstrMessageEvent) -> str: + """从消息事件中获取平台唯一 ID""" + try: + return event.get_platform_id() + except Exception: + # 后备方案：从元数据获取 + if ( + hasattr(event, "platform_meta") + and event.platform_meta + and hasattr(event.platform_meta, "id") + ): + return event.platform_meta.id + return "default" diff --git a/src/application/services/analysis_application_service.py b/src/application/services/analysis_application_service.py index f512faf..69ca8bf 100644 --- a/src/application/services/analysis_application_service.py +++ b/src/application/services/analysis_application_service.py @@ -63,25 +63,34 @@ class AnalysisApplicationService: days = self.config_manager.get_analysis_days() max_count = self.config_manager.get_max_messages() - unified_messages = await adapter.fetch_messages( + raw_messages = await adapter.fetch_messages( group_id=group_id, days=days, max_count=max_count ) - if not unified_messages: + if not raw_messages: logger.warning(f"群 {group_id} 在最近 {days} 天内无消息或无法获取") return {"success": False, "reason": "no_messages"} - # 检查最小消息阈值 - if ( - len(unified_messages) < self.config_manager.get_min_messages_threshold() - and not manual - ): + # 3. 清理消息 (Filter commands, bot messages, noise) + from ...domain.services.message_cleaner_service import MessageCleanerService + + cleaner = MessageCleanerService() + bot_self_ids = self.config_manager.get_bot_self_ids() + + # 对于自动任务，强制过滤指令；对于手动任务，也建议过滤以保持报告纯净 + unified_messages = cleaner.clean_messages( + raw_messages, bot_self_ids=bot_self_ids, filter_commands=True + ) + + # 4. 检查最小消息阈值 (在清理后进行) + threshold = self.config_manager.get_min_messages_threshold() + if len(unified_messages) < threshold and not manual: logger.info( - f"群 {group_id} 消息数 ({len(unified_messages)}) 未达到自动分析阈值" + f"群 {group_id} 有效消息数 ({len(unified_messages)}) 未达到自动分析阈值 ({threshold})" ) return {"success": False, "reason": "below_threshold"} - # 3. 基础统计 (Domain Service) + # 5. 基础统计 (Domain Service) statistics = await asyncio.to_thread( self.statistics_service.calculate_group_statistics, unified_messages ) @@ -120,7 +129,7 @@ class AnalysisApplicationService: f"{platform_id}:GroupMessage:{group_id}" if platform_id else group_id ) - if topic_enabled and user_title_enabled and golden_quote_enabled: + if topic_enabled or user_title_enabled or golden_quote_enabled: ( topics, user_titles, @@ -131,10 +140,10 @@ class AnalysisApplicationService: user_activity, umo=unified_msg_origin, top_users=top_users, + topic_enabled=topic_enabled, + user_title_enabled=user_title_enabled, + golden_quote_enabled=golden_quote_enabled, ) - else: - # 按需串行执行 (略，实际实现可补全或合并) - pass # 回填结果 statistics.golden_quotes = golden_quotes diff --git a/src/domain/services/message_cleaner_service.py b/src/domain/services/message_cleaner_service.py new file mode 100644 index 0000000..0d4f59f --- /dev/null +++ b/src/domain/services/message_cleaner_service.py @@ -0,0 +1,109 @@ +""" +消息清理服务 - 领域层 +负责过滤掉机器人消息、指令、技术性内容（如原始表情代码）及敏感内容。 +""" + +import re +from dataclasses import replace + +from ..value_objects.unified_message import ( + MessageContent, + MessageContentType, + UnifiedMessage, +) + + +class MessageCleanerService: + """消息清理服务""" + + # Discord 自定义表情正则 <:name:id> 或 + DISCORD_CUSTOM_EMOJI_PATTERN = re.compile(r"") + + # 指令匹配正则：匹配以 / 开头，或者以 @某人 / 开头的消息 + # 比如: "/group_analysis", "@bot /help", " /test" + COMMAND_PATTERN = re.compile(r"^\s*(?:<@\d+>\s+)?/") + + def clean_messages( + self, + messages: list[UnifiedMessage], + bot_self_ids: list[str] = None, + filter_commands: bool = True, + ) -> list[UnifiedMessage]: + """ + 清理并过滤消息列表。 + + Args: + messages: 原始统一格式消息列表 + bot_self_ids: 机器人自身的 ID 列表 + filter_commands: 是否过滤指令消息 + + Returns: + 清理后的消息列表 + """ + bot_ids = set(bot_self_ids or []) + cleaned_list = [] + + for msg in messages: + # 1. 过滤机器人发送的消息 + if msg.sender_id in bot_ids: + continue + + # 2. 预检指令消息（首个内容块通常是文本） + is_command = False + first_text = msg.text_content + if ( + filter_commands + and first_text + and self.COMMAND_PATTERN.match(first_text) + ): + is_command = True + + if is_command: + continue + + # 3. 清理消息内容中的技术性噪音 + cleaned_contents = [] + has_meaningful_content = False + + for content in msg.contents: + if content.type == MessageContentType.TEXT: + text = content.text or "" + + # 移除 Discord 原始表情代码 + text = self.DISCORD_CUSTOM_EMOJI_PATTERN.sub("", text) + + # 移除 @mentions 文本 (e.g. <@123456>) + text = re.sub(r"<@\d+>", "", text) + + # 清理多余空格 + text = text.strip() + + if text: + cleaned_contents.append( + MessageContent(type=MessageContentType.TEXT, text=text) + ) + has_meaningful_content = True + else: + # 其他类型（图片、回复等）暂时保留，但由后续分析器决定是否使用 + cleaned_contents.append(content) + if content.type != MessageContentType.REPLY: + has_meaningful_content = True + + # 4. 如果清理后仍有内容，则保留消息 + if has_meaningful_content: + # 重新合成 text_content 用于 LLM 分析 + new_text_content = "".join( + [ + c.text + for c in cleaned_contents + if c.type == MessageContentType.TEXT + ] + ).strip() + + # 使用 replace 创建新实例（Frozen dataclass 必须如此） + new_msg = replace( + msg, contents=tuple(cleaned_contents), text_content=new_text_content + ) + cleaned_list.append(new_msg) + + return cleaned_list diff --git a/src/domain/services/statistics_service.py b/src/domain/services/statistics_service.py index 91c9c83..073bb7f 100644 --- a/src/domain/services/statistics_service.py +++ b/src/domain/services/statistics_service.py @@ -97,7 +97,11 @@ class StatisticsService: legacy_list.append( { "time": msg.timestamp, - "sender": {"user_id": msg.sender_id}, + "sender": { + "user_id": msg.sender_id, + "nickname": msg.sender_name, + "card": msg.sender_card or "", + }, "message": [ {"type": "text", "data": {"text": msg.text_content or ""}} ], diff --git a/src/infrastructure/analysis/analyzers/golden_quote_analyzer.py b/src/infrastructure/analysis/analyzers/golden_quote_analyzer.py index 749f19c..b36cf97 100644 --- a/src/infrastructure/analysis/analyzers/golden_quote_analyzer.py +++ b/src/infrastructure/analysis/analyzers/golden_quote_analyzer.py @@ -170,40 +170,34 @@ class GoldenQuoteAnalyzer(BaseAnalyzer): def extract_interesting_messages(self, messages: list[dict]) -> list[dict]: """ - 提取圣经的文本消息 + 根据清理后的消息提取可能有意义的消息片段用于金句分析。 Args: - messages: 群聊消息列表 + messages: 已由 MessageCleaner 处理过的 legacy 消息列表 Returns: - 圣经的文本消息列表 + 提取的文本消息列表 """ - try: - interesting_messages = [] + interesting_messages = [] - for msg in messages: - sender = msg.get("sender", {}) - nickname = InfoUtils.get_user_nickname(self.config_manager, sender) - msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M") + for msg in messages: + # 获取发送者显示名 + sender = msg.get("sender", {}) + nickname = InfoUtils.get_user_nickname(self.config_manager, sender) + msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M") - for content in msg.get("message", []): - if content.get("type") == "text": - text = content.get("data", {}).get("text", "").strip() - # 过滤长度适中、可能圣经的消息 - if 5 <= len(text) <= 100 and not text.startswith( - ("http", "www", "/") - ): - interesting_messages.append( - { - "sender": nickname, - "time": msg_time, - "content": text, - "user_id": str(sender.get("user_id", "")), - } - ) + for content in msg.get("message", []): + if content.get("type") == "text": + text = content.get("data", {}).get("text", "").strip() + # 过滤掉过短或过长的噪音（已经在 cleaner 处理过一遍基本垃圾） + if 2 <= len(text) <= 500: + interesting_messages.append( + { + "sender": nickname, + "time": msg_time, + "content": text, + "user_id": str(sender.get("user_id", "")), + } + ) - return interesting_messages - - except Exception as e: - logger.error(f"提取圣经消息失败: {e}") - return [] + return interesting_messages diff --git a/src/infrastructure/analysis/analyzers/topic_analyzer.py b/src/infrastructure/analysis/analyzers/topic_analyzer.py index badf206..23be2ae 100644 --- a/src/infrastructure/analysis/analyzers/topic_analyzer.py +++ b/src/infrastructure/analysis/analyzers/topic_analyzer.py @@ -257,79 +257,38 @@ class TopicAnalyzer(BaseAnalyzer): def extract_text_messages(self, messages: list[dict]) -> list[dict]: """ - 从群聊消息中提取文本消息 + 从已清理的消息中提取文本消息用于话题分析。 Args: - messages: 群聊消息列表 + messages: 已由 MessageCleaner 处理过的 legacy 消息列表 Returns: 提取的文本消息列表 """ - logger.debug( - f"extract_text_messages 开始处理，输入消息数量: {len(messages) if messages else 0}" - ) - logger.debug(f"extract_text_messages 输入消息类型: {type(messages)}") - - if not messages: - logger.warning("extract_text_messages 收到空消息列表") - return [] - text_messages = [] - for i, msg in enumerate(messages): - logger.debug(f"处理第 {i + 1} 条消息，类型: {type(msg)}") - # 确保msg是字典类型，避免'str' object has no attribute 'get'错误 - if not isinstance(msg, dict): - logger.warning(f"跳过非字典类型的消息: {type(msg)} - {msg}") - continue + for msg in messages: + # 获取发送者显示名 + sender = msg.get("sender", {}) + nickname = InfoUtils.get_user_nickname(self.config_manager, sender) + msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M") - try: - sender = msg.get("sender", {}) - # 确保sender是字典类型，避免'str' object has no attribute 'get'错误 - if not isinstance(sender, dict): - logger.warning( - f"extract_text_messages 跳过sender非字典类型的消息: {type(sender)} - {sender}" - ) - continue + for content in msg.get("message", []): + if content.get("type") == "text": + text = content.get("data", {}).get("text", "").strip() + # 已经在 MessageCleaner 中处理过基本的垃圾内容 + if text: + # 简单的额外清理 + cleaned_text = text.replace("\n", " ").replace("\r", " ") + cleaned_text = re.sub(r"[\x00-\x1f\x7f-\x9f]", "", cleaned_text) - # 获取发送者ID并过滤机器人消息 - user_id = str(sender.get("user_id", "")) - bot_self_ids = self.config_manager.get_bot_self_ids() - - # 跳过机器人自己的消息 - if bot_self_ids and user_id in [str(uid) for uid in bot_self_ids]: - logger.debug(f"extract_text_messages 过滤掉机器人QQ号: {user_id}") - continue - - nickname = InfoUtils.get_user_nickname(self.config_manager, sender) - msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M") - - for content in msg.get("message", []): - if content.get("type") == "text": - text = content.get("data", {}).get("text", "").strip() - if text and len(text) > 2 and not text.startswith("/"): - # 清理消息内容 - text = text.replace('""', '"').replace('""', '"') - text = text.replace(""", "'").replace(""", "'") - text = text.replace("\n", " ").replace("\r", " ") - text = text.replace("\t", " ") - text = re.sub(r"[\x00-\x1f\x7f-\x9f]", "", text) - text_messages.append( - { - "sender": nickname, - "time": msg_time, - "content": text.strip(), - } - ) - except Exception as e: - logger.error(f"处理第 {i + 1} 条消息时出错: {e}", exc_info=True) - continue - - logger.debug( - f"extract_text_messages 完成，提取到 {len(text_messages)} 条文本消息" - ) - if text_messages: - logger.debug(f"extract_text_messages 第一条文本消息: {text_messages[0]}") + text_messages.append( + { + "sender": nickname, + "time": msg_time, + "content": cleaned_text.strip(), + } + ) return text_messages async def analyze_topics( diff --git a/src/infrastructure/analysis/analyzers/user_title_analyzer.py b/src/infrastructure/analysis/analyzers/user_title_analyzer.py index d6ae0d1..c671821 100644 --- a/src/infrastructure/analysis/analyzers/user_title_analyzer.py +++ b/src/infrastructure/analysis/analyzers/user_title_analyzer.py @@ -178,16 +178,15 @@ class UserTitleAnalyzer(BaseAnalyzer): for user_id, stats in user_analysis.items(): user_id_str = str(user_id) - # 过滤机器人自己的消息 + # 过滤机器人由 MessageCleaner 已处理，此处仅作为二级防御 if bot_self_ids and user_id_str in [str(uid) for uid in bot_self_ids]: - logger.debug(f"过滤掉机器人ID: {user_id}") continue - # 只处理活跃用户 + # 只处理活跃用户 (top_users 或 消息数>=5) if user_id_str not in target_user_ids: continue - # 分析用户特征 + # 分析用户特征 (此处已基于已清理的 stats) night_messages = sum(stats["hours"][h] for h in range(6)) avg_chars = ( stats["char_count"] / stats["message_count"] @@ -198,7 +197,7 @@ class UserTitleAnalyzer(BaseAnalyzer): user_summaries.append( { "name": stats["nickname"], - "user_id": user_id_str, # 使用 user_id + "user_id": user_id_str, "message_count": stats["message_count"], "avg_chars": round(avg_chars, 1), "emoji_ratio": round( diff --git a/src/infrastructure/analysis/llm_analyzer.py b/src/infrastructure/analysis/llm_analyzer.py index e9a33d9..3bedb8c 100644 --- a/src/infrastructure/analysis/llm_analyzer.py +++ b/src/infrastructure/analysis/llm_analyzer.py @@ -156,15 +156,21 @@ class LLMAnalyzer: user_analysis: dict, umo: str = None, top_users: list[dict] = None, + topic_enabled: bool = True, + user_title_enabled: bool = True, + golden_quote_enabled: bool = True, ) -> tuple[list[SummaryTopic], list[UserTitle], list[GoldenQuote], TokenUsage]: """ - 并发执行所有分析任务（话题、用户称号、金句） + 并发执行所有分析任务（话题、用户称号、金句），支持按需启用。 Args: messages: 群聊消息列表 user_analysis: 用户分析统计 umo: 模型唯一标识符 top_users: 活跃用户列表(可选) + topic_enabled: 是否启用话题分析 + user_title_enabled: 是否启用用户称号分析 + golden_quote_enabled: 是否启用金句分析 Returns: (话题列表, 用户称号列表, 金句列表, 总Token使用统计) @@ -179,10 +185,13 @@ class LLMAnalyzer: else: session_id = timestamp - logger.info(f"开始并发执行所有分析任务，会话ID: {session_id}") + logger.info( + f"开始并发执行分析任务 (话题:{topic_enabled}, 称号:{user_title_enabled}, 金句:{golden_quote_enabled})，会话ID: {session_id}" + ) # 保存原始消息数据 (Debug Mode) if self.config_manager.get_debug_mode(): + # ... (保持原有的调试保存代码) try: import json from pathlib import Path @@ -202,44 +211,57 @@ class LLMAnalyzer: msg_file_path = debug_dir / f"{session_id}_messages.json" with open(msg_file_path, "w", encoding="utf-8") as f: json.dump(messages, f, ensure_ascii=False, indent=2) - logger.info(f"已保存原始消息数据到: {msg_file_path}") - except Exception as e: - logger.error(f"保存原始消息数据失败: {e}", exc_info=True) + except Exception: + pass - # 并发执行三个分析任务 - results = await asyncio.gather( - self.topic_analyzer.analyze_topics(messages, umo, session_id), - self.user_title_analyzer.analyze_user_titles( - messages, user_analysis, umo, top_users, session_id - ), - self.golden_quote_analyzer.analyze_golden_quotes( - messages, umo, session_id - ), - return_exceptions=True, - ) + # 构建并发任务列表 + tasks = [] + task_names = [] + + if topic_enabled: + tasks.append( + self.topic_analyzer.analyze_topics(messages, umo, session_id) + ) + task_names.append("topic") + + if user_title_enabled: + tasks.append( + self.user_title_analyzer.analyze_user_titles( + messages, user_analysis, umo, top_users, session_id + ) + ) + task_names.append("user_title") + + if golden_quote_enabled: + tasks.append( + self.golden_quote_analyzer.analyze_golden_quotes( + messages, umo, session_id + ) + ) + task_names.append("golden_quote") + + if not tasks: + return [], [], [], TokenUsage() + + results = await asyncio.gather(*tasks, return_exceptions=True) # 处理结果 topics, topic_usage = [], TokenUsage() user_titles, title_usage = [], TokenUsage() golden_quotes, quote_usage = [], TokenUsage() - # 话题分析结果 - if isinstance(results[0], Exception): - logger.error(f"话题分析失败: {results[0]}") - else: - topics, topic_usage = results[0] + for i, result in enumerate(results): + name = task_names[i] + if isinstance(result, Exception): + logger.error(f"分析任务 {name} 失败: {result}") + continue - # 用户称号分析结果 - if isinstance(results[1], Exception): - logger.error(f"用户称号分析失败: {results[1]}") - else: - user_titles, title_usage = results[1] - - # 金句分析结果 - if isinstance(results[2], Exception): - logger.error(f"金句分析失败: {results[2]}") - else: - golden_quotes, quote_usage = results[2] + if name == "topic": + topics, topic_usage = result + elif name == "user_title": + user_titles, title_usage = result + elif name == "golden_quote": + golden_quotes, quote_usage = result # 合并Token使用统计 total_usage = TokenUsage( diff --git a/src/infrastructure/analysis/utils/info_utils.py b/src/infrastructure/analysis/utils/info_utils.py index bf07545..f62ae81 100644 --- a/src/infrastructure/analysis/utils/info_utils.py +++ b/src/infrastructure/analysis/utils/info_utils.py @@ -8,6 +8,14 @@ class InfoUtils: """ enable_user_card = config_manager.get_enable_user_card() if enable_user_card: - return sender.get("card", "") or sender.get("nickname", "") + return ( + sender.get("card", "") + or sender.get("nickname", "") + or str(sender.get("user_id", "")) + ) else: - return sender.get("nickname", "") or sender.get("card", "") + return ( + sender.get("nickname", "") + or sender.get("card", "") + or str(sender.get("user_id", "")) + ) diff --git a/src/infrastructure/analysis/utils/json_utils.py b/src/infrastructure/analysis/utils/json_utils.py index 3bb7272..f9c2a1f 100644 --- a/src/infrastructure/analysis/utils/json_utils.py +++ b/src/infrastructure/analysis/utils/json_utils.py @@ -6,7 +6,7 @@ JSON处理工具模块 import json import re -from ...utils.logger import logger +from ....utils.logger import logger def fix_json(text: str) -> str: diff --git a/src/infrastructure/platform/bot_manager.py b/src/infrastructure/platform/bot_manager.py index 42c34d3..be8f80d 100644 --- a/src/infrastructure/platform/bot_manager.py +++ b/src/infrastructure/platform/bot_manager.py @@ -105,9 +105,6 @@ class BotManager: def _refresh_from_stored_platforms(self): """尝试从已存储的平台对象中刷新 bot 实例 (Lazy Load)""" for platform_id, platform in self._platforms.items(): - if platform_id in self._bot_instances: - continue - bot_client = None # 优先尝试 get_client() if hasattr(platform, "get_client"): @@ -121,6 +118,13 @@ class BotManager: bot_client = platform.client if bot_client: + # 检查是否已存在且是否发生变化（防止重复创建适配器） + old_client = self._bot_instances.get(platform_id) + + # 如果 client 对象没变且已经有适配器，跳过 + if bot_client is old_client and platform_id in self._adapters: + continue + platform_name = None if hasattr(platform, "metadata"): # 优先使用 type @@ -129,6 +133,16 @@ class BotManager: elif hasattr(platform.metadata, "name"): platform_name = platform.metadata.name + # 兼容不同版本的元数据获取 + if not platform_name: + meta = getattr(platform, "meta", None) + if callable(meta): + try: + metadata = meta() + platform_name = getattr(metadata, "name", None) + except Exception: + pass + # 后备检测：如果不支持名称 if not platform_name or not PlatformAdapterFactory.is_supported( str(platform_name) @@ -138,7 +152,7 @@ class BotManager: platform_name = detected self.set_bot_instance(bot_client, platform_id, platform_name) - logger.info(f"懒加载发现平台 {platform_id} 的 bot 实例") + logger.info(f"已刷新/发现平台 {platform_id} 的 bot 实例 (变动或懒加载)") def get_all_bot_instances(self) -> dict: """获取所有已加载的bot实例 {platform_id: bot_instance}""" @@ -213,17 +227,29 @@ class BotManager: 这是 DDD 架构操作的主要方法。 """ if platform_id: + # 无论是否存在适配器，都尝试检测一次 client 是否有变（如重启后 session 变化） + if platform_id in self._platforms: + self._refresh_from_stored_platforms() + return self._adapters.get(platform_id) if self._adapters: if len(self._adapters) == 1: return list(self._adapters.values())[0] - logger.error( + logger.warning( f"存在多个适配器 {list(self._adapters.keys())}，但未指定 platform_id。" ) return None + # 如果没有任何适配器，尝试全局刷新一次 + self._refresh_from_stored_platforms() + if self._adapters: + if platform_id: + return self._adapters.get(platform_id) + if len(self._adapters) == 1: + return list(self._adapters.values())[0] + return None def get_all_adapters(self) -> dict: @@ -373,17 +399,22 @@ class BotManager: def update_from_event(self, event): """从事件更新bot实例（用于手动命令）""" - if hasattr(event, "bot") and event.bot: + # 兼容不同平台的 bot 实例属性名 (OneBot 使用 bot, Discord 使用 client) + bot_instance = getattr(event, "bot", None) or getattr(event, "client", None) + + if bot_instance: # 从事件中获取平台ID platform_id = None - if hasattr(event, "platform") and isinstance(event.platform, str): + if hasattr(event, "get_platform_id"): + platform_id = event.get_platform_id() + elif hasattr(event, "platform_meta") and hasattr(event.platform_meta, "id"): + platform_id = event.platform_meta.id + elif hasattr(event, "platform") and isinstance(event.platform, str): platform_id = event.platform - elif hasattr(event, "metadata") and hasattr(event.metadata, "id"): - platform_id = event.metadata.id - self.set_bot_instance(event.bot, platform_id) + self.set_bot_instance(bot_instance, platform_id) # 每次都尝试从bot实例提取ID - bot_self_id = self._extract_bot_self_id(event.bot) + bot_self_id = self._extract_bot_self_id(bot_instance) if bot_self_id: # 将单个ID转换为列表，保持统一处理 self.set_bot_self_ids([bot_self_id])