diff --git a/main.py b/main.py index 0ab834f..b7efee4 100644 --- a/main.py +++ b/main.py @@ -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( diff --git a/src/analysis/statistics.py b/src/analysis/statistics.py deleted file mode 100644 index 27a42a4..0000000 --- a/src/analysis/statistics.py +++ /dev/null @@ -1,149 +0,0 @@ -""" -统计分析模块 -负责用户活跃度分析和其他统计功能 -""" - -import re -from collections import defaultdict -from datetime import datetime - -from .utils import InfoUtils - - -class UserAnalyzer: - """用户分析器""" - - # Discord 自定义表情正则 <:name:id> 或 - DISCORD_CUSTOM_EMOJI_PATTERN = r"" - - # 简单的 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), - } diff --git a/src/application/__init__.py b/src/application/__init__.py index 4eb7b2f..55c8c18 100644 --- a/src/application/__init__.py +++ b/src/application/__init__.py @@ -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", diff --git a/src/application/analysis_orchestrator.py b/src/application/analysis_orchestrator.py deleted file mode 100644 index ac07ed3..0000000 --- a/src/application/analysis_orchestrator.py +++ /dev/null @@ -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) diff --git a/src/application/services/analysis_application_service.py b/src/application/services/analysis_application_service.py new file mode 100644 index 0000000..fa264d9 --- /dev/null +++ b/src/application/services/analysis_application_service.py @@ -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, + } diff --git a/src/core/__init__.py b/src/core/__init__.py deleted file mode 100644 index 0999975..0000000 --- a/src/core/__init__.py +++ /dev/null @@ -1,8 +0,0 @@ -""" -核心功能模块 -""" - -from .config import ConfigManager -from .message_handler import MessageHandler - -__all__ = ["ConfigManager", "MessageHandler"] diff --git a/src/core/config.py b/src/core/config.py deleted file mode 100644 index 49365e8..0000000 --- a/src/core/config.py +++ /dev/null @@ -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}") diff --git a/src/core/message_handler.py b/src/core/message_handler.py deleted file mode 100644 index e91f294..0000000 --- a/src/core/message_handler.py +++ /dev/null @@ -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(), - ) diff --git a/src/core/message_sender.py b/src/core/message_sender.py deleted file mode 100644 index 98582c4..0000000 --- a/src/core/message_sender.py +++ /dev/null @@ -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}" - ) diff --git a/src/models/data_models.py b/src/domain/models/data_models.py similarity index 100% rename from src/models/data_models.py rename to src/domain/models/data_models.py diff --git a/src/domain/repositories/analysis_repository.py b/src/domain/repositories/analysis_repository.py new file mode 100644 index 0000000..97999a7 --- /dev/null +++ b/src/domain/repositories/analysis_repository.py @@ -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 diff --git a/src/domain/repositories/message_repository.py b/src/domain/repositories/message_repository.py index 872b4b3..c7b7019 100644 --- a/src/domain/repositories/message_repository.py +++ b/src/domain/repositories/message_repository.py @@ -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, diff --git a/src/domain/repositories/report_repository.py b/src/domain/repositories/report_repository.py new file mode 100644 index 0000000..6278e46 --- /dev/null +++ b/src/domain/repositories/report_repository.py @@ -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 diff --git a/src/domain/services/analysis_domain_service.py b/src/domain/services/analysis_domain_service.py new file mode 100644 index 0000000..8813179 --- /dev/null +++ b/src/domain/services/analysis_domain_service.py @@ -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> 或 + DISCORD_CUSTOM_EMOJI_PATTERN = r"" + + # 简单的 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), + } diff --git a/src/domain/services/statistics_service.py b/src/domain/services/statistics_service.py new file mode 100644 index 0000000..91c9c83 --- /dev/null +++ b/src/domain/services/statistics_service.py @@ -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 diff --git a/src/analysis/__init__.py b/src/infrastructure/analysis/__init__.py similarity index 100% rename from src/analysis/__init__.py rename to src/infrastructure/analysis/__init__.py diff --git a/src/analysis/analyzers/__init__.py b/src/infrastructure/analysis/analyzers/__init__.py similarity index 100% rename from src/analysis/analyzers/__init__.py rename to src/infrastructure/analysis/analyzers/__init__.py diff --git a/src/analysis/analyzers/base_analyzer.py b/src/infrastructure/analysis/analyzers/base_analyzer.py similarity index 98% rename from src/analysis/analyzers/base_analyzer.py rename to src/infrastructure/analysis/analyzers/base_analyzer.py index ae7e8c9..94843e3 100644 --- a/src/analysis/analyzers/base_analyzer.py +++ b/src/infrastructure/analysis/analyzers/base_analyzer.py @@ -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, diff --git a/src/analysis/analyzers/golden_quote_analyzer.py b/src/infrastructure/analysis/analyzers/golden_quote_analyzer.py similarity index 98% rename from src/analysis/analyzers/golden_quote_analyzer.py rename to src/infrastructure/analysis/analyzers/golden_quote_analyzer.py index afb597c..749f19c 100644 --- a/src/analysis/analyzers/golden_quote_analyzer.py +++ b/src/infrastructure/analysis/analyzers/golden_quote_analyzer.py @@ -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 diff --git a/src/analysis/analyzers/topic_analyzer.py b/src/infrastructure/analysis/analyzers/topic_analyzer.py similarity index 99% rename from src/analysis/analyzers/topic_analyzer.py rename to src/infrastructure/analysis/analyzers/topic_analyzer.py index 17952c4..badf206 100644 --- a/src/analysis/analyzers/topic_analyzer.py +++ b/src/infrastructure/analysis/analyzers/topic_analyzer.py @@ -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 diff --git a/src/analysis/analyzers/user_title_analyzer.py b/src/infrastructure/analysis/analyzers/user_title_analyzer.py similarity index 98% rename from src/analysis/analyzers/user_title_analyzer.py rename to src/infrastructure/analysis/analyzers/user_title_analyzer.py index 751f2c3..d6ae0d1 100644 --- a/src/analysis/analyzers/user_title_analyzer.py +++ b/src/infrastructure/analysis/analyzers/user_title_analyzer.py @@ -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 diff --git a/src/analysis/llm_analyzer.py b/src/infrastructure/analysis/llm_analyzer.py similarity index 98% rename from src/analysis/llm_analyzer.py rename to src/infrastructure/analysis/llm_analyzer.py index 1a3236f..e9a33d9 100644 --- a/src/analysis/llm_analyzer.py +++ b/src/infrastructure/analysis/llm_analyzer.py @@ -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 diff --git a/src/analysis/utils/__init__.py b/src/infrastructure/analysis/utils/__init__.py similarity index 100% rename from src/analysis/utils/__init__.py rename to src/infrastructure/analysis/utils/__init__.py diff --git a/src/analysis/utils/info_utils.py b/src/infrastructure/analysis/utils/info_utils.py similarity index 100% rename from src/analysis/utils/info_utils.py rename to src/infrastructure/analysis/utils/info_utils.py diff --git a/src/analysis/utils/json_utils.py b/src/infrastructure/analysis/utils/json_utils.py similarity index 100% rename from src/analysis/utils/json_utils.py rename to src/infrastructure/analysis/utils/json_utils.py diff --git a/src/analysis/utils/llm_utils.py b/src/infrastructure/analysis/utils/llm_utils.py similarity index 100% rename from src/analysis/utils/llm_utils.py rename to src/infrastructure/analysis/utils/llm_utils.py diff --git a/src/infrastructure/config/config_manager.py b/src/infrastructure/config/config_manager.py index 23a2e63..e34508b 100644 --- a/src/infrastructure/config/config_manager.py +++ b/src/infrastructure/config/config_manager.py @@ -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}") diff --git a/src/core/history_manager.py b/src/infrastructure/persistence/history_manager.py similarity index 87% rename from src/core/history_manager.py rename to src/infrastructure/persistence/history_manager.py index 0dfdd4f..aba14cc 100644 --- a/src/core/history_manager.py +++ b/src/infrastructure/persistence/history_manager.py @@ -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 diff --git a/src/infrastructure/platform/adapters/discord_adapter.py b/src/infrastructure/platform/adapters/discord_adapter.py index e65a2b4..567a7d6 100644 --- a/src/infrastructure/platform/adapters/discord_adapter.py +++ b/src/infrastructure/platform/adapters/discord_adapter.py @@ -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: diff --git a/src/infrastructure/platform/adapters/onebot_adapter.py b/src/infrastructure/platform/adapters/onebot_adapter.py index 98d6265..915098b 100644 --- a/src/infrastructure/platform/adapters/onebot_adapter.py +++ b/src/infrastructure/platform/adapters/onebot_adapter.py @@ -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: diff --git a/src/core/bot_manager.py b/src/infrastructure/platform/bot_manager.py similarity index 90% rename from src/core/bot_manager.py rename to src/infrastructure/platform/bot_manager.py index bc93119..42c34d3 100644 --- a/src/core/bot_manager.py +++ b/src/infrastructure/platform/bot_manager.py @@ -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 diff --git a/src/reports/__init__.py b/src/infrastructure/reporting/__init__.py similarity index 100% rename from src/reports/__init__.py rename to src/infrastructure/reporting/__init__.py diff --git a/src/reports/dispatcher.py b/src/infrastructure/reporting/dispatcher.py similarity index 100% rename from src/reports/dispatcher.py rename to src/infrastructure/reporting/dispatcher.py diff --git a/src/reports/generators.py b/src/infrastructure/reporting/generators.py similarity index 99% rename from src/reports/generators.py rename to src/infrastructure/reporting/generators.py index df62c28..70f185d 100644 --- a/src/reports/generators.py +++ b/src/infrastructure/reporting/generators.py @@ -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): diff --git a/src/reports/templates.py b/src/infrastructure/reporting/templates.py similarity index 99% rename from src/reports/templates.py rename to src/infrastructure/reporting/templates.py index e6ba93e..d277ee5 100644 --- a/src/reports/templates.py +++ b/src/infrastructure/reporting/templates.py @@ -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: diff --git a/src/reports/templates/format/activity_chart.html b/src/infrastructure/reporting/templates/format/activity_chart.html similarity index 100% rename from src/reports/templates/format/activity_chart.html rename to src/infrastructure/reporting/templates/format/activity_chart.html diff --git a/src/reports/templates/format/image_template.html b/src/infrastructure/reporting/templates/format/image_template.html similarity index 100% rename from src/reports/templates/format/image_template.html rename to src/infrastructure/reporting/templates/format/image_template.html diff --git a/src/reports/templates/format/pdf_template.html b/src/infrastructure/reporting/templates/format/pdf_template.html similarity index 100% rename from src/reports/templates/format/pdf_template.html rename to src/infrastructure/reporting/templates/format/pdf_template.html diff --git a/src/reports/templates/format/quote_item.html b/src/infrastructure/reporting/templates/format/quote_item.html similarity index 100% rename from src/reports/templates/format/quote_item.html rename to src/infrastructure/reporting/templates/format/quote_item.html diff --git a/src/reports/templates/format/topic_item.html b/src/infrastructure/reporting/templates/format/topic_item.html similarity index 100% rename from src/reports/templates/format/topic_item.html rename to src/infrastructure/reporting/templates/format/topic_item.html diff --git a/src/reports/templates/format/user_title_item.html b/src/infrastructure/reporting/templates/format/user_title_item.html similarity index 100% rename from src/reports/templates/format/user_title_item.html rename to src/infrastructure/reporting/templates/format/user_title_item.html diff --git a/src/reports/templates/retro_futurism/activity_chart.html b/src/infrastructure/reporting/templates/retro_futurism/activity_chart.html similarity index 100% rename from src/reports/templates/retro_futurism/activity_chart.html rename to src/infrastructure/reporting/templates/retro_futurism/activity_chart.html diff --git a/src/reports/templates/retro_futurism/image_template.html b/src/infrastructure/reporting/templates/retro_futurism/image_template.html similarity index 100% rename from src/reports/templates/retro_futurism/image_template.html rename to src/infrastructure/reporting/templates/retro_futurism/image_template.html diff --git a/src/reports/templates/retro_futurism/pdf_template.html b/src/infrastructure/reporting/templates/retro_futurism/pdf_template.html similarity index 100% rename from src/reports/templates/retro_futurism/pdf_template.html rename to src/infrastructure/reporting/templates/retro_futurism/pdf_template.html diff --git a/src/reports/templates/retro_futurism/quote_item.html b/src/infrastructure/reporting/templates/retro_futurism/quote_item.html similarity index 100% rename from src/reports/templates/retro_futurism/quote_item.html rename to src/infrastructure/reporting/templates/retro_futurism/quote_item.html diff --git a/src/reports/templates/retro_futurism/topic_item.html b/src/infrastructure/reporting/templates/retro_futurism/topic_item.html similarity index 100% rename from src/reports/templates/retro_futurism/topic_item.html rename to src/infrastructure/reporting/templates/retro_futurism/topic_item.html diff --git a/src/reports/templates/retro_futurism/user_title_item.html b/src/infrastructure/reporting/templates/retro_futurism/user_title_item.html similarity index 100% rename from src/reports/templates/retro_futurism/user_title_item.html rename to src/infrastructure/reporting/templates/retro_futurism/user_title_item.html diff --git a/src/reports/templates/scrapbook/activity_chart.html b/src/infrastructure/reporting/templates/scrapbook/activity_chart.html similarity index 100% rename from src/reports/templates/scrapbook/activity_chart.html rename to src/infrastructure/reporting/templates/scrapbook/activity_chart.html diff --git a/src/reports/templates/scrapbook/activity_chart_pdf.html b/src/infrastructure/reporting/templates/scrapbook/activity_chart_pdf.html similarity index 100% rename from src/reports/templates/scrapbook/activity_chart_pdf.html rename to src/infrastructure/reporting/templates/scrapbook/activity_chart_pdf.html diff --git a/src/reports/templates/scrapbook/image_template.html b/src/infrastructure/reporting/templates/scrapbook/image_template.html similarity index 100% rename from src/reports/templates/scrapbook/image_template.html rename to src/infrastructure/reporting/templates/scrapbook/image_template.html diff --git a/src/reports/templates/scrapbook/pdf_template.html b/src/infrastructure/reporting/templates/scrapbook/pdf_template.html similarity index 100% rename from src/reports/templates/scrapbook/pdf_template.html rename to src/infrastructure/reporting/templates/scrapbook/pdf_template.html diff --git a/src/reports/templates/scrapbook/quote_item.html b/src/infrastructure/reporting/templates/scrapbook/quote_item.html similarity index 100% rename from src/reports/templates/scrapbook/quote_item.html rename to src/infrastructure/reporting/templates/scrapbook/quote_item.html diff --git a/src/reports/templates/scrapbook/topic_item.html b/src/infrastructure/reporting/templates/scrapbook/topic_item.html similarity index 100% rename from src/reports/templates/scrapbook/topic_item.html rename to src/infrastructure/reporting/templates/scrapbook/topic_item.html diff --git a/src/reports/templates/scrapbook/user_title_item.html b/src/infrastructure/reporting/templates/scrapbook/user_title_item.html similarity index 100% rename from src/reports/templates/scrapbook/user_title_item.html rename to src/infrastructure/reporting/templates/scrapbook/user_title_item.html diff --git a/src/reports/templates/simple/activity_chart.html b/src/infrastructure/reporting/templates/simple/activity_chart.html similarity index 100% rename from src/reports/templates/simple/activity_chart.html rename to src/infrastructure/reporting/templates/simple/activity_chart.html diff --git a/src/reports/templates/simple/image_template.html b/src/infrastructure/reporting/templates/simple/image_template.html similarity index 100% rename from src/reports/templates/simple/image_template.html rename to src/infrastructure/reporting/templates/simple/image_template.html diff --git a/src/reports/templates/simple/pdf_template.html b/src/infrastructure/reporting/templates/simple/pdf_template.html similarity index 100% rename from src/reports/templates/simple/pdf_template.html rename to src/infrastructure/reporting/templates/simple/pdf_template.html diff --git a/src/reports/templates/simple/quote_item.html b/src/infrastructure/reporting/templates/simple/quote_item.html similarity index 100% rename from src/reports/templates/simple/quote_item.html rename to src/infrastructure/reporting/templates/simple/quote_item.html diff --git a/src/reports/templates/simple/topic_item.html b/src/infrastructure/reporting/templates/simple/topic_item.html similarity index 100% rename from src/reports/templates/simple/topic_item.html rename to src/infrastructure/reporting/templates/simple/topic_item.html diff --git a/src/reports/templates/simple/user_title_item.html b/src/infrastructure/reporting/templates/simple/user_title_item.html similarity index 100% rename from src/reports/templates/simple/user_title_item.html rename to src/infrastructure/reporting/templates/simple/user_title_item.html diff --git a/src/scheduler/__init__.py b/src/infrastructure/scheduler/__init__.py similarity index 100% rename from src/scheduler/__init__.py rename to src/infrastructure/scheduler/__init__.py diff --git a/src/scheduler/auto_scheduler.py b/src/infrastructure/scheduler/auto_scheduler.py similarity index 58% rename from src/scheduler/auto_scheduler.py rename to src/infrastructure/scheduler/auto_scheduler.py index 2e63980..c10514a 100644 --- a/src/scheduler/auto_scheduler.py +++ b/src/infrastructure/scheduler/auto_scheduler.py @@ -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]]: """ diff --git a/src/scheduler/retry.py b/src/infrastructure/scheduler/retry.py similarity index 64% rename from src/scheduler/retry.py rename to src/infrastructure/scheduler/retry.py index 672ed26..bcd1816 100644 --- a/src/scheduler/retry.py +++ b/src/infrastructure/scheduler/retry.py @@ -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) diff --git a/src/visualization/__init__.py b/src/infrastructure/visualization/__init__.py similarity index 100% rename from src/visualization/__init__.py rename to src/infrastructure/visualization/__init__.py diff --git a/src/visualization/activity_charts.py b/src/infrastructure/visualization/activity_charts.py similarity index 98% rename from src/visualization/activity_charts.py rename to src/infrastructure/visualization/activity_charts.py index 168544c..11a9fae 100644 --- a/src/visualization/activity_charts.py +++ b/src/infrastructure/visualization/activity_charts.py @@ -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: diff --git a/src/models/__init__.py b/src/models/__init__.py deleted file mode 100644 index bc05798..0000000 --- a/src/models/__init__.py +++ /dev/null @@ -1,13 +0,0 @@ -""" -数据模型模块 -""" - -from .data_models import ( - GoldenQuote, - GroupStatistics, - SummaryTopic, - TokenUsage, - UserTitle, -) - -__all__ = ["SummaryTopic", "UserTitle", "GoldenQuote", "TokenUsage", "GroupStatistics"] diff --git a/src/utils/helpers.py b/src/utils/helpers.py deleted file mode 100644 index 12cec9e..0000000 --- a/src/utils/helpers.py +++ /dev/null @@ -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