fix: 分析器和消息处理

This commit is contained in:
SXP-Simon
2026-02-09 14:15:43 +08:00
parent e1ad6c9fc9
commit f8412559ef
11 changed files with 314 additions and 161 deletions
@@ -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
@@ -0,0 +1,109 @@
"""
消息清理服务 - 领域层
负责过滤掉机器人消息、指令、技术性内容(如原始表情代码)及敏感内容。
"""
import re
from dataclasses import replace
from ..value_objects.unified_message import (
MessageContent,
MessageContentType,
UnifiedMessage,
)
class MessageCleanerService:
"""消息清理服务"""
# Discord 自定义表情正则 <:name:id> 或 <a:name:id>
DISCORD_CUSTOM_EMOJI_PATTERN = re.compile(r"<a?:.+?:\d+>")
# 指令匹配正则:匹配以 / 开头,或者以 @某人 / 开头的消息
# 比如: "/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
+5 -1
View File
@@ -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 ""}}
],
@@ -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
@@ -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(
@@ -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(
+54 -32
View File
@@ -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(
@@ -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", ""))
)
@@ -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:
+42 -11
View File
@@ -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])