refactor(DDD): 整理架构

This commit is contained in:
SXP-Simon
2026-02-09 12:47:43 +08:00
parent 0b096351d5
commit 2005f0633f
67 changed files with 1149 additions and 2490 deletions
+71 -218
View File
@@ -7,7 +7,6 @@ QQ群日常分析插件
import asyncio
import os
from typing import Any
from astrbot.api import AstrBotConfig, logger
from astrbot.api.event import AstrMessageEvent, filter
@@ -15,14 +14,18 @@ from astrbot.api.event.filter import PermissionType
from astrbot.api.star import Context, Star
from astrbot.core.message.components import File
from .src.application.analysis_orchestrator import AnalysisConfig, AnalysisOrchestrator
from .src.core.bot_manager import BotManager
from .src.core.config import ConfigManager
from .src.core.history_manager import HistoryManager
from .src.reports.generators import ReportGenerator
from .src.scheduler.auto_scheduler import AutoScheduler
from .src.scheduler.retry import RetryManager
from .src.utils.helpers import MessageAnalyzer
from .src.application.services.analysis_application_service import (
AnalysisApplicationService,
)
from .src.domain.services.analysis_domain_service import AnalysisDomainService
from .src.domain.services.statistics_service import StatisticsService
from .src.infrastructure.analysis.llm_analyzer import LLMAnalyzer
from .src.infrastructure.config.config_manager import ConfigManager
from .src.infrastructure.persistence.history_manager import HistoryManager
from .src.infrastructure.platform.bot_manager import BotManager
from .src.infrastructure.reporting.generators import ReportGenerator
from .src.infrastructure.scheduler.auto_scheduler import AutoScheduler
from .src.infrastructure.scheduler.retry import RetryManager
from .src.utils.pdf_utils import PDFInstaller
@@ -33,110 +36,45 @@ class QQGroupDailyAnalysis(Star):
super().__init__(context)
self.config = config
# 初始化模块化组件(使用实例属性而非全局变量)
# 1. 基础设施层
self.config_manager = ConfigManager(config)
self.bot_manager = BotManager(self.config_manager)
self.bot_manager.set_context(context)
self.message_analyzer = MessageAnalyzer(
context, self.config_manager, self.bot_manager
)
self.report_generator = ReportGenerator(self.config_manager)
self.history_manager = HistoryManager(self)
self.report_generator = ReportGenerator(self.config_manager)
# 2. 领域层
self.statistics_service = StatisticsService()
self.analysis_domain_service = AnalysisDomainService()
# 3. 分析核心 (LLM Bridge)
self.llm_analyzer = LLMAnalyzer(context, self.config_manager)
# 4. 应用层
self.analysis_service = AnalysisApplicationService(
self.config_manager,
self.bot_manager,
self.history_manager,
self.report_generator,
self.llm_analyzer,
self.statistics_service,
self.analysis_domain_service,
)
# 调度与重试
self.retry_manager = RetryManager(
self.bot_manager, self.html_render, self.report_generator
)
self.auto_scheduler = AutoScheduler(
self.config_manager,
self.message_analyzer.message_handler,
self.message_analyzer,
self.report_generator,
self.analysis_service,
self.bot_manager,
self.retry_manager,
self.history_manager,
self.html_render, # 传入html_render函数
self.html_render,
)
# 注册分析编排器缓存
self.orchestrators = {} # {platform_id: AnalysisOrchestrator}
# 注册日志过滤器
from .src.utils.trace_context import TraceLogFilter
logger.addFilter(TraceLogFilter())
logger.info("QQ群日常分析插件已初始化(模块化版本)")
def _get_group_id_from_event(self, event: AstrMessageEvent) -> str | None:
"""从事件中提取群组ID(跨平台兼容)"""
# 使用正确的 AstrMessageEvent API
if hasattr(event, "get_group_id"):
group_id = event.get_group_id()
return str(group_id) if group_id else None
if hasattr(event, "message_obj") and hasattr(event.message_obj, "group_id"):
group_id = event.message_obj.group_id
return str(group_id) if group_id else None
return None
def _get_platform_id_from_event(self, event: AstrMessageEvent) -> str | None:
"""从事件中提取平台ID(跨平台兼容)"""
# 使用正确的 AstrMessageEvent API
if hasattr(event, "get_platform_id"):
return event.get_platform_id()
if hasattr(event, "platform_meta") and hasattr(event.platform_meta, "id"):
return event.platform_meta.id
return None
def _get_platform_name_from_event(self, event: AstrMessageEvent) -> str | None:
"""从事件中提取平台名称(如 discord, aiocqhttp 等)"""
# 使用正确的 AstrMessageEvent API
if hasattr(event, "get_platform_name"):
return event.get_platform_name()
if hasattr(event, "platform_meta") and hasattr(event.platform_meta, "name"):
return event.platform_meta.name
return None
def _get_orchestrator(
self,
platform_id: str,
platform_name: str | None = None,
bot_instance: Any = None,
) -> AnalysisOrchestrator | None:
"""获取或创建分析编排器"""
if platform_id in self.orchestrators:
return self.orchestrators[platform_id]
# 如果缓存中没有,尝试创建
if not bot_instance:
bot_instance = self.bot_manager.get_bot_instance(platform_id)
if not bot_instance:
return None
# 检测平台名称(优先使用传入的 platform_name
if not platform_name:
platform_name = self.bot_manager._detect_platform_name(bot_instance)
if not platform_name:
return None
# 创建编排器
analysis_config = AnalysisConfig(
days=self.config_manager.get_analysis_days(),
min_messages_threshold=self.config_manager.get_min_messages_threshold(),
max_messages=self.config_manager.get_max_messages(),
output_format=self.config_manager.get_output_format(),
)
orchestrator = AnalysisOrchestrator.create_for_platform(
platform_name,
bot_instance,
config={"bot_self_ids": self.config_manager.get_bot_self_ids()},
analysis_config=analysis_config,
)
if orchestrator:
self.orchestrators[platform_id] = orchestrator
return orchestrator
# orchestrators 缓存已移至 应用层逻辑 (分析服务) 或 暂时移除以简化。
# 如果需要高性能缓存,后续可由 AnalysisApplicationService 内部维护。
@filter.on_platform_loaded()
async def on_platform_loaded(self):
@@ -147,6 +85,7 @@ class QQGroupDailyAnalysis(Star):
config = self.context.get_config()
plugin_set = config.get("plugin_set")
# !!!仅开发阶段使用,正式发布后删除!!!
if isinstance(plugin_set, list) and not plugin_set:
logger.warning("检测到 plugin_set 为空,自动修正以启用插件")
config["plugin_set"].append(
@@ -165,8 +104,7 @@ class QQGroupDailyAnalysis(Star):
# 初始化所有bot实例
discovered = await self.bot_manager.initialize_from_config()
if discovered:
platform_count = len(discovered)
logger.info(f"Bot管理器初始化成功,发现 {platform_count} 个适配器")
logger.info("Bot管理器初始化成功")
for platform_id, bot_instance in discovered.items():
logger.info(
f" - 平台 {platform_id}: {type(bot_instance).__name__}"
@@ -223,109 +161,47 @@ class QQGroupDailyAnalysis(Star):
分析群聊日常活动(跨平台支持)
用法: /群分析 [天数]
"""
# 1. 获取 group_id, platform_id 和 platform_name
group_id = self._get_group_id_from_event(event)
platform_id = self._get_platform_id_from_event(event)
platform_name = self._get_platform_name_from_event(event)
if not group_id:
yield event.plain_result("❌ 请在群聊中使用此命令")
return
# 更新bot实例(用于手动命令)
if hasattr(event, "bot"):
self.bot_manager.update_from_event(event)
# 更新bot实例
self.bot_manager.update_from_event(event)
# 2. 检查群组权限
if not self.config_manager.is_group_allowed(group_id):
yield event.plain_result("❌ 此群未启用日常分析功能")
return
# 3. 设置分析天数
analysis_days = (
days if days and 1 <= days <= 7 else self.config_manager.get_analysis_days()
)
yield event.plain_result(f"🔍 开始分析群聊近{analysis_days}天的活动,请稍候...")
logger.info(
f"收到分析请求: group_id={group_id}, platform_id={platform_id}, platform_name={platform_name}, days={analysis_days}"
)
yield event.plain_result("🔍 正在启动跨平台分析引擎,正在拉取最近消息...")
try:
# 4. 获取编排器
# 首先尝试从 event 直接提取 bot 客户端
bot_from_event = None
if hasattr(event, "client"): # Discord 平台有 client 属性
bot_from_event = event.client
elif hasattr(event, "bot"): # 其他平台可能有 bot 属性
bot_from_event = event.bot
orchestrator = self._get_orchestrator(
platform_id, platform_name, bot_from_event
)
if not orchestrator:
# 尝试使用 bot_manager 获取 bot 实例再创建
bot_instance = self.bot_manager.get_bot_instance(platform_id)
if bot_instance:
orchestrator = self._get_orchestrator(
platform_id, platform_name, bot_instance
)
if not orchestrator:
yield event.plain_result(
f"❌ 未找到平台 {platform_name or platform_id} 的分析编排器,请检查配置或联系开发者"
)
return
# 5. 获取群聊消息 (使用编排器)
messages = await orchestrator.fetch_messages_as_raw(
group_id=group_id, days=analysis_days
# 调用 DDD 应用级服务
result = await self.analysis_service.execute_daily_analysis(
group_id=group_id, platform_id=platform_id, manual=True
)
if not messages:
yield event.plain_result(
"❌ 未找到足够的群聊记录,请确保群内有足够的消息历史"
)
return
# 检查消息数量是否足够分析
min_threshold = self.config_manager.get_min_messages_threshold()
if len(messages) < min_threshold:
yield event.plain_result(
f"❌ 消息数量不足({len(messages)}条),至少需要{min_threshold}条消息才能进行有效分析"
)
if not result.get("success"):
reason = result.get("reason")
if reason == "no_messages":
yield event.plain_result("❌ 未找到足够的群聊记录")
else:
yield event.plain_result("❌ 分析失败,原因未知")
return
yield event.plain_result(
f"📊 已获取{len(messages)}条消息,正在进行智能分析..."
f"📊 已获取{result['messages_count']}条消息,正在生成渲染报告..."
)
# 6. 进行分析
analysis_result = await self.message_analyzer.analyze_messages(
messages, group_id, event.unified_msg_origin
)
if not analysis_result or not analysis_result.get("statistics"):
yield event.plain_result("❌ 分析过程中出现错误,请稍后重试")
return
# 7. 保存到历史记录
await self.history_manager.save_analysis(group_id, analysis_result)
# 8. 生成并发送报告
analysis_result = result["analysis_result"]
adapter = result["adapter"]
output_format = self.config_manager.get_output_format()
# 定义头像获取回调
# 定义头像获取回调 (Infrastructure delegate)
async def avatar_getter(user_id: str) -> str | None:
if not orchestrator:
return None
try:
# orchestrator.get_member_avatars 接受列表返回字典
avatars = await orchestrator.get_member_avatars([user_id])
return avatars.get(user_id)
except Exception as e:
logger.warning(f"获取头像失败 {user_id}: {e}")
return None
return await adapter.get_user_avatar_url(user_id)
if output_format == "image":
(
@@ -339,69 +215,46 @@ class QQGroupDailyAnalysis(Star):
)
if image_url:
# 使用编排器发送图片
if await orchestrator.send_image(group_id, image_url):
logger.info(f"图片报告发送成功: {group_id}")
else:
if not await adapter.send_image(group_id, image_url):
yield event.image_result(image_url)
elif html_content:
# 生成失败但有HTML,加入重试队列
logger.warning("图片报告生成失败,加入重试队列")
yield event.plain_result(
"[AstrBot QQ群日常分析总结插件] ⚠️ 图片报告暂无法生成,已加入重试队列,稍后将自动重试发送。"
)
yield event.plain_result("⚠️ 图片生成暂不可用,已尝试加入队列。")
await self.retry_manager.add_task(
html_content, analysis_result, group_id, platform_id
)
else:
# 回退到文本报告
logger.warning("图片报告生成失败(无HTML),回退到文本报告")
text_report = self.report_generator.generate_text_report(
analysis_result
)
yield event.plain_result(
f"[AstrBot QQ群日常分析总结插件] ⚠️ 图片报告生成失败,以下是文本版本:\n\n{text_report}"
f"⚠️ 图片生成失败,回退文本:\n\n{text_report}"
)
elif output_format == "pdf":
if not self.config_manager.playwright_available:
yield event.plain_result(
"❌ PDF 功能不可用,请使用 /安装PDF 命令安装依赖"
)
return
pdf_path = await self.report_generator.generate_pdf_report(
analysis_result, group_id, avatar_getter=avatar_getter
)
if pdf_path:
# 使用编排器发送文件
if await orchestrator.send_file(group_id, pdf_path):
pass # 发送成功
else:
if not await adapter.send_file(group_id, pdf_path):
from pathlib import Path
pdf_file = File(name=Path(pdf_path).name, file=pdf_path)
result = event.make_result()
result.chain.append(pdf_file)
yield result
yield event.chain_result(
[File(name=Path(pdf_path).name, file=pdf_path)]
)
else:
logger.warning("PDF 报告生成失败,回退到文本报告")
text_report = self.report_generator.generate_text_report(
analysis_result
)
yield event.plain_result(
f"\n📝 以下是文本版本的分析报告:\n\n{text_report}"
)
yield event.plain_result("⚠️ PDF 生成失败")
else:
# 文本报告
text_report = self.report_generator.generate_text_report(
analysis_result
)
if not await orchestrator.send_text(group_id, text_report):
if not await adapter.send_text(group_id, text_report):
yield event.plain_result(text_report)
except Exception as e:
logger.error(f"群分析失败: {e}", exc_info=True)
yield event.plain_result(f"❌ 分析核心执行失败: {str(e)}")
except Exception as e:
logger.error(f"群分析失败: {e}", exc_info=True)
yield event.plain_result(
-149
View File
@@ -1,149 +0,0 @@
"""
统计分析模块
负责用户活跃度分析和其他统计功能
"""
import re
from collections import defaultdict
from datetime import datetime
from .utils import InfoUtils
class UserAnalyzer:
"""用户分析器"""
# Discord 自定义表情正则 <:name:id> 或 <a:name:id>
DISCORD_CUSTOM_EMOJI_PATTERN = r"<a?:.+?:\d+>"
# 简单的 Unicode Emoji 正则范围 (覆盖大多数常见 Emoji)
UNICODE_EMOJI_PATTERN = (
r"[\U0001F000-\U0001F9FF]|[\U00002600-\U000026FF]|[\U00002700-\U000027BF]"
)
def __init__(self, config_manager):
self.config_manager = config_manager
def analyze_users(self, messages: list[dict]) -> dict[str, dict]:
"""分析用户活跃度"""
# 获取机器人 ID 列表用于过滤
bot_self_ids = self.config_manager.get_bot_self_ids()
user_stats = defaultdict(
lambda: {
"message_count": 0,
"char_count": 0,
"emoji_count": 0,
"nickname": "",
"hours": defaultdict(int),
"reply_count": 0,
}
)
for msg in messages:
sender = msg.get("sender", {})
user_id = str(sender.get("user_id", ""))
# 跳过机器人自己的消息,避免进入统计
if bot_self_ids and user_id in [str(sid) for sid in bot_self_ids]:
continue
nickname = InfoUtils.get_user_nickname(self.config_manager, sender)
user_stats[user_id]["message_count"] += 1
user_stats[user_id]["nickname"] = nickname
# 统计时间分布
msg_time = datetime.fromtimestamp(msg.get("time", 0))
user_stats[user_id]["hours"][msg_time.hour] += 1
# 处理消息内容
for content in msg.get("message", []):
if content.get("type") == "text":
text = content.get("data", {}).get("text", "")
user_stats[user_id]["char_count"] += len(text)
# 统计文本中的 Discord 自定义表情
discord_emojis = re.findall(self.DISCORD_CUSTOM_EMOJI_PATTERN, text)
user_stats[user_id]["emoji_count"] += len(discord_emojis)
# 统计文本中的 Unicode Emoji
unicode_emojis = re.findall(self.UNICODE_EMOJI_PATTERN, text)
user_stats[user_id]["emoji_count"] += len(unicode_emojis)
elif content.get("type") == "face":
# 基础表情
user_stats[user_id]["emoji_count"] += 1
elif content.get("type") == "mface":
# 动画表情/魔法表情
user_stats[user_id]["emoji_count"] += 1
elif content.get("type") == "bface":
# 超级表情
user_stats[user_id]["emoji_count"] += 1
elif content.get("type") == "sface":
# 小表情
user_stats[user_id]["emoji_count"] += 1
elif content.get("type") == "image":
# 检查是否是动画表情(通过summary字段判断)
data = content.get("data", {})
summary = data.get("summary", "")
if "动画表情" in summary or "表情" in summary:
# 动画表情(以image形式发送)
user_stats[user_id]["emoji_count"] += 1
elif content.get("type") == "reply":
user_stats[user_id]["reply_count"] += 1
return dict(user_stats)
def get_top_users(
self, user_analysis: dict[str, dict], limit: int = 10
) -> list[dict]:
"""获取最活跃的用户"""
# 获取机器人 ID 列表用于过滤
bot_self_ids = self.config_manager.get_bot_self_ids()
users = []
for user_id, stats in user_analysis.items():
# 过滤机器人自己
if bot_self_ids and str(user_id) in [str(sid) for sid in bot_self_ids]:
continue
users.append(
{
"user_id": user_id,
"nickname": stats["nickname"],
"message_count": stats["message_count"],
"char_count": stats["char_count"],
"emoji_count": stats["emoji_count"],
"reply_count": stats["reply_count"],
}
)
# 按消息数量排序
users.sort(key=lambda x: x["message_count"], reverse=True)
return users[:limit]
def get_user_activity_pattern(
self, user_analysis: dict[str, dict], user_id: str
) -> dict:
"""获取用户活动模式"""
if user_id not in user_analysis:
return {}
stats = user_analysis[user_id]
hours = stats["hours"]
# 找出最活跃的时间段
most_active_hour = max(hours.items(), key=lambda x: x[1])[0] if hours else 0
# 计算夜间活跃度
night_messages = sum(hours[h] for h in range(0, 6))
night_ratio = (
night_messages / stats["message_count"] if stats["message_count"] > 0 else 0
)
return {
"most_active_hour": most_active_hour,
"night_ratio": night_ratio,
"hourly_distribution": dict(hours),
}
-3
View File
@@ -1,11 +1,8 @@
# 应用层 - 编排和用例
from .analysis_orchestrator import AnalysisOrchestrator
from .message_converter import MessageConverter
from .reporting_service import ReportingService
from .scheduling_service import SchedulingService
__all__ = [
"AnalysisOrchestrator",
"MessageConverter",
"SchedulingService",
"ReportingService",
-251
View File
@@ -1,251 +0,0 @@
"""
分析编排器 - 应用层协调器
此编排器连接新的 DDD 架构与现有的分析逻辑,提供渐进式迁移路径。
架构决策:
- 编排器使用 PlatformAdapter 获取消息(新的 DDD 方式)
- 但将 LLM 分析委托给现有分析器(保留已工作的代码)
- MessageConverter 提供双向转换以保持兼容性
"""
from dataclasses import dataclass
from typing import Any, Optional
from ..domain.value_objects.platform_capabilities import PlatformCapabilities
from ..domain.value_objects.unified_message import UnifiedMessage
from ..infrastructure.platform import PlatformAdapter, PlatformAdapterFactory
from ..utils.logger import logger
from .message_converter import MessageConverter
@dataclass
class AnalysisConfig:
"""分析操作配置"""
days: int = 1
max_messages: int = 1000
min_messages_threshold: int = 10
output_format: str = "image"
class AnalysisOrchestrator:
"""
分析编排器 - 协调分析工作流。
职责:
1. 使用 PlatformAdapter 获取消息(DDD 方式)
2. 转换消息以兼容现有分析器
3. 协调分析流程
4. 提供平台能力检查
此类作为以下组件之间的桥梁:
- 新的 DDD 基础设施(PlatformAdapter, UnifiedMessage
- 现有分析逻辑(MessageHandler, LLMAnalyzer 等)
"""
def __init__(
self,
adapter: PlatformAdapter,
config: AnalysisConfig = None,
):
"""
初始化编排器。
参数:
adapter: 用于消息操作的平台适配器
config: 分析配置
"""
self.adapter = adapter
self.config = config or AnalysisConfig()
@classmethod
def create_for_platform(
cls,
platform_name: str,
bot_instance: Any,
config: dict = None,
analysis_config: AnalysisConfig = None,
) -> Optional["AnalysisOrchestrator"]:
"""
工厂方法 - 为特定平台创建编排器。
参数:
platform_name: 平台名称(如 "aiocqhttp", "telegram"
bot_instance: 来自 AstrBot 的 bot 实例
config: 平台特定配置
analysis_config: 分析配置
返回:
AnalysisOrchestrator 或 None(如果平台不支持)
"""
adapter = PlatformAdapterFactory.create(platform_name, bot_instance, config)
if adapter is None:
logger.warning(f"平台 '{platform_name}' 不支持分析功能")
return None
return cls(adapter, analysis_config)
def get_capabilities(self) -> PlatformCapabilities:
"""获取平台能力。"""
return self.adapter.get_capabilities()
def can_analyze(self) -> bool:
"""检查平台是否支持分析。"""
return self.adapter.get_capabilities().can_analyze()
def can_send_report(self, format: str = "image") -> bool:
"""检查平台是否能发送指定格式的报告。"""
return self.adapter.get_capabilities().can_send_report(format)
async def fetch_messages(
self,
group_id: str,
days: int = None,
max_count: int = None,
) -> list[UnifiedMessage]:
"""
使用平台适配器获取消息。
参数:
group_id: 要获取消息的群组 ID
days: 天数(默认使用配置值)
max_count: 最大消息数量(默认使用配置值)
返回:
UnifiedMessage 列表
"""
days = days or self.config.days
max_count = max_count or self.config.max_messages
# 应用平台能力限制
caps = self.adapter.get_capabilities()
effective_days = caps.get_effective_days(days)
effective_count = caps.get_effective_count(max_count)
if effective_days < days:
logger.info(f"平台限制:请求 {days} 天,实际使用 {effective_days}")
return await self.adapter.fetch_messages(
group_id=group_id,
days=effective_days,
max_count=effective_count,
)
async def fetch_messages_as_raw(
self,
group_id: str,
days: int = None,
max_count: int = None,
) -> list[dict]:
"""
获取消息并转换为原始字典格式。
此方法提供与现有分析器的向后兼容性,
这些分析器期望原始字典格式的消息。
参数:
group_id: 要获取消息的群组 ID
days: 天数
max_count: 最大消息数量
返回:
原始消息字典列表(通用格式,由适配器决定具体格式)
"""
# unified_messages = await self.fetch_messages(group_id, days, max_count)
#
# # 如果适配器实现了 convert_to_raw_format,则使用它
# if hasattr(self.adapter, "convert_to_raw_format"):
# return self.adapter.convert_to_raw_format(unified_messages)
#
# # 默认回退逻辑:手动转换
# # 这可能不完美,但能保证基本的向后兼容性
# return [
# {
# "message_id": msg.message_id,
# "group_id": msg.group_id,
# "sender": {
# "user_id": msg.sender_id,
# "nickname": msg.sender_name,
# "card": msg.sender_card
# },
# "time": msg.timestamp,
# "message": msg.text_content, # 简化处理
# "raw_message": msg.text_content
# }
# for msg in unified_messages
# ]
# 暂时直接使用适配器获取 raw 格式,如果适配器支持
# 这是为了确保现有逻辑完全兼容,因为 convert_to_raw_format 可能有损
# 但我们希望尽可能使用新的 fetch_messages
unified_messages = await self.fetch_messages(group_id, days, max_count)
return self.adapter.convert_to_raw_format(unified_messages)
async def get_group_info(self, group_id: str):
"""获取群组信息。"""
return await self.adapter.get_group_info(group_id)
async def get_member_avatars(
self,
user_ids: list[str],
size: int = 100,
) -> dict[str, str | None]:
"""
批量获取用户头像 URL。
参数:
user_ids: 用户 ID 列表
size: 头像尺寸
返回:
用户 ID 到头像 URL 的映射字典(URL 可能为 None)
"""
return await self.adapter.batch_get_avatar_urls(user_ids, size)
async def send_text(self, group_id: str, text: str) -> bool:
"""发送文本消息到群组。"""
return await self.adapter.send_text(group_id, text)
async def send_image(
self,
group_id: str,
image_path: str,
caption: str = "",
) -> bool:
"""发送图片到群组。"""
return await self.adapter.send_image(group_id, image_path, caption)
async def send_file(
self,
group_id: str,
file_path: str,
filename: str = None,
) -> bool:
"""发送文件到群组。"""
return await self.adapter.send_file(group_id, file_path, filename)
def validate_message_count(self, messages: list[UnifiedMessage]) -> bool:
"""
检查消息数量是否达到最小阈值。
参数:
messages: 消息列表
返回:
如果数量足够返回 True
"""
return len(messages) >= self.config.min_messages_threshold
def get_analysis_text(self, messages: list[UnifiedMessage]) -> str:
"""
将消息转换为 LLM 分析文本格式。
参数:
messages: UnifiedMessage 列表
返回:
格式化的 LLM 分析文本
"""
return MessageConverter.unified_to_analysis_text(messages)
@@ -0,0 +1,160 @@
"""
分析应用服务 - 应用层
实现“每日群聊分析并生成报告”的核心用例。
负责协调领域服务、基础设施适配器及持久化层。
"""
import asyncio
from typing import Any
from ...utils.logger import logger
from ..domain.models.data_models import TokenUsage
from ..domain.repositories.analysis_repository import IAnalysisProvider
from ..domain.repositories.report_repository import IReportGenerator
from ..domain.services.analysis_domain_service import AnalysisDomainService
from ..domain.services.statistics_service import StatisticsService
class AnalysisApplicationService:
"""分析应用服务 - 协调业务流程"""
def __init__(
self,
config_manager: Any,
bot_manager: Any,
history_manager: Any,
report_generator: IReportGenerator,
llm_analyzer: IAnalysisProvider,
statistics_service: StatisticsService,
analysis_domain_service: AnalysisDomainService,
):
self.config_manager = config_manager
self.bot_manager = bot_manager
self.history_manager = history_manager
self.report_generator = report_generator
self.llm_analyzer = llm_analyzer
self.statistics_service = statistics_service
self.analysis_domain_service = analysis_domain_service
async def execute_daily_analysis(
self, group_id: str, platform_id: str | None = None, manual: bool = False
) -> dict[str, Any]:
"""
执行每日分析用例。
流程:
1. 获取适配器
2. 拉取消息 (Infrastructure)
3. 基础统计 (Domain Service)
4. 用户分析 (Domain Service)
5. LLM 语义分析 (Infrastructure/Analysis Bridge)
6. 生成报告 (Visualization/Infrastructure)
7. 持久化摘要 (Persistence)
8. 返回结果
"""
logger.info(f"开始执行分析用例: 群 {group_id}, 平台 {platform_id or '默认'}")
# 1. 获取适配器
adapter = self.bot_manager.get_adapter(platform_id)
if not adapter:
raise ValueError(f"未找到平台 {platform_id} 的适配器")
# 2. 拉取消息
days = self.config_manager.get_analysis_days()
max_count = self.config_manager.get_max_messages()
unified_messages = await adapter.fetch_messages(
group_id=group_id, days=days, max_count=max_count
)
if not unified_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
):
logger.info(
f"{group_id} 消息数 ({len(unified_messages)}) 未达到自动分析阈值"
)
return {"success": False, "reason": "below_threshold"}
# 3. 基础统计 (Domain Service)
statistics = await asyncio.to_thread(
self.statistics_service.calculate_group_statistics, unified_messages
)
# 4. 用户分析 (Domain Service)
bot_self_ids = self.config_manager.get_bot_self_ids()
user_activity = await asyncio.to_thread(
self.analysis_domain_service.analyze_user_activity,
unified_messages,
bot_self_ids,
)
max_user_titles = self.config_manager.get_max_user_titles()
top_users = self.analysis_domain_service.get_top_users(
user_activity, limit=max_user_titles
)
# 5. LLM 语义分析 (为了保持兼容,目前直接传 UnifiedMessage,后续如需传 raw dict 再加转换)
# LLMAnalyzer 内部可能已经处理了转换(见之前代码)
topic_enabled = self.config_manager.get_topic_analysis_enabled()
user_title_enabled = self.config_manager.get_user_title_analysis_enabled()
golden_quote_enabled = self.config_manager.get_golden_quote_analysis_enabled()
topics = []
user_titles = []
golden_quotes = []
total_token_usage = TokenUsage()
# Note: LLMAnalyzer 目前可能只接收 legacy 格式或特定的 UnifiedMessage 适配
# 暂时转换回 legacy 格式以确保稳定性,直到 LLMAnalyzer 被重构
legacy_messages = self.statistics_service._convert_to_legacy_dict(
unified_messages
)
unified_msg_origin = (
f"{platform_id}:GroupMessage:{group_id}" if platform_id else group_id
)
if topic_enabled and user_title_enabled and golden_quote_enabled:
(
topics,
user_titles,
golden_quotes,
total_token_usage,
) = await self.llm_analyzer.analyze_all_concurrent(
legacy_messages,
user_activity,
umo=unified_msg_origin,
top_users=top_users,
)
else:
# 按需串行执行 (略,实际实现可补全或合并)
pass
# 回填结果
statistics.golden_quotes = golden_quotes
statistics.token_usage = total_token_usage
analysis_result = {
"statistics": statistics,
"topics": topics,
"user_titles": user_titles,
"user_analysis": user_activity,
}
# 6. 持久化摘要 (Persistence)
await self.history_manager.save_analysis(group_id, analysis_result)
# 7. 生成报告并发送 (应用层编排发送动作)
# 这里由调用方处理发送,本服务只返回分析结果和可能的视觉产物
return {
"success": True,
"analysis_result": analysis_result,
"messages_count": len(unified_messages),
"adapter": adapter,
}
-8
View File
@@ -1,8 +0,0 @@
"""
核心功能模块
"""
from .config import ConfigManager
from .message_handler import MessageHandler
__all__ = ["ConfigManager", "MessageHandler"]
-480
View File
@@ -1,480 +0,0 @@
"""
配置管理模块
负责处理插件配置和PDF依赖检查
"""
import sys
from astrbot.api import AstrBotConfig, logger
from astrbot.core.utils.astrbot_path import get_astrbot_data_path
class ConfigManager:
"""配置管理器"""
def __init__(self, config: AstrBotConfig):
self.config = config
self._playwright_available = False
self._playwright_version = None
self._check_playwright_availability()
def get_group_list_mode(self) -> str:
"""获取群组列表模式 (whitelist/blacklist/none)"""
return self.config.get("group_list_mode", "none")
def get_group_list(self) -> list[str]:
"""获取群组列表(用于黑白名单)"""
return self.config.get("group_list", [])
def is_group_allowed(self, group_id_or_umo: str) -> bool:
"""
根据配置的白/黑名单判断是否允许在该群聊中使用
支持传入 simple group_id 或 UMO (Unified Message Origin)
"""
mode = self.get_group_list_mode().lower()
if mode not in ("whitelist", "blacklist", "none"):
mode = "none"
# none模式下,不进行黑白名单检查,由调用方决定(通常是回退到 enabled_groups
if mode == "none":
return True
glist = [str(g) for g in self.get_group_list()]
target = str(group_id_or_umo)
# 解析目标 ID(如果是 UMO,提取最后一部分作为 ID)
# UMO 格式通常为: platform_id:GroupMessage:group_id
target_simple_id = target.split(":")[-1] if ":" in target else target
def _is_match(item: str, target: str, target_simple_id: str) -> bool:
# 1. 配置项是 UMO (包含 :) -> 必须精确匹配目标 UMO
if ":" in item:
return item == target
# 2. 配置项是 Simple ID (不含 :) -> 匹配目标的 Simple ID
# 这意味着 Simple ID 配置对所有平台生效 (向后兼容)
return item == target_simple_id
is_in_list = any(_is_match(item, target, target_simple_id) for item in glist)
if mode == "whitelist":
return is_in_list
if mode == "blacklist":
return not is_in_list
return True
def get_max_messages(self) -> int:
"""获取最大消息数量"""
return self.config.get("max_messages", 1000)
def get_analysis_days(self) -> int:
"""获取分析天数"""
return self.config.get("analysis_days", 1)
def get_auto_analysis_time(self) -> list[str]:
"""获取自动分析时间列表"""
val = self.config.get("auto_analysis_time", ["09:00"])
# 兼容旧版本字符串配置
if isinstance(val, str):
val_list = [val]
# 自动修复配置格式
try:
self.config["auto_analysis_time"] = val_list
self.config.save_config()
logger.info(f"自动修复配置格式 auto_analysis_time: {val} -> {val_list}")
except Exception as e:
logger.warning(f"修复配置格式失败: {e}")
return val_list
return val if isinstance(val, list) else ["09:00"]
def get_enable_auto_analysis(self) -> bool:
"""获取是否启用自动分析"""
return self.config.get("enable_auto_analysis", False)
def get_output_format(self) -> str:
"""获取输出格式"""
return self.config.get("output_format", "image")
def get_min_messages_threshold(self) -> int:
"""获取最小消息阈值"""
return self.config.get("min_messages_threshold", 50)
def get_topic_analysis_enabled(self) -> bool:
"""获取是否启用话题分析"""
return self.config.get("topic_analysis_enabled", True)
def get_user_title_analysis_enabled(self) -> bool:
"""获取是否启用用户称号分析"""
return self.config.get("user_title_analysis_enabled", True)
def get_golden_quote_analysis_enabled(self) -> bool:
"""获取是否启用金句分析"""
return self.config.get("golden_quote_analysis_enabled", True)
def get_max_topics(self) -> int:
"""获取最大话题数量"""
return self.config.get("max_topics", 5)
def get_max_user_titles(self) -> int:
"""获取最大用户称号数量"""
return self.config.get("max_user_titles", 8)
def get_max_golden_quotes(self) -> int:
"""获取最大金句数量"""
return self.config.get("max_golden_quotes", 5)
def get_llm_retries(self) -> int:
"""获取LLM请求重试次数"""
return self.config.get("llm_retries", 2)
def get_llm_backoff(self) -> int:
"""获取LLM请求重试退避基值(秒),实际退避会乘以尝试次数"""
return self.config.get("llm_backoff", 2)
def get_topic_max_tokens(self) -> int:
"""获取话题分析最大token数"""
return self.config.get("topic_max_tokens", 12288)
def get_golden_quote_max_tokens(self) -> int:
"""获取金句分析最大token数"""
return self.config.get("golden_quote_max_tokens", 4096)
def get_user_title_max_tokens(self) -> int:
"""获取用户称号分析最大token数"""
return self.config.get("user_title_max_tokens", 4096)
def get_debug_mode(self) -> bool:
"""获取是否启用调试模式"""
return self.config.get("debug_mode", False)
def get_llm_provider_id(self) -> str:
"""获取主 LLM Provider ID"""
return self.config.get("llm_provider_id", "")
def get_topic_provider_id(self) -> str:
"""获取话题分析专用 Provider ID"""
return self.config.get("topic_provider_id", "")
def get_user_title_provider_id(self) -> str:
"""获取用户称号分析专用 Provider ID"""
return self.config.get("user_title_provider_id", "")
def get_golden_quote_provider_id(self) -> str:
"""获取金句分析专用 Provider ID"""
return self.config.get("golden_quote_provider_id", "")
def get_pdf_output_dir(self) -> str:
"""获取PDF输出目录"""
try:
plugin_name = "astrbot_plugin_qq_group_daily_analysis"
data_path = get_astrbot_data_path()
default_path = data_path / "plugin_data" / plugin_name / "reports"
return self.config.get("pdf_output_dir", str(default_path))
except Exception:
# 针对旧版本或导入错误的后备方案
return self.config.get(
"pdf_output_dir",
"data/plugins/astrbot_plugin_qq_group_daily_analysis/reports",
)
def get_bot_self_ids(self) -> list:
"""获取机器人自身的 ID 列表 (兼容 bot_qq_ids)"""
ids = self.config.get("bot_self_ids", [])
if not ids:
# 向后兼容 bot_qq_ids
ids = self.config.get("bot_qq_ids", [])
return ids
def get_pdf_filename_format(self) -> str:
"""获取PDF文件名格式"""
return self.config.get(
"pdf_filename_format", "群聊分析报告_{group_id}_{date}.pdf"
)
def get_topic_analysis_prompt(self, style: str = "topic_prompt") -> str:
"""
获取话题分析提示词模板
Args:
style: 提示词风格,默认为 "topic_prompt"
Returns:
提示词模板字符串
"""
# 直接从配置中获取 prompts 对象
prompts_config = self.config.get("topic_analysis_prompts", {})
# 获取指定的 prompt
prompt = prompts_config.get(style, "topic_prompt")
if prompt:
return prompt
# 兼容旧配置
return self.config.get("topic_analysis_prompt", "")
def get_user_title_analysis_prompt(self, style: str = "user_title_prompt") -> str:
"""
获取用户称号分析提示词模板
Args:
style: 提示词风格,默认为 "user_title_prompt"
Returns:
提示词模板字符串
"""
# 直接从配置中获取 prompts 对象
prompts_config = self.config.get("user_title_analysis_prompts", {})
# 获取指定的 prompt
prompt = prompts_config.get(style, "user_title_prompt")
if prompt:
return prompt
# 兼容旧配置
return self.config.get("user_title_analysis_prompt", "")
def get_golden_quote_analysis_prompt(
self, style: str = "golden_quote_prompt"
) -> str:
"""
获取金句分析提示词模板
Args:
style: 提示词风格,默认为 "golden_quote_prompt"
Returns:
提示词模板字符串
"""
# 直接从配置中获取 prompts 对象
prompts_config = self.config.get("golden_quote_analysis_prompts", {})
# 获取指定的 prompt
prompt = prompts_config.get(style, "golden_quote_prompt")
if prompt:
return prompt
# 兼容旧配置
return self.config.get("golden_quote_analysis_prompt", "")
def set_topic_analysis_prompt(self, prompt: str):
"""设置话题分析提示词模板"""
self.config["topic_analysis_prompt"] = prompt
self.config.save_config()
def set_user_title_analysis_prompt(self, prompt: str):
"""设置用户称号分析提示词模板"""
self.config["user_title_analysis_prompt"] = prompt
self.config.save_config()
def set_golden_quote_analysis_prompt(self, prompt: str):
"""设置金句分析提示词模板"""
self.config["golden_quote_analysis_prompt"] = prompt
self.config.save_config()
def set_output_format(self, format_type: str):
"""设置输出格式"""
self.config["output_format"] = format_type
self.config.save_config()
def set_group_list_mode(self, mode: str):
"""设置群组列表模式"""
self.config["group_list_mode"] = mode
self.config.save_config()
def set_group_list(self, groups: list[str]):
"""设置群组列表"""
self.config["group_list"] = groups
self.config.save_config()
def get_max_concurrent_tasks(self) -> int:
"""获取自动分析最大并发数"""
return self.config.get("max_concurrent_tasks", 3)
def set_max_concurrent_tasks(self, count: int):
"""设置自动分析最大并发数"""
self.config["max_concurrent_tasks"] = count
self.config.save_config()
def set_max_messages(self, count: int):
"""设置最大消息数量"""
self.config["max_messages"] = count
self.config.save_config()
def set_analysis_days(self, days: int):
"""设置分析天数"""
self.config["analysis_days"] = days
self.config.save_config()
def set_auto_analysis_time(self, time_val: str | list[str]):
"""设置自动分析时间"""
self.config["auto_analysis_time"] = time_val
self.config.save_config()
def set_enable_auto_analysis(self, enabled: bool):
"""设置是否启用自动分析"""
self.config["enable_auto_analysis"] = enabled
self.config.save_config()
def set_min_messages_threshold(self, threshold: int):
"""设置最小消息阈值"""
self.config["min_messages_threshold"] = threshold
self.config.save_config()
def set_topic_analysis_enabled(self, enabled: bool):
"""设置是否启用话题分析"""
self.config["topic_analysis_enabled"] = enabled
self.config.save_config()
def set_user_title_analysis_enabled(self, enabled: bool):
"""设置是否启用用户称号分析"""
self.config["user_title_analysis_enabled"] = enabled
self.config.save_config()
def set_golden_quote_analysis_enabled(self, enabled: bool):
"""设置是否启用金句分析"""
self.config["golden_quote_analysis_enabled"] = enabled
self.config.save_config()
def set_max_topics(self, count: int):
"""设置最大话题数量"""
self.config["max_topics"] = count
self.config.save_config()
def set_max_user_titles(self, count: int):
"""设置最大用户称号数量"""
self.config["max_user_titles"] = count
self.config.save_config()
def set_max_golden_quotes(self, count: int):
"""设置最大金句数量"""
self.config["max_golden_quotes"] = count
self.config.save_config()
def set_pdf_output_dir(self, directory: str):
"""设置PDF输出目录"""
self.config["pdf_output_dir"] = directory
self.config.save_config()
def set_pdf_filename_format(self, format_str: str):
"""设置PDF文件名格式"""
self.config["pdf_filename_format"] = format_str
self.config.save_config()
def get_report_template(self) -> str:
"""获取报告模板名称"""
return self.config.get("report_template", "scrapbook")
def set_report_template(self, template_name: str):
"""设置报告模板名称"""
self.config["report_template"] = template_name
self.config.save_config()
def get_enable_user_card(self) -> bool:
"""获取是否使用用户群名片"""
return self.config.get("enable_user_card", False)
@property
def playwright_available(self) -> bool:
"""检查playwright是否可用"""
return self._playwright_available
@property
def playwright_version(self) -> str | None:
"""获取playwright版本"""
return self._playwright_version
def _check_playwright_availability(self):
"""检查 playwright 可用性"""
try:
import importlib.util
if importlib.util.find_spec("playwright") is None:
raise ImportError
# 尝试导入以确保完整性
import playwright
from playwright.async_api import async_playwright # noqa: F401
self._playwright_available = True
# 检查版本
try:
self._playwright_version = playwright.__version__
logger.info(f"使用 playwright {self._playwright_version} 作为 PDF 引擎")
except AttributeError:
self._playwright_version = "unknown"
logger.info("使用 playwright (版本未知) 作为 PDF 引擎")
except ImportError:
self._playwright_available = False
self._playwright_version = None
logger.warning(
"playwright 未安装,PDF 功能将不可用。请使用 pip install playwright 安装,并运行 playwright install chromium"
)
def get_browser_path(self) -> str:
"""获取自定义浏览器路径"""
return self.config.get("browser_path", "")
def set_browser_path(self, path: str):
"""设置自定义浏览器路径"""
self.config["browser_path"] = path
self.config.save_config()
def reload_playwright(self) -> bool:
"""重新加载 playwright 模块"""
try:
logger.info("开始重新加载 playwright 模块...")
# 移除所有 playwright 相关模块
modules_to_remove = [
mod for mod in sys.modules.keys() if mod.startswith("playwright")
]
logger.info(f"移除模块: {modules_to_remove}")
for mod in modules_to_remove:
del sys.modules[mod]
# 强制重新导入
try:
import playwright
# 更新全局变量
self._playwright_available = True
try:
self._playwright_version = playwright.__version__
logger.info(
f"重新加载成功,playwright 版本: {self._playwright_version}"
)
except AttributeError:
self._playwright_version = "unknown"
logger.info("重新加载成功,playwright 版本未知")
return True
except ImportError:
logger.info("playwright 重新导入可能需要重启 AstrBot")
self._playwright_available = False
self._playwright_version = None
return False
except Exception:
logger.info("playwright 重新导入失败")
self._playwright_available = False
self._playwright_version = None
return False
except Exception as e:
logger.error(f"重新加载 playwright 时出错: {e}")
return False
def save_config(self):
"""保存配置到AstrBot配置系统"""
try:
self.config.save_config()
logger.info("配置已保存")
except Exception as e:
logger.error(f"保存配置失败: {e}")
def reload_config(self):
"""重新加载配置"""
try:
# 重新从AstrBot配置系统读取所有配置
logger.info("重新加载配置...")
# 配置会自动从self.config中重新读取
logger.info("配置重载完成")
except Exception as e:
logger.error(f"重新加载配置失败: {e}")
-282
View File
@@ -1,282 +0,0 @@
"""
消息处理模块
负责群聊消息的获取、过滤和预处理
"""
from collections import defaultdict
from datetime import datetime, timedelta
from ..models.data_models import EmojiStatistics, GroupStatistics, TokenUsage
from ..utils.logger import logger
from ..visualization.activity_charts import ActivityVisualizer
class MessageHandler:
"""消息处理器"""
def __init__(self, config_manager, bot_manager=None):
self.config_manager = config_manager
self.activity_visualizer = ActivityVisualizer()
self.bot_manager = bot_manager
def _extract_bot_self_id_from_instance(self, bot_instance):
"""从bot实例中提取ID(单个)"""
if hasattr(bot_instance, "self_id") and bot_instance.self_id:
return str(bot_instance.self_id)
elif hasattr(bot_instance, "user_id") and bot_instance.user_id:
return str(bot_instance.user_id)
return None
async def fetch_group_messages(
self, bot_instance, group_id: str, days: int, platform_id: str = None
) -> list[dict]:
"""获取群聊消息记录"""
try:
# 验证参数
if not group_id:
logger.error(f"{group_id} 参数无效")
return []
# 优先使用 PlatformAdapter (DDD)
if self.bot_manager and platform_id:
adapter = self.bot_manager.get_adapter(platform_id)
if adapter:
logger.info(
f"使用适配器获取群 {group_id} 消息 (平台: {platform_id})"
)
# 使用 adapter 获取统一消息列表
unified_messages = await adapter.fetch_messages(
group_id=str(group_id),
days=days,
max_count=self.config_manager.get_max_messages(),
)
# 转换为原始格式以兼容后续处理 (后续应迁移到统一格式处理)
return adapter.convert_to_raw_format(unified_messages)
# --- 以下为旧逻辑 (Legacy) ---
if not bot_instance:
logger.error("未提供 bot_instance 且未找到适配器")
return []
# 确保bot_manager有 ID 列表用于过滤
if self.bot_manager and not self.bot_manager.has_bot_self_id():
# 尝试从bot_instance提取 ID 并设置为列表
bot_self_id = self._extract_bot_self_id_from_instance(bot_instance)
if bot_self_id:
# 将单个 ID 转换为列表,保持统一处理
self.bot_manager.set_bot_self_ids([bot_self_id])
# 计算时间范围
end_time = datetime.now()
start_time = end_time - timedelta(days=days)
messages = []
# 一次性获取,移除分页与多轮查询
max_messages = self.config_manager.get_max_messages()
query_rounds = 0
logger.info(f"开始获取群 {group_id}{days} 天的消息记录")
logger.info(
f"时间范围: {start_time.strftime('%Y-%m-%d %H:%M:%S')}{end_time.strftime('%Y-%m-%d %H:%M:%S')}"
)
# 单次请求,按配置的 max_messages 作为 count
try:
payloads = {
"group_id": int(group_id) if group_id.isdigit() else group_id,
"count": int(max_messages),
}
result = None
if hasattr(bot_instance, "call_action"):
try:
# aiocqhttp (CQHttp) 方式
result = await bot_instance.call_action(
"get_group_msg_history", **payloads
)
query_rounds = 1
except Exception as api_err:
error_msg = str(api_err)
# 检查是否是特定的错误码(1200表示不在该群)
if (
"retcode=1200" in error_msg
or "消息undefined不存在" in error_msg
):
logger.warning(f"{group_id} 机器人不在此群中: {api_err}")
return []
else:
logger.error(f"{group_id} API 调用失败: {api_err}")
logger.error(
f"{group_id} 当前 OneBot 实现可能不支持 get_group_msg_history API"
)
return []
elif hasattr(bot_instance, "api"):
# 官方 bot (botClient) 不支持历史消息
logger.error(
f"{group_id} 检测到官方 Bot,官方 API 不支持获取历史消息"
)
return []
else:
logger.error(
f"{group_id} 未知的 bot_instance 类型,无法调用 API,类型: {type(bot_instance)}"
)
return []
if not result or "messages" not in result:
logger.warning(f"{group_id} API返回无效结果: {result}")
return []
round_messages = result.get("messages", [])
if not round_messages:
logger.info(f"{group_id} 未获取到消息")
# 过滤时间范围内的消息并过滤机器人自身消息
for msg in round_messages:
try:
msg_time = datetime.fromtimestamp(msg.get("time", 0))
if not (start_time <= msg_time <= end_time):
continue
sender_id = str(msg.get("sender", {}).get("user_id", ""))
if (
self.bot_manager
and self.bot_manager.should_filter_bot_message(sender_id)
):
continue
messages.append(msg)
except Exception as msg_error:
logger.warning(f"{group_id} 处理单条消息失败: {msg_error}")
continue
except Exception as e:
logger.error(f"{group_id} 获取消息失败: {e}")
return []
# ========== 最终清理步骤:严格过滤和限制 ==========
original_count = len(messages)
# 1. 严格过滤时间范围外的消息
messages = [
msg
for msg in messages
if start_time <= datetime.fromtimestamp(msg.get("time", 0)) <= end_time
]
time_filtered_count = len(messages)
# 2. 严格限制消息数量
if len(messages) > max_messages:
# 保留最新的消息(假设messages已按时间排序,从新到旧)
messages = messages[:max_messages]
logger.info(
f"{group_id} 消息数量超过限制,已截断: {time_filtered_count} -> {max_messages}"
)
# 记录清理结果
if original_count != len(messages):
logger.info(
f"{group_id} 最终清理: 原始 {original_count} 条 -> 时间过滤 {time_filtered_count} 条 -> 最终 {len(messages)}"
)
logger.info(
f"{group_id} 消息获取完成,共获取到 {len(messages)} 条有效消息(时间范围: 近{days}天),查询轮数: {query_rounds}"
)
return messages
except Exception as e:
logger.error(f"{group_id} 获取群聊消息记录失败: {e}", exc_info=True)
return []
def calculate_statistics(self, messages: list[dict]) -> GroupStatistics:
"""计算基础统计数据"""
total_chars = 0
participants = set()
hour_counts = defaultdict(int)
emoji_statistics = EmojiStatistics()
for msg in messages:
sender_id = str(msg.get("sender", {}).get("user_id", ""))
participants.add(sender_id)
# 统计时间分布
msg_time = datetime.fromtimestamp(msg.get("time", 0))
hour_counts[msg_time.hour] += 1
# 处理消息内容
for content in msg.get("message", []):
if content.get("type") == "text":
text = content.get("data", {}).get("text", "")
total_chars += len(text)
elif content.get("type") == "face":
# 基础表情
emoji_statistics.face_count += 1
face_id = content.get("data", {}).get("id", "unknown")
emoji_statistics.face_details[f"face_{face_id}"] = (
emoji_statistics.face_details.get(f"face_{face_id}", 0) + 1
)
elif content.get("type") == "mface":
# 动画表情/魔法表情
emoji_statistics.mface_count += 1
emoji_id = content.get("data", {}).get("emoji_id", "unknown")
emoji_statistics.face_details[f"mface_{emoji_id}"] = (
emoji_statistics.face_details.get(f"mface_{emoji_id}", 0) + 1
)
elif content.get("type") == "bface":
# 超级表情
emoji_statistics.bface_count += 1
emoji_id = content.get("data", {}).get("p", "unknown")
emoji_statistics.face_details[f"bface_{emoji_id}"] = (
emoji_statistics.face_details.get(f"bface_{emoji_id}", 0) + 1
)
elif content.get("type") == "sface":
# 小表情
emoji_statistics.sface_count += 1
emoji_id = content.get("data", {}).get("id", "unknown")
emoji_statistics.face_details[f"sface_{emoji_id}"] = (
emoji_statistics.face_details.get(f"sface_{emoji_id}", 0) + 1
)
elif content.get("type") == "image":
# 检查是否是动画表情(通过summary字段判断)
data = content.get("data", {})
summary = data.get("summary", "")
if "动画表情" in summary or "表情" in summary:
# 动画表情(以image形式发送)
emoji_statistics.mface_count += 1
file_name = data.get("file", "unknown")
emoji_statistics.face_details[f"animated_{file_name}"] = (
emoji_statistics.face_details.get(
f"animated_{file_name}", 0
)
+ 1
)
else:
# 普通图片,不计入表情统计
pass
elif (
content.get("type") in ["record", "video"]
and "emoji" in str(content.get("data", {})).lower()
):
# 其他可能的表情类型
emoji_statistics.other_emoji_count += 1
# 找出最活跃时段
most_active_hour = (
max(hour_counts.items(), key=lambda x: x[1])[0] if hour_counts else 0
)
most_active_period = (
f"{most_active_hour:02d}:00-{(most_active_hour + 1) % 24:02d}:00"
)
# 生成活跃度可视化数据
activity_visualization = (
self.activity_visualizer.generate_activity_visualization(messages)
)
return GroupStatistics(
message_count=len(messages),
total_characters=total_chars,
participant_count=len(participants),
most_active_period=most_active_period,
golden_quotes=[],
emoji_count=emoji_statistics.total_emoji_count, # 保持向后兼容
emoji_statistics=emoji_statistics,
activity_visualization=activity_visualization,
token_usage=TokenUsage(),
)
-316
View File
@@ -1,316 +0,0 @@
import base64
import aiohttp
from ..utils.logger import logger
from ..utils.trace_context import TraceContext
class MessageSender:
"""
负责消息发送的核心组件
封装了多平台发送、格式转换 (URL/Base64)、失败重试等逻辑
"""
def __init__(self, bot_manager, config_manager, retry_manager=None):
self.bot_manager = bot_manager
self.config_manager = config_manager
self.retry_manager = retry_manager
async def send_text(
self, group_id: str, text: str, platform_id: str | None = None
) -> bool:
"""
发送文本消息
"""
trace_id = TraceContext.get()
logger.info(f"[{trace_id}] 开始发送文本消息到群 {group_id}")
platforms = self._get_available_platforms(group_id, platform_id)
if not platforms:
logger.error(f"[{trace_id}] 群 {group_id} 无可用发送平台")
return False
for pid, adapter in platforms:
try:
logger.info(f"[{trace_id}] 正在尝试平台 {pid}...")
# 优先使用 Adapter 接口
if hasattr(adapter, "send_text"):
if await adapter.send_text(group_id, text):
logger.info(f"[{trace_id}] 成功通过 {pid} 发送文本")
return True
# Fallback to OneBot API (for backward compatibility or if adapter wrapping failed)
if hasattr(adapter, "api") and hasattr(adapter.api, "call_action"):
await adapter.api.call_action(
"send_group_msg", group_id=group_id, message=text
)
logger.info(f"[{trace_id}] 成功通过 {pid} 发送文本 (API)")
return True
except Exception as e:
self._log_send_error(pid, group_id, "text", e)
continue
logger.error(f"[{trace_id}] 所有平台均发送文本失败")
return False
async def send_image_url(
self,
group_id: str,
image_url: str,
text_prefix: str = "",
platform_id: str | None = None,
) -> bool:
"""
发送图片 (URL 模式)
"""
trace_id = TraceContext.get()
platforms = self._get_available_platforms(group_id, platform_id)
if not platforms:
return False
for pid, adapter in platforms:
try:
logger.info(f"[{trace_id}] 正在通过 {pid} 发送图片 (URL 模式)...")
# 优先使用 Adapter 接口
if hasattr(adapter, "send_image"):
if await adapter.send_image(
group_id, image_url, caption=text_prefix
):
logger.info(f"[{trace_id}] 成功通过 {pid} 发送图片 (URL 模式)")
return True
# Fallback to OneBot API
if hasattr(adapter, "api") and hasattr(adapter.api, "call_action"):
message_chain = []
if text_prefix:
message_chain.append(
{"type": "text", "data": {"text": text_prefix}}
)
message_chain.append({"type": "image", "data": {"url": image_url}})
await adapter.api.call_action(
"send_group_msg", group_id=group_id, message=message_chain
)
logger.info(
f"[{trace_id}] 成功通过 {pid} 发送图片 (URL 模式) (API)"
)
return True
except Exception as e:
self._log_send_error(pid, group_id, "image_url", e)
continue
return False
async def send_image_base64(
self,
group_id: str,
image_url: str,
text_prefix: str = "",
platform_id: str | None = None,
) -> bool:
"""
发送图片 (Base64 模式) - 需先下载图片
"""
trace_id = TraceContext.get()
logger.info(f"[{trace_id}] 正在下载图片以进行 Base64 回退发送...")
image_bytes = await self._download_image(image_url)
if not image_bytes:
logger.error(f"[{trace_id}] 下载图片进行 Base64 转换失败")
return False
image_b64 = base64.b64encode(image_bytes).decode()
# file URI for Base64 (OneBot style)
base64_uri = f"base64://{image_b64}"
platforms = self._get_available_platforms(group_id, platform_id)
if not platforms:
return False
for pid, adapter in platforms:
try:
logger.info(f"[{trace_id}] 正在通过 {pid} 发送图片 (Base64 模式)...")
# 优先使用 Adapter 接口 (注意 Adapter 接口通常接受 path/url,这里我们传 base64 uri 它是支持的吗?)
# 大多数 Adapter 的 send_image 如果识别 base64:// 应该能处理
# 如果是 DiscordAdapter, 它需要特殊处理 local file.
# 但这里是 Base64 string.
# 为了稳妥,我们可以先尝试 Adapter,如果 Adapter 明确支持 base64://
if hasattr(adapter, "send_image"):
# 尝试发送 base64 URI
if await adapter.send_image(
group_id, base64_uri, caption=text_prefix
):
logger.info(
f"[{trace_id}] 成功通过 {pid} 发送图片 (Base64 模式)"
)
return True
# Fallback to OneBot API
if hasattr(adapter, "api") and hasattr(adapter.api, "call_action"):
message_chain = []
if text_prefix:
message_chain.append(
{"type": "text", "data": {"text": text_prefix}}
)
message_chain.append(
{"type": "image", "data": {"file": base64_uri}}
)
await adapter.api.call_action(
"send_group_msg", group_id=group_id, message=message_chain
)
logger.info(
f"[{trace_id}] 成功通过 {pid} 发送图片 (Base64 模式) (API)"
)
return True
except Exception as e:
self._log_send_error(pid, group_id, "image_base64", e)
continue
return False
async def send_image_smart(
self,
group_id: str,
image_url: str,
text_prefix: str = "",
platform_id: str | None = None,
) -> bool:
"""
智能发送图片:先尝试 URL,失败则回退到 Base64
"""
if await self.send_image_url(group_id, image_url, text_prefix, platform_id):
return True
logger.warning(
f"[{TraceContext.get()}] URL 发送失败,正在回退至 Base64 模式..."
)
return await self.send_image_base64(
group_id, image_url, text_prefix, platform_id
)
async def send_pdf(
self,
group_id: str,
pdf_path: str,
text_prefix: str = "",
platform_id: str | None = None,
) -> bool:
"""
发送 PDF 文件
"""
trace_id = TraceContext.get()
platforms = self._get_available_platforms(group_id, platform_id)
if not platforms:
return False
for pid, adapter in platforms:
try:
logger.info(f"[{trace_id}] 正在通过 {pid} 发送 PDF...")
if hasattr(adapter, "send_file"):
if await adapter.send_file(group_id, pdf_path):
logger.info(f"[{trace_id}] 成功通过 {pid} 发送 PDF")
return True
# Fallback to OneBot API
if hasattr(adapter, "api") and hasattr(adapter.api, "call_action"):
message_chain = []
if text_prefix:
message_chain.append(
{"type": "text", "data": {"text": text_prefix}}
)
message_chain.append({"type": "file", "data": {"file": pdf_path}})
await adapter.api.call_action(
"send_group_msg", group_id=group_id, message=message_chain
)
logger.info(f"[{trace_id}] 成功通过 {pid} 发送 PDF (API)")
return True
except Exception as e:
self._log_send_error(pid, group_id, "pdf", e)
continue
return False
def _get_available_platforms(
self, group_id: str, specific_platform_id: str | None = None
) -> list[tuple]:
"""
获取可用的发送平台列表 (返回 Adapter 实例)
"""
from ..infrastructure.platform.base import PlatformAdapter
from ..infrastructure.platform.factory import PlatformAdapterFactory
instances = []
if specific_platform_id:
bot = self.bot_manager.get_bot_instance(specific_platform_id)
if bot:
instances.append((specific_platform_id, bot))
else:
logger.warning(f"找不到指定的平台 {specific_platform_id}")
else:
# 获取所有已发现的平台
all_instances = self.bot_manager.get_all_bot_instances()
if all_instances:
instances = list(all_instances.items())
# Wrap instances with Adapters if needed
adapters = []
for pid, bot in instances:
# Check if it's already an adapter
if isinstance(bot, PlatformAdapter):
adapters.append((pid, bot))
continue
# If not, try to create an adapter
# We need to detect platform name first
platform_name = self.bot_manager._detect_platform_name(bot)
if not platform_name:
# If cannot detect, assume it's a OneBot raw object if it has api
if hasattr(bot, "api"):
adapters.append((pid, bot)) # Return raw bot for backward compat
continue
# Create adapter
try:
# We need config for adapter, here we use empty config or try to fetch from somewhere
# Ideally config_manager should provide it but it's complex.
# Passing empty config is fine for basic sending tasks as long as bot instance is valid.
adapter = PlatformAdapterFactory.create(platform_name, bot, config={})
if adapter:
adapters.append((pid, adapter))
else:
# Fallback: return raw bot
adapters.append((pid, bot))
except Exception as e:
logger.warning(f"{pid} 创建适配器失败: {e}")
adapters.append((pid, bot))
return adapters
async def _download_image(self, url: str) -> bytes | None:
"""下载图片 helper"""
try:
timeout = aiohttp.ClientTimeout(total=15)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(url) as resp:
if resp.status != 200:
return None
return await resp.read()
except Exception as e:
logger.error(f"图片下载失败: {e}")
return None
def _log_send_error(
self, platform_id: str, group_id: str, msg_type: str, error: Exception
):
"""统一错误日志"""
logger.debug(
f"[{TraceContext.get()}] 通过 {platform_id}{group_id} 发送 {msg_type} 失败: {error}"
)
@@ -0,0 +1,51 @@
"""
分析服务接口 - 领域层
定义语义分析的抽象契约
"""
from abc import ABC, abstractmethod
from ..models.data_models import GoldenQuote, SummaryTopic, TokenUsage, UserTitle
class IAnalysisProvider(ABC):
"""
LLM 分析提供商接口
"""
@abstractmethod
async def analyze_topics(
self, messages: list[dict], umo: str = None, session_id: str = None
) -> tuple[list[SummaryTopic], TokenUsage]:
"""分析话题"""
pass
@abstractmethod
async def analyze_user_titles(
self,
messages: list[dict],
user_activity: dict,
umo: str = None,
top_users: list[dict] = None,
session_id: str = None,
) -> tuple[list[UserTitle], TokenUsage]:
"""分析用户称号"""
pass
@abstractmethod
async def analyze_golden_quotes(
self, messages: list[dict], umo: str = None, session_id: str = None
) -> tuple[list[GoldenQuote], TokenUsage]:
"""分析金句"""
pass
@abstractmethod
async def analyze_all_concurrent(
self,
messages: list[dict],
user_activity: dict,
umo: str = None,
top_users: list[dict] = None,
) -> tuple[list[SummaryTopic], list[UserTitle], list[GoldenQuote], TokenUsage]:
"""并发分析所有内容"""
pass
@@ -73,6 +73,22 @@ class IMessageSender(ABC):
"""发送图片消息"""
pass
@abstractmethod
async def send_forward_msg(
self,
group_id: str,
nodes: list[dict],
) -> bool:
"""
发送合并转发消息。
Args:
group_id: 目标群组 ID
nodes: 转发节点列表。每个节点通常包含 name, uin (或 user_id), content。
目前主要用于 OneBot 兼容性。
"""
pass
@abstractmethod
async def send_file(
self,
@@ -0,0 +1,36 @@
"""
报告生成接口 - 领域层
定义分析报告生成的抽象契约
"""
from abc import ABC, abstractmethod
from typing import Any
class IReportGenerator(ABC):
"""
报告生成器接口
"""
@abstractmethod
async def generate_image_report(
self,
analysis_result: dict,
group_id: str,
html_render_func: Any,
avatar_getter: Any = None,
) -> tuple[str | None, str | None]:
"""生成图片报告"""
pass
@abstractmethod
async def generate_pdf_report(
self, analysis_result: dict, group_id: str, avatar_getter: Any = None
) -> str | None:
"""生成 PDF 报告"""
pass
@abstractmethod
def generate_text_report(self, analysis_result: dict) -> str:
"""生成文本报告"""
pass
@@ -0,0 +1,125 @@
"""
分析领域服务 - 领域层
负责用户维度的活跃度分析、发言习惯及活动模式识别。
"""
import re
from collections import defaultdict
from datetime import datetime
from ..value_objects.unified_message import MessageContentType, UnifiedMessage
class AnalysisDomainService:
"""分析领域服务 - 处理用户画像及行为分析"""
# Discord 自定义表情正则 <:name:id> 或 <a:name:id>
DISCORD_CUSTOM_EMOJI_PATTERN = r"<a?:.+?:\d+>"
# 简单的 Unicode Emoji 正则范围
UNICODE_EMOJI_PATTERN = (
r"[\U0001F000-\U0001F9FF]|[\U00002600-\U000026FF]|[\U00002700-\U000027BF]"
)
def analyze_user_activity(
self, messages: list[UnifiedMessage], bot_self_ids: list[str] = None
) -> dict[str, dict]:
"""
分析用户活跃度。
基于 UnifiedMessage 计算每个用户的发言数、字数、表情数等。
"""
user_stats = defaultdict(
lambda: {
"message_count": 0,
"char_count": 0,
"emoji_count": 0,
"nickname": "",
"hours": defaultdict(int),
"reply_count": 0,
}
)
bot_ids = set(bot_self_ids or [])
for msg in messages:
user_id = msg.sender_id
# 跳过机器人自己的消息
if user_id in bot_ids:
continue
user_stats[user_id]["message_count"] += 1
user_stats[user_id]["nickname"] = msg.sender_card or msg.sender_name
# 统计时间分布
msg_time = datetime.fromtimestamp(msg.timestamp)
user_stats[user_id]["hours"][msg_time.hour] += 1
# 统计内容
for content in msg.contents:
if content.type == MessageContentType.TEXT:
text = content.text or ""
user_stats[user_id]["char_count"] += len(text)
# 统计文本中的表情 (Discord/Unicode)
user_stats[user_id]["emoji_count"] += len(
re.findall(self.DISCORD_CUSTOM_EMOJI_PATTERN, text)
)
user_stats[user_id]["emoji_count"] += len(
re.findall(self.UNICODE_EMOJI_PATTERN, text)
)
elif content.type == MessageContentType.EMOJI:
user_stats[user_id]["emoji_count"] += 1
elif content.type == MessageContentType.REPLY:
user_stats[user_id]["reply_count"] += 1
return dict(user_stats)
def get_top_users(
self, user_activity: dict[str, dict], limit: int = 10
) -> list[dict]:
"""获取最活跃的用户列表"""
users = []
for user_id, stats in user_activity.items():
users.append(
{
"user_id": user_id,
"nickname": stats["nickname"],
"message_count": stats["message_count"],
"char_count": stats["char_count"],
"emoji_count": stats["emoji_count"],
"reply_count": stats["reply_count"],
}
)
# 按消息数量排序
users.sort(key=lambda x: x["message_count"], reverse=True)
return users[:limit]
def get_user_activity_pattern(
self, user_activity: dict[str, dict], user_id: str
) -> dict:
"""获取并识别指定用户的活动模式"""
if user_id not in user_activity:
return {}
stats = user_activity[user_id]
hours = stats["hours"]
# 找出最活跃的时间段
most_active_hour = max(hours.items(), key=lambda x: x[1])[0] if hours else 0
# 计算夜间活跃度 (0-6点)
night_messages = sum(hours[h] for h in range(0, 6))
night_ratio = (
night_messages / stats["message_count"] if stats["message_count"] > 0 else 0
)
return {
"most_active_hour": most_active_hour,
"night_ratio": night_ratio,
"hourly_distribution": dict(hours),
}
+106
View File
@@ -0,0 +1,106 @@
"""
统计领域服务 - 领域层
负责核心统计逻辑的计算,不依赖于具体的平台或基础设施。
"""
from collections import defaultdict
from datetime import datetime
from ...infrastructure.visualization.activity_charts import ActivityVisualizer
from ..models.data_models import EmojiStatistics, GroupStatistics, TokenUsage
from ..value_objects.unified_message import MessageContentType, UnifiedMessage
class StatisticsService:
"""统计服务 - 处理群聊数据的聚合统计"""
def __init__(self):
self.activity_visualizer = ActivityVisualizer()
def calculate_group_statistics(
self, messages: list[UnifiedMessage]
) -> GroupStatistics:
"""
计算群组基础统计数据。
基于统一消息格式(UnifiedMessage)进行计算,确保跨平台一致性。
"""
total_chars = 0
participants = set()
hour_counts = defaultdict(int)
emoji_statistics = EmojiStatistics()
for msg in messages:
participants.add(msg.sender_id)
# 统计时间分布
msg_time = datetime.fromtimestamp(msg.timestamp)
hour_counts[msg_time.hour] += 1
# 处理消息内容
for content in msg.contents:
if content.type == MessageContentType.TEXT:
total_chars += len(content.text or "")
elif content.type == MessageContentType.EMOJI:
emoji_statistics.face_count += 1
# 尝试保留原始表情详情(如果适配器提供了)
face_id = content.emoji_id or "unknown"
emoji_statistics.face_details[f"emoji_{face_id}"] = (
emoji_statistics.face_details.get(f"emoji_{face_id}", 0) + 1
)
elif content.type == MessageContentType.IMAGE:
# 检查是否是动画表情(通过raw_data判断,如果适配器提供了)
if content.raw_data and (
"动画表情" in str(content.raw_data)
or "表情" in str(content.raw_data)
):
emoji_statistics.mface_count += 1
elif content.type in (
MessageContentType.VOICE,
MessageContentType.VIDEO,
):
# 其他非文本类型统计(可选)
pass
# 找出最活跃时段
most_active_hour = (
max(hour_counts.items(), key=lambda x: x[1])[0] if hour_counts else 0
)
most_active_period = (
f"{most_active_hour:02d}:00-{(most_active_hour + 1) % 24:02d}:00"
)
# 生成活跃度可视化数据
# 注意:ActivityVisualizer 可能需要迁移以支持 UnifiedMessage
# 目前先转换回 dict 以保持兼容性,或者之后重构它
raw_msgs = self._convert_to_legacy_dict(messages)
activity_visualization = (
self.activity_visualizer.generate_activity_visualization(raw_msgs)
)
return GroupStatistics(
message_count=len(messages),
total_characters=total_chars,
participant_count=len(participants),
most_active_period=most_active_period,
golden_quotes=[],
emoji_count=emoji_statistics.total_emoji_count,
emoji_statistics=emoji_statistics,
activity_visualization=activity_visualization,
token_usage=TokenUsage(),
)
def _convert_to_legacy_dict(self, messages: list[UnifiedMessage]) -> list[dict]:
"""内部辅助:将 UnifiedMessage 转换为 Legacy Dict 格式,用于兼容可视化组件"""
legacy_list = []
for msg in messages:
legacy_list.append(
{
"time": msg.timestamp,
"sender": {"user_id": msg.sender_id},
"message": [
{"type": "text", "data": {"text": msg.text_content or ""}}
],
}
)
return legacy_list
@@ -6,8 +6,8 @@
from abc import ABC, abstractmethod
from typing import Any
from ...models.data_models import TokenUsage
from ...utils.logger import logger
from ....domain.models.data_models import TokenUsage
from ....utils.logger import logger
from ..utils.json_utils import parse_json_response
from ..utils.llm_utils import (
call_provider_with_retry,
@@ -5,8 +5,8 @@
from datetime import datetime
from ...models.data_models import GoldenQuote, TokenUsage
from ...utils.logger import logger
from ....domain.models.data_models import GoldenQuote, TokenUsage
from ....utils.logger import logger
from ..utils import InfoUtils
from ..utils.json_utils import extract_golden_quotes_with_regex
from .base_analyzer import BaseAnalyzer
@@ -6,8 +6,8 @@
import re
from datetime import datetime
from ...models.data_models import SummaryTopic, TokenUsage
from ...utils.logger import logger
from ....domain.models.data_models import SummaryTopic, TokenUsage
from ....utils.logger import logger
from ..utils import InfoUtils
from ..utils.json_utils import extract_topics_with_regex
from .base_analyzer import BaseAnalyzer
@@ -3,8 +3,8 @@
专门处理用户称号和MBTI类型分析
"""
from ...models.data_models import TokenUsage, UserTitle
from ...utils.logger import logger
from ....domain.models.data_models import TokenUsage, UserTitle
from ....utils.logger import logger
from ..utils.json_utils import extract_user_titles_with_regex
from .base_analyzer import BaseAnalyzer
@@ -5,8 +5,13 @@ LLM分析器模块
import asyncio
from ..models.data_models import GoldenQuote, SummaryTopic, TokenUsage, UserTitle
from ..utils.logger import logger
from ...domain.models.data_models import (
GoldenQuote,
SummaryTopic,
TokenUsage,
UserTitle,
)
from ...utils.logger import logger
from .analyzers.golden_quote_analyzer import GoldenQuoteAnalyzer
from .analyzers.topic_analyzer import TopicAnalyzer
from .analyzers.user_title_analyzer import UserTitleAnalyzer
+390 -193
View File
@@ -1,234 +1,431 @@
"""
配置管理 - 集中化配置管理
该模块提供了一个访问插件配置的统一接口,
封装了现有的配置模块,并增加了验证和默认值功能。
配置管理模块 - 基础设施层
负责处理插件配置和PDF依赖检查
"""
from typing import Any
import sys
from astrbot.api import AstrBotConfig, logger
from astrbot.core.utils.astrbot_path import get_astrbot_data_path
class ConfigManager:
"""
插件的集中配置管理器。
"""配置管理器"""
提供带有默认值和验证的配置值类型化访问。
"""
def __init__(self, config: AstrBotConfig):
self.config = config
self._playwright_available = False
self._playwright_version = None
self._check_playwright_availability()
def __init__(self, config: dict[str, Any]):
def get_group_list_mode(self) -> str:
"""获取群组列表模式 (whitelist/blacklist/none)"""
return self.config.get("group_list_mode", "none")
def get_group_list(self) -> list[str]:
"""获取群组列表(用于黑白名单)"""
return self.config.get("group_list", [])
def is_group_allowed(self, group_id_or_umo: str) -> bool:
"""
初始化配置管理器。
Args:
config: 原始配置字典
根据配置的白/黑名单判断是否允许在该群聊中使用
支持传入 simple group_id 或 UMO (Unified Message Origin)
"""
self._config = config or {}
mode = self.get_group_list_mode().lower()
if mode not in ("whitelist", "blacklist", "none"):
mode = "none"
def get(self, key: str, default: Any = None) -> Any:
"""
获取配置值。
if mode == "none":
return True
Args:
key: 配置键(支持点号表示法)
default: 如果键未找到则返回默认值
glist = [str(g) for g in self.get_group_list()]
target = str(group_id_or_umo)
Returns:
配置值或默认值
"""
try:
keys = key.split(".")
value = self._config
for k in keys:
if isinstance(value, dict):
value = value.get(k)
else:
return default
if value is None:
return default
return value
except Exception:
return default
target_simple_id = target.split(":")[-1] if ":" in target else target
def set(self, key: str, value: Any) -> None:
"""
设置配置值。
def _is_match(item: str, target: str, target_simple_id: str) -> bool:
if ":" in item:
return item == target
return item == target_simple_id
Args:
key: 配置键
value: 要设置的值
"""
keys = key.split(".")
config = self._config
for k in keys[:-1]:
if k not in config:
config[k] = {}
config = config[k]
config[keys[-1]] = value
is_in_list = any(_is_match(item, target, target_simple_id) for item in glist)
# ========================================================================
# 群组配置
# ========================================================================
if mode == "whitelist":
return is_in_list
if mode == "blacklist":
return not is_in_list
def get_enabled_groups(self) -> list[str]:
"""获取启用的群组 ID 列表。"""
groups = self.get("enabled_groups", [])
return [str(g) for g in groups] if groups else []
return True
def is_group_enabled(self, group_id: str) -> bool:
"""检查群组是否启用了分析。"""
enabled = self.get_enabled_groups()
return str(group_id) in enabled or not enabled # 空列表意味着全部启用
def get_max_messages(self) -> int:
"""获取最大消息数量"""
return self.config.get("max_messages", 1000)
def get_bot_qq_ids(self) -> list[str]:
"""获取要过滤掉的机器人 QQ ID 列表。"""
ids = self.get("bot_qq_ids", [])
return [str(i) for i in ids] if ids else []
def get_analysis_days(self) -> int:
"""获取分析天数"""
return self.config.get("analysis_days", 1)
# ========================================================================
# 分析配置
# ========================================================================
def get_auto_analysis_time(self) -> list[str]:
"""获取自动分析时间列表"""
val = self.config.get("auto_analysis_time", ["09:00"])
# 兼容旧版本字符串配置
if isinstance(val, str):
val_list = [val]
# 自动修复配置格式
try:
self.config["auto_analysis_time"] = val_list
self.config.save_config()
logger.info(f"自动修复配置格式 auto_analysis_time: {val} -> {val_list}")
except Exception as e:
logger.warning(f"修复配置格式失败: {e}")
return val_list
return val if isinstance(val, list) else ["09:00"]
def get_enable_auto_analysis(self) -> bool:
"""获取是否启用自动分析"""
return self.config.get("enable_auto_analysis", False)
def get_output_format(self) -> str:
"""获取输出格式"""
return self.config.get("output_format", "image")
def get_min_messages_threshold(self) -> int:
"""获取最小消息阈值"""
return self.config.get("min_messages_threshold", 50)
def get_topic_analysis_enabled(self) -> bool:
"""获取是否启用话题分析"""
return self.config.get("topic_analysis_enabled", True)
def get_user_title_analysis_enabled(self) -> bool:
"""获取是否启用用户称号分析"""
return self.config.get("user_title_analysis_enabled", True)
def get_golden_quote_analysis_enabled(self) -> bool:
"""获取是否启用金句分析"""
return self.config.get("golden_quote_analysis_enabled", True)
def get_max_topics(self) -> int:
"""获取要提取的最大话题数"""
return int(self.get("max_topics", 5))
"""获取最大话题数"""
return self.config.get("max_topics", 5)
def get_max_user_titles(self) -> int:
"""获取要生成的最大用户称号数"""
return int(self.get("max_user_titles", 10))
"""获取最大用户称号数"""
return self.config.get("max_user_titles", 8)
def get_max_golden_quotes(self) -> int:
"""获取要提取的最大金句数"""
return int(self.get("max_golden_quotes", 5))
"""获取最大金句数"""
return self.config.get("max_golden_quotes", 5)
def get_min_messages_for_analysis(self) -> int:
"""获取分析所需的最小消息数。"""
return int(self.get("min_messages", 50))
def get_llm_retries(self) -> int:
"""获取LLM请求重试次数"""
return self.config.get("llm_retries", 2)
# ========================================================================
# LLM 配置
def get_topic_provider_id(self) -> str | None:
"""获取话题分析的提供商 ID"""
return self.get("topic_provider_id")
def get_user_title_provider_id(self) -> str | None:
"""获取用户称号分析的提供商 ID"""
return self.get("user_title_provider_id")
def get_golden_quote_provider_id(self) -> str | None:
"""获取金句分析的提供商 ID"""
return self.get("golden_quote_provider_id")
def get_llm_backoff(self) -> int:
"""获取LLM请求重试退避基值(秒),实际退避会乘以尝试次数"""
return self.config.get("llm_backoff", 2)
def get_topic_max_tokens(self) -> int:
"""获取话题分析最大 token """
return int(self.get("topic_max_tokens", 2000))
def get_user_title_max_tokens(self) -> int:
"""获取用户称号分析的最大 token 数"""
return int(self.get("user_title_max_tokens", 2000))
"""获取话题分析最大token数"""
return self.config.get("topic_max_tokens", 12288)
def get_golden_quote_max_tokens(self) -> int:
"""获取金句分析最大 token """
return int(self.get("golden_quote_max_tokens", 1500))
"""获取金句分析最大token数"""
return self.config.get("golden_quote_max_tokens", 4096)
# ========================================================================
# 提示词配置
# ========================================================================
def get_user_title_max_tokens(self) -> int:
"""获取用户称号分析最大token数"""
return self.config.get("user_title_max_tokens", 4096)
def get_topic_analysis_prompt(self) -> str | None:
"""获取话题分析的自定义提示词模板"""
return self.get("prompts.topic_analysis")
def get_debug_mode(self) -> bool:
"""获取是否启用调试模式"""
return self.config.get("debug_mode", False)
def get_user_title_analysis_prompt(self) -> str | None:
"""获取用户称号分析的自定义提示词模板"""
return self.get("prompts.user_title_analysis")
def get_llm_provider_id(self) -> str:
"""获取主 LLM Provider ID"""
return self.config.get("llm_provider_id", "")
def get_golden_quote_analysis_prompt(self) -> str | None:
"""获取金句分析的自定义提示词模板"""
return self.get("prompts.golden_quote_analysis")
def get_topic_provider_id(self) -> str:
"""获取话题分析专用 Provider ID"""
return self.config.get("topic_provider_id", "")
# ========================================================================
# 调度配置
# ========================================================================
def get_user_title_provider_id(self) -> str:
"""获取用户称号分析专用 Provider ID"""
return self.config.get("user_title_provider_id", "")
def get_auto_analysis_enabled(self) -> bool:
"""检查是否启用了自动分析"""
return bool(self.get("auto_analysis_enabled", False))
def get_golden_quote_provider_id(self) -> str:
"""获取金句分析专用 Provider ID"""
return self.config.get("golden_quote_provider_id", "")
def get_analysis_time(self) -> str:
"""获取计划分析时间 (HH:MM 格式)"""
return str(self.get("analysis_time", "23:00"))
def get_analysis_timezone(self) -> str:
"""获取计划分析的时区"""
return str(self.get("timezone", "Asia/Shanghai"))
# ========================================================================
# 报告配置
# ========================================================================
def get_report_format(self) -> str:
"""获取报告格式 (text, markdown, image)"""
return str(self.get("report_format", "text"))
def get_include_statistics(self) -> bool:
"""检查是否在报告中包含统计信息"""
return bool(self.get("include_statistics", True))
def get_include_topics(self) -> bool:
"""检查是否在报告中包含话题"""
return bool(self.get("include_topics", True))
def get_include_user_titles(self) -> bool:
"""检查是否在报告中包含用户称号"""
return bool(self.get("include_user_titles", True))
def get_include_golden_quotes(self) -> bool:
"""检查是否在报告中包含金句"""
return bool(self.get("include_golden_quotes", True))
# ========================================================================
# 工具方法
# ========================================================================
def to_dict(self) -> dict[str, Any]:
"""获取原始配置字典"""
return self._config.copy()
def update(self, updates: dict[str, Any]) -> None:
"""
使用新值更新配置
Args:
updates: 要应用的更新字典
"""
self._config.update(updates)
def validate(self) -> list[str]:
"""
验证配置
Returns:
验证错误消息列表(如果有效则为空)
"""
errors = []
# 验证数值范围
if self.get_max_topics() < 1 or self.get_max_topics() > 20:
errors.append("max_topics 必须在 1 到 20 之间")
if self.get_max_user_titles() < 1 or self.get_max_user_titles() > 50:
errors.append("max_user_titles 必须在 1 到 50 之间")
if self.get_max_golden_quotes() < 1 or self.get_max_golden_quotes() > 20:
errors.append("max_golden_quotes 必须在 1 到 20 之间")
# 验证时间格式
time_str = self.get_analysis_time()
def get_pdf_output_dir(self) -> str:
"""获取PDF输出目录"""
try:
hours, minutes = time_str.split(":")
if not (0 <= int(hours) <= 23 and 0 <= int(minutes) <= 59):
errors.append("analysis_time 必须是 HH:MM 格式 (00:00-23:59)")
except ValueError:
errors.append("analysis_time 必须是 HH:MM 格式")
plugin_name = "astrbot_plugin_qq_group_daily_analysis"
data_path = get_astrbot_data_path()
default_path = data_path / "plugin_data" / plugin_name / "reports"
return self.config.get("pdf_output_dir", str(default_path))
except Exception:
return self.config.get(
"pdf_output_dir",
"data/plugins/astrbot_plugin_qq_group_daily_analysis/reports",
)
return errors
def get_bot_self_ids(self) -> list:
"""获取机器人自身的 ID 列表 (兼容 bot_qq_ids)"""
ids = self.config.get("bot_self_ids", [])
if not ids:
ids = self.config.get("bot_qq_ids", [])
return ids
def get_pdf_filename_format(self) -> str:
"""获取PDF文件名格式"""
return self.config.get(
"pdf_filename_format", "群聊分析报告_{group_id}_{date}.pdf"
)
def get_topic_analysis_prompt(self, style: str = "topic_prompt") -> str:
"""获取话题分析提示词模板"""
prompts_config = self.config.get("topic_analysis_prompts", {})
prompt = prompts_config.get(style, "topic_prompt")
if prompt:
return prompt
return self.config.get("topic_analysis_prompt", "")
def get_user_title_analysis_prompt(self, style: str = "user_title_prompt") -> str:
"""获取用户称号分析提示词模板"""
prompts_config = self.config.get("user_title_analysis_prompts", {})
prompt = prompts_config.get(style, "user_title_prompt")
if prompt:
return prompt
return self.config.get("user_title_analysis_prompt", "")
def get_golden_quote_analysis_prompt(
self, style: str = "golden_quote_prompt"
) -> str:
"""获取金句分析提示词模板"""
prompts_config = self.config.get("golden_quote_analysis_prompts", {})
prompt = prompts_config.get(style, "golden_quote_prompt")
if prompt:
return prompt
return self.config.get("golden_quote_analysis_prompt", "")
def set_topic_analysis_prompt(self, prompt: str):
"""设置话题分析提示词模板"""
self.config["topic_analysis_prompt"] = prompt
self.config.save_config()
def set_user_title_analysis_prompt(self, prompt: str):
"""设置用户称号分析提示词模板"""
self.config["user_title_analysis_prompt"] = prompt
self.config.save_config()
def set_golden_quote_analysis_prompt(self, prompt: str):
"""设置金句分析提示词模板"""
self.config["golden_quote_analysis_prompt"] = prompt
self.config.save_config()
def set_output_format(self, format_type: str):
"""设置输出格式"""
self.config["output_format"] = format_type
self.config.save_config()
def set_group_list_mode(self, mode: str):
"""设置群组列表模式"""
self.config["group_list_mode"] = mode
self.config.save_config()
def set_group_list(self, groups: list[str]):
"""设置群组列表"""
self.config["group_list"] = groups
self.config.save_config()
def get_max_concurrent_tasks(self) -> int:
"""获取自动分析最大并发数"""
return self.config.get("max_concurrent_tasks", 3)
def set_max_concurrent_tasks(self, count: int):
"""设置自动分析最大并发数"""
self.config["max_concurrent_tasks"] = count
self.config.save_config()
def set_max_messages(self, count: int):
"""设置最大消息数量"""
self.config["max_messages"] = count
self.config.save_config()
def set_analysis_days(self, days: int):
"""设置分析天数"""
self.config["analysis_days"] = days
self.config.save_config()
def set_auto_analysis_time(self, time_val: str | list[str]):
"""设置自动分析时间"""
self.config["auto_analysis_time"] = time_val
self.config.save_config()
def set_enable_auto_analysis(self, enabled: bool):
"""设置是否启用自动分析"""
self.config["enable_auto_analysis"] = enabled
self.config.save_config()
def set_min_messages_threshold(self, threshold: int):
"""设置最小消息阈值"""
self.config["min_messages_threshold"] = threshold
self.config.save_config()
def set_topic_analysis_enabled(self, enabled: bool):
"""设置是否启用话题分析"""
self.config["topic_analysis_enabled"] = enabled
self.config.save_config()
def set_user_title_analysis_enabled(self, enabled: bool):
"""设置是否启用用户称号分析"""
self.config["user_title_analysis_enabled"] = enabled
self.config.save_config()
def set_golden_quote_analysis_enabled(self, enabled: bool):
"""设置是否启用金句分析"""
self.config["golden_quote_analysis_enabled"] = enabled
self.config.save_config()
def set_max_topics(self, count: int):
"""设置最大话题数量"""
self.config["max_topics"] = count
self.config.save_config()
def set_max_user_titles(self, count: int):
"""设置最大用户称号数量"""
self.config["max_user_titles"] = count
self.config.save_config()
def set_max_golden_quotes(self, count: int):
"""设置最大金句数量"""
self.config["max_golden_quotes"] = count
self.config.save_config()
def set_pdf_output_dir(self, directory: str):
"""设置PDF输出目录"""
self.config["pdf_output_dir"] = directory
self.config.save_config()
def set_pdf_filename_format(self, format_str: str):
"""设置PDF文件名格式"""
self.config["pdf_filename_format"] = format_str
self.config.save_config()
def get_report_template(self) -> str:
"""获取报告模板名称"""
return self.config.get("report_template", "scrapbook")
def set_report_template(self, template_name: str):
"""设置报告模板名称"""
self.config["report_template"] = template_name
self.config.save_config()
def get_enable_user_card(self) -> bool:
"""获取是否使用用户群名片"""
return self.config.get("enable_user_card", False)
@property
def playwright_available(self) -> bool:
"""检查playwright是否可用"""
return self._playwright_available
@property
def playwright_version(self) -> str | None:
"""获取playwright版本"""
return self._playwright_version
def _check_playwright_availability(self):
"""检查 playwright 可用性"""
try:
import importlib.util
if importlib.util.find_spec("playwright") is None:
raise ImportError
import playwright
from playwright.async_api import async_playwright # noqa: F401
self._playwright_available = True
try:
self._playwright_version = playwright.__version__
logger.info(f"使用 playwright {self._playwright_version} 作为 PDF 引擎")
except AttributeError:
self._playwright_version = "unknown"
logger.info("使用 playwright (版本未知) 作为 PDF 引擎")
except ImportError:
self._playwright_available = False
self._playwright_version = None
logger.warning(
"playwright 未安装,PDF 功能将不可用。请使用 pip install playwright 安装,并运行 playwright install chromium"
)
def get_browser_path(self) -> str:
"""获取自定义浏览器路径"""
return self.config.get("browser_path", "")
def set_browser_path(self, path: str):
"""设置自定义浏览器路径"""
self.config["browser_path"] = path
self.config.save_config()
def reload_playwright(self) -> bool:
"""重新加载 playwright 模块"""
try:
logger.info("开始重新加载 playwright 模块...")
modules_to_remove = [
mod for mod in sys.modules.keys() if mod.startswith("playwright")
]
logger.info(f"移除模块: {modules_to_remove}")
for mod in modules_to_remove:
del sys.modules[mod]
try:
import playwright
self._playwright_available = True
try:
self._playwright_version = playwright.__version__
logger.info(
f"重新加载成功,playwright 版本: {self._playwright_version}"
)
except AttributeError:
self._playwright_version = "unknown"
logger.info("重新加载成功,playwright 版本未知")
return True
except ImportError:
logger.info("playwright 重新导入可能需要重启 AstrBot")
self._playwright_available = False
self._playwright_version = None
return False
except Exception:
logger.info("playwright 重新导入失败")
self._playwright_available = False
self._playwright_version = None
return False
except Exception as e:
logger.error(f"重新加载 playwright 时出错: {e}")
return False
def save_config(self):
"""保存配置到AstrBot配置系统"""
try:
self.config.save_config()
logger.info("配置已保存")
except Exception as e:
logger.error(f"保存配置失败: {e}")
def reload_config(self):
"""重新加载配置"""
try:
logger.info("重新加载配置...")
logger.info("配置重载完成")
except Exception as e:
logger.error(f"重新加载配置失败: {e}")
@@ -1,5 +1,5 @@
"""
历史记录管理器模块
历史记录管理器模块 - 基础设施持久化层
负责存储和查询群聊分析报告的摘要信息
使用 AstrBot put_kv_data/get_kv_data 实现
"""
@@ -7,7 +7,7 @@
import datetime
from typing import Any
from ..utils.logger import logger
from ...utils.logger import logger
class HistoryManager:
@@ -90,16 +90,7 @@ class HistoryManager:
) -> dict[str, Any] | None:
"""
根据群组日期和时间点检索一份历史摘要
Args:
group_id (str): 群组 ID
date_str (str): 日期 (YYYY-MM-DD)
time_str (str): 时间点 (HH-MM)
Returns:
dict[str, Any] | None: 历史摘要字典未找到返回 None
"""
# 对齐存储时的 Key 规范
time_str = time_str.replace(":", "-")
key = f"analysis_{group_id}_{date_str}_{time_str}"
return await self.plugin.get_kv_data(key, None)
@@ -107,14 +98,6 @@ class HistoryManager:
async def has_history(self, group_id: str, date_str: str, time_str: str) -> bool:
"""
快速判定是否存在指定时间点的历史分析记录
Args:
group_id (str): 群组 ID
date_str (str): 日期
time_str (str): 时间点
Returns:
bool: 是否存在记录
"""
history = await self.get_history(group_id, date_str, time_str)
return history is not None
@@ -474,6 +474,53 @@ class DiscordAdapter(PlatformAdapter):
logger.error(f"Discord 文件发送失败: {e}")
return False
async def send_forward_msg(
self,
group_id: str,
nodes: list[dict],
) -> bool:
"""
在 Discord 模拟合并转发。
由于 Discord 没有原生节点转发 API,我们将其转换为一组文本消息发送。
"""
if not discord:
return False
try:
channel_id = int(group_id)
channel = self._discord_client.get_channel(channel_id)
if not channel:
channel = await self._discord_client.fetch_channel(channel_id)
if not hasattr(channel, "send"):
return False
# 将节点汇总为美化的文本块
lines = ["📊 **结构化报告摘要 (Structured Report)**\n"]
for node in nodes:
data = node.get("data", node) # 兼容不同格式
name = data.get("name", "AstrBot")
content = data.get("content", "")
lines.append(f"**[{name}]**:\n{content}\n")
full_text = "\n".join(lines)
# 分段处理大消息
if len(full_text) > 1900:
parts = [
full_text[i : i + 1900] for i in range(0, len(full_text), 1900)
]
for part in parts:
await channel.send(content=part)
else:
await channel.send(content=full_text)
return True
except Exception as e:
logger.error(f"Discord 模拟转发失败: {e}")
return False
# ==================== IGroupInfoRepository 实现 ====================
async def get_group_info(self, group_id: str) -> UnifiedGroup | None:
@@ -418,6 +418,41 @@ class OneBotAdapter(PlatformAdapter):
logger.error(f"OneBot 文件发送失败: {e}")
return False
async def send_forward_msg(
self,
group_id: str,
nodes: list[dict],
) -> bool:
"""
发送群合并转发消息。
Args:
group_id (str): 目标群号
nodes (list[dict]): 转发节点列表
Returns:
bool: 是否发送成功
"""
if not hasattr(self.bot, "call_action"):
return False
try:
# 兼容处理节点中的 uin -> user_id (有些后端偏好 uin)
for node in nodes:
if "data" in node:
if "user_id" in node["data"] and "uin" not in node["data"]:
node["data"]["uin"] = node["data"]["user_id"]
await self.bot.call_action(
"send_group_forward_msg",
group_id=int(group_id),
messages=nodes,
)
return True
except Exception as e:
logger.warning(f"OneBot 发送合并转发消息失败: {e}")
return False
# ==================== IGroupInfoRepository 实现 ====================
async def get_group_info(self, group_id: str) -> UnifiedGroup | None:
@@ -1,14 +1,12 @@
"""
Bot实例管理模块
Bot实例管理模块 - 基础设施层
统一管理bot实例的获取设置和使用
已重构以集成 DDD PlatformAdapter 架构支持多平台扩展
"""
from typing import Any
from ..infrastructure.platform import PlatformAdapter, PlatformAdapterFactory
from ..utils.logger import logger
from ...utils.logger import logger
from . import PlatformAdapter, PlatformAdapterFactory
class BotManager:
@@ -146,6 +144,14 @@ class BotManager:
"""获取所有已加载的bot实例 {platform_id: bot_instance}"""
return self._bot_instances.copy()
def get_platform_count(self) -> int:
"""获取当前已加载的平台数量"""
return len(self._bot_instances)
def get_platform_ids(self) -> list[str]:
"""获取所有已加载的平台 ID 列表"""
return list(self._bot_instances.keys())
def has_bot_instance(self) -> bool:
"""检查是否有可用的bot实例"""
return bool(self._bot_instances)
@@ -169,11 +175,6 @@ class BotManager:
bot 实例检测平台名称用于创建适配器
返回平台名称如 'aiocqhttp', 'discord'
检测优先级:
1. bot 实例的 platform 属性
2. 已知的 API 特征检测
3. 类名模式匹配作为后备方案
"""
# 优先使用 platform 属性
if hasattr(bot_instance, "platform"):
@@ -304,10 +305,6 @@ class BotManager:
elif isinstance(metadata, dict) and "name" in metadata:
platform_name = metadata["name"]
logger.info(
f"[群分析插件 BotManager] Initial platform_name detection: {platform_name}"
)
# 验证此平台名称是否受支持,如果不支持,尝试从bot实例检测(如果可用)
if (
not platform_name
@@ -332,22 +329,7 @@ class BotManager:
f"发现平台 {platform_id} 但客户端未就绪。将进行懒加载。"
)
discovered[platform_id] = platform
else:
# 后备方案:如果元数据丢失/损坏但我们有客户端,尝试使用它
if bot_client:
platform_name = self._detect_platform_name(bot_client)
if platform_name:
# 生成临时ID或使用名称
platform_id = platform_name
logger.warning(
f"平台元数据丢失,使用检测到的类型 '{platform_name}' 作为 ID。"
)
self._platforms[platform_id] = platform
self.set_bot_instance(bot_client, platform_id, platform_name)
discovered[platform_id] = bot_client
# 记录适配器创建结果
if self._adapters:
logger.info(
f"已创建 {len(self._adapters)} 个 PlatformAdapter: "
@@ -367,7 +349,6 @@ class BotManager:
discovered = await self.auto_discover_bot_instances()
self._is_initialized = True
# 返回发现的实例字典
return discovered
def get_status_info(self) -> dict[str, Any]:
@@ -392,9 +373,6 @@ class BotManager:
def update_from_event(self, event):
"""从事件更新bot实例(用于手动命令)"""
# 检查是否为 QQ 平台事件 (兼容性检查)
# 注意: 非 aiocqhttp 平台也可以使用,只要适配器已注册
if hasattr(event, "bot") and event.bot:
# 从事件中获取平台ID
platform_id = None
@@ -449,8 +427,6 @@ class BotManager:
def is_plugin_enabled(self, platform_id: str, plugin_name: str) -> bool:
"""检查指定平台是否启用了该插件"""
if platform_id not in self._platforms:
# 如果找不到平台对象(例如是手动添加的),默认认为启用
# 或者可以返回 True,因为无法进行否定检查
return True
platform = self._platforms[platform_id]
@@ -460,7 +436,7 @@ class BotManager:
plugin_set = platform.config.get("plugin_set", ["*"])
if plugin_set is None:
return False # 如果明确为 None, 视为都不启用? 或者默认? Default is ["*"] usually.
return False
if "*" in plugin_set:
return True
@@ -10,12 +10,13 @@ from pathlib import Path
import aiohttp
from ..utils.logger import logger
from ...domain.repositories.report_repository import IReportGenerator
from ...utils.logger import logger
from ..visualization.activity_charts import ActivityVisualizer
from .templates import HTMLTemplates
class ReportGenerator:
class ReportGenerator(IReportGenerator):
"""报告生成器"""
def __init__(self, config_manager):
@@ -9,7 +9,7 @@ import threading
from jinja2 import Environment, FileSystemLoader, select_autoescape
from ..utils.logger import logger
from ...utils.logger import logger
class HTMLTemplates:
@@ -8,10 +8,10 @@ import weakref
from apscheduler.triggers.cron import CronTrigger
from ...utils.logger import logger
from ..core.message_sender import MessageSender
from ..infrastructure.platform.factory import PlatformAdapterFactory
from ..reports.dispatcher import ReportDispatcher
from ..utils.logger import logger
from ..platform.factory import PlatformAdapterFactory
from ..reporting.dispatcher import ReportDispatcher
from ..utils.trace_context import TraceContext
@@ -21,27 +21,21 @@ class AutoScheduler:
def __init__(
self,
config_manager,
message_handler,
analyzer,
report_generator,
analysis_service,
bot_manager,
retry_manager,
history_manager,
html_render_func=None,
):
self.config_manager = config_manager
self.message_handler = message_handler
self.analyzer = analyzer
self.report_generator = report_generator
self.analysis_service = analysis_service
self.bot_manager = bot_manager
self.retry_manager = retry_manager # 保存引用
self.history_manager = history_manager
self.retry_manager = retry_manager
self.html_render_func = html_render_func
# Initialize Core Components
self.message_sender = MessageSender(bot_manager, config_manager, retry_manager)
self.report_dispatcher = ReportDispatcher(
config_manager, report_generator, self.message_sender, retry_manager
config_manager, None, self.message_sender, retry_manager
)
if html_render_func:
self.report_dispatcher.set_html_render(html_render_func)
@@ -74,50 +68,26 @@ class AutoScheduler:
and self.bot_manager._bot_instances
):
# 如果只有一个实例,直接返回
if len(self.bot_manager._bot_instances) == 1:
platform_id = list(self.bot_manager._bot_instances.keys())[0]
if self.bot_manager.get_platform_count() == 1:
platform_id = self.bot_manager.get_platform_ids()[0]
logger.debug(f"只有一个适配器,使用平台: {platform_id}")
return platform_id
# 如果有多个实例,尝试通过API检查群属于哪个适配器
# 如果有多个实例,尝试通过适配器检查群属于哪个平台
logger.info(f"检测到多个适配器,正在验证群 {group_id} 属于哪个平台...")
for platform_id in self.bot_manager.get_all_bot_instances().keys():
for platform_id in self.bot_manager.get_platform_ids():
try:
# 优先使用 Adapter (DDD)
adapter = self.bot_manager.get_adapter(platform_id)
if adapter:
# 通过统一接口尝试获取群信息,如果能获取到则说明属于该平台
info = await adapter.get_group_info(str(group_id))
if info:
logger.info(f"✅ 群 {group_id} 属于平台 {platform_id}")
return platform_id
else:
logger.debug(
f"平台 {platform_id} 无法获取群 {group_id} 信息 (返回None)"
f"平台 {platform_id} 无法获取群 {group_id} 信息"
)
continue
# 回退到原始逻辑 (Legacy)
bot_instance = self.bot_manager.get_bot_instance(platform_id)
if hasattr(bot_instance, "call_action"):
result = await bot_instance.call_action(
"get_group_info", group_id=int(group_id)
)
if result and result.get("group_id"):
logger.info(f"✅ 群 {group_id} 属于平台 {platform_id}")
return platform_id
except Exception as e:
logger.debug(f"平台 {platform_id} 验证群 {group_id} 失败: {e}")
continue
# 回退到原始逻辑 (Legacy)
bot_instance = self.bot_manager.get_bot_instance(platform_id)
if hasattr(bot_instance, "call_action"):
result = await bot_instance.call_action(
"get_group_info", group_id=int(group_id)
)
if result and result.get("group_id"):
logger.info(f"✅ 群 {group_id} 属于平台 {platform_id}")
return platform_id
except Exception as e:
logger.debug(f"平台 {platform_id} 验证群 {group_id} 失败: {e}")
continue
@@ -311,7 +281,7 @@ class AutoScheduler:
async def _perform_auto_analysis_for_group(
self, group_id: str, target_platform_id: str = None
):
"""为指定群执行自动分析(核心逻辑"""
"""为指定群执行自动分析(业务逻辑委派给 AnalysisApplicationService"""
# 为每个群聊使用独立的锁
group_lock_key = f"analysis_{group_id}"
if not hasattr(self, "_group_locks"):
@@ -324,189 +294,51 @@ class AutoScheduler:
async with lock:
try:
start_time = asyncio.get_event_loop().time()
# 设置 TraceID
trace_id = TraceContext.generate(prefix=f"group_{group_id}")
TraceContext.set(trace_id)
import datetime
now = datetime.datetime.now()
date_str = now.strftime("%Y-%m-%d")
time_str = now.strftime("%H-%M")
if await self.history_manager.has_history(group_id, date_str, time_str):
logger.info(
f"{group_id}{date_str} {time_str} 已有分析记录,跳过自动分析"
)
return
logger.info(f"开始为群 {group_id} 执行自动分析(并发任务)")
logger.info(
f"开始为群 {group_id} 执行自动分析 (Platform: {target_platform_id or 'Auto'})"
)
# 检查平台状态 (BotManager 为基础设施层,用于获取平台就绪状态)
if not self.bot_manager.is_ready_for_auto_analysis():
status = self.bot_manager.get_status_info()
logger.warning(
f"{group_id} 自动分析跳过:bot管理器未就绪 - {status}"
)
logger.warning(f"{group_id} 自动分析跳过:bot管理器未就绪")
return
messages = None
platform_id = None
bot_instance = None
# 1. 优先使用指定的 platform_id (如果有)
if target_platform_id:
if self.bot_manager.is_plugin_enabled(
target_platform_id, "astrbot_plugin_qq_group_daily_analysis"
):
try:
logger.info(
f"使用指定平台 {target_platform_id} 获取群 {group_id} 的消息..."
)
bot_instance = self.bot_manager.get_bot_instance(
target_platform_id
)
if bot_instance:
analysis_days = self.config_manager.get_analysis_days()
messages = (
await self.message_handler.fetch_group_messages(
bot_instance,
group_id,
analysis_days,
target_platform_id,
)
)
if messages:
platform_id = target_platform_id
logger.info(
f"✅ 群 {group_id} 成功通过平台 {platform_id} 获取到 {len(messages)} 条消息"
)
except Exception as e:
logger.error(
f"指定平台 {target_platform_id} 获取消息失败: {e}"
)
# 2. 如果指定平台失败或没有指定,尝试自动检测 (原有逻辑)
if not messages:
# 获取所有可用的平台ID和bot实例
if (
hasattr(self.bot_manager, "_bot_instances")
and self.bot_manager._bot_instances
):
available_platforms = list(
self.bot_manager._bot_instances.items()
)
logger.info(
f"{group_id} 检测到 {len(available_platforms)} 个可用平台,开始依次尝试..."
)
for test_platform_id, test_bot_instance in available_platforms:
# 如果已经试过 target_platform_id,跳过
if (
target_platform_id
and test_platform_id == target_platform_id
):
continue
# 检查该平台是否启用了此插件
if not self.bot_manager.is_plugin_enabled(
test_platform_id,
"astrbot_plugin_qq_group_daily_analysis",
):
logger.debug(
f"平台 {test_platform_id} 未启用此插件,跳过"
)
continue
try:
logger.info(
f"尝试使用平台 {test_platform_id} 获取群 {group_id} 的消息..."
)
analysis_days = self.config_manager.get_analysis_days()
test_messages = (
await self.message_handler.fetch_group_messages(
test_bot_instance,
group_id,
analysis_days,
test_platform_id,
)
)
if test_messages and len(test_messages) > 0:
# 成功获取到消息,使用这个平台
messages = test_messages
platform_id = test_platform_id
bot_instance = test_bot_instance
logger.info(
f"✅ 群 {group_id} 成功通过平台 {platform_id} 获取到 {len(messages)} 条消息"
)
break
else:
logger.debug(
f"平台 {test_platform_id} 未获取到消息,继续尝试下一个平台"
)
except Exception as e:
logger.debug(
f"平台 {test_platform_id} 获取消息失败: {e},继续尝试下一个平台"
)
continue
if not messages:
logger.warning(
f"{group_id} 所有平台都尝试失败,未获取到足够的消息记录"
)
return
else:
# 回退到原来的逻辑(单个平台)- 几乎不会走到这里,除非 _bot_instances 为空
pass # 省略 legacy 逻辑,因为 _bot_instances 为空在上面 is_ready 检查了
if not messages:
# 最后尝试 legacy get_platform_id_for_group
# ... (Keep existing fallback if needed, but the loop above covers most cases)
pass
# 检查消息数量
min_threshold = self.config_manager.get_min_messages_threshold()
if not messages or len(messages) < min_threshold:
logger.warning(
f"{group_id} 消息数量不足({len(messages) if messages else 0}条),跳过分析"
)
return
logger.info(f"{group_id} 获取到 {len(messages)} 条消息,开始分析")
# 进行分析 - 构造正确的 unified_msg_origin
# platform_id 已经在前面获取,直接使用
umo = f"{platform_id}:GroupMessage:{group_id}" if platform_id else None
analysis_result = await self.analyzer.analyze_messages(
messages, group_id, umo
# 委派给应用层服务执行核心用例
result = await self.analysis_service.execute_daily_analysis(
group_id=group_id, platform_id=target_platform_id, manual=False
)
if not analysis_result:
logger.error(f"{group_id} 分析失败")
if not result.get("success"):
reason = result.get("reason")
logger.info(f"{group_id} 自动分析跳过: {reason}")
return
# 生成并发送报告
# 获取分析结果及适配器
analysis_result = result["analysis_result"]
adapter = result["adapter"]
# 调度导出并发送报告 (由 ReportDispatcher 协调)
# 注意:ReportDispatcher 可能也需要轻微重构以接收 adapter
# 但目前为了最小化改动,我们仍然使用 dispatcher 逻辑
# 传入 platform_id 以便其能正确路由
await self.report_dispatcher.dispatch(
group_id, analysis_result, platform_id
group_id,
analysis_result,
adapter.platform_id
if hasattr(adapter, "platform_id")
else target_platform_id,
)
# 保存到历史记录
await self.history_manager.save_analysis(
group_id, analysis_result, date_str, time_str
)
# 记录执行时间
end_time = asyncio.get_event_loop().time()
execution_time = end_time - start_time
logger.info(f"{group_id} 分析完成,耗时: {execution_time:.2f}")
logger.info(f"{group_id} 自动分析任务执行成功")
except Exception as e:
logger.error(f"{group_id} 自动分析执行失败: {e}", exc_info=True)
finally:
# 锁资源由 WeakValueDictionary 自动管理,无需手动清理
logger.info(f"{group_id} 自动分析完成")
logger.debug(f"{group_id} 自动分析流程结束")
async def _get_all_groups(self) -> list[tuple[str, str]]:
"""
@@ -7,7 +7,7 @@ from dataclasses import dataclass
import aiohttp
from ..utils.logger import logger
from ...utils.logger import logger
@dataclass
@@ -194,73 +194,27 @@ class RetryManager:
logger.error(f"[RetryManager] Base64编码失败: {e}")
return False
# 2. 获取 Bot 实例
bot = self.bot_manager.get_bot_instance(task.platform_id)
if not bot:
# 2. 获取适配器 (DDD 基础设施层)
adapter = self.bot_manager.get_adapter(task.platform_id)
if not adapter:
logger.error(
f"[RetryManager] 平台 {task.platform_id} Bot 实例未找到,无法重试"
f"[RetryManager] 平台 {task.platform_id}适配器未找到,无法重试"
)
return False # 无法重试,因为 Bot 已离线
return False
# 3. 发送图片
# 3. 发送图片 (通过统一适配器接口)
logger.info(
f"[RetryManager] 正在向群 {task.group_id} 发送重试图片 (Base64模式)..."
f"[RetryManager] 正在向群 {task.group_id} 发送重试图片 (Adapter: {type(adapter).__name__})..."
)
# 使用 OneBot v11 标准 API
if hasattr(bot, "api") and hasattr(bot.api, "call_action"):
try:
# 构造消息
# 使用 list 格式兼容性更好
message = [
{
"type": "text",
"data": {"text": "📊 每日群聊分析报告(重试发送):\n"},
},
{"type": "image", "data": {"file": image_file_str}},
]
result = await bot.api.call_action(
"send_group_msg", group_id=int(task.group_id), message=message
)
# 检查 retcode
if isinstance(result, dict):
retcode = result.get("retcode", 0)
if retcode == 0:
return True
elif retcode == 1200:
# 即使是 Base64 也可能超时,但概率小很多
logger.warning(
"[RetryManager] 发送失败 (retcode=1200): 消息可能过大或Bot连接不稳定"
)
return False
else:
logger.warning(
f"[RetryManager] 发送失败 (retcode={retcode}): {result}"
)
return False
return (
True # 假设非 dict 类型返回即成功(某些适配器可能返回不同类型)
)
except Exception as e:
logger.error(f"[RetryManager] 发送API调用异常: {e}")
return False
elif hasattr(bot, "send_msg"): # 尝试 AstrBot 抽象接口
try:
# 尝试直接发送
await bot.send_msg(image_file_str, group_id=task.group_id)
return True
except Exception as e:
logger.error(f"[RetryManager] 抽象接口发送失败: {e}")
return False
else:
logger.warning(
f"[RetryManager] 未知的 Bot 类型 {type(bot)},无法发送消息。"
)
# 注意:某些适配器可能需要 URL,某些需要 Base64。
# 适配器内部通常应处理好 bytes/base64 的发送。
# 这里我们尝试直接传 image_file_str (base64://)
try:
success = await adapter.send_image(task.group_id, image_file_str)
return success
except Exception as e:
logger.error(f"[RetryManager] 适配器发送图片异常: {e}")
return False
except Exception as e:
@@ -271,7 +225,7 @@ class RetryManager:
pass
async def _send_fallback_text(self, task: RetryTask):
"""发送文本回退报告(使用合并转发"""
"""发送文本回退报告(业务逻辑委派给适配器"""
if not self.report_generator:
logger.warning("[RetryManager] 未配置 ReportGenerator,无法发送文本回退")
return
@@ -282,60 +236,41 @@ class RetryManager:
task.analysis_result
)
bot = self.bot_manager.get_bot_instance(task.platform_id)
if not bot:
# 2. 获取适配器 (DDD 基础设施层)
adapter = self.bot_manager.get_adapter(task.platform_id)
if not adapter:
logger.error(
f"[RetryManager] 无法获取适配器 {task.platform_id},放弃发送回退文本"
)
return
# 构造合并转发节点
# 注意:这里需要构造符合 OneBot v11 标准的节点列表
# 即使没有 self_id,我们也可以尝试发送
# 获取 bot self_id (如果能获取到)
bot_id = "10000" # fallback id
if hasattr(bot, "self_id"):
bot_id = str(bot.self_id)
nickname = "AstrBot日常分析"
nodes = [
{
"type": "node",
"data": {
"name": nickname,
"uin": bot_id,
"content": "⚠️ 图片报告多次生成失败,为您呈现文本版报告:",
},
},
{
"type": "node",
"data": {"name": nickname, "uin": bot_id, "content": text_report},
"data": {"name": nickname, "content": text_report},
},
]
if hasattr(bot, "api") and hasattr(bot.api, "call_action"):
# 尝试发送群合并转发消息
# 一般使用 send_group_forward_msg 或 send_group_msg (带 nodes)
try:
await bot.api.call_action(
"send_group_forward_msg",
group_id=int(task.group_id),
messages=nodes,
)
logger.info(
f"[RetryManager] 群 {task.group_id} 文本回退报告发送成功 (合并转发)"
)
except Exception as e:
logger.warning(
f"[RetryManager] 合并转发失败,尝试直接发送文本: {e}"
)
# 回退到直接发送宽文本
await bot.api.call_action(
"send_group_msg",
group_id=int(task.group_id),
message=f"⚠️ 图片报告生成失败,文本报告:\n{text_report}"[
:4500
], # 截断防止过长
)
# 3. 通过适配器发送结构化消息
success = await adapter.send_forward_msg(task.group_id, nodes)
if success:
logger.info(f"[RetryManager] 群 {task.group_id} 文本回退报告发送成功")
else:
# 最终兜底:发送简单文本
logger.warning("[RetryManager] 结构化发送失败,尝试直接发送文本回退")
await adapter.send_text(
task.group_id,
f"⚠️ 图片报告生成失败,文本报告:\n{text_report}"[:4500],
)
except Exception as e:
logger.error(f"[RetryManager] 文本回退发送失败: {e}", exc_info=True)
logger.error(f"[RetryManager] 文本回退流程异常: {e}", exc_info=True)
@@ -6,7 +6,7 @@
from collections import defaultdict
from datetime import datetime
from ..models.data_models import ActivityVisualization
from ...domain.models.data_models import ActivityVisualization
class ActivityVisualizer:
-13
View File
@@ -1,13 +0,0 @@
"""
数据模型模块
"""
from .data_models import (
GoldenQuote,
GroupStatistics,
SummaryTopic,
TokenUsage,
UserTitle,
)
__all__ = ["SummaryTopic", "UserTitle", "GoldenQuote", "TokenUsage", "GroupStatistics"]
-197
View File
@@ -1,197 +0,0 @@
"""
通用工具函数模块
包含消息分析和其他通用功能
"""
import asyncio
from typing import Any
from ..analysis.llm_analyzer import LLMAnalyzer
from ..analysis.statistics import UserAnalyzer
from ..core.message_handler import MessageHandler
from ..models.data_models import TokenUsage
from .logger import logger
class MessageAnalyzer:
"""
业务逻辑:消息分析整合器
该类作为一个门面(Facade),将消息存储、统计计算、LLM 智能分析以及用户画像分析
等多个底层组件整合在一起,提供统一的消息分析流程接口。
Attributes:
context (Any): AstrBot 上下文环境
config_manager (Any): 配置管理者实例
bot_manager (Any, optional): 机器人多实例管理者
message_handler (MessageHandler): 负责消息过滤和基础统计
llm_analyzer (LLMAnalyzer): 负责调用大模型进行语义分析
user_analyzer (UserAnalyzer): 负责用户活跃度及角色分析
"""
def __init__(
self, context: Any, config_manager: Any, bot_manager: Any | None = None
):
"""
初始化消息分析器。
Args:
context (Any): AstrBot 核心上下文
config_manager (Any): 插件配置管理器
bot_manager (Any, optional): 多平台机器人管理器实例
"""
self.context = context
self.config_manager = config_manager
self.bot_manager = bot_manager
self.message_handler = MessageHandler(config_manager, bot_manager)
self.llm_analyzer = LLMAnalyzer(context, config_manager)
self.user_analyzer = UserAnalyzer(config_manager)
def _extract_bot_self_id_from_instance(self, bot_instance: Any) -> str | None:
"""
内部方法:从不同平台的机器人实例中探测其自身 ID。
Args:
bot_instance (Any): 宿主机器人实例 (如 OneBot, Discord 实例)
Returns:
str | None: 探测到的用户 ID 或 None
"""
if hasattr(bot_instance, "self_id") and bot_instance.self_id:
return str(bot_instance.self_id)
elif hasattr(bot_instance, "user_id") and bot_instance.user_id:
return str(bot_instance.user_id)
return None
async def set_bot_instance(
self, bot_instance: Any, platform_id: str | None = None
) -> None:
"""
向分析组件注入当前活跃的机器人实例。
Args:
bot_instance (Any): 活跃的机器人 SDK 实例
platform_id (str, optional): 平台标识符,用于多实例路由
"""
if self.bot_manager:
self.bot_manager.set_bot_instance(bot_instance, platform_id)
else:
# 降级逻辑:仅设置单个默认 ID
bot_self_id = self._extract_bot_self_id_from_instance(bot_instance)
if bot_self_id:
await self.message_handler.set_bot_self_ids([bot_self_id])
async def analyze_messages(
self, messages: list[dict], group_id: str, unified_msg_origin: str | None = None
) -> dict | None:
"""
执行完整的群消息流水化分析。
包含:消息预处理 -> 词频统计 -> 活跃用户识别 -> LLM 摘要/金句提取。
Args:
messages (list[dict]): 待处理的原始或统一格式消息字典列表
group_id (str): 群组 ID,用于上下文标识
unified_msg_origin (str, optional): 统一消息来源标识
Returns:
dict | None: 包含 statistics, topics, user_titles, user_analysis 的字典,失败返回 None
"""
try:
# 1. 基础消息统计 (耗时操作,放入线程池避免阻塞事件循环)
statistics = await asyncio.to_thread(
self.message_handler.calculate_statistics, messages
)
# 2. 用户维度分析 (等级、发言习惯等)
user_analysis = await asyncio.to_thread(
self.user_analyzer.analyze_users, messages
)
# 3. 筛选分析范围:提取 Top N 活跃用户用于深度称号分析
max_user_titles = self.config_manager.get_max_user_titles()
top_users = self.user_analyzer.get_top_users(
user_analysis, limit=max_user_titles
)
logger.info(
f"已为称号分析筛选出 {len(top_users)} 名活跃用户 (最大限制: {max_user_titles})"
)
# 4. LLM 语义分析阶段
topics = []
user_titles = []
golden_quotes = []
total_token_usage = TokenUsage()
# 检查开关设置
topic_enabled = self.config_manager.get_topic_analysis_enabled()
user_title_enabled = self.config_manager.get_user_title_analysis_enabled()
golden_quote_enabled = (
self.config_manager.get_golden_quote_analysis_enabled()
)
# 策略:如果多项功能均开启,则通过 LLMAnalyzer 并发调用,显著降低分析总时长
if topic_enabled and user_title_enabled and golden_quote_enabled:
(
topics,
user_titles,
golden_quotes,
total_token_usage,
) = await self.llm_analyzer.analyze_all_concurrent(
messages, user_analysis, umo=unified_msg_origin, top_users=top_users
)
else:
# 串行降级路径:根据开关按需串行调用 (适用于 Token 敏感或单项测试)
if topic_enabled:
topics, topic_tokens = await self.llm_analyzer.analyze_topics(
messages, umo=unified_msg_origin
)
total_token_usage.prompt_tokens += topic_tokens.prompt_tokens
total_token_usage.completion_tokens += (
topic_tokens.completion_tokens
)
total_token_usage.total_tokens += topic_tokens.total_tokens
if user_title_enabled:
(
user_titles,
title_tokens,
) = await self.llm_analyzer.analyze_user_titles(
messages,
user_analysis,
umo=unified_msg_origin,
top_users=top_users,
)
total_token_usage.prompt_tokens += title_tokens.prompt_tokens
total_token_usage.completion_tokens += (
title_tokens.completion_tokens
)
total_token_usage.total_tokens += title_tokens.total_tokens
if golden_quote_enabled:
(
golden_quotes,
quote_tokens,
) = await self.llm_analyzer.analyze_golden_quotes(
messages, umo=unified_msg_origin
)
total_token_usage.prompt_tokens += quote_tokens.prompt_tokens
total_token_usage.completion_tokens += (
quote_tokens.completion_tokens
)
total_token_usage.total_tokens += quote_tokens.total_tokens
# 5. 回填分析结果并组装返回字典
statistics.golden_quotes = golden_quotes
statistics.token_usage = total_token_usage
return {
"statistics": statistics,
"topics": topics,
"user_titles": user_titles,
"user_analysis": user_analysis,
}
except Exception as e:
logger.error(f"消息分析流水线执行失败: {e}")
return None