diff --git a/README.md b/README.md index b568892..ec6cb75 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ # QQ群日常分析插件 -[![Plugin Version](https://img.shields.io/badge/Latest_Version-v1.6.0-blue.svg?style=for-the-badge&color=76bad9)](https://github.com/SXP-Simon/astrbot-qq-group-daily-analysis) +[![Plugin Version](https://img.shields.io/badge/Latest_Version-v1.7.0-blue.svg?style=for-the-badge&color=76bad9)](https://github.com/SXP-Simon/astrbot-qq-group-daily-analysis) [![AstrBot](https://img.shields.io/badge/AstrBot-Plugin-ff69b4?style=for-the-badge)](https://github.com/AstrBotDevs/AstrBot) [![License](https://img.shields.io/badge/License-MIT-green.svg?style=for-the-badge)](LICENSE) @@ -137,6 +137,10 @@ _✨ 一个基于AstrBot的智能群聊分析插件，能够生成精美的群 ### v1.4.0 - PDF 版本情况说明更新 +### v1.7.0 +- 修复提示 +- 解耦化 + ## 许可证 MIT License diff --git a/main.py b/main.py index c5893e7..1c6452b 100644 --- a/main.py +++ b/main.py @@ -1,142 +1,34 @@ -""" +""" QQ群日常分析插件 基于群聊记录生成精美的日常分析报告，包含话题总结、用户画像、统计数据等 + +重构版本 - 使用模块化架构 """ -import json import asyncio -import base64 -import aiohttp -import subprocess -import sys -import os -from datetime import datetime, timedelta -from typing import List, Dict, Optional +from typing import Optional from pathlib import Path -from dataclasses import dataclass, field -from collections import defaultdict from astrbot.api.event import filter -from astrbot.api.star import Context, Star, register +from astrbot.api.star import Context, Star from astrbot.api import logger, AstrBotConfig from astrbot.core.platform.sources.aiocqhttp.aiocqhttp_message_event import AiocqhttpMessageEvent from astrbot.core.message.components import File from astrbot.core.star.filter.permission import PermissionType -# PDF 生成相关导入 -PYPPETEER_AVAILABLE = False -PYPPETEER_VERSION = None - -def check_pyppeteer_availability(): - """检查 pyppeteer 可用性""" - global PYPPETEER_AVAILABLE, PYPPETEER_VERSION - try: - import pyppeteer - from pyppeteer import launch - PYPPETEER_AVAILABLE = True - - # 检查版本 - try: - PYPPETEER_VERSION = pyppeteer.__version__ - logger.info(f"使用 pyppeteer {PYPPETEER_VERSION} 作为 PDF 引擎") - except AttributeError: - PYPPETEER_VERSION = "unknown" - logger.info("使用 pyppeteer (版本未知) 作为 PDF 引擎") - - return True - except ImportError: - PYPPETEER_AVAILABLE = False - PYPPETEER_VERSION = None - logger.warning("pyppeteer 未安装，PDF 功能将不可用。请使用 /安装PDF 命令安装 pyppeteer==1.0.2") - return False - -def reload_pyppeteer(): - """重新加载 pyppeteer 模块""" - global PYPPETEER_AVAILABLE, PYPPETEER_VERSION - import sys - import importlib - - try: - logger.info("开始重新加载 pyppeteer 模块...") - - # 移除所有 pyppeteer 相关模块 - modules_to_remove = [mod for mod in sys.modules.keys() if mod.startswith('pyppeteer')] - logger.info(f"移除模块: {modules_to_remove}") - for mod in modules_to_remove: - del sys.modules[mod] - - # 强制重新导入 - try: - import pyppeteer - from pyppeteer import launch - - # 更新全局变量 - PYPPETEER_AVAILABLE = True - try: - PYPPETEER_VERSION = pyppeteer.__version__ - logger.info(f"重新加载成功，pyppeteer 版本: {PYPPETEER_VERSION}") - except AttributeError: - PYPPETEER_VERSION = "unknown" - logger.info("重新加载成功，pyppeteer 版本未知") - - return True - - except ImportError as e: - logger.error(f"重新导入 pyppeteer 失败: {e}") - PYPPETEER_AVAILABLE = False - PYPPETEER_VERSION = None - return False - - except Exception as e: - logger.error(f"重新加载 pyppeteer 时出错: {e}") - return False - -# 初始检查 -check_pyppeteer_availability() +# 导入重构后的模块 +from .src.core.config import ConfigManager +from .src.reports.generators import ReportGenerator +from .src.scheduler.auto_scheduler import AutoScheduler +from .src.utils.pdf_utils import PDFInstaller +from .src.utils.helpers import MessageAnalyzer -@dataclass -class SummaryTopic: - """话题总结数据结构""" - topic: str - contributors: List[str] - detail: str - - -@dataclass -class UserTitle: - """用户称号数据结构""" - name: str - qq: int - title: str - mbti: str - reason: str - - -@dataclass -class GoldenQuote: - """群聊金句数据结构""" - content: str - sender: str - reason: str - -@dataclass -class TokenUsage: - """Token使用统计""" - prompt_tokens: int = 0 - completion_tokens: int = 0 - total_tokens: int = 0 - -@dataclass -class GroupStatistics: - """群聊统计数据结构""" - message_count: int - total_characters: int - participant_count: int - most_active_period: str - golden_quotes: List[GoldenQuote] - emoji_count: int - token_usage: TokenUsage = field(default_factory=TokenUsage) +# 全局变量 +config_manager = None +message_analyzer = None +report_generator = None +auto_scheduler = None class QQGroupDailyAnalysis(Star): @@ -144,37 +36,84 @@ class QQGroupDailyAnalysis(Star): super().__init__(context) self.config = config - # 直接从AstrBot配置系统读取配置 - self.enabled_groups = config.get("enabled_groups", []) - self.max_messages = config.get("max_messages", 1000) - self.analysis_days = config.get("analysis_days", 1) - self.auto_analysis_time = config.get("auto_analysis_time", "09:00") - self.enable_auto_analysis = config.get("enable_auto_analysis", False) - self.output_format = config.get("output_format", "image") + # 初始化模块化组件 + global config_manager, message_analyzer, report_generator, auto_scheduler - self.min_messages_threshold = config.get("min_messages_threshold", 50) - self.topic_analysis_enabled = config.get("topic_analysis_enabled", True) - self.user_title_analysis_enabled = config.get("user_title_analysis_enabled", True) - self.max_topics = config.get("max_topics", 5) - self.max_user_titles = config.get("max_user_titles", 8) - self.max_golden_quotes = config.get("max_golden_quotes", 5) - self.max_query_rounds = config.get("max_query_rounds", 35) + config_manager = ConfigManager(config) + message_analyzer = MessageAnalyzer(context, config_manager) + report_generator = ReportGenerator(config_manager) + auto_scheduler = AutoScheduler( + config_manager, + message_analyzer.message_handler, + message_analyzer, + report_generator, + self.html_render # 传入html_render函数 + ) - # PDF 相关配置 - self.pdf_output_dir = config.get("pdf_output_dir", "data/plugins/astrbot-qq-group-daily-analysis/reports") - - self.pdf_filename_format = config.get("pdf_filename_format", "群聊分析报告_{group_id}_{date}.pdf") - - # 启动定时任务 - self.scheduler_task = None - self.bot_instance = None # 保存bot实例用于自动分析 - self.bot_qq_id = None # 保存机器人QQ号，用于过滤机器人消息 - - # 延迟启动定时任务，给系统时间初始化 - if self.enable_auto_analysis: + # 延迟启动自动调度器，给系统时间初始化 + if config_manager.get_enable_auto_analysis(): asyncio.create_task(self._delayed_start_scheduler()) - - logger.info("QQ群日常分析插件已初始化") + + logger.info("QQ群日常分析插件已初始化（模块化版本）") + + async def _delayed_start_scheduler(self): + """延迟启动调度器，给系统时间初始化""" + try: + # 等待10秒让系统完全初始化 + await asyncio.sleep(10) + + # 尝试获取bot实例 + bot_instance = await self._get_bot_instance() + if bot_instance: + auto_scheduler.set_bot_instance(bot_instance) + logger.info("已为自动调度器设置bot实例") + else: + logger.info("暂时未获取到bot实例，定时任务仍会启动") + + # 启动调度器 + await auto_scheduler.start_scheduler() + + except Exception as e: + logger.error(f"延迟启动调度器失败: {e}") + + async def _get_bot_instance(self): + """从Context获取bot实例""" + try: + # 简化的获取逻辑，尝试常见的几种方式 + if hasattr(self.context, 'get_platforms') and callable(self.context.get_platforms): + platforms = self.context.get_platforms() + for platform in platforms: + if hasattr(platform, 'bot') and platform.bot: + logger.info(f"从平台获取到bot实例") + return platform.bot + + # 尝试从context的platforms属性获取 + if hasattr(self.context, 'platforms') and self.context.platforms: + for platform in self.context.platforms: + if hasattr(platform, 'bot') and platform.bot: + logger.info(f"从平台列表获取到bot实例") + return platform.bot + + logger.info("暂时无法获取bot实例") + return None + + except Exception as e: + logger.error(f"获取bot实例失败: {e}") + return None + + async def _reload_config_and_restart_scheduler(self): + """重新加载配置并重启调度器""" + try: + # 重新加载配置 + config_manager.reload_config() + logger.info(f"重新加载配置: 自动分析={config_manager.get_enable_auto_analysis()}") + + # 重启调度器 + await auto_scheduler.restart_scheduler() + logger.info("配置重载和调度器重启完成") + + except Exception as e: + logger.error(f"重新加载配置失败: {e}") @filter.command("群分析") @filter.permission_type(PermissionType.ADMIN) @@ -186,49 +125,47 @@ class QQGroupDailyAnalysis(Star): if not isinstance(event, AiocqhttpMessageEvent): yield event.plain_result("❌ 此功能仅支持QQ群聊") return - + group_id = event.get_group_id() if not group_id: yield event.plain_result("❌ 请在群聊中使用此命令") return - # 保存bot实例用于自动分析 - self.bot_instance = event.bot - - # 获取机器人QQ号 - await self._get_bot_qq_id() - + # 设置bot实例 + auto_scheduler.set_bot_instance(event.bot) + await message_analyzer.set_bot_instance(event.bot) + # 检查群组权限 - if self.enabled_groups and group_id not in self.enabled_groups: + enabled_groups = config_manager.get_enabled_groups() + if enabled_groups and group_id not in enabled_groups: yield event.plain_result("❌ 此群未启用日常分析功能") return - - # 设置分析天数 - analysis_days = days if days and 1 <= days <= 7 else self.analysis_days - + analysis_days = days if days and 1 <= days <= 7 else config_manager.get_analysis_days() + yield event.plain_result(f"🔍 开始分析群聊近{analysis_days}天的活动，请稍候...") # 调试：输出当前配置 - logger.info(f"当前输出格式配置: {self.output_format}") + logger.info(f"当前输出格式配置: {config_manager.get_output_format()}") try: # 获取群聊消息 - messages = await self._fetch_group_messages(event, analysis_days) + messages = await message_analyzer.message_handler.fetch_group_messages(event.bot, group_id, analysis_days) if not messages: yield event.plain_result("❌ 未找到足够的群聊记录，请确保群内有足够的消息历史") return # 检查消息数量是否足够分析 - if len(messages) < self.min_messages_threshold: - yield event.plain_result(f"❌ 消息数量不足（{len(messages)}条），至少需要{self.min_messages_threshold}条消息才能进行有效分析") + min_threshold = config_manager.get_min_messages_threshold() + if len(messages) < min_threshold: + yield event.plain_result(f"❌ 消息数量不足（{len(messages)}条），至少需要{min_threshold}条消息才能进行有效分析") return yield event.plain_result(f"📊 已获取{len(messages)}条消息，正在进行智能分析...") # 进行分析 - analysis_result = await self._analyze_messages(messages, group_id) + analysis_result = await message_analyzer.analyze_messages(messages, group_id) # 检查分析结果 if not analysis_result or not analysis_result.get("statistics"): @@ -236,31 +173,29 @@ class QQGroupDailyAnalysis(Star): return # 生成报告 - if self.output_format == "image": - image_url = await self._generate_image_report(analysis_result, group_id) + output_format = config_manager.get_output_format() + if output_format == "image": + image_url = await report_generator.generate_image_report(analysis_result, group_id, self.html_render) if image_url: yield event.image_result(image_url) else: # 如果图片生成失败，回退到文本报告 logger.warning("图片报告生成失败，回退到文本报告") - text_report = await self._generate_text_report(analysis_result) + text_report = report_generator.generate_text_report(analysis_result) yield event.plain_result(f"⚠️ 图片报告生成失败，以下是文本版本：\n\n{text_report}") - elif self.output_format == "pdf": - if not PYPPETEER_AVAILABLE: + elif output_format == "pdf": + if not config_manager.pyppeteer_available: yield event.plain_result("❌ PDF 功能不可用，请使用 /安装PDF 命令安装 pyppeteer==1.0.2") return - # yield event.plain_result("📄 正在生成 PDF 报告，请稍候...") - # yield event.plain_result("💡 首次使用可能需要下载 Chromium 浏览器，请耐心等待...") - - pdf_path = await self._generate_pdf_report(analysis_result, group_id) + pdf_path = await report_generator.generate_pdf_report(analysis_result, group_id) if pdf_path: # 发送 PDF 文件 + 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.plain_result(f"✅ PDF 报告已生成并发送") else: # 如果 PDF 生成失败，提供详细的错误信息和解决方案 yield event.plain_result("❌ PDF 报告生成失败") @@ -271,10 +206,10 @@ class QQGroupDailyAnalysis(Star): # 回退到文本报告 logger.warning("PDF 报告生成失败，回退到文本报告") - text_report = await self._generate_text_report(analysis_result) + text_report = report_generator.generate_text_report(analysis_result) yield event.plain_result(f"\n📝 以下是文本版本的分析报告：\n\n{text_report}") else: - text_report = await self._generate_text_report(analysis_result) + text_report = report_generator.generate_text_report(analysis_result) yield event.plain_result(text_report) except Exception as e: @@ -298,12 +233,14 @@ class QQGroupDailyAnalysis(Star): return if not format_type: - yield event.plain_result(f"""📊 当前输出格式: {self.output_format} + current_format = config_manager.get_output_format() + pdf_status = '✅' if config_manager.pyppeteer_available else '❌ (需安装 pyppeteer)' + yield event.plain_result(f"""📊 当前输出格式: {current_format} 可用格式: • image - 图片格式 (默认) • text - 文本格式 -• pdf - PDF 格式 {'✅' if PYPPETEER_AVAILABLE else '❌ (需安装 pyppeteer)'} +• pdf - PDF 格式 {pdf_status} 用法: /设置格式 [格式名称]""") return @@ -313,13 +250,11 @@ class QQGroupDailyAnalysis(Star): yield event.plain_result("❌ 无效的格式类型，支持: image, text, pdf") return - if format_type == "pdf" and not PYPPETEER_AVAILABLE: + if format_type == "pdf" and not config_manager.pyppeteer_available: yield event.plain_result("❌ PDF 格式不可用，请使用 /安装PDF 命令安装 pyppeteer==1.0.2") return - self.output_format = format_type - self.config["output_format"] = format_type - self.config.save_config() + config_manager.set_output_format(format_type) yield event.plain_result(f"✅ 输出格式已设置为: {format_type}") @filter.command("安装PDF") @@ -329,8 +264,6 @@ class QQGroupDailyAnalysis(Star): 安装 PDF 功能依赖 用法: /安装PDF """ - global PYPPETEER_AVAILABLE - if not isinstance(event, AiocqhttpMessageEvent): yield event.plain_result("❌ 此功能仅支持QQ群聊") return @@ -338,62 +271,9 @@ class QQGroupDailyAnalysis(Star): yield event.plain_result("🔄 开始安装 PDF 功能依赖，请稍候...") try: - # 检查是否已安装 - if PYPPETEER_AVAILABLE: - yield event.plain_result("✅ pyppeteer 已安装，正在检查 Chromium...") - - # 检查 Chromium - try: - import pyppeteer - # 尝试获取 Chromium 路径 - try: - chromium_path = pyppeteer.executablePath() - if Path(chromium_path).exists(): - yield event.plain_result("✅ PDF 功能已完全可用！") - return - except Exception: - # executablePath() 可能失败，说明 Chromium 未安装 - pass - - yield event.plain_result("🔄 Chromium 未安装，正在下载...") - success = await self._install_chromium() - if success: - yield event.plain_result("✅ PDF 功能安装完成！") - else: - yield event.plain_result("❌ Chromium 安装失败，请检查网络连接。\n💡 可尝试手动安装：在 Python 中运行 'import pyppeteer; await pyppeteer.launch()'") - return - except Exception as e: - yield event.plain_result(f"⚠️ 检查 Chromium 时出错: {e}") - - # 尝试安装更新版本的 pyppeteer - yield event.plain_result("📦 正在安装/更新 pyppeteer 库...") - - # 强制安装稳定版本的 pyppeteer (1.0.2) - yield event.plain_result("🔄 强制安装 pyppeteer 稳定版本 (1.0.2)...") - yield event.plain_result("� 使用 1.0.2 版本可避免 Chromium 下载问题") - success = await self._install_package("pyppeteer==1.0.2") - - if not success: - yield event.plain_result("❌ pyppeteer 安装失败") - yield event.plain_result("🔧 请尝试手动安装稳定版本：") - yield event.plain_result(" pip install pyppeteer==1.0.2") - yield event.plain_result("💡 如果仍然失败，请检查网络连接或使用代理") - return - - yield event.plain_result("✅ pyppeteer 安装成功！") - - # 重新检查可用性 - reload_success = reload_pyppeteer() - - if reload_success: - yield event.plain_result("🎉 PDF 功能安装完成！") - yield event.plain_result(f"✅ pyppeteer {PYPPETEER_VERSION} 已成功加载") - yield event.plain_result("💡 现在可以使用 /设置格式 pdf 启用 PDF 报告") - yield event.plain_result("📝 注意：首次生成 PDF 时会自动下载 Chromium") - else: - yield event.plain_result("⚠️ pyppeteer 重新加载失败") - yield event.plain_result("🔄 需要你手动去重启一次 AstrBot 程序") - yield event.plain_result("💡 pyppeteer 包已安装，但需要重启程序才能生效") + # 使用模块化的PDF安装器 + result = await PDFInstaller.install_pyppeteer(config_manager) + yield event.plain_result(result) except Exception as e: logger.error(f"安装 PDF 依赖失败: {e}", exc_info=True) @@ -414,2316 +294,79 @@ class QQGroupDailyAnalysis(Star): if not isinstance(event, AiocqhttpMessageEvent): yield event.plain_result("❌ 此功能仅支持QQ群聊") return - + group_id = event.get_group_id() if not group_id: yield event.plain_result("❌ 请在群聊中使用此命令") return - + if action == "enable": - if group_id not in self.enabled_groups: - self.enabled_groups.append(group_id) - self.config["enabled_groups"] = self.enabled_groups - self.config.save_config() + enabled_groups = config_manager.get_enabled_groups() + if group_id not in enabled_groups: + config_manager.add_enabled_group(group_id) yield event.plain_result("✅ 已为当前群启用日常分析功能") - # 重新加载配置并启动定时任务 - await self._reload_config_and_restart_scheduler() + # 重新启动定时任务 + await auto_scheduler.restart_scheduler() else: yield event.plain_result("ℹ️ 当前群已启用日常分析功能") elif action == "disable": - if group_id in self.enabled_groups: - self.enabled_groups.remove(group_id) - self.config["enabled_groups"] = self.enabled_groups - self.config.save_config() + enabled_groups = config_manager.get_enabled_groups() + if group_id in enabled_groups: + config_manager.remove_enabled_group(group_id) yield event.plain_result("✅ 已为当前群禁用日常分析功能") + + # 重新启动定时任务 + await auto_scheduler.restart_scheduler() else: yield event.plain_result("ℹ️ 当前群未启用日常分析功能") elif action == "reload": - # 重新加载配置 - await self._reload_config_and_restart_scheduler() + # 重新启动定时任务 + await auto_scheduler.restart_scheduler() yield event.plain_result("✅ 已重新加载配置并重启定时任务") elif action == "test": # 测试自动分析功能 - if group_id not in self.enabled_groups: + enabled_groups = config_manager.get_enabled_groups() + if group_id not in enabled_groups: yield event.plain_result("❌ 请先启用当前群的分析功能") return yield event.plain_result("🧪 开始测试自动分析功能...") - # 保存bot实例 - self.bot_instance = event.bot + # 设置bot实例 + auto_scheduler.set_bot_instance(event.bot) # 执行自动分析 try: - await self._perform_auto_analysis_for_group(group_id) + await auto_scheduler._perform_auto_analysis_for_group(group_id) yield event.plain_result("✅ 自动分析测试完成，请查看群消息") except Exception as e: yield event.plain_result(f"❌ 自动分析测试失败: {str(e)}") else: # status - status = "已启用" if group_id in self.enabled_groups else "未启用" - auto_status = "已启用" if self.enable_auto_analysis else "未启用" - scheduler_status = "运行中" if hasattr(self, 'scheduler_task') and self.scheduler_task and not self.scheduler_task.done() else "未运行" + enabled_groups = config_manager.get_enabled_groups() + status = "已启用" if group_id in enabled_groups else "未启用" + auto_status = "已启用" if config_manager.get_enable_auto_analysis() else "未启用" + auto_time = config_manager.get_auto_analysis_time() + + pdf_status = PDFInstaller.get_pdf_status(config_manager) + output_format = config_manager.get_output_format() + min_threshold = config_manager.get_min_messages_threshold() + max_rounds = config_manager.get_max_query_rounds() - if PYPPETEER_AVAILABLE: - pdf_status = f"可用 (pyppeteer {PYPPETEER_VERSION})" - else: - pdf_status = "不可用 (使用 /安装PDF 命令安装)" yield event.plain_result(f"""📊 当前群分析功能状态: • 群分析功能: {status} -• 自动分析: {auto_status} ({self.auto_analysis_time}) -• 定时任务: {scheduler_status} -• 输出格式: {self.output_format} +• 自动分析: {auto_status} ({auto_time}) +• 输出格式: {output_format} • PDF 功能: {pdf_status} -• 最小消息数: {self.min_messages_threshold} -• 最大查询轮数: {self.max_query_rounds} +• 最小消息数: {min_threshold} +• 最大查询轮数: {max_rounds} 💡 可用命令: enable, disable, status, reload, test 💡 支持的输出格式: image, text, pdf 💡 其他命令: /设置格式, /安装PDF""") - async def _get_bot_qq_id(self): - """获取机器人QQ号""" - try: - if self.bot_instance and not self.bot_qq_id: - login_info = await self.bot_instance.api.call_action("get_login_info") - self.bot_qq_id = str(login_info.get("user_id", "")) - logger.info(f"获取到机器人QQ号: {self.bot_qq_id}") - except Exception as e: - logger.error(f"获取机器人QQ号失败: {e}") - - async def _install_package(self, package_name: str) -> bool: - """安装 Python 包""" - try: - process = await asyncio.create_subprocess_exec( - sys.executable, "-m", "pip", "install", package_name, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE - ) - - stdout, stderr = await process.communicate() - - if process.returncode == 0: - logger.info(f"成功安装包: {package_name}") - return True - else: - logger.error(f"安装包 {package_name} 失败: {stderr.decode()}") - return False - - except Exception as e: - logger.error(f"安装包 {package_name} 时出错: {e}") - return False - - async def _install_chromium(self) -> bool: - """安装 Chromium 浏览器""" - try: - # 确保 pyppeteer 可用 - if not PYPPETEER_AVAILABLE: - logger.error("pyppeteer 不可用，无法安装 Chromium") - return False - - # 尝试直接启动浏览器，这会触发自动下载 - logger.info("尝试通过启动浏览器来触发 Chromium 下载") - - import pyppeteer - from pyppeteer import launch - browser = await launch(headless=True, args=['--no-sandbox']) - await browser.close() - - logger.info("成功安装并测试 Chromium") - return True - - except Exception as e: - logger.error(f"通过启动浏览器安装 Chromium 失败: {e}") - - # 备用方法：尝试命令行安装 - try: - logger.info("尝试命令行安装方法") - process = await asyncio.create_subprocess_exec( - sys.executable, "-c", - "import pyppeteer; import asyncio; asyncio.run(pyppeteer.launch())", - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE - ) - - stdout, stderr = await process.communicate() - - if process.returncode == 0: - logger.info("成功通过命令行安装 Chromium") - return True - else: - logger.error(f"命令行安装失败: {stderr.decode()}") - return False - - except Exception as e2: - logger.error(f"命令行安装 Chromium 时出错: {e2}") - return False - - async def _fetch_group_messages_unified(self, client, group_id: str, days: int) -> List[Dict]: - """统一的群聊消息获取方法""" - try: - if not client or not group_id: - logger.error(f"群 {group_id} 无效的客户端或群组ID") - return [] - - # 计算时间范围 - end_time = datetime.now() - start_time = end_time - timedelta(days=days) - - messages = [] - message_seq = 0 - query_rounds = 0 - max_rounds = self.max_query_rounds # 从配置读取最大查询轮数 - consecutive_failures = 0 - max_failures = 3 # 最大连续失败次数 - - 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')}") - - while len(messages) < self.max_messages and query_rounds < max_rounds: - try: - payloads = { - "group_id": group_id, - "message_seq": message_seq, - "count": 200, - "reverseOrder": True, - } - - result = await client.api.call_action("get_group_msg_history", **payloads) - - if not result or "messages" not in result: - logger.warning(f"群 {group_id} API返回无效结果: {result}") - consecutive_failures += 1 - if consecutive_failures >= max_failures: - break - continue - - round_messages = result.get("messages", []) - - if not round_messages: - logger.info(f"群 {group_id} 没有更多消息，结束获取") - break - - # 重置失败计数 - consecutive_failures = 0 - - # 过滤时间范围内的消息 - valid_messages_in_round = 0 - oldest_msg_time = None - - for msg in round_messages: - try: - msg_time = datetime.fromtimestamp(msg.get("time", 0)) - oldest_msg_time = msg_time # 记录最老的消息时间 - - # 过滤掉机器人自己的消息 - sender_id = str(msg.get("sender", {}).get("user_id", "")) - if self.bot_qq_id and sender_id == self.bot_qq_id: - continue - - if msg_time >= start_time and msg_time <= end_time: - messages.append(msg) - valid_messages_in_round += 1 - except Exception as msg_error: - logger.warning(f"群 {group_id} 处理单条消息失败: {msg_error}") - continue - - # 如果最老的消息时间已经超出范围，停止获取 - if oldest_msg_time and oldest_msg_time < start_time: - logger.info(f"群 {group_id} 已获取到时间范围外的消息，停止获取。共获取 {len(messages)} 条消息") - break - - if valid_messages_in_round == 0: - logger.warning(f"群 {group_id} 本轮未获取到有效消息") - break - - message_seq = round_messages[0]["message_id"] - query_rounds += 1 - - # 添加延迟避免请求过快 - if query_rounds % 5 == 0: - await asyncio.sleep(0.5) - - except Exception as e: - logger.error(f"群 {group_id} 获取消息失败 (第{query_rounds+1}轮): {e}") - consecutive_failures += 1 - if consecutive_failures >= max_failures: - logger.error(f"群 {group_id} 连续失败 {max_failures} 次，停止获取") - break - await asyncio.sleep(1) # 失败后等待1秒再重试 - - logger.info(f"群 {group_id} 消息获取完成，共获取 {len(messages)} 条消息，查询轮数: {query_rounds}") - return messages - - except Exception as e: - logger.error(f"群 {group_id} 获取群聊消息记录失败: {e}", exc_info=True) - return [] - - async def _fetch_group_messages(self, event: AiocqhttpMessageEvent, days: int) -> List[Dict]: - """获取群聊消息记录（手动分析）""" - return await self._fetch_group_messages_unified(event.bot, event.get_group_id(), days) - - async def _analyze_messages(self, messages: List[Dict], group_id: str) -> Dict: - """分析消息内容""" - # 基础统计 - stats = self._calculate_statistics(messages) - - # 用户活跃度分析 - user_analysis = self._analyze_users(messages) - - # 话题分析（根据配置决定是否启用） - topics = [] - topics_token_usage = TokenUsage() - if self.topic_analysis_enabled: - topics, topics_token_usage = await self._analyze_topics(messages) - - # 用户称号分析（根据配置决定是否启用） - user_titles = [] - titles_token_usage = TokenUsage() - if self.user_title_analysis_enabled: - user_titles, titles_token_usage = await self._analyze_user_titles(messages, user_analysis) - - # 群聊金句分析 - golden_quotes, quotes_token_usage = await self._analyze_golden_quotes(messages) - stats.golden_quotes = golden_quotes - - # 汇总token使用情况 - stats.token_usage.prompt_tokens = (topics_token_usage.prompt_tokens + - titles_token_usage.prompt_tokens + - quotes_token_usage.prompt_tokens) - stats.token_usage.completion_tokens = (topics_token_usage.completion_tokens + - titles_token_usage.completion_tokens + - quotes_token_usage.completion_tokens) - stats.token_usage.total_tokens = (topics_token_usage.total_tokens + - titles_token_usage.total_tokens + - quotes_token_usage.total_tokens) - - return { - "group_id": group_id, - "analysis_time": datetime.now().isoformat(), - "statistics": stats, - "user_analysis": user_analysis, - "topics": topics, - "user_titles": user_titles, - "message_count": len(messages) - } - - def _calculate_statistics(self, messages: List[Dict]) -> GroupStatistics: - """计算基础统计数据""" - total_chars = 0 - participants = set() - hour_counts = defaultdict(int) - emoji_count = 0 - - 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_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" - - return GroupStatistics( - message_count=len(messages), - total_characters=total_chars, - participant_count=len(participants), - most_active_period=most_active_period, - golden_quotes=[], # 将在后续LLM分析中填充 - emoji_count=emoji_count, - token_usage=TokenUsage() # 初始化为空，将在LLM分析后填充 - ) - - def _analyze_users(self, messages: List[Dict]) -> Dict[str, Dict]: - """分析用户活跃度""" - 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", "")) - nickname = sender.get("nickname", "") or sender.get("card", "") - - 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) - elif content.get("type") == "face": - user_stats[user_id]["emoji_count"] += 1 - elif content.get("type") == "reply": - user_stats[user_id]["reply_count"] += 1 - - return dict(user_stats) - - def _render_html_template(self, template: str, data: Dict) -> str: - """简单的 HTML 模板渲染""" - result = template - - # 调试：记录渲染数据 - logger.info(f"渲染数据键: {list(data.keys())}") - - for key, value in data.items(): - placeholder = f"{{{key}}}" # 修正：使用单大括号 - # 调试：记录替换过程 - if placeholder in result: - logger.debug(f"替换 {placeholder} -> {str(value)[:100]}...") - result = result.replace(placeholder, str(value)) - - # 检查是否还有未替换的占位符 - import re - remaining_placeholders = re.findall(r'\{[^}]+\}', result) - if remaining_placeholders: - logger.warning(f"未替换的占位符: {remaining_placeholders[:10]}") - - return result - - async def _get_user_avatar(self, user_id: str) -> Optional[str]: - """获取用户头像的base64编码""" - try: - avatar_url = f"https://q4.qlogo.cn/headimg_dl?dst_uin={user_id}&spec=640" - async with aiohttp.ClientSession() as client: - response = await client.get(avatar_url) - response.raise_for_status() - avatar_data = await response.read() - # 转换为base64编码 - avatar_base64 = base64.b64encode(avatar_data).decode('utf-8') - return f"data:image/jpeg;base64,{avatar_base64}" - except Exception as e: - logger.error(f"获取用户头像失败 {user_id}: {e}") - return None - - async def _analyze_topics(self, messages: List[Dict]) -> tuple[List[SummaryTopic], TokenUsage]: - """使用LLM分析话题""" - try: - # 提取文本消息 - text_messages = [] - for msg in messages: - sender = msg.get("sender", {}) - nickname = sender.get("nickname", "") or sender.get("card", "") - msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M") - - for content in msg.get("message", []): - if content.get("type") == "text": - text = content.get("data", {}).get("text", "").strip() - if text and len(text) > 2 and not text.startswith(("/")): # 过滤太短的消息和对机器人的命令 - text_messages.append({ - "sender": nickname, - "time": msg_time, - "content": text - }) - - if not text_messages: - return [], TokenUsage() - - # # 限制消息数量以避免token过多 - # if len(text_messages) > 100: - # # 均匀采样 - # step = len(text_messages) // 100 - # text_messages = text_messages[::step] - - # 构建LLM提示词，清理消息内容 - def clean_message_content(content): - """清理消息内容，移除可能影响JSON解析的字符""" - import re - # 替换中文引号 - content = content.replace('"', '"').replace('"', '"') - content = content.replace(''', "'").replace(''', "'") - # 移除或替换其他特殊字符 - content = content.replace('\n', ' ').replace('\r', ' ') - content = content.replace('\t', ' ') - # 移除可能的控制字符 - content = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', content) - return content.strip() - - messages_text = "\n".join([ - f"[{msg['time']}] {msg['sender']}: {clean_message_content(msg['content'])}" - for msg in text_messages - ]) - - prompt = f""" -你是一个帮我进行群聊信息总结的助手，生成总结内容时，你需要严格遵守下面的几个准则： -请分析接下来提供的群聊记录，提取出最多{self.max_topics}个主要话题。 - -对于每个话题，请提供： -1. 话题名称（突出主题内容，尽量简明扼要） -2. 主要参与者（最多5人） -3. 话题详细描述（包含关键信息和结论） - -注意： -- 对于比较有价值的点，稍微用一两句话详细讲讲，比如不要生成 “Nolan 和 SOV 讨论了 galgame 中关于性符号的衍生情况” 这种宽泛的内容，而是生成更加具体的讨论内容，让其他人只看这个消息就能知道讨论中有价值的，有营养的信息。 -- 对于其中的部分信息，你需要特意提到主题施加的主体是谁，是哪个群友做了什么事情，而不要直接生成和群友没有关系的语句。 -- 对于每一条总结，尽量讲清楚前因后果，以及话题的结论，是什么，为什么，怎么做，如果用户没有讲到细节，则可以不用这么做。 - -群聊记录： -{messages_text} - -重要：必须返回标准JSON格式，严格遵守以下规则： -1. 只使用英文双引号 " 不要使用中文引号 " " -2. 字符串内容中的引号必须转义为 \" -3. 多个对象之间用逗号分隔 -4. 数组元素之间用逗号分隔 -5. 不要在JSON外添加任何文字说明 -6. 描述内容避免使用特殊符号，用普通文字表达 - -请严格按照以下JSON格式返回，确保可以被标准JSON解析器解析： -[ - {{ - "topic": "话题名称", - "contributors": ["用户1", "用户2"], - "detail": "话题描述内容" - }}, - {{ - "topic": "另一个话题", - "contributors": ["用户3", "用户4"], - "detail": "另一个话题的描述" - }} -] - -注意：返回的内容必须是纯JSON，不要包含markdown代码块标记或其他格式 -""" - - # 调用LLM - provider = self.context.get_using_provider() - if not provider: - logger.warning("未配置LLM提供商，跳过话题分析") - return [], TokenUsage() - - response = await provider.text_chat( - prompt=prompt, - max_tokens=10000, # 增加token限制以避免响应被截断 - temperature=0.6 - ) - - # 提取token使用统计 - token_usage = TokenUsage() - if response.raw_completion and hasattr(response.raw_completion, 'usage'): - usage = response.raw_completion.usage - token_usage.prompt_tokens = usage.prompt_tokens - token_usage.completion_tokens = usage.completion_tokens - token_usage.total_tokens = usage.total_tokens - - # 解析响应 - if hasattr(response, 'completion_text'): - result_text = response.completion_text - else: - result_text = str(response) - - # 尝试解析JSON - try: - import re - # 提取JSON部分 - json_match = re.search(r'\[.*?\]', result_text, re.DOTALL) - if json_match: - json_text = json_match.group() - logger.debug(f"话题分析JSON原文: {json_text[:500]}...") - - # 强化JSON清理和修复 - def fix_json(text): - # 移除markdown代码块标记 - text = re.sub(r'```json\s*', '', text) - text = re.sub(r'```\s*$', '', text) - - # 基础清理 - text = text.replace('\n', ' ').replace('\r', ' ') - text = re.sub(r'\s+', ' ', text) - - # 替换中文引号为英文引号 - text = text.replace('"', '"').replace('"', '"') - text = text.replace(''', "'").replace(''', "'") - - # 处理字符串内容中的特殊字符 - # 转义字符串内的双引号 - def escape_quotes_in_strings(match): - content = match.group(1) - # 转义内部的双引号 - content = content.replace('"', '\\"') - return f'"{content}"' - - # 先处理字段值中的引号 - text = re.sub(r'"([^"]*(?:"[^"]*)*)"', escape_quotes_in_strings, text) - - # 修复截断的JSON - if not text.endswith(']'): - last_complete = text.rfind('}') - if last_complete > 0: - text = text[:last_complete + 1] + ']' - - # 修复常见的JSON格式问题 - # 1. 修复缺失的逗号 - text = re.sub(r'}\s*{', '}, {', text) - - # 2. 确保字段名有引号 - text = re.sub(r'([{,]\s*)([a-zA-Z_][a-zA-Z0-9_]*)\s*:', r'\1"\2":', text) - - # 3. 移除多余的逗号 - text = re.sub(r',\s*}', '}', text) - text = re.sub(r',\s*]', ']', text) - - return text - - json_text = fix_json(json_text) - logger.debug(f"修复后的JSON: {json_text[:300]}...") - - topics_data = json.loads(json_text) - topics = [SummaryTopic(**topic) for topic in topics_data[:self.max_topics]] - logger.info(f"话题分析成功，解析到 {len(topics)} 个话题") - return topics, token_usage - else: - logger.warning(f"话题分析响应中未找到JSON格式，响应内容: {result_text[:200]}...") - except json.JSONDecodeError as e: - logger.error(f"话题分析JSON解析失败: {e}") - logger.debug(f"修复后的JSON: {json_text if 'json_text' in locals() else 'N/A'}") - logger.debug(f"原始响应: {result_text}") - - # 如果JSON解析失败，尝试用正则表达式提取话题信息 - try: - logger.info("JSON解析失败，尝试正则表达式提取话题...") - topics = [] - - # 更强的正则表达式提取话题信息，处理转义字符 - # 匹配每个完整的话题对象 - topic_pattern = r'\{\s*"topic":\s*"([^"]+)"\s*,\s*"contributors":\s*\[([^\]]+)\]\s*,\s*"detail":\s*"([^"]*(?:\\.[^"]*)*)"\s*\}' - matches = re.findall(topic_pattern, result_text, re.DOTALL) - - if not matches: - # 尝试更宽松的匹配 - topic_pattern = r'"topic":\s*"([^"]+)"[^}]*"contributors":\s*\[([^\]]+)\][^}]*"detail":\s*"([^"]*(?:\\.[^"]*)*)"' - matches = re.findall(topic_pattern, result_text, re.DOTALL) - - for match in matches[:self.max_topics]: - topic_name = match[0].strip() - contributors_str = match[1].strip() - detail = match[2].strip() - - # 清理detail中的转义字符 - detail = detail.replace('\\"', '"').replace('\\n', ' ').replace('\\t', ' ') - - # 解析参与者列表 - contributors = [] - for contrib in re.findall(r'"([^"]+)"', contributors_str): - contributors.append(contrib.strip()) - - if not contributors: - contributors = ["群友"] - - topics.append(SummaryTopic( - topic=topic_name, - contributors=contributors[:5], # 最多5个参与者 - detail=detail # 限制长度 - )) - - if topics: - logger.info(f"正则表达式提取成功，获得 {len(topics)} 个话题") - return topics, token_usage - else: - # 最后的降级方案 - logger.info("正则表达式提取失败，使用默认话题...") - return [SummaryTopic( - topic="群聊讨论", - contributors=["群友"], - detail="今日群聊内容丰富，涵盖多个话题" - )] - except Exception as regex_e: - logger.error(f"正则表达式提取失败: {regex_e}") - # 最终降级方案 - return [SummaryTopic( - topic="群聊讨论", - contributors=["群友"], - detail="今日群聊内容丰富，涵盖多个话题" - )] - - except Exception as e: - logger.error(f"话题分析处理失败: {e}") - - return [], token_usage - - except Exception as e: - logger.error(f"话题分析失败: {e}") - return [], TokenUsage() - - async def _analyze_user_titles(self, messages: List[Dict], user_analysis: Dict) -> tuple[List[UserTitle], TokenUsage]: - """使用LLM分析用户称号""" - try: - # 准备用户数据 - user_summaries = [] - for user_id, stats in user_analysis.items(): - if stats["message_count"] < 5: # 过滤活跃度太低的用户 - continue - - # 分析用户特征 - night_messages = sum(stats["hours"][h] for h in range(0, 6)) - day_messages = stats["message_count"] - night_messages - avg_chars = stats["char_count"] / stats["message_count"] if stats["message_count"] > 0 else 0 - - user_summaries.append({ - "name": stats["nickname"], - "qq": int(user_id), - "message_count": stats["message_count"], - "avg_chars": round(avg_chars, 1), - "emoji_ratio": round(stats["emoji_count"] / stats["message_count"], 2), - "night_ratio": round(night_messages / stats["message_count"], 2), - "reply_ratio": round(stats["reply_count"] / stats["message_count"], 2) - }) - - if not user_summaries: - return [], TokenUsage() - - # 按消息数量排序，取前N名（根据配置） - user_summaries.sort(key=lambda x: x["message_count"], reverse=True) - user_summaries = user_summaries[:self.max_user_titles] - - # 构建LLM提示词 - users_text = "\n".join([ - f"- {user['name']} (QQ:{user['qq']}): " - f"发言{user['message_count']}条, 平均{user['avg_chars']}字, " - f"表情比例{user['emoji_ratio']}, 夜间发言比例{user['night_ratio']}, " - f"回复比例{user['reply_ratio']}" - for user in user_summaries - ]) - - prompt = f""" -请为以下群友分配合适的称号和MBTI类型。每个人只能有一个称号，每个称号只能给一个人。 - -可选称号： -- 龙王: 发言频繁但内容轻松的人 -- 技术专家: 经常讨论技术话题的人 -- 夜猫子: 经常在深夜发言的人 -- 表情包军火库: 经常发表情的人 -- 沉默终结者: 经常开启话题的人 -- 评论家: 平均发言长度很长的人 -- 阳角: 在群里很有影响力的人 -- 互动达人: 经常回复别人的人 -- ... (你可以自行进行拓展添加) - -用户数据： -{users_text} - -请以JSON格式返回，格式如下： -[ - {{ - "name": "用户名", - "qq": 123456789, - "title": "称号", - "mbti": "MBTI类型", - "reason": "获得此称号的原因" - }} -] -""" - - # 调用LLM - provider = self.context.get_using_provider() - if not provider: - logger.warning("未配置LLM提供商，跳过用户称号分析") - return [], TokenUsage() - - response = await provider.text_chat( - prompt=prompt, - max_tokens=1500, - temperature=0.5 - ) - - # 提取token使用统计 - token_usage = TokenUsage() - if response.raw_completion and hasattr(response.raw_completion, 'usage'): - usage = response.raw_completion.usage - token_usage.prompt_tokens = usage.prompt_tokens - token_usage.completion_tokens = usage.completion_tokens - token_usage.total_tokens = usage.total_tokens - - # 解析响应 - if hasattr(response, 'completion_text'): - result_text = response.completion_text - else: - result_text = str(response) - - # 尝试解析JSON - try: - import re - json_match = re.search(r'\[.*\]', result_text, re.DOTALL) - if json_match: - titles_data = json.loads(json_match.group()) - return [UserTitle(**title) for title in titles_data], token_usage - except: - pass - - return [], token_usage - - except Exception as e: - logger.error(f"用户称号分析失败: {e}") - return [], TokenUsage() - - async def _analyze_golden_quotes(self, messages: List[Dict]) -> tuple[List[GoldenQuote], TokenUsage]: - """使用LLM分析群聊金句""" - try: - # 提取有趣的文本消息 - interesting_messages = [] - for msg in messages: - sender = msg.get("sender", {}) - nickname = sender.get("nickname", "") or sender.get("card", "") - msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M") - - for content in msg.get("message", []): - if content.get("type") == "text": - text = content.get("data", {}).get("text", "").strip() - # 过滤长度适中、可能有趣的消息 - if 5 <= len(text) <= 100 and not text.startswith(("http", "www", "/")): - interesting_messages.append({ - "sender": nickname, - "time": msg_time, - "content": text - }) - - if not interesting_messages: - return [], TokenUsage() - - # # 限制消息数量以避免token过多 - # if len(interesting_messages) > 50: - # # 均匀采样 - # step = len(interesting_messages) // 50 - # interesting_messages = interesting_messages[::step] - - # 构建LLM提示词 - messages_text = "\n".join([ - f"[{msg['time']}] {msg['sender']}: {msg['content']}" - for msg in interesting_messages - ]) - - # 计算金句数量，默认5句，但可以根据配置调整 - self.max_golden_quotes - - prompt = f""" -请从以下群聊记录中挑选出{self.max_golden_quotes}句最具冲击力、最令人惊叹的"金句"。这些金句需满足： -- 核心标准：**逆天的神人发言**，即具备颠覆常识的脑洞、逻辑跳脱的表达或强烈反差感的原创内容 -- 典型特征：包含某些争议话题元素、夸张类比、反常规结论、一本正经的"胡说八道"或突破语境的清奇思路，并且具备一定的冲击力，让人印象深刻。 - -对于每个金句，请提供： -1. 原文内容（完整保留发言细节） -2. 发言人昵称 -3. 选择理由（具体说明其"逆天"之处，如逻辑颠覆点/脑洞角度/反差感/争议话题元素） - -此外，我将对你进行严格约束： -- 优先筛选 **逆天指数最高** 的内容：发情、性压抑话题 > 争议话题 > 元素级 > 颠覆认知级 > 逻辑跳脱级 > 趣味调侃级，剔除单纯玩梗或网络热词堆砌的普通发言 -- 重点标记包含极端类比、反常识论证或无厘头结论的内容，并且包含一定的争议话题元素。 - -群聊记录： -{messages_text} - -请以JSON格式返回，格式如下： -[ - {{ - "content": "金句原文", - "sender": "发言人昵称", - "reason": "选择这句话的理由（需明确说明逆天特质）" - }} -] -""" - - # 调用LLM - provider = self.context.get_using_provider() - if not provider: - logger.warning("未配置LLM提供商，跳过金句分析") - return [], TokenUsage() - - response = await provider.text_chat( - prompt=prompt, - max_tokens=1500, - temperature=0.7 - ) - - # 提取token使用统计 - token_usage = TokenUsage() - if response.raw_completion and hasattr(response.raw_completion, 'usage'): - usage = response.raw_completion.usage - token_usage.prompt_tokens = usage.prompt_tokens - token_usage.completion_tokens = usage.completion_tokens - token_usage.total_tokens = usage.total_tokens - - # 解析响应 - if hasattr(response, 'completion_text'): - result_text = response.completion_text - else: - result_text = str(response) - - # 尝试解析JSON - try: - import re - json_match = re.search(r'\[.*\]', result_text, re.DOTALL) - if json_match: - quotes_data = json.loads(json_match.group()) - return [GoldenQuote(**quote) for quote in quotes_data[:self.max_golden_quotes]], token_usage - except: - pass - - return [], token_usage - - except Exception as e: - logger.error(f"金句分析失败: {e}") - return [], TokenUsage() - - async def _html_to_pdf(self, html_content: str, output_path: str) -> bool: - """将 HTML 内容转换为 PDF 文件""" - try: - # 确保 pyppeteer 可用 - if not PYPPETEER_AVAILABLE: - logger.error("pyppeteer 不可用，无法生成 PDF") - return False - - # 动态导入 pyppeteer - import pyppeteer - from pyppeteer import launch - - # 尝试启动浏览器，如果 Chromium 不存在会自动下载 - logger.info("启动浏览器进行 PDF 转换") - - # 配置浏览器启动参数，避免 Chromium 下载问题 - launch_options = { - 'headless': True, - 'args': [ - '--no-sandbox', - '--disable-setuid-sandbox', - '--disable-dev-shm-usage', - '--disable-gpu', - '--no-first-run', - '--disable-extensions', - '--disable-default-apps' - ] - } - - # 如果是 Windows 系统，尝试使用系统 Chrome - if sys.platform.startswith('win'): - # 常见的 Chrome 安装路径 - chrome_paths = [ - r"C:\Program Files\Google\Chrome\Application\chrome.exe", - r"C:\Program Files (x86)\Google\Chrome\Application\chrome.exe", - r"C:\Users\{}\AppData\Local\Google\Chrome\Application\chrome.exe".format(os.environ.get('USERNAME', '')), - ] - - for chrome_path in chrome_paths: - if Path(chrome_path).exists(): - launch_options['executablePath'] = chrome_path - logger.info(f"使用系统 Chrome: {chrome_path}") - break - - browser = await launch(**launch_options) - page = await browser.newPage() - - # 设置页面内容 (pyppeteer 1.0.2 版本的 API) - await page.setContent(html_content) - # 等待页面加载完成 - try: - await page.waitForSelector('body', {'timeout': 10000}) - except Exception: - # 如果等待失败，继续执行（可能页面已经加载完成） - pass - - # 导出 PDF - await page.pdf({ - 'path': output_path, - 'format': 'A4', - 'printBackground': True, - 'margin': { - 'top': '10mm', - 'right': '10mm', - 'bottom': '10mm', - 'left': '10mm' - }, - 'scale': 0.8 - }) - - await browser.close() - logger.info(f"PDF 生成成功: {output_path}") - return True - - except Exception as e: - error_msg = str(e) - if "Chromium downloadable not found" in error_msg: - logger.error("Chromium 下载失败，建议安装 pyppeteer2 或使用系统 Chrome") - elif "No usable sandbox" in error_msg: - logger.error("沙盒权限问题，已尝试禁用沙盒") - else: - logger.error(f"HTML 转 PDF 失败: {e}") - return False - - async def _generate_pdf_report(self, analysis_result: Dict, group_id: str) -> Optional[str]: - """生成 PDF 格式的分析报告""" - try: - #确保输出目录存在 - output_dir = Path(self.pdf_output_dir) - output_dir.mkdir(parents=True, exist_ok=True) - - # 生成文件名 - current_date = datetime.now().strftime('%Y%m%d') - filename = self.pdf_filename_format.format( - group_id=group_id, - date=current_date - ) - pdf_path = output_dir / filename - - # 准备渲染数据 - render_data = await self._prepare_render_data(analysis_result) - logger.info(f"PDF 渲染数据准备完成，包含 {len(render_data)} 个字段") - - # 生成 HTML 内容 - html_content = self._render_html_template(self._get_pdf_html_template(), render_data) - logger.info(f"HTML 内容生成完成，长度: {len(html_content)} 字符") - - # 转换为 PDF - success = await self._html_to_pdf(html_content, str(pdf_path)) - - if success: - return str(pdf_path.absolute()) - else: - return None - - except Exception as e: - logger.error(f"生成 PDF 报告失败: {e}") - return None - - async def _generate_image_report(self, analysis_result: Dict, group_id: str) -> Optional[str]: - """生成图片格式的分析报告""" - try: - # 准备渲染数据 - render_payload = await self._prepare_render_data(analysis_result) - - # 使用AstrBot内置的HTML渲染服务 - image_url = await self.html_render(self._get_html_template(), render_payload) - return image_url - - except Exception as e: - logger.error(f"生成图片报告失败: {e}") - return None - - - - async def _prepare_render_data(self, analysis_result: Dict) -> Dict: - """准备渲染数据""" - stats = analysis_result["statistics"] - topics = analysis_result["topics"] - user_titles = analysis_result["user_titles"] - - # 构建话题HTML - topics_html = "" - for i, topic in enumerate(topics[:self.max_topics], 1): - contributors_str = "、".join(topic.contributors) - topics_html += f""" -
-
- {i} - {topic.topic} -
-
参与者: {contributors_str}
-
{topic.detail}
-
- """ - - # 构建用户称号HTML（包含头像） - titles_html = "" - for title in user_titles[:self.max_user_titles]: - # 获取用户头像 - avatar_data = await self._get_user_avatar(str(title.qq)) - avatar_html = f'头像' if avatar_data else '
👤
' - - titles_html += f""" -
-
- {avatar_html} -
-
{title.name}
-
-
{title.title}
-
{title.mbti}
-
-
-
-
{title.reason}
-
- """ - - # 构建金句HTML - quotes_html = "" - for quote in stats.golden_quotes[:self.max_golden_quotes]: - quotes_html += f""" -
-
"{quote.content}"
-
—— {quote.sender}
-
{quote.reason}
-
- """ - - # 返回扁平化的渲染数据 - return { - "current_date": datetime.now().strftime('%Y年%m月%d日'), - "current_datetime": datetime.now().strftime('%Y-%m-%d %H:%M:%S'), - "message_count": stats.message_count, - "participant_count": stats.participant_count, - "total_characters": stats.total_characters, - "emoji_count": stats.emoji_count, - "most_active_period": stats.most_active_period, - "topics_html": topics_html, - "titles_html": titles_html, - "quotes_html": quotes_html, - "total_tokens": stats.token_usage.total_tokens, - "prompt_tokens": stats.token_usage.prompt_tokens, - "completion_tokens": stats.token_usage.completion_tokens - } - - def _get_html_template(self) -> str: - """获取HTML模板""" - return """ - - - - - - 群聊日常分析报告 - - - - -
-
-

📊 群聊日常分析报告

-
{{ current_date }}
-
- -
- -
-

📈 基础统计

-
-
-
{{ message_count }}
-
消息总数
-
-
-
{{ participant_count }}
-
参与人数
-
-
-
{{ total_characters }}
-
总字符数
-
-
-
{{ emoji_count }}
-
表情数量
-
-
- -
-
{{ most_active_period }}
-
最活跃时段
-
-
- - -
-

💬 热门话题

-
- {{ topics_html | safe }} -
-
- - -
-

🏆 群友称号

-
- {{ titles_html | safe }} -
-
- - -
-

💬 群圣经

- {{ quotes_html | safe }} -
-
- -
- 由 AstrBot QQ群日常分析插件 生成 | {{ current_datetime }} | SXP-Simon/astrbot-qq-group-daily-analysis
- - 🤖 AI分析消耗：{{ total_tokens }} tokens (输入: {{ prompt_tokens }}, 输出: {{ completion_tokens }}) - -
-
- - - """ - - - - - async def _generate_text_report(self, analysis_result: Dict) -> str: - """生成文本格式的分析报告""" - stats = analysis_result["statistics"] - topics = analysis_result["topics"] - user_titles = analysis_result["user_titles"] - - report = f""" -🎯 群聊日常分析报告 -📅 {datetime.now().strftime('%Y年%m月%d日')} - -📊 基础统计 -• 消息总数: {stats.message_count} -• 参与人数: {stats.participant_count} -• 总字符数: {stats.total_characters} -• 表情数量: {stats.emoji_count} -• 最活跃时段: {stats.most_active_period} - -💬 热门话题 -""" - - for i, topic in enumerate(topics[:self.max_topics], 1): - contributors_str = "、".join(topic.contributors) - report += f"{i}. {topic.topic}\n" - report += f" 参与者: {contributors_str}\n" - report += f" {topic.detail}\n\n" - - report += "🏆 群友称号\n" - for title in user_titles[:self.max_user_titles]: - report += f"• {title.name} - {title.title} ({title.mbti})\n" - report += f" {title.reason}\n\n" - - report += "💬 群圣经\n" - for i, quote in enumerate(stats.golden_quotes[:self.max_golden_quotes], 1): - report += f"{i}. \"{quote.content}\" —— {quote.sender}\n" - report += f" {quote.reason}\n\n" - - return report - - - - async def _reload_config_and_restart_scheduler(self): - """重新加载配置并重启调度器""" - try: - # 重新从配置系统读取配置 - self.enabled_groups = self.config.get("enabled_groups", []) - self.enable_auto_analysis = self.config.get("enable_auto_analysis", False) - self.auto_analysis_time = self.config.get("auto_analysis_time", "09:00") - logger.info(f"重新加载配置: 自动分析={self.enable_auto_analysis}") - - # 停止现有的调度器 - if hasattr(self, 'scheduler_task') and self.scheduler_task and not self.scheduler_task.done(): - self.scheduler_task.cancel() - logger.info("已停止现有的定时任务") - - # 如果启用了自动分析，启动新的调度器 - if self.enable_auto_analysis: - self.scheduler_task = asyncio.create_task(self._start_scheduler()) - logger.info("已启动新的定时任务") - - except Exception as e: - logger.error(f"重新加载配置失败: {e}") - - async def _start_scheduler(self): - """启动定时任务调度器""" - logger.info(f"启动定时任务调度器，自动分析时间: {self.auto_analysis_time}") - - while True: - try: - now = datetime.now() - target_time = datetime.strptime(self.auto_analysis_time, "%H:%M").replace( - year=now.year, month=now.month, day=now.day - ) - - # 如果今天的目标时间已过，设置为明天 - if now >= target_time: - target_time += timedelta(days=1) - - # 计算等待时间 - wait_seconds = (target_time - now).total_seconds() - logger.info(f"定时分析将在 {target_time.strftime('%Y-%m-%d %H:%M:%S')} 执行，等待 {wait_seconds:.0f} 秒") - - # 等待到目标时间 - await asyncio.sleep(wait_seconds) - - # 执行自动分析 - if self.enable_auto_analysis: - logger.info("开始执行定时分析") - await self._run_auto_analysis() - else: - logger.info("自动分析已禁用，跳过执行") - break - - except Exception as e: - logger.error(f"定时任务调度器错误: {e}") - # 等待5分钟后重试 - await asyncio.sleep(300) - - async def _run_auto_analysis(self): - """执行自动分析""" - try: - logger.info("开始执行自动群聊分析") - - # 为每个启用的群执行分析 - for group_id in self.enabled_groups: - try: - logger.info(f"为群 {group_id} 执行自动分析") - - # 这里需要模拟一个事件对象来调用分析功能 - # 由于自动分析没有真实的用户事件，我们直接调用内部方法 - await self._perform_auto_analysis_for_group(group_id) - - except Exception as e: - logger.error(f"群 {group_id} 自动分析失败: {e}") - - except Exception as e: - logger.error(f"自动分析执行失败: {e}") - - async def _perform_auto_analysis_for_group(self, group_id: str): - """为指定群执行自动分析""" - try: - # 尝试获取bot实例 - if not self.bot_instance: - self.bot_instance = await self._get_bot_instance() - - if not self.bot_instance: - logger.warning(f"群 {group_id} 自动分析跳过：未获取到bot实例，请先手动触发一次分析") - return - - # 确保有机器人QQ号 - if not self.bot_qq_id: - await self._get_bot_qq_id() - - logger.info(f"开始为群 {group_id} 执行自动分析") - - # 获取群聊消息 - messages = await self._fetch_group_messages_for_auto(group_id) - if not messages: - logger.warning(f"群 {group_id} 未获取到足够的消息记录") - return - - # 检查消息数量 - if len(messages) < self.min_messages_threshold: - logger.warning(f"群 {group_id} 消息数量不足（{len(messages)}条），跳过分析") - return - - logger.info(f"群 {group_id} 获取到 {len(messages)} 条消息，开始分析") - - # 进行分析 - analysis_result = await self._analyze_messages(messages, group_id) - if not analysis_result: - logger.error(f"群 {group_id} 分析失败") - return - - # 生成报告 - if self.output_format == "image": - image_url = await self._generate_image_report(analysis_result, group_id) - if image_url: - # 发送分析报告到群 - await self._send_auto_analysis_result(group_id, image_url) - logger.info(f"群 {group_id} 自动分析完成，已发送图片报告") - else: - logger.error(f"群 {group_id} 图片报告生成失败") - # 图片生成失败时回退到文本报告 - text_report = await self._generate_text_report(analysis_result) - await self._send_auto_analysis_text(group_id, text_report) - logger.info(f"群 {group_id} 图片报告生成失败，已发送文本报告") - elif self.output_format == "pdf": - if not PYPPETEER_AVAILABLE: - logger.warning(f"群 {group_id} PDF功能不可用，回退到文本报告") - text_report = await self._generate_text_report(analysis_result) - await self._send_auto_analysis_text(group_id, text_report) - logger.info(f"群 {group_id} PDF功能不可用，已发送文本报告") - else: - pdf_path = await self._generate_pdf_report(analysis_result, group_id) - if pdf_path: - # 发送PDF文件到群 - await self._send_auto_analysis_pdf(group_id, pdf_path) - logger.info(f"群 {group_id} 自动分析完成，已发送PDF报告") - else: - logger.error(f"群 {group_id} PDF报告生成失败，回退到文本报告") - text_report = await self._generate_text_report(analysis_result) - await self._send_auto_analysis_text(group_id, text_report) - logger.info(f"群 {group_id} PDF报告生成失败，已发送文本报告") - else: - text_report = await self._generate_text_report(analysis_result) - await self._send_auto_analysis_text(group_id, text_report) - logger.info(f"群 {group_id} 自动分析完成，已发送文本报告") - - except Exception as e: - logger.error(f"群 {group_id} 自动分析执行失败: {e}", exc_info=True) - - async def _fetch_group_messages_for_auto(self, group_id: str) -> List[Dict]: - """为自动分析获取群聊消息（使用统一方法）""" - if not self.bot_instance: - logger.error(f"群 {group_id} 获取消息失败：缺少bot实例") - return [] - - return await self._fetch_group_messages_unified(self.bot_instance, group_id, self.analysis_days) - - async def _send_auto_analysis_result(self, group_id: str, image_url: str): - """发送自动分析的图片结果到群""" - try: - if not self.bot_instance: - return - - # 发送图片消息到群 - await self.bot_instance.api.call_action( - "send_group_msg", - group_id=group_id, - message=[{ - "type": "text", - "data": {"text": "📊 每日群聊分析报告已生成："} - }, { - "type": "image", - "data": {"url": image_url} - }] - ) - - except Exception as e: - logger.error(f"发送自动分析结果到群 {group_id} 失败: {e}") - - async def _send_auto_analysis_text(self, group_id: str, text_report: str): - """发送自动分析的文本结果到群""" - try: - if not self.bot_instance: - return - - # 发送文本消息到群 - await self.bot_instance.api.call_action( - "send_group_msg", - group_id=group_id, - message=f"📊 每日群聊分析报告：\n\n{text_report}" - ) - - except Exception as e: - logger.error(f"发送自动分析文本到群 {group_id} 失败: {e}") - - async def _send_auto_analysis_pdf(self, group_id: str, pdf_path: str): - """发送自动分析的PDF结果到群""" - try: - if not self.bot_instance: - return - - # 发送PDF文件到群 - await self.bot_instance.api.call_action( - "send_group_msg", - group_id=group_id, - message=[{ - "type": "text", - "data": {"text": "📊 每日群聊分析报告已生成："} - }, { - "type": "file", - "data": {"file": pdf_path} - }] - ) - - except Exception as e: - logger.error(f"发送自动分析PDF到群 {group_id} 失败: {e}") - # 如果发送PDF失败，尝试发送提示信息 - try: - await self.bot_instance.api.call_action( - "send_group_msg", - group_id=group_id, - message=f"📊 每日群聊分析报告已生成，但发送PDF文件失败。PDF文件路径：{pdf_path}" - ) - except Exception as e2: - logger.error(f"发送PDF失败提示到群 {group_id} 也失败: {e2}") - - async def _get_bot_instance(self): - """从Context获取bot实例""" - try: - # 如果已经有保存的实例，直接返回 - if self.bot_instance: - return self.bot_instance - - logger.info("尝试获取bot实例...") - - # 简化的获取逻辑，尝试常见的几种方式 - if hasattr(self.context, 'get_platforms') and callable(self.context.get_platforms): - platforms = self.context.get_platforms() - for platform in platforms: - if hasattr(platform, 'bot') and platform.bot: - logger.info(f"从平台获取到bot实例") - return platform.bot - - # 尝试从context的platforms属性获取 - if hasattr(self.context, 'platforms') and self.context.platforms: - for platform in self.context.platforms: - if hasattr(platform, 'bot') and platform.bot: - logger.info(f"从平台列表获取到bot实例") - return platform.bot - - logger.info("暂时无法获取bot实例，等待用户手动触发分析") - return None - - except Exception as e: - logger.error(f"获取bot实例失败: {e}") - return None - - async def _delayed_start_scheduler(self): - """延迟启动调度器，给系统时间初始化""" - try: - # 等待10秒让系统完全初始化 - await asyncio.sleep(10) - - # 尝试获取bot实例 - self.bot_instance = await self._get_bot_instance() - - if self.bot_instance: - logger.info("成功获取bot实例，启动定时任务") - # 获取机器人QQ号 - await self._get_bot_qq_id() - else: - logger.info("暂时未获取到bot实例，定时任务仍会启动。首次手动触发分析后将自动获取bot实例") - - # 启动调度器 - self.scheduler_task = asyncio.create_task(self._start_scheduler()) - - except Exception as e: - logger.error(f"延迟启动调度器失败: {e}") - - def _get_pdf_html_template(self) -> str: - """获取 PDF 专用的 HTML 模板""" - return """ - - - - - - 群聊日常分析报告 - - - -
-
-

📊 群聊日常分析报告

-
{current_date}
-
- -
-

📈 基础统计

-
-
-
{message_count}
-
消息总数
-
-
-
{participant_count}
-
参与人数
-
-
-
{total_characters}
-
总字符数
-
-
-
{emoji_count}
-
表情数量
-
-
- -
-
{most_active_period}
-
最活跃时段
-
-
- -
-

💬 热门话题

- {topics_html} -
- -
-

🏆 群友称号

- {titles_html} -
- -
-

💬 群圣经

- {quotes_html} -
- -
- 由 AstrBot QQ群日常分析插件 生成 | {current_datetime} | SXP-Simon/astrbot-qq-group-daily-analysis
- - 🤖 AI分析消耗：{total_tokens} tokens (输入: {prompt_tokens}, 输出: {completion_tokens}) - -
-
- - - """ diff --git a/metadata.yaml b/metadata.yaml index 81eac2e..923eadd 100644 --- a/metadata.yaml +++ b/metadata.yaml @@ -13,6 +13,6 @@ help: | # 插件的帮助信息 命令： /群分析 [天数] - 分析群聊活动 /分析设置 [操作] - 管理设置（enable/disable/status/test） -version: v1.6.0 # 插件版本号。格式：v1.1.1 或者 v1.1 +version: v1.7.0 # 插件版本号。格式：v1.1.1 或者 v1.1 author: SXP-Simon # 作者 repo: https://github.com/SXP-Simon/astrbot-qq-group-daily-analysis # 插件的仓库地址 diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..33c66e7 --- /dev/null +++ b/src/__init__.py @@ -0,0 +1,5 @@ +""" +QQ群日常分析插件 - 源代码包 +""" + +__author__ = "SXP-Simon" \ No newline at end of file diff --git a/src/analysis/__init__.py b/src/analysis/__init__.py new file mode 100644 index 0000000..d1d48ee --- /dev/null +++ b/src/analysis/__init__.py @@ -0,0 +1,12 @@ +""" +分析模块 +包含LLM分析和统计分析功能 +""" + +from .llm_analyzer import LLMAnalyzer +from .statistics import UserAnalyzer + +__all__ = [ + 'LLMAnalyzer', + 'UserAnalyzer' +] \ No newline at end of file diff --git a/src/analysis/llm_analyzer.py b/src/analysis/llm_analyzer.py new file mode 100644 index 0000000..45f92b4 --- /dev/null +++ b/src/analysis/llm_analyzer.py @@ -0,0 +1,470 @@ +""" +LLM分析器模块 +负责使用LLM进行话题分析、用户称号分析和金句分析 +""" + +import json +import re +from datetime import datetime +from typing import List, Dict, Tuple +from astrbot.api import logger +from ...src.models.data_models import SummaryTopic, UserTitle, GoldenQuote, TokenUsage + + +class LLMAnalyzer: + """LLM分析器""" + + def __init__(self, context, config_manager): + self.context = context + self.config_manager = config_manager + + async def analyze_topics(self, messages: List[Dict]) -> Tuple[List[SummaryTopic], TokenUsage]: + """使用LLM分析话题""" + try: + # 提取文本消息 + text_messages = [] + for msg in messages: + sender = msg.get("sender", {}) + nickname = sender.get("nickname", "") or sender.get("card", "") + msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M") + + for content in msg.get("message", []): + if content.get("type") == "text": + text = content.get("data", {}).get("text", "").strip() + if text and len(text) > 2 and not text.startswith(("/")): + text_messages.append({ + "sender": nickname, + "time": msg_time, + "content": text + }) + + if not text_messages: + return [], TokenUsage() + + # 构建LLM提示词，清理消息内容 + def clean_message_content(content): + """清理消息内容，移除可能影响JSON解析的字符""" + # 替换中文引号 + content = content.replace('"', '"').replace('"', '"') + content = content.replace(''', "'").replace(''', "'") + # 移除或替换其他特殊字符 + content = content.replace('\n', ' ').replace('\r', ' ') + content = content.replace('\t', ' ') + # 移除可能的控制字符 + content = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', content) + return content.strip() + + messages_text = "\n".join([ + f"[{msg['time']}] {msg['sender']}: {clean_message_content(msg['content'])}" + for msg in text_messages + ]) + + max_topics = self.config_manager.get_max_topics() + prompt = f""" +你是一个帮我进行群聊信息总结的助手，生成总结内容时，你需要严格遵守下面的几个准则： +请分析接下来提供的群聊记录，提取出最多{max_topics}个主要话题。 + +对于每个话题，请提供： +1. 话题名称（突出主题内容，尽量简明扼要） +2. 主要参与者（最多5人） +3. 话题详细描述（包含关键信息和结论） + +注意： +- 对于比较有价值的点，稍微用一两句话详细讲讲，比如不要生成 "Nolan 和 SOV 讨论了 galgame 中关于性符号的衍生情况" 这种宽泛的内容，而是生成更加具体的讨论内容，让其他人只看这个消息就能知道讨论中有价值的，有营养的信息。 +- 对于其中的部分信息，你需要特意提到主题施加的主体是谁，是哪个群友做了什么事情，而不要直接生成和群友没有关系的语句。 +- 对于每一条总结，尽量讲清楚前因后果，以及话题的结论，是什么，为什么，怎么做，如果用户没有讲到细节，则可以不用这么做。 + +群聊记录： +{messages_text} + +重要：必须返回标准JSON格式，严格遵守以下规则： +1. 只使用英文双引号 " 不要使用中文引号 " " +2. 字符串内容中的引号必须转义为 \" +3. 多个对象之间用逗号分隔 +4. 数组元素之间用逗号分隔 +5. 不要在JSON外添加任何文字说明 +6. 描述内容避免使用特殊符号，用普通文字表达 + +请严格按照以下JSON格式返回，确保可以被标准JSON解析器解析： +[ + {{ + "topic": "话题名称", + "contributors": ["用户1", "用户2"], + "detail": "话题描述内容" + }}, + {{ + "topic": "另一个话题", + "contributors": ["用户3", "用户4"], + "detail": "另一个话题的描述" + }} +] + +注意：返回的内容必须是纯JSON，不要包含markdown代码块标记或其他格式 +""" + + # 调用LLM + provider = self.context.get_using_provider() + if not provider: + logger.warning("未配置LLM提供商，跳过话题分析") + return [], TokenUsage() + + response = await provider.text_chat( + prompt=prompt, + max_tokens=10000, + temperature=0.6 + ) + + # 提取token使用统计 + token_usage = TokenUsage() + if response.raw_completion and hasattr(response.raw_completion, 'usage'): + usage = response.raw_completion.usage + token_usage.prompt_tokens = usage.prompt_tokens + token_usage.completion_tokens = usage.completion_tokens + token_usage.total_tokens = usage.total_tokens + + # 解析响应 + if hasattr(response, 'completion_text'): + result_text = response.completion_text + else: + result_text = str(response) + + # 尝试解析JSON + try: + # 提取JSON部分 + json_match = re.search(r'\[.*?\]', result_text, re.DOTALL) + if json_match: + json_text = json_match.group() + logger.debug(f"话题分析JSON原文: {json_text[:500]}...") + + # 强化JSON清理和修复 + json_text = self._fix_json(json_text) + logger.debug(f"修复后的JSON: {json_text[:300]}...") + + topics_data = json.loads(json_text) + topics = [SummaryTopic(**topic) for topic in topics_data[:max_topics]] + logger.info(f"话题分析成功，解析到 {len(topics)} 个话题") + return topics, token_usage + else: + logger.warning(f"话题分析响应中未找到JSON格式，响应内容: {result_text[:200]}...") + except json.JSONDecodeError as e: + logger.error(f"话题分析JSON解析失败: {e}") + logger.debug(f"修复后的JSON: {json_text if 'json_text' in locals() else 'N/A'}") + logger.debug(f"原始响应: {result_text}") + + # 如果JSON解析失败，尝试用正则表达式提取话题信息 + topics = self._extract_topics_with_regex(result_text, max_topics) + if topics: + logger.info(f"正则表达式提取成功，获得 {len(topics)} 个话题") + return topics, token_usage + else: + # 最后的降级方案 + logger.info("正则表达式提取失败，使用默认话题...") + return [SummaryTopic( + topic="群聊讨论", + contributors=["群友"], + detail="今日群聊内容丰富，涵盖多个话题" + )], token_usage + + return [], token_usage + + except Exception as e: + logger.error(f"话题分析失败: {e}") + return [], TokenUsage() + + def _fix_json(self, text: str) -> str: + """修复JSON格式问题""" + # 移除markdown代码块标记 + text = re.sub(r'```json\s*', '', text) + text = re.sub(r'```\s*$', '', text) + + # 基础清理 + text = text.replace('\n', ' ').replace('\r', ' ') + text = re.sub(r'\s+', ' ', text) + + # 替换中文引号为英文引号 + text = text.replace('"', '"').replace('"', '"') + text = text.replace(''', "'").replace(''', "'") + + # 处理字符串内容中的特殊字符 + # 转义字符串内的双引号 + def escape_quotes_in_strings(match): + content = match.group(1) + # 转义内部的双引号 + content = content.replace('"', '\\"') + return f'"{content}"' + + # 先处理字段值中的引号 + text = re.sub(r'"([^"]*(?:"[^"]*)*)"', escape_quotes_in_strings, text) + + # 修复截断的JSON + if not text.endswith(']'): + last_complete = text.rfind('}') + if last_complete > 0: + text = text[:last_complete + 1] + ']' + + # 修复常见的JSON格式问题 + # 1. 修复缺失的逗号 + text = re.sub(r'}\s*{', '}, {', text) + + # 2. 确保字段名有引号 + text = re.sub(r'([{,]\s*)([a-zA-Z_][a-zA-Z0-9_]*)\s*:', r'\1"\2":', text) + + # 3. 移除多余的逗号 + text = re.sub(r',\s*}', '}', text) + text = re.sub(r',\s*]', ']', text) + + return text + + def _extract_topics_with_regex(self, result_text: str, max_topics: int) -> List[SummaryTopic]: + """使用正则表达式提取话题信息""" + try: + topics = [] + + # 更强的正则表达式提取话题信息，处理转义字符 + # 匹配每个完整的话题对象 + topic_pattern = r'\{\s*"topic":\s*"([^"]+)"\s*,\s*"contributors":\s*\[([^\]]+)\]\s*,\s*"detail":\s*"([^"]*(?:\\.[^"]*)*)"\s*\}' + matches = re.findall(topic_pattern, result_text, re.DOTALL) + + if not matches: + # 尝试更宽松的匹配 + topic_pattern = r'"topic":\s*"([^"]+)"[^}]*"contributors":\s*\[([^\]]+)\][^}]*"detail":\s*"([^"]*(?:\\.[^"]*)*)"' + matches = re.findall(topic_pattern, result_text, re.DOTALL) + + for match in matches[:max_topics]: + topic_name = match[0].strip() + contributors_str = match[1].strip() + detail = match[2].strip() + + # 清理detail中的转义字符 + detail = detail.replace('\\"', '"').replace('\\n', ' ').replace('\\t', ' ') + + # 解析参与者列表 + contributors = [] + for contrib in re.findall(r'"([^"]+)"', contributors_str): + contributors.append(contrib.strip()) + + if not contributors: + contributors = ["群友"] + + topics.append(SummaryTopic( + topic=topic_name, + contributors=contributors[:5], # 最多5个参与者 + detail=detail + )) + + return topics + except Exception as e: + logger.error(f"正则表达式提取失败: {e}") + return [] + + async def analyze_user_titles(self, messages: List[Dict], user_analysis: Dict) -> Tuple[List[UserTitle], TokenUsage]: + """使用LLM分析用户称号""" + try: + # 准备用户数据 + user_summaries = [] + for user_id, stats in user_analysis.items(): + if stats["message_count"] < 5: # 过滤活跃度太低的用户 + continue + + # 分析用户特征 + night_messages = sum(stats["hours"][h] for h in range(0, 6)) + day_messages = stats["message_count"] - night_messages + avg_chars = stats["char_count"] / stats["message_count"] if stats["message_count"] > 0 else 0 + + user_summaries.append({ + "name": stats["nickname"], + "qq": int(user_id), + "message_count": stats["message_count"], + "avg_chars": round(avg_chars, 1), + "emoji_ratio": round(stats["emoji_count"] / stats["message_count"], 2), + "night_ratio": round(night_messages / stats["message_count"], 2), + "reply_ratio": round(stats["reply_count"] / stats["message_count"], 2) + }) + + if not user_summaries: + return [], TokenUsage() + + # 按消息数量排序，取前N名 + max_user_titles = self.config_manager.get_max_user_titles() + user_summaries.sort(key=lambda x: x["message_count"], reverse=True) + user_summaries = user_summaries[:max_user_titles] + + # 构建LLM提示词 + users_text = "\n".join([ + f"- {user['name']} (QQ:{user['qq']}): " + f"发言{user['message_count']}条, 平均{user['avg_chars']}字, " + f"表情比例{user['emoji_ratio']}, 夜间发言比例{user['night_ratio']}, " + f"回复比例{user['reply_ratio']}" + for user in user_summaries + ]) + + prompt = f""" +请为以下群友分配合适的称号和MBTI类型。每个人只能有一个称号，每个称号只能给一个人。 + +可选称号： +- 龙王: 发言频繁但内容轻松的人 +- 技术专家: 经常讨论技术话题的人 +- 夜猫子: 经常在深夜发言的人 +- 表情包军火库: 经常发表情的人 +- 沉默终结者: 经常开启话题的人 +- 评论家: 平均发言长度很长的人 +- 阳角: 在群里很有影响力的人 +- 互动达人: 经常回复别人的人 +- ... (你可以自行进行拓展添加) + +用户数据： +{users_text} + +请以JSON格式返回，格式如下： +[ + {{ + "name": "用户名", + "qq": 123456789, + "title": "称号", + "mbti": "MBTI类型", + "reason": "获得此称号的原因" + }} +] +""" + + # 调用LLM + provider = self.context.get_using_provider() + if not provider: + logger.warning("未配置LLM提供商，跳过用户称号分析") + return [], TokenUsage() + + response = await provider.text_chat( + prompt=prompt, + max_tokens=1500, + temperature=0.5 + ) + + # 提取token使用统计 + token_usage = TokenUsage() + if response.raw_completion and hasattr(response.raw_completion, 'usage'): + usage = response.raw_completion.usage + token_usage.prompt_tokens = usage.prompt_tokens + token_usage.completion_tokens = usage.completion_tokens + token_usage.total_tokens = usage.total_tokens + + # 解析响应 + if hasattr(response, 'completion_text'): + result_text = response.completion_text + else: + result_text = str(response) + + # 尝试解析JSON + try: + json_match = re.search(r'\[.*\]', result_text, re.DOTALL) + if json_match: + titles_data = json.loads(json_match.group()) + return [UserTitle(**title) for title in titles_data], token_usage + except: + pass + + return [], token_usage + + except Exception as e: + logger.error(f"用户称号分析失败: {e}") + return [], TokenUsage() + + async def analyze_golden_quotes(self, messages: List[Dict]) -> Tuple[List[GoldenQuote], TokenUsage]: + """使用LLM分析群聊金句""" + try: + # 提取有趣的文本消息 + interesting_messages = [] + for msg in messages: + sender = msg.get("sender", {}) + nickname = sender.get("nickname", "") or sender.get("card", "") + msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M") + + for content in msg.get("message", []): + if content.get("type") == "text": + text = content.get("data", {}).get("text", "").strip() + # 过滤长度适中、可能有趣的消息 + if 5 <= len(text) <= 100 and not text.startswith(("http", "www", "/")): + interesting_messages.append({ + "sender": nickname, + "time": msg_time, + "content": text + }) + + if not interesting_messages: + return [], TokenUsage() + + # 构建LLM提示词 + messages_text = "\n".join([ + f"[{msg['time']}] {msg['sender']}: {msg['content']}" + for msg in interesting_messages + ]) + + # 计算金句数量 + max_golden_quotes = self.config_manager.get_max_golden_quotes() + + prompt = f""" +请从以下群聊记录中挑选出{max_golden_quotes}句最具冲击力、最令人惊叹的"金句"。这些金句需满足： +- 核心标准：**逆天的神人发言**，即具备颠覆常识的脑洞、逻辑跳脱的表达或强烈反差感的原创内容 +- 典型特征：包含某些争议话题元素、夸张类比、反常规结论、一本正经的"胡说八道"或突破语境的清奇思路，并且具备一定的冲击力，让人印象深刻。 + +对于每个金句，请提供： +1. 原文内容（完整保留发言细节） +2. 发言人昵称 +3. 选择理由（具体说明其"逆天"之处，如逻辑颠覆点/脑洞角度/反差感/争议话题元素） + +此外，我将对你进行严格约束： +- 优先筛选 **逆天指数最高** 的内容：发情、性压抑话题 > 争议话题 > 元素级 > 颠覆认知级 > 逻辑跳脱级 > 趣味调侃级，剔除单纯玩梗或网络热词堆砌的普通发言 +- 重点标记包含极端类比、反常识论证或无厘头结论的内容，并且包含一定的争议话题元素。 + +群聊记录： +{messages_text} + +请以JSON格式返回，格式如下： +[ + {{ + "content": "金句原文", + "sender": "发言人昵称", + "reason": "选择这句话的理由（需明确说明逆天特质）" + }} +] +""" + + # 调用LLM + provider = self.context.get_using_provider() + if not provider: + logger.warning("未配置LLM提供商，跳过金句分析") + return [], TokenUsage() + + response = await provider.text_chat( + prompt=prompt, + max_tokens=1500, + temperature=0.7 + ) + + # 提取token使用统计 + token_usage = TokenUsage() + if response.raw_completion and hasattr(response.raw_completion, 'usage'): + usage = response.raw_completion.usage + token_usage.prompt_tokens = usage.prompt_tokens + token_usage.completion_tokens = usage.completion_tokens + token_usage.total_tokens = usage.total_tokens + + # 解析响应 + if hasattr(response, 'completion_text'): + result_text = response.completion_text + else: + result_text = str(response) + + # 尝试解析JSON + try: + json_match = re.search(r'\[.*\]', result_text, re.DOTALL) + if json_match: + quotes_data = json.loads(json_match.group()) + return [GoldenQuote(**quote) for quote in quotes_data[:max_golden_quotes]], token_usage + except: + pass + + return [], token_usage + + except Exception as e: + logger.error(f"金句分析失败: {e}") + return [], TokenUsage() \ No newline at end of file diff --git a/src/analysis/statistics.py b/src/analysis/statistics.py new file mode 100644 index 0000000..c53db24 --- /dev/null +++ b/src/analysis/statistics.py @@ -0,0 +1,88 @@ +""" +统计分析模块 +负责用户活跃度分析和其他统计功能 +""" + +from datetime import datetime +from typing import List, Dict +from collections import defaultdict + + +class UserAnalyzer: + """用户分析器""" + + def __init__(self, config_manager): + self.config_manager = config_manager + + def analyze_users(self, messages: List[Dict]) -> Dict[str, Dict]: + """分析用户活跃度""" + 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", "")) + nickname = sender.get("nickname", "") or sender.get("card", "") + + 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) + elif content.get("type") == "face": + 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]: + """获取最活跃的用户""" + users = [] + for user_id, stats in user_analysis.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_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) + } \ No newline at end of file diff --git a/src/core/__init__.py b/src/core/__init__.py new file mode 100644 index 0000000..9e7dda8 --- /dev/null +++ b/src/core/__init__.py @@ -0,0 +1,11 @@ +""" +核心功能模块 +""" + +from .config import ConfigManager +from .message_handler import MessageHandler + +__all__ = [ + 'ConfigManager', + 'MessageHandler' +] \ No newline at end of file diff --git a/src/core/config.py b/src/core/config.py new file mode 100644 index 0000000..7bb908b --- /dev/null +++ b/src/core/config.py @@ -0,0 +1,263 @@ +""" +配置管理模块 +负责处理插件配置和PDF依赖检查 +""" + +import sys +import importlib +from pathlib import Path +from typing import Optional, List +from astrbot.api import logger, AstrBotConfig + + +class ConfigManager: + """配置管理器""" + + def __init__(self, config: AstrBotConfig): + self.config = config + self._pyppeteer_available = False + self._pyppeteer_version = None + self._check_pyppeteer_availability() + + def get_enabled_groups(self) -> List[str]: + """获取启用的群组列表""" + return self.config.get("enabled_groups", []) + + 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) -> str: + """获取自动分析时间""" + return self.config.get("auto_analysis_time", "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_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_max_query_rounds(self) -> int: + """获取最大查询轮数""" + return self.config.get("max_query_rounds", 35) + + def get_pdf_output_dir(self) -> str: + """获取PDF输出目录""" + return self.config.get("pdf_output_dir", "data/plugins/astrbot-qq-group-daily-analysis/reports") + + def get_pdf_filename_format(self) -> str: + """获取PDF文件名格式""" + return self.config.get("pdf_filename_format", "群聊分析报告_{group_id}_{date}.pdf") + + def set_output_format(self, format_type: str): + """设置输出格式""" + self.config["output_format"] = format_type + self.config.save_config() + + def set_enabled_groups(self, groups: List[str]): + """设置启用的群组列表""" + self.config["enabled_groups"] = groups + 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_str: str): + """设置自动分析时间""" + self.config["auto_analysis_time"] = time_str + 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_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_max_query_rounds(self, rounds: int): + """设置最大查询轮数""" + self.config["max_query_rounds"] = rounds + 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 add_enabled_group(self, group_id: str): + """添加启用的群组""" + enabled_groups = self.get_enabled_groups() + if group_id not in enabled_groups: + enabled_groups.append(group_id) + self.config["enabled_groups"] = enabled_groups + self.config.save_config() + + def remove_enabled_group(self, group_id: str): + """移除启用的群组""" + enabled_groups = self.get_enabled_groups() + if group_id in enabled_groups: + enabled_groups.remove(group_id) + self.config["enabled_groups"] = enabled_groups + self.config.save_config() + + @property + def pyppeteer_available(self) -> bool: + """检查pyppeteer是否可用""" + return self._pyppeteer_available + + @property + def pyppeteer_version(self) -> Optional[str]: + """获取pyppeteer版本""" + return self._pyppeteer_version + + def _check_pyppeteer_availability(self): + """检查 pyppeteer 可用性""" + try: + import pyppeteer + from pyppeteer import launch + self._pyppeteer_available = True + + # 检查版本 + try: + self._pyppeteer_version = pyppeteer.__version__ + logger.info(f"使用 pyppeteer {self._pyppeteer_version} 作为 PDF 引擎") + except AttributeError: + self._pyppeteer_version = "unknown" + logger.info("使用 pyppeteer (版本未知) 作为 PDF 引擎") + + except ImportError: + self._pyppeteer_available = False + self._pyppeteer_version = None + logger.warning("pyppeteer 未安装，PDF 功能将不可用。请使用 /安装PDF 命令安装 pyppeteer==1.0.2") + + def reload_pyppeteer(self) -> bool: + """重新加载 pyppeteer 模块""" + try: + logger.info("开始重新加载 pyppeteer 模块...") + + # 移除所有 pyppeteer 相关模块 + modules_to_remove = [mod for mod in sys.modules.keys() if mod.startswith('pyppeteer')] + logger.info(f"移除模块: {modules_to_remove}") + for mod in modules_to_remove: + del sys.modules[mod] + + # 强制重新导入 + try: + import pyppeteer + from pyppeteer import launch + + # 更新全局变量 + self._pyppeteer_available = True + try: + self._pyppeteer_version = pyppeteer.__version__ + logger.info(f"重新加载成功，pyppeteer 版本: {self._pyppeteer_version}") + except AttributeError: + self._pyppeteer_version = "unknown" + logger.info("重新加载成功，pyppeteer 版本未知") + + return True + + except ImportError as e: + logger.info(f"pyppeteer 重新导入需要重启 AstrBot 才能生效") + logger.info("💡 提示：pyppeteer 安装成功，但需要重启 AstrBot 后才能使用 PDF 功能") + self._pyppeteer_available = False + self._pyppeteer_version = None + return False + except Exception as e: + logger.info(f"pyppeteer 重新导入需要重启 AstrBot 才能生效") + logger.info("💡 提示：pyppeteer 安装成功，但需要重启 AstrBot 后才能使用 PDF 功能") + self._pyppeteer_available = False + self._pyppeteer_version = None + return False + + except Exception as e: + logger.error(f"重新加载 pyppeteer 时出错: {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}") \ No newline at end of file diff --git a/src/core/message_handler.py b/src/core/message_handler.py new file mode 100644 index 0000000..00376c4 --- /dev/null +++ b/src/core/message_handler.py @@ -0,0 +1,167 @@ +""" +消息处理模块 +负责群聊消息的获取、过滤和预处理 +""" + +import asyncio +from datetime import datetime, timedelta +from typing import List, Dict, Optional +from collections import defaultdict +from astrbot.api import logger +from ...src.models.data_models import GroupStatistics, TokenUsage + + +class MessageHandler: + """消息处理器""" + + def __init__(self, config_manager): + self.config_manager = config_manager + self.bot_qq_id = None + + async def set_bot_qq_id(self, bot_instance): + """设置机器人QQ号""" + try: + if bot_instance and not self.bot_qq_id: + login_info = await bot_instance.api.call_action("get_login_info") + self.bot_qq_id = str(login_info.get("user_id", "")) + logger.info(f"获取到机器人QQ号: {self.bot_qq_id}") + except Exception as e: + logger.error(f"获取机器人QQ号失败: {e}") + + async def fetch_group_messages(self, bot_instance, group_id: str, days: int) -> List[Dict]: + """获取群聊消息记录""" + try: + if not bot_instance or not group_id: + logger.error(f"群 {group_id} 无效的客户端或群组ID") + return [] + + # 计算时间范围 + end_time = datetime.now() + start_time = end_time - timedelta(days=days) + + messages = [] + message_seq = 0 + query_rounds = 0 + max_rounds = self.config_manager.get_max_query_rounds() + max_messages = self.config_manager.get_max_messages() + consecutive_failures = 0 + max_failures = 3 + + 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')}") + + while len(messages) < max_messages and query_rounds < max_rounds: + try: + payloads = { + "group_id": group_id, + "message_seq": message_seq, + "count": 200, + "reverseOrder": True, + } + + result = await bot_instance.api.call_action("get_group_msg_history", **payloads) + + if not result or "messages" not in result: + logger.warning(f"群 {group_id} API返回无效结果: {result}") + consecutive_failures += 1 + if consecutive_failures >= max_failures: + break + continue + + round_messages = result.get("messages", []) + + if not round_messages: + logger.info(f"群 {group_id} 没有更多消息，结束获取") + break + + # 重置失败计数 + consecutive_failures = 0 + + # 过滤时间范围内的消息 + valid_messages_in_round = 0 + oldest_msg_time = None + + for msg in round_messages: + try: + msg_time = datetime.fromtimestamp(msg.get("time", 0)) + oldest_msg_time = msg_time + + # 过滤掉机器人自己的消息 + sender_id = str(msg.get("sender", {}).get("user_id", "")) + if self.bot_qq_id and sender_id == self.bot_qq_id: + continue + + if msg_time >= start_time and msg_time <= end_time: + messages.append(msg) + valid_messages_in_round += 1 + except Exception as msg_error: + logger.warning(f"群 {group_id} 处理单条消息失败: {msg_error}") + continue + + # 如果最老的消息时间已经超出范围，停止获取 + if oldest_msg_time and oldest_msg_time < start_time: + logger.info(f"群 {group_id} 已获取到时间范围外的消息，停止获取。共获取 {len(messages)} 条消息") + break + + if valid_messages_in_round == 0: + logger.warning(f"群 {group_id} 本轮未获取到有效消息") + break + + message_seq = round_messages[0]["message_id"] + query_rounds += 1 + + # 添加延迟避免请求过快 + if query_rounds % 5 == 0: + await asyncio.sleep(0.5) + + except Exception as e: + logger.error(f"群 {group_id} 获取消息失败 (第{query_rounds+1}轮): {e}") + consecutive_failures += 1 + if consecutive_failures >= max_failures: + logger.error(f"群 {group_id} 连续失败 {max_failures} 次，停止获取") + break + await asyncio.sleep(1) + + logger.info(f"群 {group_id} 消息获取完成，共获取 {len(messages)} 条消息，查询轮数: {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_count = 0 + + 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_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" + + 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_count, + token_usage=TokenUsage() + ) \ No newline at end of file diff --git a/src/models/__init__.py b/src/models/__init__.py new file mode 100644 index 0000000..5a42962 --- /dev/null +++ b/src/models/__init__.py @@ -0,0 +1,19 @@ +""" +数据模型模块 +""" + +from .data_models import ( + SummaryTopic, + UserTitle, + GoldenQuote, + TokenUsage, + GroupStatistics +) + +__all__ = [ + 'SummaryTopic', + 'UserTitle', + 'GoldenQuote', + 'TokenUsage', + 'GroupStatistics' +] \ No newline at end of file diff --git a/src/models/data_models.py b/src/models/data_models.py new file mode 100644 index 0000000..1c71c2e --- /dev/null +++ b/src/models/data_models.py @@ -0,0 +1,53 @@ +""" +数据模型定义 +包含所有分析相关的数据结构 +""" + +from dataclasses import dataclass, field +from typing import List + + +@dataclass +class SummaryTopic: + """话题总结数据结构""" + topic: str + contributors: List[str] + detail: str + + +@dataclass +class UserTitle: + """用户称号数据结构""" + name: str + qq: int + title: str + mbti: str + reason: str + + +@dataclass +class GoldenQuote: + """群聊金句数据结构""" + content: str + sender: str + reason: str + + +@dataclass +class TokenUsage: + """Token使用统计""" + prompt_tokens: int = 0 + completion_tokens: int = 0 + total_tokens: int = 0 + + +@dataclass +class GroupStatistics: + """群聊统计数据结构""" + message_count: int + total_characters: int + participant_count: int + most_active_period: str + golden_quotes: List[GoldenQuote] + emoji_count: int + token_usage: TokenUsage = field(default_factory=TokenUsage) \ No newline at end of file diff --git a/src/reports/__init__.py b/src/reports/__init__.py new file mode 100644 index 0000000..267df1e --- /dev/null +++ b/src/reports/__init__.py @@ -0,0 +1,12 @@ +""" +报告生成模块 +包含HTML、PDF、文本报告生成功能 +""" + +from .generators import ReportGenerator +from .templates import HTMLTemplates + +__all__ = [ + 'ReportGenerator', + 'HTMLTemplates' +] \ No newline at end of file diff --git a/src/reports/generators.py b/src/reports/generators.py new file mode 100644 index 0000000..86fd3b4 --- /dev/null +++ b/src/reports/generators.py @@ -0,0 +1,323 @@ +""" +报告生成器模块 +负责生成各种格式的分析报告 +""" + +import base64 +import aiohttp +from datetime import datetime +from typing import Dict, Optional +from pathlib import Path +from astrbot.api import logger +from .templates import HTMLTemplates + + +class ReportGenerator: + """报告生成器""" + + def __init__(self, config_manager): + self.config_manager = config_manager + + async def generate_image_report(self, analysis_result: Dict, group_id: str, html_render_func) -> Optional[str]: + """生成图片格式的分析报告""" + try: + # 准备渲染数据 + render_payload = await self._prepare_render_data(analysis_result) + + # 使用AstrBot内置的HTML渲染服务（直接传递模板和数据） + image_url = await html_render_func(HTMLTemplates.get_image_template(), render_payload) + return image_url + + except Exception as e: + logger.error(f"生成图片报告失败: {e}") + return None + + async def generate_pdf_report(self, analysis_result: Dict, group_id: str) -> Optional[str]: + """生成PDF格式的分析报告""" + try: + # 确保输出目录存在 + output_dir = Path(self.config_manager.get_pdf_output_dir()) + output_dir.mkdir(parents=True, exist_ok=True) + + # 生成文件名 + current_date = datetime.now().strftime('%Y%m%d') + filename = self.config_manager.get_pdf_filename_format().format( + group_id=group_id, + date=current_date + ) + pdf_path = output_dir / filename + + # 准备渲染数据 + render_data = await self._prepare_render_data(analysis_result) + logger.info(f"PDF 渲染数据准备完成，包含 {len(render_data)} 个字段") + + # 生成 HTML 内容（PDF模板使用{}占位符） + html_content = self._render_html_template(HTMLTemplates.get_pdf_template(), render_data, use_jinja_style=False) + logger.info(f"HTML 内容生成完成，长度: {len(html_content)} 字符") + + # 转换为 PDF + success = await self._html_to_pdf(html_content, str(pdf_path)) + + if success: + return str(pdf_path.absolute()) + else: + return None + + except Exception as e: + logger.error(f"生成 PDF 报告失败: {e}") + return None + + def generate_text_report(self, analysis_result: Dict) -> str: + """生成文本格式的分析报告""" + stats = analysis_result["statistics"] + topics = analysis_result["topics"] + user_titles = analysis_result["user_titles"] + + report = f""" +🎯 群聊日常分析报告 +📅 {datetime.now().strftime('%Y年%m月%d日')} + +📊 基础统计 +• 消息总数: {stats.message_count} +• 参与人数: {stats.participant_count} +• 总字符数: {stats.total_characters} +• 表情数量: {stats.emoji_count} +• 最活跃时段: {stats.most_active_period} + +💬 热门话题 +""" + + max_topics = self.config_manager.get_max_topics() + for i, topic in enumerate(topics[:max_topics], 1): + contributors_str = "、".join(topic.contributors) + report += f"{i}. {topic.topic}\n" + report += f" 参与者: {contributors_str}\n" + report += f" {topic.detail}\n\n" + + report += "🏆 群友称号\n" + max_user_titles = self.config_manager.get_max_user_titles() + for title in user_titles[:max_user_titles]: + report += f"• {title.name} - {title.title} ({title.mbti})\n" + report += f" {title.reason}\n\n" + + report += "💬 群圣经\n" + max_golden_quotes = self.config_manager.get_max_golden_quotes() + for i, quote in enumerate(stats.golden_quotes[:max_golden_quotes], 1): + report += f"{i}. \"{quote.content}\" —— {quote.sender}\n" + report += f" {quote.reason}\n\n" + + return report + + async def _prepare_render_data(self, analysis_result: Dict) -> Dict: + """准备渲染数据""" + stats = analysis_result["statistics"] + topics = analysis_result["topics"] + user_titles = analysis_result["user_titles"] + + # 构建话题HTML + topics_html = "" + max_topics = self.config_manager.get_max_topics() + for i, topic in enumerate(topics[:max_topics], 1): + contributors_str = "、".join(topic.contributors) + topics_html += f""" +
+
+ {i} + {topic.topic} +
+
参与者: {contributors_str}
+
{topic.detail}
+
+ """ + + # 构建用户称号HTML（包含头像） + titles_html = "" + max_user_titles = self.config_manager.get_max_user_titles() + for title in user_titles[:max_user_titles]: + # 获取用户头像 + avatar_data = await self._get_user_avatar(str(title.qq)) + avatar_html = f'头像' if avatar_data else '
👤
' + + titles_html += f""" +
+
+ {avatar_html} +
+
{title.name}
+
+
{title.title}
+
{title.mbti}
+
+
+
+
{title.reason}
+
+ """ + + # 构建金句HTML + quotes_html = "" + max_golden_quotes = self.config_manager.get_max_golden_quotes() + for quote in stats.golden_quotes[:max_golden_quotes]: + quotes_html += f""" +
+
"{quote.content}"
+
—— {quote.sender}
+
{quote.reason}
+
+ """ + + # 返回扁平化的渲染数据 + return { + "current_date": datetime.now().strftime('%Y年%m月%d日'), + "current_datetime": datetime.now().strftime('%Y-%m-%d %H:%M:%S'), + "message_count": stats.message_count, + "participant_count": stats.participant_count, + "total_characters": stats.total_characters, + "emoji_count": stats.emoji_count, + "most_active_period": stats.most_active_period, + "topics_html": topics_html, + "titles_html": titles_html, + "quotes_html": quotes_html, + "total_tokens": stats.token_usage.total_tokens, + "prompt_tokens": stats.token_usage.prompt_tokens, + "completion_tokens": stats.token_usage.completion_tokens + } + + def _render_html_template(self, template: str, data: Dict, use_jinja_style: bool = False) -> str: + """HTML模板渲染，支持两种占位符格式 + + Args: + template: HTML模板字符串 + data: 渲染数据 + use_jinja_style: 是否使用Jinja2风格的{{ }}占位符，否则使用{}占位符 + """ + result = template + + # 调试：记录渲染数据 + logger.info(f"渲染数据键: {list(data.keys())}, 使用Jinja风格: {use_jinja_style}") + + for key, value in data.items(): + if use_jinja_style: + # 图片模板使用{{ }}占位符 + placeholder = f"{{{{ {key} }}}}" + else: + # PDF模板使用{}占位符 + placeholder = f"{{{key}}}" + + # 调试：记录替换过程 + if placeholder in result: + logger.debug(f"替换 {placeholder} -> {str(value)[:100]}...") + result = result.replace(placeholder, str(value)) + + # 检查是否还有未替换的占位符 + import re + if use_jinja_style: + remaining_placeholders = re.findall(r'\{\{[^}]+\}\}', result) + else: + remaining_placeholders = re.findall(r'\{[^}]+\}', result) + + if remaining_placeholders: + logger.warning(f"未替换的占位符: {remaining_placeholders[:10]}") + + return result + + async def _get_user_avatar(self, user_id: str) -> Optional[str]: + """获取用户头像的base64编码""" + try: + avatar_url = f"https://q4.qlogo.cn/headimg_dl?dst_uin={user_id}&spec=640" + async with aiohttp.ClientSession() as client: + response = await client.get(avatar_url) + response.raise_for_status() + avatar_data = await response.read() + # 转换为base64编码 + avatar_base64 = base64.b64encode(avatar_data).decode('utf-8') + return f"data:image/jpeg;base64,{avatar_base64}" + except Exception as e: + logger.error(f"获取用户头像失败 {user_id}: {e}") + return None + + async def _html_to_pdf(self, html_content: str, output_path: str) -> bool: + """将 HTML 内容转换为 PDF 文件""" + try: + # 确保 pyppeteer 可用 + if not self.config_manager.pyppeteer_available: + logger.error("pyppeteer 不可用，无法生成 PDF") + return False + + # 动态导入 pyppeteer + import pyppeteer + from pyppeteer import launch + import sys + import os + + # 尝试启动浏览器，如果 Chromium 不存在会自动下载 + logger.info("启动浏览器进行 PDF 转换") + + # 配置浏览器启动参数，避免 Chromium 下载问题 + launch_options = { + 'headless': True, + 'args': [ + '--no-sandbox', + '--disable-setuid-sandbox', + '--disable-dev-shm-usage', + '--disable-gpu', + '--no-first-run', + '--disable-extensions', + '--disable-default-apps' + ] + } + + # 如果是 Windows 系统，尝试使用系统 Chrome + if sys.platform.startswith('win'): + # 常见的 Chrome 安装路径 + chrome_paths = [ + r"C:\Program Files\Google\Chrome\Application\chrome.exe", + r"C:\Program Files (x86)\Google\Chrome\Application\chrome.exe", + r"C:\Users\{}\AppData\Local\Google\Chrome\Application\chrome.exe".format(os.environ.get('USERNAME', '')), + ] + + for chrome_path in chrome_paths: + if Path(chrome_path).exists(): + launch_options['executablePath'] = chrome_path + logger.info(f"使用系统 Chrome: {chrome_path}") + break + + browser = await launch(**launch_options) + page = await browser.newPage() + + # 设置页面内容 (pyppeteer 1.0.2 版本的 API) + await page.setContent(html_content) + # 等待页面加载完成 + try: + await page.waitForSelector('body', {'timeout': 10000}) + except Exception: + # 如果等待失败，继续执行（可能页面已经加载完成） + pass + + # 导出 PDF + await page.pdf({ + 'path': output_path, + 'format': 'A4', + 'printBackground': True, + 'margin': { + 'top': '10mm', + 'right': '10mm', + 'bottom': '10mm', + 'left': '10mm' + }, + 'scale': 0.8 + }) + + await browser.close() + logger.info(f"PDF 生成成功: {output_path}") + return True + + except Exception as e: + error_msg = str(e) + if "Chromium downloadable not found" in error_msg: + logger.error("Chromium 下载失败，建议安装 pyppeteer2 或使用系统 Chrome") + elif "No usable sandbox" in error_msg: + logger.error("沙盒权限问题，已尝试禁用沙盒") + else: + logger.error(f"HTML 转 PDF 失败: {e}") + return False \ No newline at end of file diff --git a/src/reports/templates.py b/src/reports/templates.py new file mode 100644 index 0000000..4077f46 --- /dev/null +++ b/src/reports/templates.py @@ -0,0 +1,200 @@ +""" +HTML模板模块 +严格按照main-backup中的实现，包含图片报告和PDF报告的不同HTML模板 +""" + + +class HTMLTemplates: + """HTML模板管理类""" + + @staticmethod + def get_image_template() -> str: + """获取图片报告的HTML模板（使用{{ }}占位符）""" + return """ + + + + + 群聊日常分析报告 + + + + +
+
+

📊 群聊日常分析报告

+
{{ current_date }}
+
+
+
+

📈 基础统计

+
+
{{ message_count }}
消息总数
+
{{ participant_count }}
参与人数
+
{{ total_characters }}
总字符数
+
{{ emoji_count }}
表情数量
+
+
+
{{ most_active_period }}
+
最活跃时段
+
+
+
+

💬 热门话题

+
{{ topics_html | safe }}
+
+
+

🏆 群友称号

+
{{ titles_html | safe }}
+
+
+

💬 群圣经

+ {{ quotes_html | safe }} +
+
+
+ 由 AstrBot QQ群日常分析插件 生成 | {{ current_datetime }} | SXP-Simon/astrbot-qq-group-daily-analysis
+ 🤖 AI分析消耗：{{ total_tokens }} tokens (输入: {{ prompt_tokens }}, 输出: {{ completion_tokens }}) +
+
+ +""" + + @staticmethod + def get_pdf_template() -> str: + """获取PDF报告的HTML模板（使用{}占位符）""" + return """ + + + + + 群聊日常分析报告 + + + +
+
+

📊 群聊日常分析报告

+
{current_date}
+
+
+

📈 基础统计

+
+
{message_count}
消息总数
+
{participant_count}
参与人数
+
{total_characters}
总字符数
+
{emoji_count}
表情数量
+
+
+
{most_active_period}
+
最活跃时段
+
+
+
+

💬 热门话题

+ {topics_html} +
+
+

🏆 群友称号

+ {titles_html} +
+
+

💬 群圣经

+ {quotes_html} +
+
+ 由 AstrBot QQ群日常分析插件 生成 | {current_datetime} | SXP-Simon/astrbot-qq-group-daily-analysis
+ 🤖 AI分析消耗：{total_tokens} tokens (输入: {prompt_tokens}, 输出: {completion_tokens}) +
+
+ +""" diff --git a/src/scheduler/__init__.py b/src/scheduler/__init__.py new file mode 100644 index 0000000..67a693f --- /dev/null +++ b/src/scheduler/__init__.py @@ -0,0 +1,10 @@ +""" +调度和自动化模块 +包含定时任务和自动分析功能 +""" + +from .auto_scheduler import AutoScheduler + +__all__ = [ + 'AutoScheduler' +] \ No newline at end of file diff --git a/src/scheduler/auto_scheduler.py b/src/scheduler/auto_scheduler.py new file mode 100644 index 0000000..3593174 --- /dev/null +++ b/src/scheduler/auto_scheduler.py @@ -0,0 +1,317 @@ +""" +自动调度器模块 +负责定时任务和自动分析功能 +""" + +import asyncio +from datetime import datetime, timedelta +from typing import Optional +from astrbot.api import logger + + +class AutoScheduler: + """自动调度器""" + + def __init__(self, config_manager, message_handler, analyzer, report_generator, html_render_func=None): + self.config_manager = config_manager + self.message_handler = message_handler + self.analyzer = analyzer + self.report_generator = report_generator + self.html_render_func = html_render_func + self.scheduler_task = None + self.bot_instance = None + self.last_execution_date = None # 记录上次执行日期，防止重复执行 + + def set_bot_instance(self, bot_instance): + """设置bot实例""" + self.bot_instance = bot_instance + # 同时设置消息处理器的bot实例 + asyncio.create_task(self.message_handler.set_bot_qq_id(bot_instance)) + + async def start_scheduler(self): + """启动定时任务调度器""" + if not self.config_manager.get_enable_auto_analysis(): + logger.info("自动分析功能未启用") + return + + # 延迟启动，给系统时间初始化 + await asyncio.sleep(10) + + logger.info(f"启动定时任务调度器，自动分析时间: {self.config_manager.get_auto_analysis_time()}") + + self.scheduler_task = asyncio.create_task(self._scheduler_loop()) + + async def stop_scheduler(self): + """停止定时任务调度器""" + if self.scheduler_task and not self.scheduler_task.done(): + self.scheduler_task.cancel() + logger.info("已停止定时任务调度器") + + async def restart_scheduler(self): + """重启定时任务调度器""" + await self.stop_scheduler() + if self.config_manager.get_enable_auto_analysis(): + await self.start_scheduler() + + async def _scheduler_loop(self): + """调度器主循环""" + while True: + try: + now = datetime.now() + target_time = datetime.strptime(self.config_manager.get_auto_analysis_time(), "%H:%M").replace( + year=now.year, month=now.month, day=now.day + ) + + # 如果今天的目标时间已过，设置为明天 + if now >= target_time: + target_time += timedelta(days=1) + + # 计算等待时间 + wait_seconds = (target_time - now).total_seconds() + logger.info(f"定时分析将在 {target_time.strftime('%Y-%m-%d %H:%M:%S')} 执行，等待 {wait_seconds:.0f} 秒") + + # 等待到目标时间 + await asyncio.sleep(wait_seconds) + + # 执行自动分析 + if self.config_manager.get_enable_auto_analysis(): + # 检查是否今天已经执行过 + today = now.date() + if self.last_execution_date == today: + logger.info(f"今天 {today} 已经执行过自动分析，跳过执行") + # 等待到明天再检查 + await asyncio.sleep(3600) # 等待1小时后再检查 + continue + + logger.info("开始执行定时分析") + await self._run_auto_analysis() + self.last_execution_date = today # 记录执行日期 + logger.info(f"定时分析执行完成，记录执行日期: {today}") + else: + logger.info("自动分析已禁用，跳过执行") + break + + except Exception as e: + logger.error(f"定时任务调度器错误: {e}") + # 等待5分钟后重试 + await asyncio.sleep(300) + + async def _run_auto_analysis(self): + """执行自动分析""" + try: + logger.info("开始执行自动群聊分析") + + # 为每个启用的群执行分析 + enabled_groups = self.config_manager.get_enabled_groups() + for group_id in enabled_groups: + try: + logger.info(f"为群 {group_id} 执行自动分析") + await self._perform_auto_analysis_for_group(group_id) + except Exception as e: + logger.error(f"群 {group_id} 自动分析失败: {e}") + + except Exception as e: + logger.error(f"自动分析执行失败: {e}") + + async def _perform_auto_analysis_for_group(self, group_id: str): + """为指定群执行自动分析""" + try: + if not self.bot_instance: + logger.warning(f"群 {group_id} 自动分析跳过：未获取到bot实例") + return + + logger.info(f"开始为群 {group_id} 执行自动分析") + + # 获取群聊消息 + analysis_days = self.config_manager.get_analysis_days() + messages = await self.message_handler.fetch_group_messages(self.bot_instance, group_id, analysis_days) + if not messages: + logger.warning(f"群 {group_id} 未获取到足够的消息记录") + return + + # 检查消息数量 + min_threshold = self.config_manager.get_min_messages_threshold() + if len(messages) < min_threshold: + logger.warning(f"群 {group_id} 消息数量不足（{len(messages)}条），跳过分析") + return + + logger.info(f"群 {group_id} 获取到 {len(messages)} 条消息，开始分析") + + # 进行分析 + analysis_result = await self.analyzer.analyze_messages(messages, group_id) + if not analysis_result: + logger.error(f"群 {group_id} 分析失败") + return + + # 生成并发送报告 + await self._send_analysis_report(group_id, analysis_result) + + except Exception as e: + logger.error(f"群 {group_id} 自动分析执行失败: {e}", exc_info=True) + + async def _send_analysis_report(self, group_id: str, analysis_result: dict): + """发送分析报告到群""" + try: + output_format = self.config_manager.get_output_format() + + if output_format == "image": + if self.html_render_func: + # 使用图片格式 + logger.info(f"群 {group_id} 自动分析使用图片报告格式") + image_url = await self.report_generator.generate_image_report(analysis_result, group_id, self.html_render_func) + if image_url: + await self._send_image_message(group_id, image_url) + logger.info(f"群 {group_id} 图片报告发送成功") + else: + # 图片生成失败，回退到文本 + logger.warning(f"群 {group_id} 图片报告生成失败，回退到文本报告") + text_report = self.report_generator.generate_text_report(analysis_result) + await self._send_text_message(group_id, f"📊 每日群聊分析报告：\n\n{text_report}") + else: + # 没有html_render函数，回退到文本报告 + logger.warning(f"群 {group_id} 缺少html_render函数，回退到文本报告") + text_report = self.report_generator.generate_text_report(analysis_result) + await self._send_text_message(group_id, f"📊 每日群聊分析报告：\n\n{text_report}") + + elif output_format == "pdf": + if not self.config_manager.pyppeteer_available: + logger.warning(f"群 {group_id} PDF功能不可用，回退到文本报告") + text_report = self.report_generator.generate_text_report(analysis_result) + await self._send_text_message(group_id, f"📊 每日群聊分析报告：\n\n{text_report}") + else: + pdf_path = await self.report_generator.generate_pdf_report(analysis_result, group_id) + if pdf_path: + await self._send_pdf_file(group_id, pdf_path) + logger.info(f"群 {group_id} 自动分析完成，已发送PDF报告") + else: + logger.error(f"群 {group_id} PDF报告生成失败，回退到文本报告") + text_report = self.report_generator.generate_text_report(analysis_result) + await self._send_text_message(group_id, f"📊 每日群聊分析报告：\n\n{text_report}") + else: + text_report = self.report_generator.generate_text_report(analysis_result) + await self._send_text_message(group_id, f"📊 每日群聊分析报告：\n\n{text_report}") + + logger.info(f"群 {group_id} 自动分析完成，已发送报告") + + except Exception as e: + logger.error(f"发送分析报告到群 {group_id} 失败: {e}") + + async def _send_image_message(self, group_id: str, image_url: str): + """发送图片消息到群""" + try: + if not self.bot_instance: + logger.error(f"群 {group_id} 发送图片失败：缺少bot实例") + return + + # 发送图片消息到群 + await self.bot_instance.api.call_action( + "send_group_msg", + group_id=group_id, + message=[{ + "type": "text", + "data": {"text": "📊 每日群聊分析报告已生成："} + }, { + "type": "image", + "data": {"url": image_url} + }] + ) + logger.info(f"群 {group_id} 图片消息发送成功") + + except Exception as e: + logger.error(f"发送图片消息到群 {group_id} 失败: {e}") + + async def _send_text_message(self, group_id: str, text_content: str): + """发送文本消息到群""" + try: + if not self.bot_instance: + logger.error(f"群 {group_id} 发送文本失败：缺少bot实例") + return + + # 发送文本消息到群 + await self.bot_instance.api.call_action( + "send_group_msg", + group_id=group_id, + message=text_content + ) + logger.info(f"群 {group_id} 文本消息发送成功") + + except Exception as e: + logger.error(f"发送文本消息到群 {group_id} 失败: {e}") + + async def _send_pdf_file(self, group_id: str, pdf_path: str): + """发送PDF文件到群""" + try: + if not self.bot_instance: + logger.error(f"群 {group_id} 发送PDF失败：缺少bot实例") + return + + # 发送PDF文件到群 + await self.bot_instance.api.call_action( + "send_group_msg", + group_id=group_id, + message=[{ + "type": "text", + "data": {"text": "📊 每日群聊分析报告已生成："} + }, { + "type": "file", + "data": {"file": pdf_path} + }] + ) + logger.info(f"群 {group_id} PDF文件发送成功") + + except Exception as e: + logger.error(f"发送PDF文件到群 {group_id} 失败: {e}") + # 发送失败提示 + try: + await self.bot_instance.api.call_action( + "send_group_msg", + group_id=group_id, + message=f"📊 每日群聊分析报告已生成，但发送PDF文件失败。PDF文件路径：{pdf_path}" + ) + except Exception as e2: + logger.error(f"发送PDF失败提示到群 {group_id} 也失败: {e2}") + + async def _send_text_message(self, group_id: str, message: str): + """发送文本消息到群""" + try: + if not self.bot_instance: + return + + await self.bot_instance.api.call_action( + "send_group_msg", + group_id=group_id, + message=message + ) + + except Exception as e: + logger.error(f"发送文本消息到群 {group_id} 失败: {e}") + + async def _send_pdf_file(self, group_id: str, pdf_path: str): + """发送PDF文件到群""" + try: + if not self.bot_instance: + return + + await self.bot_instance.api.call_action( + "send_group_msg", + group_id=group_id, + message=[{ + "type": "text", + "data": {"text": "📊 每日群聊分析报告已生成："} + }, { + "type": "file", + "data": {"file": pdf_path} + }] + ) + + except Exception as e: + logger.error(f"发送PDF文件到群 {group_id} 失败: {e}") + # 如果发送PDF失败，尝试发送提示信息 + try: + await self.bot_instance.api.call_action( + "send_group_msg", + group_id=group_id, + message=f"📊 每日群聊分析报告已生成，但发送PDF文件失败。PDF文件路径：{pdf_path}" + ) + except Exception as e2: + logger.error(f"发送PDF失败提示到群 {group_id} 也失败: {e2}") \ No newline at end of file diff --git a/src/utils/__init__.py b/src/utils/__init__.py new file mode 100644 index 0000000..c70fe49 --- /dev/null +++ b/src/utils/__init__.py @@ -0,0 +1,12 @@ +""" +工具函数模块 +包含PDF处理和通用工具函数 +""" + +from .pdf_utils import PDFInstaller +from .helpers import MessageAnalyzer + +__all__ = [ + 'PDFInstaller', + 'MessageAnalyzer' +] \ No newline at end of file diff --git a/src/utils/helpers.py b/src/utils/helpers.py new file mode 100644 index 0000000..4901a55 --- /dev/null +++ b/src/utils/helpers.py @@ -0,0 +1,76 @@ +""" +通用工具函数模块 +包含消息分析和其他通用功能 +""" + +from typing import List, Dict +from ...src.models.data_models import GroupStatistics, SummaryTopic, UserTitle, GoldenQuote, TokenUsage +from ...src.core.message_handler import MessageHandler +from ...src.analysis.llm_analyzer import LLMAnalyzer +from ...src.analysis.statistics import UserAnalyzer + + +class MessageAnalyzer: + """消息分析器 - 整合所有分析功能""" + + def __init__(self, context, config_manager): + self.context = context + self.config_manager = config_manager + self.message_handler = MessageHandler(config_manager) + self.llm_analyzer = LLMAnalyzer(context, config_manager) + self.user_analyzer = UserAnalyzer(config_manager) + + async def set_bot_instance(self, bot_instance): + """设置bot实例""" + await self.message_handler.set_bot_qq_id(bot_instance) + + async def analyze_messages(self, messages: List[Dict], group_id: str) -> Dict: + """完整的消息分析流程""" + try: + # 基础统计 + statistics = self.message_handler.calculate_statistics(messages) + + # 用户分析 + user_analysis = self.user_analyzer.analyze_users(messages) + + # LLM分析 + topics = [] + user_titles = [] + golden_quotes = [] + total_token_usage = TokenUsage() + + # 话题分析 + if self.config_manager.get_topic_analysis_enabled(): + topics, topic_tokens = await self.llm_analyzer.analyze_topics(messages) + 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 self.config_manager.get_user_title_analysis_enabled(): + user_titles, title_tokens = await self.llm_analyzer.analyze_user_titles(messages, user_analysis) + 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 + + # 金句分析 + golden_quotes, quote_tokens = await self.llm_analyzer.analyze_golden_quotes(messages) + 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 + + # 更新统计数据 + 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: + from astrbot.api import logger + logger.error(f"消息分析失败: {e}") + return None \ No newline at end of file diff --git a/src/utils/pdf_utils.py b/src/utils/pdf_utils.py new file mode 100644 index 0000000..c633696 --- /dev/null +++ b/src/utils/pdf_utils.py @@ -0,0 +1,57 @@ +""" +PDF工具模块 +负责PDF相关的安装和管理功能 +""" + +import sys +import asyncio +from astrbot.api import logger + + +class PDFInstaller: + """PDF功能安装器""" + + @staticmethod + async def install_pyppeteer(config_manager): + """安装pyppeteer依赖""" + try: + logger.info("开始安装 pyppeteer...") + + # 使用asyncio安装pyppeteer和兼容的websockets版本 + logger.info("安装 pyppeteer==1.0.2 和兼容的依赖...") + process = await asyncio.create_subprocess_exec( + sys.executable, "-m", "pip", "install", + "pyppeteer==1.0.2", "websockets==10.4", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + + stdout, stderr = await process.communicate() + + if process.returncode == 0: + logger.info("pyppeteer 安装成功") + logger.info(f"安装输出: {stdout.decode()}") + + # 重新加载pyppeteer模块 + success = config_manager.reload_pyppeteer() + if success: + return "✅ pyppeteer 安装成功！PDF 功能现已可用。" + else: + return "⚠️ pyppeteer 安装完成，但重新加载失败。请重启 AstrBot 以使用 PDF 功能。" + else: + error_msg = stderr.decode() + logger.error(f"pyppeteer 安装失败: {error_msg}") + return f"❌ pyppeteer 安装失败: {error_msg}" + + except Exception as e: + logger.error(f"安装 pyppeteer 时出错: {e}") + return f"❌ 安装过程中出错: {str(e)}" + + @staticmethod + def get_pdf_status(config_manager) -> str: + """获取PDF功能状态""" + if config_manager.pyppeteer_available: + version = config_manager.pyppeteer_version or "未知版本" + return f"✅ PDF 功能可用 (pyppeteer {version})" + else: + return "❌ PDF 功能不可用 - 需要安装 pyppeteer" \ No newline at end of file