This commit is contained in:
xiaoxi68
2025-11-09 20:28:51 +08:00
parent f42929938d
commit 4f509ae5f9
29 changed files with 1390 additions and 978 deletions
+82 -38
View File
@@ -1,4 +1,4 @@
"""
"""
QQ群日常分析插件
基于群聊记录生成精美的日常分析报告,包含话题总结、用户画像、统计数据等
@@ -7,12 +7,13 @@ QQ群日常分析插件
import asyncio
from typing import Optional
from pathlib import Path
from astrbot.api.event import filter
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.platform.sources.aiocqhttp.aiocqhttp_message_event import (
AiocqhttpMessageEvent,
)
from astrbot.core.message.components import File
from astrbot.core.star.filter.permission import PermissionType
@@ -39,7 +40,12 @@ class QQGroupDailyAnalysis(Star):
self.config = config
# 初始化模块化组件
global config_manager, bot_manager, message_analyzer, report_generator, auto_scheduler
global \
config_manager, \
bot_manager, \
message_analyzer, \
report_generator, \
auto_scheduler
config_manager = ConfigManager(config)
bot_manager = BotManager(config_manager)
@@ -52,7 +58,7 @@ class QQGroupDailyAnalysis(Star):
message_analyzer,
report_generator,
bot_manager,
self.html_render # 传入html_render函数
self.html_render, # 传入html_render函数
)
# 延迟启动自动调度器,给系统时间初始化
@@ -85,45 +91,52 @@ class QQGroupDailyAnalysis(Star):
"""插件被卸载/停用时调用,清理资源"""
try:
logger.info("开始清理QQ群日常分析插件资源...")
global auto_scheduler, bot_manager, message_analyzer, report_generator, config_manager
global \
auto_scheduler, \
bot_manager, \
message_analyzer, \
report_generator, \
config_manager
# 停止自动调度器
if auto_scheduler:
logger.info("正在停止自动调度器...")
await auto_scheduler.stop_scheduler()
logger.info("自动调度器已停止")
# 清理bot管理器资源
# if bot_manager:
# logger.info("正在清理bot管理器资源...")
# # 如果有其他需要清理的资源,可以在这里添加
# # 清理消息分析器资源
# if message_analyzer:
# logger.info("正在清理消息分析器资源...")
# # 如果有其他需要清理的资源,可以在这里添加
# # 清理报告生成器资源
# # 清理报告生成器资源
# if report_generator:
# logger.info("正在清理报告生成器资源...")
# # 如果有其他需要清理的资源,可以在这里添加
# 重置全局变量
auto_scheduler = None
bot_manager = None
message_analyzer = None
report_generator = None
config_manager = None
logger.info("QQ群日常分析插件资源清理完成")
except Exception as e:
logger.error(f"插件资源清理失败: {e}")
@filter.command("群分析")
@filter.permission_type(PermissionType.ADMIN)
async def analyze_group_daily(self, event: AiocqhttpMessageEvent, days: Optional[int] = None):
async def analyze_group_daily(
self, event: AiocqhttpMessageEvent, days: Optional[int] = None
):
"""
分析群聊日常活动
用法: /群分析 [天数]
@@ -147,7 +160,9 @@ class QQGroupDailyAnalysis(Star):
return
# 设置分析天数
analysis_days = days if days and 1 <= days <= 7 else config_manager.get_analysis_days()
analysis_days = (
days if days and 1 <= days <= 7 else config_manager.get_analysis_days()
)
yield event.plain_result(f"🔍 开始分析群聊近{analysis_days}天的活动,请稍候...")
@@ -156,21 +171,31 @@ class QQGroupDailyAnalysis(Star):
try:
# 获取群聊消息
messages = await message_analyzer.message_handler.fetch_group_messages(bot_manager.get_bot_instance(), group_id, analysis_days)
messages = await message_analyzer.message_handler.fetch_group_messages(
bot_manager.get_bot_instance(), group_id, analysis_days
)
if not messages:
yield event.plain_result("❌ 未找到足够的群聊记录,请确保群内有足够的消息历史")
yield event.plain_result(
"❌ 未找到足够的群聊记录,请确保群内有足够的消息历史"
)
return
# 检查消息数量是否足够分析
min_threshold = config_manager.get_min_messages_threshold()
if len(messages) < min_threshold:
yield event.plain_result(f"❌ 消息数量不足({len(messages)}条),至少需要{min_threshold}条消息才能进行有效分析")
yield event.plain_result(
f"❌ 消息数量不足({len(messages)}条),至少需要{min_threshold}条消息才能进行有效分析"
)
return
yield event.plain_result(f"📊 已获取{len(messages)}条消息,正在进行智能分析...")
yield event.plain_result(
f"📊 已获取{len(messages)}条消息,正在进行智能分析..."
)
# 进行分析 - 传递 unified_msg_origin 以获取正确的 LLM 提供商
analysis_result = await message_analyzer.analyze_messages(messages, group_id, event.unified_msg_origin)
analysis_result = await message_analyzer.analyze_messages(
messages, group_id, event.unified_msg_origin
)
# 检查分析结果
if not analysis_result or not analysis_result.get("statistics"):
@@ -180,23 +205,32 @@ class QQGroupDailyAnalysis(Star):
# 生成报告
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)
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 = report_generator.generate_text_report(analysis_result)
yield event.plain_result(f"⚠️ 图片报告生成失败,以下是文本版本:\n\n{text_report}")
yield event.plain_result(
f"⚠️ 图片报告生成失败,以下是文本版本:\n\n{text_report}"
)
elif output_format == "pdf":
if not config_manager.pyppeteer_available:
yield event.plain_result("❌ PDF 功能不可用,请使用 /安装PDF 命令安装 pyppeteer==1.0.2")
yield event.plain_result(
"❌ PDF 功能不可用,请使用 /安装PDF 命令安装 pyppeteer==1.0.2"
)
return
pdf_path = await report_generator.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)
@@ -212,20 +246,24 @@ class QQGroupDailyAnalysis(Star):
# 回退到文本报告
logger.warning("PDF 报告生成失败,回退到文本报告")
text_report = report_generator.generate_text_report(analysis_result)
yield event.plain_result(f"\n📝 以下是文本版本的分析报告:\n\n{text_report}")
yield event.plain_result(
f"\n📝 以下是文本版本的分析报告:\n\n{text_report}"
)
else:
text_report = report_generator.generate_text_report(analysis_result)
yield event.plain_result(text_report)
except Exception as e:
logger.error(f"群分析失败: {e}", exc_info=True)
yield event.plain_result(f"❌ 分析失败: {str(e)}。请检查网络连接和LLM配置,或联系管理员")
yield event.plain_result(
f"❌ 分析失败: {str(e)}。请检查网络连接和LLM配置,或联系管理员"
)
@filter.command("设置格式")
@filter.permission_type(PermissionType.ADMIN)
async def set_output_format(self, event: AiocqhttpMessageEvent, format_type: str = ""):
async def set_output_format(
self, event: AiocqhttpMessageEvent, format_type: str = ""
):
"""
设置分析报告输出格式
用法: /设置格式 [image|text|pdf]
@@ -241,7 +279,9 @@ class QQGroupDailyAnalysis(Star):
if not format_type:
current_format = config_manager.get_output_format()
pdf_status = '' if config_manager.pyppeteer_available else '❌ (需安装 pyppeteer)'
pdf_status = (
"" if config_manager.pyppeteer_available else "❌ (需安装 pyppeteer)"
)
yield event.plain_result(f"""📊 当前输出格式: {current_format}
可用格式:
@@ -258,7 +298,9 @@ class QQGroupDailyAnalysis(Star):
return
if format_type == "pdf" and not config_manager.pyppeteer_available:
yield event.plain_result("❌ PDF 格式不可用,请使用 /安装PDF 命令安装 pyppeteer==1.0.2")
yield event.plain_result(
"❌ PDF 格式不可用,请使用 /安装PDF 命令安装 pyppeteer==1.0.2"
)
return
config_manager.set_output_format(format_type)
@@ -281,7 +323,7 @@ class QQGroupDailyAnalysis(Star):
# 安装 pyppeteer
result = await PDFInstaller.install_pyppeteer(config_manager)
yield event.plain_result(result)
# 提供系统依赖安装指导
system_deps_result = await PDFInstaller.install_system_deps()
yield event.plain_result(system_deps_result)
@@ -292,7 +334,9 @@ class QQGroupDailyAnalysis(Star):
@filter.command("分析设置")
@filter.permission_type(PermissionType.ADMIN)
async def analysis_settings(self, event: AiocqhttpMessageEvent, action: str = "status"):
async def analysis_settings(
self, event: AiocqhttpMessageEvent, action: str = "status"
):
"""
管理分析设置
用法: /分析设置 [enable|disable|status|reload|test]
@@ -360,7 +404,9 @@ class QQGroupDailyAnalysis(Star):
else: # status
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_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)
@@ -379,5 +425,3 @@ class QQGroupDailyAnalysis(Star):
💡 可用命令: enable, disable, status, reload, test
💡 支持的输出格式: image, text, pdf (图片和PDF包含活跃度可视化)
💡 其他命令: /设置格式, /安装PDF""")
+1 -1
View File
@@ -2,4 +2,4 @@
QQ群日常分析插件 - 源代码包
"""
__author__ = "SXP-Simon"
__author__ = "SXP-Simon"
+1 -4
View File
@@ -6,7 +6,4 @@
from .llm_analyzer import LLMAnalyzer
from .statistics import UserAnalyzer
__all__ = [
'LLMAnalyzer',
'UserAnalyzer'
]
__all__ = ["LLMAnalyzer", "UserAnalyzer"]
+1 -6
View File
@@ -8,9 +8,4 @@ from .topic_analyzer import TopicAnalyzer
from .user_title_analyzer import UserTitleAnalyzer
from .golden_quote_analyzer import GoldenQuoteAnalyzer
__all__ = [
'BaseAnalyzer',
'TopicAnalyzer',
'UserTitleAnalyzer',
'GoldenQuoteAnalyzer'
]
__all__ = ["BaseAnalyzer", "TopicAnalyzer", "UserTitleAnalyzer", "GoldenQuoteAnalyzer"]
+76 -54
View File
@@ -4,184 +4,206 @@
"""
from abc import ABC, abstractmethod
from typing import List, Dict, Tuple, Any, Optional
from datetime import datetime
from typing import List, Dict, Tuple, Any
from astrbot.api import logger
from ...models.data_models import TokenUsage
from ..utils.json_utils import parse_json_response
from ..utils.llm_utils import call_provider_with_retry, extract_token_usage, extract_response_text
import re
from ..utils.llm_utils import (
call_provider_with_retry,
extract_token_usage,
extract_response_text,
)
class BaseAnalyzer(ABC):
"""
基础分析器抽象类
定义所有分析器的通用接口和流程
"""
def __init__(self, context, config_manager):
"""
初始化基础分析器
Args:
context: AstrBot上下文对象
config_manager: 配置管理器
"""
self.context = context
self.config_manager = config_manager
@abstractmethod
def get_data_type(self) -> str:
"""
获取数据类型标识
Returns:
数据类型字符串
"""
pass
@abstractmethod
def get_max_count(self) -> int:
"""
获取最大提取数量
Returns:
最大数量
"""
pass
@abstractmethod
def build_prompt(self, data: Any) -> str:
"""
构建LLM提示词
Args:
data: 输入数据
Returns:
提示词字符串
"""
pass
@abstractmethod
def extract_with_regex(self, result_text: str, max_count: int) -> List[Dict]:
"""
使用正则表达式提取数据
Args:
result_text: LLM响应文本
max_count: 最大提取数量
Returns:
提取到的数据列表
"""
pass
@abstractmethod
def create_data_objects(self, data_list: List[Dict]) -> List[Any]:
"""
创建数据对象列表
Args:
data_list: 原始数据列表
Returns:
数据对象列表
"""
pass
async def analyze(self, data: Any, umo: str = None) -> Tuple[List[Any], TokenUsage]:
"""
统一的分析流程
Args:
data: 输入数据
umo: 模型唯一标识符
Returns:
(分析结果列表, Token使用统计)
"""
try:
# 1. 构建提示词
logger.debug(f"{self.get_data_type()}分析开始构建prompt,输入数据类型: {type(data)}")
logger.debug(f"{self.get_data_type()}分析输入数据长度: {len(data) if hasattr(data, '__len__') else 'N/A'}")
logger.debug(
f"{self.get_data_type()}分析开始构建prompt输入数据类型: {type(data)}"
)
logger.debug(
f"{self.get_data_type()}分析输入数据长度: {len(data) if hasattr(data, '__len__') else 'N/A'}"
)
prompt = self.build_prompt(data)
logger.info(f"开始{self.get_data_type()}分析,构建提示词完成")
logger.debug(f"{self.get_data_type()}分析prompt长度: {len(prompt) if prompt else 0}")
logger.debug(f"{self.get_data_type()}分析prompt前100字符: {prompt[:100] if prompt else 'None'}...")
logger.debug(
f"{self.get_data_type()}分析prompt长度: {len(prompt) if prompt else 0}"
)
logger.debug(
f"{self.get_data_type()}分析prompt前100字符: {prompt[:100] if prompt else 'None'}..."
)
# 检查 prompt 是否为空
if not prompt or not prompt.strip():
logger.warning(f"{self.get_data_type()}分析: prompt 为空或只包含空白字符,跳过LLM调用")
logger.warning(
f"{self.get_data_type()}分析: prompt 为空或只包含空白字符,跳过LLM调用"
)
return [], TokenUsage()
# 2. 调用LLM
max_tokens = self.get_max_tokens()
temperature = self.get_temperature()
response = await call_provider_with_retry(
self.context, self.config_manager, prompt,
max_tokens, temperature, umo
self.context, self.config_manager, prompt, max_tokens, temperature, umo
)
if response is None:
logger.error(f"{self.get_data_type()}分析调用LLM失败: provider返回None(重试失败)")
logger.error(
f"{self.get_data_type()}分析调用LLM失败: provider返回None(重试失败)"
)
return [], TokenUsage()
# 3. 提取token使用统计
token_usage_dict = extract_token_usage(response)
token_usage = TokenUsage(
prompt_tokens=token_usage_dict["prompt_tokens"],
completion_tokens=token_usage_dict["completion_tokens"],
total_tokens=token_usage_dict["total_tokens"]
total_tokens=token_usage_dict["total_tokens"],
)
# 4. 提取响应文本
result_text = extract_response_text(response)
logger.debug(f"{self.get_data_type()}分析原始响应: {result_text[:500]}...")
# 5. 尝试JSON解析
success, parsed_data, error_msg = parse_json_response(result_text, self.get_data_type())
success, parsed_data, error_msg = parse_json_response(
result_text, self.get_data_type()
)
if success and parsed_data:
# JSON解析成功,创建数据对象
data_objects = self.create_data_objects(parsed_data)
logger.info(f"{self.get_data_type()}分析成功,解析到 {len(data_objects)} 条数据")
logger.info(
f"{self.get_data_type()}分析成功,解析到 {len(data_objects)} 条数据"
)
return data_objects, token_usage
# 6. JSON解析失败,使用正则表达式降级
logger.warning(f"{self.get_data_type()}JSON解析失败,尝试正则表达式提取: {error_msg}")
logger.warning(
f"{self.get_data_type()}JSON解析失败,尝试正则表达式提取: {error_msg}"
)
regex_data = self.extract_with_regex(result_text, self.get_max_count())
if regex_data:
logger.info(f"{self.get_data_type()}正则表达式提取成功,获得 {len(regex_data)} 条数据")
logger.info(
f"{self.get_data_type()}正则表达式提取成功,获得 {len(regex_data)} 条数据"
)
data_objects = self.create_data_objects(regex_data)
return data_objects, token_usage
else:
# 最后的降级方案 - 两种方法都失败
logger.error(f"{self.get_data_type()}分析失败: JSON解析和正则表达式提取均未成功,返回空列表")
logger.error(
f"{self.get_data_type()}分析失败: JSON解析和正则表达式提取均未成功,返回空列表"
)
return [], token_usage
except Exception as e:
logger.error(f"{self.get_data_type()}分析失败: {e}", exc_info=True)
return [], TokenUsage()
def get_max_tokens(self) -> int:
"""
获取最大token数,子类可重写
Returns:
最大token数
"""
return 10000
def get_temperature(self) -> float:
"""
获取温度参数,子类可重写
Returns:
温度参数
"""
return 0.6
+53 -56
View File
@@ -12,59 +12,56 @@ from ..utils.json_utils import extract_golden_quotes_with_regex
from ..utils import InfoUtils
class GoldenQuoteAnalyzer(BaseAnalyzer):
"""
金句分析器
专门处理群聊金句的提取和分析
"""
def get_data_type(self) -> str:
"""获取数据类型标识"""
return "金句"
def get_max_count(self) -> int:
"""获取最大金句数量"""
return self.config_manager.get_max_golden_quotes()
def get_max_tokens(self) -> int:
"""获取最大token数"""
return self.config_manager.get_golden_quote_max_tokens()
def get_temperature(self) -> float:
"""获取温度参数"""
return 0.7
def build_prompt(self, messages: List[Dict]) -> str:
"""
构建金句分析提示词
Args:
messages: 群聊的文本消息列表
Returns:
提示词字符串
"""
if not messages:
return ""
# 构建消息文本
messages_text = "\n".join([
f"[{msg['time']}] {msg['sender']}: {msg['content']}"
for msg in messages
])
messages_text = "\n".join(
[f"[{msg['time']}] {msg['sender']}: {msg['content']}" for msg in messages]
)
max_golden_quotes = self.get_max_count()
# 从配置读取 prompt 模板(默认使用 "default" 风格)
prompt_template = self.config_manager.get_golden_quote_analysis_prompt()
if prompt_template:
# 使用配置中的 prompt 并替换变量
try:
prompt = prompt_template.format(
max_golden_quotes=max_golden_quotes,
messages_text=messages_text
max_golden_quotes=max_golden_quotes, messages_text=messages_text
)
logger.info("使用配置中的金句分析提示词")
return prompt
@@ -72,117 +69,117 @@ class GoldenQuoteAnalyzer(BaseAnalyzer):
logger.warning(f"金句分析提示词变量格式错误: {e}")
except Exception as e:
logger.warning(f"应用金句分析提示词失败: {e}")
logger.warning("未找到有效的金句分析提示词配置,请检查配置文件")
return ""
def extract_with_regex(self, result_text: str, max_count: int) -> List[Dict]:
"""
使用正则表达式提取金句信息
Args:
result_text: LLM响应文本
max_count: 最大提取数量
Returns:
金句数据列表
"""
return extract_golden_quotes_with_regex(result_text, max_count)
def create_data_objects(self, quotes_data: List[Dict]) -> List[GoldenQuote]:
"""
创建金句对象列表
Args:
quotes_data: 原始金句数据列表
Returns:
GoldenQuote对象列表
"""
try:
quotes = []
max_quotes = self.get_max_count()
for quote_data in quotes_data[:max_quotes]:
# 确保数据格式正确
content = quote_data.get("content", "").strip()
sender = quote_data.get("sender", "").strip()
reason = quote_data.get("reason", "").strip()
# 验证必要字段
if not content or not sender or not reason:
logger.warning(f"金句数据格式不完整,跳过: {quote_data}")
continue
quotes.append(GoldenQuote(
content=content,
sender=sender,
reason=reason
))
quotes.append(
GoldenQuote(content=content, sender=sender, reason=reason)
)
return quotes
except Exception as e:
logger.error(f"创建金句对象失败: {e}")
return []
def extract_interesting_messages(self, messages: List[Dict]) -> List[Dict]:
"""
提取圣经的文本消息
Args:
messages: 群聊消息列表
Returns:
圣经的文本消息列表
"""
try:
interesting_messages = []
for msg in messages:
sender = msg.get("sender", {})
nickname = InfoUtils.get_user_nickname(self.config_manager, sender)
msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M")
for 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 5 <= len(text) <= 100 and not text.startswith(
("http", "www", "/")
):
interesting_messages.append(
{"sender": nickname, "time": msg_time, "content": text}
)
return interesting_messages
except Exception as e:
logger.error(f"提取圣经消息失败: {e}")
return []
async def analyze_golden_quotes(self, messages: List[Dict], umo: str = None) -> Tuple[List[GoldenQuote], TokenUsage]:
async def analyze_golden_quotes(
self, messages: List[Dict], umo: str = None
) -> Tuple[List[GoldenQuote], TokenUsage]:
"""
分析群聊金句
Args:
messages: 群聊消息列表
umo: 模型唯一标识符
Returns:
(金句列表, Token使用统计)
"""
try:
# 提取圣经的文本消息
interesting_messages = self.extract_interesting_messages(messages)
if not interesting_messages:
logger.info("没有符合条件的圣经消息,返回空结果")
return [], TokenUsage()
logger.info(f"开始从 {len(interesting_messages)} 条圣经消息中提取金句")
return await self.analyze(interesting_messages, umo)
except Exception as e:
logger.error(f"金句分析失败: {e}")
return [], TokenUsage()
return [], TokenUsage()
+162 -117
View File
@@ -18,30 +18,30 @@ class TopicAnalyzer(BaseAnalyzer):
话题分析器
专门处理群聊话题的提取和分析
"""
def get_data_type(self) -> str:
"""获取数据类型标识"""
return "话题"
def get_max_count(self) -> int:
"""获取最大话题数量"""
return self.config_manager.get_max_topics()
def get_max_tokens(self) -> int:
"""获取最大token数"""
return self.config_manager.get_topic_max_tokens()
def get_temperature(self) -> float:
"""获取温度参数"""
return 0.6
def build_prompt(self, messages: List[Dict]) -> str:
"""
构建话题分析提示词
Args:
messages: 群聊消息列表
Returns:
提示词字符串
"""
@@ -49,49 +49,61 @@ class TopicAnalyzer(BaseAnalyzer):
if not isinstance(messages, list):
logger.error(f"build_prompt 期望列表,但收到: {type(messages)}")
return ""
# 检查消息列表是否为空
if not messages:
logger.warning("build_prompt 收到空消息列表")
return ""
# 提取文本消息
text_messages = []
for i, msg in enumerate(messages):
logger.debug(f"build_prompt 处理第 {i+1} 条消息,类型: {type(msg)}")
logger.debug(f"build_prompt 处理第 {i + 1} 条消息,类型: {type(msg)}")
# 确保msg是字典类型,避免'str' object has no attribute 'get'错误
if not isinstance(msg, dict):
logger.warning(f"build_prompt 跳过非字典类型的消息: {type(msg)} - {msg}")
logger.warning(
f"build_prompt 跳过非字典类型的消息: {type(msg)} - {msg}"
)
continue
try:
sender = msg.get("sender", {})
# 确保sender是字典类型,避免'str' object has no attribute 'get'错误
if not isinstance(sender, dict):
logger.warning(f"build_prompt 跳过sender非字典类型的消息: {type(sender)} - {sender}")
logger.warning(
f"build_prompt 跳过sender非字典类型的消息: {type(sender)} - {sender}"
)
continue
nickname = InfoUtils.get_user_nickname(self.config_manager, sender)
msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M")
message_list = msg.get("message", [])
logger.debug(f"build_prompt 消息 {i+1} 的 message 字段类型: {type(message_list)}, 长度: {len(message_list) if hasattr(message_list, '__len__') else 'N/A'}")
logger.debug(
f"build_prompt 消息 {i + 1} 的 message 字段类型: {type(message_list)}, 长度: {len(message_list) if hasattr(message_list, '__len__') else 'N/A'}"
)
# 提取文本内容,可能分布在多个 content 中
text_parts = []
for j, content in enumerate(message_list):
logger.debug(f"build_prompt 处理消息 {i+1} 的内容 {j+1}, 类型: {type(content)}")
logger.debug(
f"build_prompt 处理消息 {i + 1} 的内容 {j + 1}, 类型: {type(content)}"
)
if not isinstance(content, dict):
logger.warning(f"build_prompt 跳过非字典类型的内容: {type(content)} - {content}")
logger.warning(
f"build_prompt 跳过非字典类型的内容: {type(content)} - {content}"
)
continue
content_type = content.get("type", "")
logger.debug(f"build_prompt 内容类型: {content_type}")
if content_type == "text":
text = content.get("data", {}).get("text", "").strip()
logger.debug(f"build_prompt 提取到的文本: '{text}' (长度: {len(text)})")
logger.debug(
f"build_prompt 提取到的文本: '{text}' (长度: {len(text)})"
)
if text:
text_parts.append(text)
elif content_type == "at":
@@ -108,53 +120,62 @@ class TopicAnalyzer(BaseAnalyzer):
reply_text = f"[回复:{reply_id}]"
text_parts.append(reply_text)
logger.debug(f"build_prompt 提取到回复消息: {reply_text}")
# 合并所有文本部分
combined_text = "".join(text_parts).strip()
logger.debug(f"build_prompt 合并后的文本: '{combined_text}' (长度: {len(combined_text)})")
if combined_text and len(combined_text) > 2 and not combined_text.startswith("/"):
logger.debug(
f"build_prompt 合并后的文本: '{combined_text}' (长度: {len(combined_text)})"
)
if (
combined_text
and len(combined_text) > 2
and not combined_text.startswith("/")
):
# 清理消息内容
cleaned_text = combined_text.replace('', '"').replace('', '"')
cleaned_text = cleaned_text.replace('', "'").replace('', "'")
cleaned_text = cleaned_text.replace('\n', ' ').replace('\r', ' ')
cleaned_text = cleaned_text.replace('\t', ' ')
cleaned_text = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', cleaned_text)
cleaned_text = combined_text.replace("", '"').replace("", '"')
cleaned_text = cleaned_text.replace("", "'").replace("", "'")
cleaned_text = cleaned_text.replace("\n", " ").replace("\r", " ")
cleaned_text = cleaned_text.replace("\t", " ")
cleaned_text = re.sub(r"[\x00-\x1f\x7f-\x9f]", "", cleaned_text)
logger.debug(f"build_prompt 清理后的文本: '{cleaned_text}'")
text_messages.append({
"sender": nickname,
"time": msg_time,
"content": cleaned_text
})
text_messages.append(
{"sender": nickname, "time": msg_time, "content": cleaned_text}
)
else:
logger.debug(f"build_prompt 跳过文本: '{combined_text}' (长度不足或以/开头)")
logger.debug(
f"build_prompt 跳过文本: '{combined_text}' (长度不足或以/开头)"
)
except Exception as e:
logger.error(f"build_prompt 处理第 {i+1} 条消息时出错: {e}", exc_info=True)
logger.error(
f"build_prompt 处理第 {i + 1} 条消息时出错: {e}", exc_info=True
)
continue
if not text_messages:
logger.warning("build_prompt 没有提取到有效的文本消息,返回空prompt")
return ""
# 构建消息文本
messages_text = "\n".join([
f"[{msg['time']}] {msg['sender']}: {msg['content']}"
for msg in text_messages
])
messages_text = "\n".join(
[
f"[{msg['time']}] {msg['sender']}: {msg['content']}"
for msg in text_messages
]
)
max_topics = self.get_max_count()
# 从配置读取 prompt 模板(默认使用 "default" 风格)
prompt_template = self.config_manager.get_topic_analysis_prompt()
if prompt_template:
# 使用配置中的 prompt 并替换变量
try:
prompt = prompt_template.format(
max_topics=max_topics,
messages_text=messages_text
max_topics=max_topics, messages_text=messages_text
)
logger.info("使用配置中的话题分析提示词")
return prompt
@@ -162,181 +183,205 @@ class TopicAnalyzer(BaseAnalyzer):
logger.warning(f"话题分析提示词变量格式错误: {e}")
except Exception as e:
logger.warning(f"应用话题分析提示词失败: {e}")
logger.warning("未找到有效的话题分析提示词配置,请检查配置文件")
return ""
def extract_with_regex(self, result_text: str, max_topics: int) -> List[Dict]:
"""
使用正则表达式提取话题信息
Args:
result_text: LLM响应文本
max_topics: 最大话题数量
Returns:
话题数据列表
"""
return extract_topics_with_regex(result_text, max_topics)
def create_data_objects(self, topics_data: List[Dict]) -> List[SummaryTopic]:
"""
创建话题对象列表
Args:
topics_data: 原始话题数据列表
Returns:
SummaryTopic对象列表
"""
logger.debug(f"create_data_objects 开始处理,输入数据数量: {len(topics_data) if topics_data else 0}")
logger.debug(
f"create_data_objects 开始处理,输入数据数量: {len(topics_data) if topics_data else 0}"
)
logger.debug(f"输入数据类型: {type(topics_data)}")
try:
topics = []
max_topics = self.get_max_count()
logger.debug(f"处理前 {max_topics} 条话题数据")
for i, topic_data in enumerate(topics_data[:max_topics]):
logger.debug(f"处理第 {i+1} 条话题数据,类型: {type(topic_data)}")
logger.debug(f"处理第 {i + 1} 条话题数据,类型: {type(topic_data)}")
# 确保topic_data是字典类型,避免'str' object has no attribute 'get'错误
if not isinstance(topic_data, dict):
logger.warning(f"跳过非字典类型的话题数据: {type(topic_data)} - {topic_data}")
logger.warning(
f"跳过非字典类型的话题数据: {type(topic_data)} - {topic_data}"
)
continue
try:
# 确保数据格式正确
topic_name = topic_data.get("topic", "").strip()
contributors = topic_data.get("contributors", [])
detail = topic_data.get("detail", "").strip()
logger.debug(f"话题数据 - 名称: {topic_name}, 参与者: {contributors}, 详情: {detail[:50]}...")
logger.debug(
f"话题数据 - 名称: {topic_name}, 参与者: {contributors}, 详情: {detail[:50]}..."
)
# 验证必要字段
if not topic_name or not detail:
logger.warning(f"话题数据格式不完整,跳过: {topic_data}")
continue
# 确保参与者列表有效
if not contributors or not isinstance(contributors, list):
contributors = ["群友"]
else:
# 清理参与者名称
contributors = [str(c).strip() for c in contributors if c and str(c).strip()] or ["群友"]
topics.append(SummaryTopic(
topic=topic_name,
contributors=contributors[:5], # 最多5个参与者
detail=detail
))
contributors = [
str(c).strip() for c in contributors if c and str(c).strip()
] or ["群友"]
topics.append(
SummaryTopic(
topic=topic_name,
contributors=contributors[:5], # 最多5个参与者
detail=detail,
)
)
except Exception as e:
logger.error(f"处理第 {i+1} 条话题数据时出错: {e}", exc_info=True)
logger.error(f"处理第 {i + 1} 条话题数据时出错: {e}", exc_info=True)
continue
logger.debug(f"create_data_objects 完成,创建了 {len(topics)} 个话题对象")
return topics
except Exception as e:
logger.error(f"创建话题对象失败: {e}", exc_info=True)
return []
def extract_text_messages(self, messages: List[Dict]) -> List[Dict]:
"""
从群聊消息中提取文本消息
Args:
messages: 群聊消息列表
Returns:
提取的文本消息列表
"""
logger.debug(f"extract_text_messages 开始处理,输入消息数量: {len(messages) if messages else 0}")
logger.debug(
f"extract_text_messages 开始处理,输入消息数量: {len(messages) if messages else 0}"
)
logger.debug(f"extract_text_messages 输入消息类型: {type(messages)}")
if not messages:
logger.warning("extract_text_messages 收到空消息列表")
return []
text_messages = []
for i, msg in enumerate(messages):
logger.debug(f"处理第 {i+1} 条消息,类型: {type(msg)}")
logger.debug(f"处理第 {i + 1} 条消息,类型: {type(msg)}")
# 确保msg是字典类型,避免'str' object has no attribute 'get'错误
if not isinstance(msg, dict):
logger.warning(f"跳过非字典类型的消息: {type(msg)} - {msg}")
continue
try:
sender = msg.get("sender", {})
# 确保sender是字典类型,避免'str' object has no attribute 'get'错误
if not isinstance(sender, dict):
logger.warning(f"extract_text_messages 跳过sender非字典类型的消息: {type(sender)} - {sender}")
logger.warning(
f"extract_text_messages 跳过sender非字典类型的消息: {type(sender)} - {sender}"
)
continue
nickname = InfoUtils.get_user_nickname(self.config_manager, sender)
msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M")
for content in msg.get("message", []):
if content.get("type") == "text":
text = content.get("data", {}).get("text", "").strip()
if text and len(text) > 2 and not text.startswith("/"):
# 清理消息内容
text = text.replace('""', '"').replace('""', '"')
text = text.replace(''', "'").replace(''', "'")
text = text.replace('\n', ' ').replace('\r', ' ')
text = text.replace('\t', ' ')
text = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', text)
text_messages.append({
"sender": nickname,
"time": msg_time,
"content": text.strip()
})
text = text.replace(""", "'").replace(""", "'")
text = text.replace("\n", " ").replace("\r", " ")
text = text.replace("\t", " ")
text = re.sub(r"[\x00-\x1f\x7f-\x9f]", "", text)
text_messages.append(
{
"sender": nickname,
"time": msg_time,
"content": text.strip(),
}
)
except Exception as e:
logger.error(f"处理第 {i+1} 条消息时出错: {e}", exc_info=True)
logger.error(f"处理第 {i + 1} 条消息时出错: {e}", exc_info=True)
continue
logger.debug(f"extract_text_messages 完成,提取到 {len(text_messages)} 条文本消息")
logger.debug(
f"extract_text_messages 完成,提取到 {len(text_messages)} 条文本消息"
)
if text_messages:
logger.debug(f"extract_text_messages 第一条文本消息: {text_messages[0]}")
return text_messages
async def analyze_topics(self, messages: List[Dict], umo: str = None) -> Tuple[List[SummaryTopic], TokenUsage]:
async def analyze_topics(
self, messages: List[Dict], umo: str = None
) -> Tuple[List[SummaryTopic], TokenUsage]:
"""
分析群聊话题
Args:
messages: 群聊消息列表
umo: 模型唯一标识符
Returns:
(话题列表, Token使用统计)
"""
try:
logger.debug(f"analyze_topics 开始处理,消息数量: {len(messages) if messages else 0}")
logger.debug(
f"analyze_topics 开始处理,消息数量: {len(messages) if messages else 0}"
)
logger.debug(f"消息类型: {type(messages)}")
if messages:
logger.debug(f"第一条消息类型: {type(messages[0]) if messages else ''}")
logger.debug(
f"第一条消息类型: {type(messages[0]) if messages else ''}"
)
logger.debug(f"第一条消息内容: {messages[0] if messages else ''}")
# 检查是否有有效的文本消息
text_messages = self.extract_text_messages(messages)
logger.debug(f"提取到 {len(text_messages)} 条文本消息")
if not text_messages:
logger.info("没有有效的文本消息,返回空结果")
return [], TokenUsage()
logger.info(f"开始分析 {len(text_messages)} 条文本消息中的话题")
logger.debug(f"文本消息类型: {type(text_messages)}")
if text_messages:
logger.debug(f"第一条文本消息类型: {type(text_messages[0])}")
logger.debug(f"第一条文本消息内容: {text_messages[0]}")
# 直接传入原始消息,让 build_prompt 方法处理
return await self.analyze(messages, umo)
except Exception as e:
logger.error(f"话题分析失败: {e}", exc_info=True)
return [], TokenUsage()
return [], TokenUsage()
+102 -77
View File
@@ -15,93 +15,93 @@ class UserTitleAnalyzer(BaseAnalyzer):
用户称号分析器
专门处理用户称号分配和MBTI类型分析
"""
def get_data_type(self) -> str:
"""获取数据类型标识"""
return "用户称号"
def get_max_count(self) -> int:
"""获取最大用户称号数量"""
return self.config_manager.get_max_user_titles()
def get_max_tokens(self) -> int:
"""获取最大token数"""
return self.config_manager.get_user_title_max_tokens()
def get_temperature(self) -> float:
"""获取温度参数"""
return 0.5
def build_prompt(self, user_data: Dict) -> str:
"""
构建用户称号分析提示词
Args:
user_data: 用户数据字典,包含用户统计信息
Returns:
提示词字符串
"""
user_summaries = user_data.get("user_summaries", [])
if not user_summaries:
return ""
# 构建用户数据文本
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
])
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 模板(默认使用 "default" 风格)
prompt_template = self.config_manager.get_user_title_analysis_prompt()
if prompt_template:
# 使用配置中的 prompt 并替换变量
try:
prompt = prompt_template.format(
users_text=users_text
)
prompt = prompt_template.format(users_text=users_text)
logger.info("使用配置中的用户称号分析提示词")
return prompt
except KeyError as e:
logger.warning(f"用户称号分析提示词变量格式错误: {e}")
except Exception as e:
logger.warning(f"应用用户称号分析提示词失败: {e}")
logger.warning("未找到有效的用户称号分析提示词配置,请检查配置文件")
return ""
def extract_with_regex(self, result_text: str, max_count: int) -> List[Dict]:
"""
使用正则表达式提取用户称号信息
Args:
result_text: LLM响应文本
max_count: 最大提取数量
Returns:
用户称号数据列表
"""
return extract_user_titles_with_regex(result_text, max_count)
def create_data_objects(self, titles_data: List[Dict]) -> List[UserTitle]:
"""
创建用户称号对象列表
Args:
titles_data: 原始用户称号数据列表
Returns:
UserTitle对象列表
"""
try:
titles = []
max_titles = self.get_max_count()
for title_data in titles_data[:max_titles]:
# 确保数据格式正确
name = title_data.get("name", "").strip()
@@ -109,122 +109,147 @@ class UserTitleAnalyzer(BaseAnalyzer):
title = title_data.get("title", "").strip()
mbti = title_data.get("mbti", "").strip()
reason = title_data.get("reason", "").strip()
# 验证必要字段
if not name or not title or not mbti or not reason:
logger.warning(f"用户称号数据格式不完整,跳过: {title_data}")
continue
# 验证QQ号格式
try:
qq = int(qq)
except (ValueError, TypeError):
logger.warning(f"QQ号格式无效,跳过: {qq}")
continue
titles.append(UserTitle(
name=name,
qq=qq,
title=title,
mbti=mbti,
reason=reason
))
titles.append(
UserTitle(name=name, qq=qq, title=title, mbti=mbti, reason=reason)
)
return titles
except Exception as e:
logger.error(f"创建用户称号对象失败: {e}")
return []
def prepare_user_data(self, messages: List[Dict], user_analysis: Dict, top_users: List[Dict] = None) -> Dict:
def prepare_user_data(
self, messages: List[Dict], user_analysis: Dict, top_users: List[Dict] = None
) -> Dict:
"""
准备用户数据
Args:
messages: 群聊消息列表
user_analysis: 用户分析统计
top_users: 活跃用户列表(从get_top_users获取)
Returns:
准备好的用户数据字典
"""
try:
# 获取机器人QQ号用于过滤
bot_qq_id = self.config_manager.get_bot_qq_id()
user_summaries = []
# 如果提供了top_users列表,只分析这些活跃用户
if top_users:
logger.info(f"使用get_top_users筛选出的 {len(top_users)} 个活跃用户进行称号分析")
target_user_ids = {str(user['user_id']) for user in top_users}
logger.info(
f"使用get_top_users筛选出的 {len(top_users)} 个活跃用户进行称号分析"
)
target_user_ids = {str(user["user_id"]) for user in top_users}
else:
# 兼容旧逻辑:如果没有提供top_users,则使用所有消息数>=5的用户
logger.info("未提供活跃用户列表,使用消息数>=5的用户")
target_user_ids = {user_id for user_id, stats in user_analysis.items()
if stats["message_count"] >= 5}
target_user_ids = {
user_id
for user_id, stats in user_analysis.items()
if stats["message_count"] >= 5
}
for user_id, stats in user_analysis.items():
# 过滤机器人自己的消息
if bot_qq_id and str(user_id) == str(bot_qq_id):
logger.debug(f"过滤掉机器人QQ号: {user_id}")
continue
# 只处理活跃用户
if user_id not in target_user_ids:
continue
# 分析用户特征
night_messages = sum(stats["hours"][h] for h in range(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) if stats["message_count"] > 0 else 0,
"night_ratio": round(night_messages / stats["message_count"], 2) if stats["message_count"] > 0 else 0,
"reply_ratio": round(stats["reply_count"] / stats["message_count"], 2) if stats["message_count"] > 0 else 0
})
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
)
if stats["message_count"] > 0
else 0,
"night_ratio": round(night_messages / stats["message_count"], 2)
if stats["message_count"] > 0
else 0,
"reply_ratio": round(
stats["reply_count"] / stats["message_count"], 2
)
if stats["message_count"] > 0
else 0,
}
)
if not user_summaries:
return {"user_summaries": []}
# 按消息数量排序
user_summaries.sort(key=lambda x: x["message_count"], reverse=True)
return {"user_summaries": user_summaries}
except Exception as e:
logger.error(f"准备用户数据失败: {e}")
return {"user_summaries": []}
async def analyze_user_titles(self, messages: List[Dict], user_analysis: Dict, umo: str = None, top_users: List[Dict] = None) -> Tuple[List[UserTitle], TokenUsage]:
async def analyze_user_titles(
self,
messages: List[Dict],
user_analysis: Dict,
umo: str = None,
top_users: List[Dict] = None,
) -> Tuple[List[UserTitle], TokenUsage]:
"""
分析用户称号
Args:
messages: 群聊消息列表
user_analysis: 用户分析统计
umo: 模型唯一标识符
top_users: 活跃用户列表(从get_top_users获取,可选)
Returns:
(用户称号列表, Token使用统计)
"""
try:
# 准备用户数据,传入活跃用户列表
user_data = self.prepare_user_data(messages, user_analysis, top_users)
if not user_data["user_summaries"]:
logger.info("没有符合条件的用户,返回空结果")
return [], TokenUsage()
logger.info(f"开始分析 {len(user_data['user_summaries'])} 个活跃用户的称号")
return await self.analyze(user_data, umo)
except Exception as e:
logger.error(f"用户称号分析失败: {e}")
return [], TokenUsage()
return [], TokenUsage()
+79 -46
View File
@@ -12,7 +12,6 @@ from .analyzers.user_title_analyzer import UserTitleAnalyzer
from .analyzers.golden_quote_analyzer import GoldenQuoteAnalyzer
from .utils.llm_utils import call_provider_with_retry
from .utils.json_utils import fix_json
from .utils.json_utils import extract_topics_with_regex, extract_user_titles_with_regex, extract_golden_quotes_with_regex
class LLMAnalyzer:
@@ -21,32 +20,34 @@ class LLMAnalyzer:
作为统一入口,协调各个专门的分析器进行不同类型的分析
保持向后兼容性,提供原有的接口
"""
def __init__(self, context, config_manager):
"""
初始化LLM分析器
Args:
context: AstrBot上下文对象
config_manager: 配置管理器
"""
self.context = context
self.config_manager = config_manager
# 初始化各个专门的分析器
self.topic_analyzer = TopicAnalyzer(context, config_manager)
self.user_title_analyzer = UserTitleAnalyzer(context, config_manager)
self.golden_quote_analyzer = GoldenQuoteAnalyzer(context, config_manager)
async def analyze_topics(self, messages: List[Dict], umo: str = None) -> Tuple[List[SummaryTopic], TokenUsage]:
async def analyze_topics(
self, messages: List[Dict], umo: str = None
) -> Tuple[List[SummaryTopic], TokenUsage]:
"""
使用LLM分析话题
保持原有接口,委托给专门的TopicAnalyzer处理
Args:
messages: 群聊消息列表
umo: 模型唯一标识符
Returns:
(话题列表, Token使用统计)
"""
@@ -56,37 +57,47 @@ class LLMAnalyzer:
except Exception as e:
logger.error(f"话题分析失败: {e}")
return [], TokenUsage()
async def analyze_user_titles(self, messages: List[Dict], user_analysis: Dict, umo: str = None, top_users: List[Dict] = None) -> Tuple[List[UserTitle], TokenUsage]:
async def analyze_user_titles(
self,
messages: List[Dict],
user_analysis: Dict,
umo: str = None,
top_users: List[Dict] = None,
) -> Tuple[List[UserTitle], TokenUsage]:
"""
使用LLM分析用户称号
保持原有接口,委托给专门的UserTitleAnalyzer处理
Args:
messages: 群聊消息列表
user_analysis: 用户分析统计
umo: 模型唯一标识符
top_users: 活跃用户列表(可选)
Returns:
(用户称号列表, Token使用统计)
"""
try:
logger.info("开始用户称号分析")
return await self.user_title_analyzer.analyze_user_titles(messages, user_analysis, umo, top_users)
return await self.user_title_analyzer.analyze_user_titles(
messages, user_analysis, umo, top_users
)
except Exception as e:
logger.error(f"用户称号分析失败: {e}")
return [], TokenUsage()
async def analyze_golden_quotes(self, messages: List[Dict], umo: str = None) -> Tuple[List[GoldenQuote], TokenUsage]:
async def analyze_golden_quotes(
self, messages: List[Dict], umo: str = None
) -> Tuple[List[GoldenQuote], TokenUsage]:
"""
使用LLM分析群聊金句
保持原有接口,委托给专门的GoldenQuoteAnalyzer处理
Args:
messages: 群聊消息列表
umo: 模型唯一标识符
Returns:
(金句列表, Token使用统计)
"""
@@ -96,98 +107,120 @@ class LLMAnalyzer:
except Exception as e:
logger.error(f"金句分析失败: {e}")
return [], TokenUsage()
async def analyze_all_concurrent(self, messages: List[Dict], user_analysis: Dict, umo: str = None, top_users: List[Dict] = None) -> Tuple[List[SummaryTopic], List[UserTitle], List[GoldenQuote], TokenUsage]:
async def analyze_all_concurrent(
self,
messages: List[Dict],
user_analysis: Dict,
umo: str = None,
top_users: List[Dict] = None,
) -> Tuple[List[SummaryTopic], List[UserTitle], List[GoldenQuote], TokenUsage]:
"""
并发执行所有分析任务(话题、用户称号、金句)
Args:
messages: 群聊消息列表
user_analysis: 用户分析统计
umo: 模型唯一标识符
top_users: 活跃用户列表(可选)
Returns:
(话题列表, 用户称号列表, 金句列表, 总Token使用统计)
"""
try:
logger.info("开始并发执行所有分析任务")
# 并发执行三个分析任务
results = await asyncio.gather(
self.topic_analyzer.analyze_topics(messages, umo),
self.user_title_analyzer.analyze_user_titles(messages, user_analysis, umo, top_users),
self.user_title_analyzer.analyze_user_titles(
messages, user_analysis, umo, top_users
),
self.golden_quote_analyzer.analyze_golden_quotes(messages, umo),
return_exceptions=True
return_exceptions=True,
)
# 处理结果
topics, topic_usage = [], TokenUsage()
user_titles, title_usage = [], TokenUsage()
golden_quotes, quote_usage = [], TokenUsage()
# 话题分析结果
if isinstance(results[0], Exception):
logger.error(f"话题分析失败: {results[0]}")
else:
topics, topic_usage = results[0]
# 用户称号分析结果
if isinstance(results[1], Exception):
logger.error(f"用户称号分析失败: {results[1]}")
else:
user_titles, title_usage = results[1]
# 金句分析结果
if isinstance(results[2], Exception):
logger.error(f"金句分析失败: {results[2]}")
else:
golden_quotes, quote_usage = results[2]
# 合并Token使用统计
total_usage = TokenUsage(
prompt_tokens=topic_usage.prompt_tokens + title_usage.prompt_tokens + quote_usage.prompt_tokens,
completion_tokens=topic_usage.completion_tokens + title_usage.completion_tokens + quote_usage.completion_tokens,
total_tokens=topic_usage.total_tokens + title_usage.total_tokens + quote_usage.total_tokens
prompt_tokens=topic_usage.prompt_tokens
+ title_usage.prompt_tokens
+ quote_usage.prompt_tokens,
completion_tokens=topic_usage.completion_tokens
+ title_usage.completion_tokens
+ quote_usage.completion_tokens,
total_tokens=topic_usage.total_tokens
+ title_usage.total_tokens
+ quote_usage.total_tokens,
)
logger.info(
f"并发分析完成 - 话题: {len(topics)}, 称号: {len(user_titles)}, 金句: {len(golden_quotes)}"
)
logger.info(f"并发分析完成 - 话题: {len(topics)}, 称号: {len(user_titles)}, 金句: {len(golden_quotes)}")
return topics, user_titles, golden_quotes, total_usage
except Exception as e:
logger.error(f"并发分析失败: {e}")
return [], [], [], TokenUsage()
# 向后兼容的方法,保持原有调用方式
async def _call_provider_with_retry(self, provider, prompt: str, max_tokens: int,
temperature: float, umo: str = None):
async def _call_provider_with_retry(
self,
provider,
prompt: str,
max_tokens: int,
temperature: float,
umo: str = None,
):
"""
向后兼容的LLM调用方法
现在委托给llm_utils模块处理
Args:
provider: LLM服务商实例或None
prompt: 输入的提示语
max_tokens: 最大生成token数
temperature: 采样温度
umo: 指定使用的模型唯一标识符
Returns:
LLM生成的结果
"""
return await call_provider_with_retry(self.context, self.config_manager,
prompt, max_tokens, temperature, umo)
return await call_provider_with_retry(
self.context, self.config_manager, prompt, max_tokens, temperature, umo
)
def _fix_json(self, text: str) -> str:
"""
向后兼容的JSON修复方法
现在委托给json_utils模块处理
Args:
text: 需要修复的JSON文本
Returns:
修复后的JSON文本
"""
return fix_json(text)
+35 -26
View File
@@ -19,24 +19,26 @@ class UserAnalyzer:
"""分析用户活跃度"""
# 获取机器人QQ号用于过滤
bot_qq_id = self.config_manager.get_bot_qq_id()
user_stats = defaultdict(lambda: {
"message_count": 0,
"char_count": 0,
"emoji_count": 0,
"nickname": "",
"hours": defaultdict(int),
"reply_count": 0
})
user_stats = defaultdict(
lambda: {
"message_count": 0,
"char_count": 0,
"emoji_count": 0,
"nickname": "",
"hours": defaultdict(int),
"reply_count": 0,
}
)
for msg in messages:
sender = msg.get("sender", {})
user_id = str(sender.get("user_id", ""))
# 跳过机器人自己的消息,避免进入统计
if bot_qq_id and user_id == str(bot_qq_id):
continue
nickname = InfoUtils.get_user_nickname(self.config_manager, sender)
user_stats[user_id]["message_count"] += 1
@@ -75,31 +77,37 @@ class UserAnalyzer:
return dict(user_stats)
def get_top_users(self, user_analysis: Dict[str, Dict], limit: int = 10) -> List[Dict]:
def get_top_users(
self, user_analysis: Dict[str, Dict], limit: int = 10
) -> List[Dict]:
"""获取最活跃的用户"""
# 获取机器人QQ号用于过滤
bot_qq_id = self.config_manager.get_bot_qq_id()
users = []
for user_id, stats in user_analysis.items():
# 过滤机器人自己
if bot_qq_id and str(user_id) == str(bot_qq_id):
continue
users.append({
"user_id": user_id,
"nickname": stats["nickname"],
"message_count": stats["message_count"],
"char_count": stats["char_count"],
"emoji_count": stats["emoji_count"],
"reply_count": stats["reply_count"]
})
users.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:
def get_user_activity_pattern(
self, user_analysis: Dict[str, Dict], user_id: str
) -> Dict:
"""获取用户活动模式"""
if user_id not in user_analysis:
return {}
@@ -112,11 +120,12 @@ class UserAnalyzer:
# 计算夜间活跃度
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
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)
"hourly_distribution": dict(hours),
}
+12 -14
View File
@@ -8,30 +8,28 @@ from .json_utils import (
parse_json_response,
extract_topics_with_regex,
extract_user_titles_with_regex,
extract_golden_quotes_with_regex
extract_golden_quotes_with_regex,
)
from .llm_utils import (
call_provider_with_retry,
extract_token_usage,
extract_response_text
extract_response_text,
)
from .info_utils import InfoUtils
__all__ = [
# JSON处理工具
'fix_json',
'parse_json_response',
'extract_topics_with_regex',
'extract_user_titles_with_regex',
'extract_golden_quotes_with_regex',
"fix_json",
"parse_json_response",
"extract_topics_with_regex",
"extract_user_titles_with_regex",
"extract_golden_quotes_with_regex",
# LLM工具
'call_provider_with_retry',
'extract_token_usage',
'extract_response_text',
"call_provider_with_retry",
"extract_token_usage",
"extract_response_text",
# 信息工具
'InfoUtils'
]
"InfoUtils",
]
+89 -89
View File
@@ -1,4 +1,3 @@
"""
JSON处理工具模块
提供JSON解析、修复和正则提取功能
@@ -6,41 +5,41 @@ JSON处理工具模块
import json
import re
from typing import List, Dict, Tuple, Any, Optional
from typing import List, Dict, Tuple, Optional
from astrbot.api import logger
def fix_json(text: str) -> str:
"""
修复JSON格式问题,包括中文符号替换
Args:
text: 需要修复的JSON文本
Returns:
修复后的JSON文本
"""
try:
# 1. 移除markdown代码块标记
text = re.sub(r'```json\s*', '', text)
text = re.sub(r'```\s*$', '', text)
text = re.sub(r"```json\s*", "", text)
text = re.sub(r"```\s*$", "", text)
# 2. 基础清理
text = text.replace('\n', ' ').replace('\r', ' ')
text = re.sub(r'\s+', ' ', text)
text = text.replace("\n", " ").replace("\r", " ")
text = re.sub(r"\s+", " ", text)
# 3. 替换中文符号为英文符号(修复)
# 中文引号 -> 英文引号
text = text.replace('', '"').replace('', '"')
text = text.replace('', "'").replace('', "'")
text = text.replace("", '"').replace("", '"')
text = text.replace("", "'").replace("", "'")
# 中文逗号 -> 英文逗号
text = text.replace('', ',')
text = text.replace("", ",")
# 中文冒号 -> 英文冒号
text = text.replace('', ':')
text = text.replace("", ":")
# 中文括号 -> 英文括号
text = text.replace('', '(').replace('', ')')
text = text.replace('', '[').replace('', ']')
text = text.replace("", "(").replace("", ")")
text = text.replace("", "[").replace("", "]")
# 4. 处理字符串内容中的特殊字符
# 转义字符串内的双引号
def escape_quotes_in_strings(match):
@@ -48,71 +47,73 @@ def fix_json(text: str) -> str:
# 转义内部的双引号
content = content.replace('"', '\\"')
return f'"{content}"'
# 先处理字段值中的引号
text = re.sub(r'"([^"]*(?:"[^"]*)*)"', escape_quotes_in_strings, text)
# 5. 修复截断的JSON
if not text.endswith(']'):
last_complete = text.rfind('}')
if not text.endswith("]"):
last_complete = text.rfind("}")
if last_complete > 0:
text = text[:last_complete + 1] + ']'
text = text[: last_complete + 1] + "]"
# 6. 修复常见的JSON格式问题
# 1. 修复缺失的逗号
text = re.sub(r'}\s*{', '}, {', text)
text = re.sub(r"}\s*{", "}, {", text)
# 2. 确保字段名有引号(仅在对象开始或逗号后,避免破坏字符串值)
def quote_field_names(match):
prefix = match.group(1)
key = match.group(2)
return f'{prefix}"{key}":'
# 只在 { 或 , 后面匹配字段名,避免在字符串值中误匹配
text = re.sub(r'([{,]\s*)([a-zA-Z_][a-zA-Z0-9_]*)\s*:', quote_field_names, text)
text = re.sub(r"([{,]\s*)([a-zA-Z_][a-zA-Z0-9_]*)\s*:", quote_field_names, text)
# 3. 移除多余的逗号
text = re.sub(r',\s*}', '}', text)
text = re.sub(r',\s*]', ']', text)
text = re.sub(r",\s*}", "}", text)
text = re.sub(r",\s*]", "]", text)
return text.strip()
except Exception as e:
logger.error(f"JSON修复失败: {e}")
return text
def parse_json_response(result_text: str, data_type: str) -> Tuple[bool, Optional[List[Dict]], Optional[str]]:
def parse_json_response(
result_text: str, data_type: str
) -> Tuple[bool, Optional[List[Dict]], Optional[str]]:
"""
统一的JSON解析方法
Args:
result_text: LLM返回的原始文本
data_type: 数据类型 ('topics' | 'user_titles' | 'golden_quotes')
Returns:
(成功标志, 解析后的数据列表, 错误消息)
"""
try:
# 1. 提取JSON部分
json_match = re.search(r'\[.*?\]', result_text, re.DOTALL)
json_match = re.search(r"\[.*?\]", result_text, re.DOTALL)
if not json_match:
error_msg = f"{data_type}响应中未找到JSON格式"
logger.warning(error_msg)
return False, None, error_msg
json_text = json_match.group()
logger.debug(f"{data_type}分析JSON原文: {json_text[:500]}...")
# 2. 修复JSON
json_text = fix_json(json_text)
logger.debug(f"{data_type}修复后的JSON: {json_text[:300]}...")
# 3. 解析JSON
data = json.loads(json_text)
logger.info(f"{data_type}分析成功,解析到 {len(data)} 条数据")
return True, data, None
except json.JSONDecodeError as e:
error_msg = f"{data_type}JSON解析失败: {e}"
logger.warning(error_msg)
@@ -127,11 +128,11 @@ def parse_json_response(result_text: str, data_type: str) -> Tuple[bool, Optiona
def extract_topics_with_regex(result_text: str, max_topics: int) -> List[Dict]:
"""
使用正则表达式提取话题信息
Args:
result_text: 需要提取的文本
max_topics: 最大话题数量
Returns:
话题数据列表
"""
@@ -140,33 +141,38 @@ def extract_topics_with_regex(result_text: str, max_topics: int) -> List[Dict]:
# 匹配每个完整的话题对象
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)
topics = []
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', ' ')
detail = detail.replace('\\"', '"').replace("\\n", " ").replace("\\t", " ")
# 解析参与者列表
contributors = [contrib.strip() for contrib in re.findall(r'"([^"]+)"', contributors_str)] or ["群友"]
topics.append({
"topic": topic_name,
"contributors": contributors[:5], # 最多5个参与者
"detail": detail
})
contributors = [
contrib.strip()
for contrib in re.findall(r'"([^"]+)"', contributors_str)
] or ["群友"]
topics.append(
{
"topic": topic_name,
"contributors": contributors[:5], # 最多5个参与者
"detail": detail,
}
)
logger.info(f"话题正则表达式提取成功,提取到 {len(topics)} 条有效话题内容")
return topics
except Exception as e:
logger.error(f"话题正则表达式提取失败: {e}")
return []
@@ -175,47 +181,43 @@ def extract_topics_with_regex(result_text: str, max_topics: int) -> List[Dict]:
def extract_user_titles_with_regex(result_text: str, max_count: int) -> List[Dict]:
"""
使用正则表达式提取用户称号信息
Args:
result_text: 需要提取的文本
max_count: 最大提取数量
Returns:
用户称号数据列表
"""
try:
titles = []
# 正则模式:匹配完整的用户称号对象
pattern = r'\{\s*"name":\s*"([^"]+)"\s*,\s*"qq":\s*(\d+)\s*,\s*"title":\s*"([^"]+)"\s*,\s*"mbti":\s*"([^"]+)"\s*,\s*"reason":\s*"([^"]*(?:\\.[^"]*)*)"\s*\}'
matches = re.findall(pattern, result_text, re.DOTALL)
if not matches:
# 尝试更宽松的匹配(字段顺序可变)
pattern = r'"name":\s*"([^"]+)"[^}]*"qq":\s*(\d+)[^}]*"title":\s*"([^"]+)"[^}]*"mbti":\s*"([^"]+)"[^}]*"reason":\s*"([^"]*(?:\\.[^"]*)*)"'
matches = re.findall(pattern, result_text, re.DOTALL)
for match in matches[:max_count]:
name = match[0].strip()
qq = int(match[1])
title = match[2].strip()
mbti = match[3].strip()
reason = match[4].strip()
# 清理转义字符
reason = reason.replace('\\"', '"').replace('\\n', ' ').replace('\\t', ' ')
titles.append({
"name": name,
"qq": qq,
"title": title,
"mbti": mbti,
"reason": reason
})
reason = reason.replace('\\"', '"').replace("\\n", " ").replace("\\t", " ")
titles.append(
{"name": name, "qq": qq, "title": title, "mbti": mbti, "reason": reason}
)
logger.info(f"用户称号正则表达式提取成功,提取到 {len(titles)} 条有效用户称号")
return titles
except Exception as e:
logger.error(f"用户称号正则表达式提取失败: {e}")
return []
@@ -224,44 +226,42 @@ def extract_user_titles_with_regex(result_text: str, max_count: int) -> List[Dic
def extract_golden_quotes_with_regex(result_text: str, max_count: int) -> List[Dict]:
"""
使用正则表达式提取金句信息
Args:
result_text: 需要提取的文本
max_count: 最大提取数量
Returns:
金句数据列表
"""
try:
quotes = []
# 正则模式:匹配完整的金句对象
pattern = r'\{\s*"content":\s*"([^"]*(?:\\.[^"]*)*)"\s*,\s*"sender":\s*"([^"]+)"\s*,\s*"reason":\s*"([^"]*(?:\\.[^"]*)*)"\s*\}'
matches = re.findall(pattern, result_text, re.DOTALL)
if not matches:
# 尝试更宽松的匹配(字段顺序可变)
pattern = r'"content":\s*"([^"]*(?:\\.[^"]*)*)"[^}]*"sender":\s*"([^"]+)"[^}]*"reason":\s*"([^"]*(?:\\.[^"]*)*)"'
matches = re.findall(pattern, result_text, re.DOTALL)
for match in matches[:max_count]:
content = match[0].strip()
sender = match[1].strip()
reason = match[2].strip()
# 清理转义字符
content = content.replace('\\"', '"').replace('\\n', ' ').replace('\\t', ' ')
reason = reason.replace('\\"', '"').replace('\\n', ' ').replace('\\t', ' ')
quotes.append({
"content": content,
"sender": sender,
"reason": reason
})
content = (
content.replace('\\"', '"').replace("\\n", " ").replace("\\t", " ")
)
reason = reason.replace('\\"', '"').replace("\\n", " ").replace("\\t", " ")
quotes.append({"content": content, "sender": sender, "reason": reason})
logger.info(f"金句正则表达式提取成功,提取到 {len(quotes)} 条有效金句")
return quotes
except Exception as e:
logger.error(f"金句正则表达式提取失败: {e}")
return []
+90 -55
View File
@@ -9,11 +9,17 @@ from astrbot.api import logger
import aiohttp
async def call_provider_with_retry(context, config_manager, prompt: str, max_tokens: int,
temperature: float, umo: str = None) -> Optional[Any]:
async def call_provider_with_retry(
context,
config_manager,
prompt: str,
max_tokens: int,
temperature: float,
umo: str = None,
) -> Optional[Any]:
"""
调用LLM提供者,带超时、重试与退避。支持自定义服务商。
Args:
context: AstrBot上下文对象
config_manager: 配置管理器
@@ -21,59 +27,80 @@ async def call_provider_with_retry(context, config_manager, prompt: str, max_tok
max_tokens: 最大生成token数
temperature: 采样温度
umo: 指定使用的模型唯一标识符
Returns:
LLM生成的结果,失败时返回None
"""
timeout = config_manager.get_llm_timeout()
retries = config_manager.get_llm_retries()
backoff = config_manager.get_llm_backoff()
# 获取自定义服务商参数
custom_api_key = config_manager.get_custom_api_key()
custom_api_base = config_manager.get_custom_api_base_url()
custom_model = config_manager.get_custom_model_name()
last_exc = None
for attempt in range(1, retries + 1):
try:
if custom_api_key and custom_api_base and custom_model:
logger.info(f"使用自定义LLM提供商: {custom_api_base} model={custom_model}, max_tokens={max_tokens}, temperature={temperature}")
logger.debug(f"自定义LLM提供商 prompt 长度: {len(prompt) if prompt else 0}")
logger.debug(f"自定义LLM提供商 prompt 前100字符: {prompt[:100] if prompt else 'None'}...")
logger.info(
f"使用自定义LLM提供商: {custom_api_base} model={custom_model}, max_tokens={max_tokens}, temperature={temperature}"
)
logger.debug(
f"自定义LLM提供商 prompt 长度: {len(prompt) if prompt else 0}"
)
logger.debug(
f"自定义LLM提供商 prompt 前100字符: {prompt[:100] if prompt else 'None'}..."
)
# 检查 prompt 是否为空
if not prompt or not prompt.strip():
logger.error("自定义LLM提供商: prompt 为空或只包含空白字符,无法发送请求")
logger.error(
"自定义LLM提供商: prompt 为空或只包含空白字符,无法发送请求"
)
return None
async with aiohttp.ClientSession() as session:
headers = {
"Authorization": f"Bearer {custom_api_key}",
"Content-Type": "application/json"
"Content-Type": "application/json",
}
payload = {
"model": custom_model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens,
"temperature": temperature
"temperature": temperature,
}
aio_timeout = aiohttp.ClientTimeout(total=timeout)
async with session.post(custom_api_base, json=payload, headers=headers, timeout=aio_timeout) as resp:
async with session.post(
custom_api_base,
json=payload,
headers=headers,
timeout=aio_timeout,
) as resp:
if resp.status != 200:
error_text = await resp.text()
logger.error(f"自定义LLM服务商请求失败: HTTP {resp.status}, 内容: {error_text}")
logger.error(
f"自定义LLM服务商请求失败: HTTP {resp.status}, 内容: {error_text}"
)
try:
response_json = await resp.json()
except Exception as json_err:
error_text = await resp.text()
logger.error(f"自定义LLM服务商响应JSON解析失败: {json_err}, 内容: {error_text}")
logger.error(
f"自定义LLM服务商响应JSON解析失败: {json_err}, 内容: {error_text}"
)
return None
# 兼容 OpenAI 格式,安全访问嵌套字段
content = None
try:
choices = response_json.get("choices")
if choices and isinstance(choices, list) and len(choices) > 0:
if (
choices
and isinstance(choices, list)
and len(choices) > 0
):
message = choices[0].get("message")
if message and isinstance(message, dict):
content = message.get("content")
@@ -81,17 +108,21 @@ async def call_provider_with_retry(context, config_manager, prompt: str, max_tok
logger.error(f"自定义LLM响应格式异常: {response_json}")
return None
except Exception as key_err:
logger.error(f"自定义LLM响应结构解析失败: {key_err}, 响应内容: {response_json}")
logger.error(
f"自定义LLM响应结构解析失败: {key_err}, 响应内容: {response_json}"
)
return None
# 构造一个兼容原有逻辑的对象
class CustomResponse:
completion_text = content
raw_completion = response_json
return CustomResponse()
else:
# 确保使用当前指定的模型
provider = context.get_using_provider(umo=umo)
provider_id = 'unknown'
provider_id = "unknown"
if provider:
try:
meta = provider.meta()
@@ -99,23 +130,33 @@ async def call_provider_with_retry(context, config_manager, prompt: str, max_tok
except Exception as e:
logger.debug(f"获取提供商ID失败: {e}")
logger.info(f"获取到的 provider ID: {provider_id}")
if not provider or provider_id == 'unknown':
if not provider or provider_id == "unknown":
logger.warning(f"获取的提供商不正确 (Provider ID: {provider_id})")
logger.info(f"使用LLM provider: {provider}, max_tokens={max_tokens}, temperature={temperature}")
logger.info(
f"使用LLM provider: {provider}, max_tokens={max_tokens}, temperature={temperature}"
)
if not provider:
logger.error("provider 为空,无法调用 text_chat,直接返回 None")
return None
logger.debug(f"LLM provider prompt 长度: {len(prompt) if prompt else 0}")
logger.debug(f"LLM provider prompt 前100字符: {prompt[:100] if prompt else 'None'}...")
logger.debug(
f"LLM provider prompt 长度: {len(prompt) if prompt else 0}"
)
logger.debug(
f"LLM provider prompt 前100字符: {prompt[:100] if prompt else 'None'}..."
)
# 检查 prompt 是否为空
if not prompt or not prompt.strip():
logger.error("LLM provider: prompt 为空或只包含空白字符,无法调用 text_chat")
logger.error(
"LLM provider: prompt 为空或只包含空白字符,无法调用 text_chat"
)
return None
coro = provider.text_chat(prompt=prompt, max_tokens=max_tokens, temperature=temperature)
coro = provider.text_chat(
prompt=prompt, max_tokens=max_tokens, temperature=temperature
)
return await asyncio.wait_for(coro, timeout=timeout)
except asyncio.TimeoutError as e:
last_exc = e
@@ -126,7 +167,7 @@ async def call_provider_with_retry(context, config_manager, prompt: str, max_tok
# 若非最后一次,等待退避后重试
if attempt < retries:
await asyncio.sleep(backoff * attempt)
# 最终仍失败,记录错误并返回 None 由调用方处理降级,避免抛出异常
logger.error(f"LLM请求全部重试失败: {last_exc}")
return None
@@ -135,55 +176,49 @@ async def call_provider_with_retry(context, config_manager, prompt: str, max_tok
def extract_token_usage(response) -> Optional[dict]:
"""
从LLM响应中提取token使用统计
Args:
response: LLM响应对象
Returns:
Token使用统计字典,包含prompt_tokens, completion_tokens, total_tokens
"""
try:
token_usage = {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0
}
token_usage = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
# 安全地提取 usage,避免 response.raw_completion.usage 为 None 导致的 AttributeError
usage = None
if getattr(response, 'raw_completion', None) is not None:
usage = getattr(response.raw_completion, 'usage', None)
if getattr(response, "raw_completion", None) is not None:
usage = getattr(response.raw_completion, "usage", None)
if usage:
token_usage["prompt_tokens"] = getattr(usage, 'prompt_tokens', 0) or 0
token_usage["completion_tokens"] = getattr(usage, 'completion_tokens', 0) or 0
token_usage["total_tokens"] = getattr(usage, 'total_tokens', 0) or 0
token_usage["prompt_tokens"] = getattr(usage, "prompt_tokens", 0) or 0
token_usage["completion_tokens"] = (
getattr(usage, "completion_tokens", 0) or 0
)
token_usage["total_tokens"] = getattr(usage, "total_tokens", 0) or 0
return token_usage
except Exception as e:
logger.error(f"提取token使用统计失败: {e}")
return {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0
}
return {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
def extract_response_text(response) -> str:
"""
从LLM响应中提取文本内容
Args:
response: LLM响应对象
Returns:
响应文本内容
"""
try:
if hasattr(response, 'completion_text'):
if hasattr(response, "completion_text"):
return response.completion_text
else:
return str(response)
except Exception as e:
logger.error(f"提取响应文本失败: {e}")
return ""
return ""
+1 -4
View File
@@ -5,7 +5,4 @@
from .config import ConfigManager
from .message_handler import MessageHandler
__all__ = [
'ConfigManager',
'MessageHandler'
]
__all__ = ["ConfigManager", "MessageHandler"]
+22 -23
View File
@@ -5,20 +5,21 @@ Bot实例管理模块
from typing import Dict, Any
class BotManager:
"""Bot实例管理器 - 统一管理所有bot相关操作"""
def __init__(self, config_manager):
self.config_manager = config_manager
self._bot_instance = None
self._bot_qq_id = None
self._context = None
self._is_initialized = False
def set_context(self, context):
"""设置AstrBot上下文"""
self._context = context
def set_bot_instance(self, bot_instance):
"""设置bot实例"""
if bot_instance:
@@ -29,48 +30,46 @@ class BotManager:
if bot_qq_id:
self._bot_qq_id = str(bot_qq_id)
def set_bot_qq_id(self, bot_qq_id: str):
"""设置bot QQ号"""
if bot_qq_id:
self._bot_qq_id = str(bot_qq_id)
def get_bot_instance(self):
"""获取当前bot实例"""
return self._bot_instance
def has_bot_instance(self) -> bool:
"""检查是否有可用的bot实例"""
return self._bot_instance is not None
def has_bot_qq_id(self) -> bool:
"""检查是否有配置的bot QQ号"""
return self._bot_qq_id is not None
def is_ready_for_auto_analysis(self) -> bool:
"""检查是否准备好进行自动分析"""
return self.has_bot_instance() and self.has_bot_qq_id()
async def auto_discover_bot_instance(self):
"""自动发现可用的bot实例"""
if not self._context or not hasattr(self._context, 'platform_manager'):
if not self._context or not hasattr(self._context, "platform_manager"):
return None
platforms = getattr(self._context.platform_manager, 'platform_insts', [])
platforms = getattr(self._context.platform_manager, "platform_insts", [])
for platform in platforms:
# 获取bot实例
bot_client = None
if hasattr(platform, 'get_client'):
if hasattr(platform, "get_client"):
bot_client = platform.get_client()
elif hasattr(platform, 'bot'):
elif hasattr(platform, "bot"):
bot_client = platform.bot
if bot_client:
self.set_bot_instance(bot_client)
return bot_client
return None
async def initialize_from_config(self):
"""从配置初始化bot管理器"""
# 设置配置的bot QQ号
@@ -84,19 +83,19 @@ class BotManager:
# 返回是否成功初始化(至少有bot实例)
return self.has_bot_instance()
def get_status_info(self) -> Dict[str, Any]:
"""获取bot管理器状态信息"""
return {
"has_bot_instance": self.has_bot_instance(),
"has_bot_qq_id": self.has_bot_qq_id(),
"bot_qq_id": self._bot_qq_id,
"ready_for_auto_analysis": self.is_ready_for_auto_analysis()
"ready_for_auto_analysis": self.is_ready_for_auto_analysis(),
}
def update_from_event(self, event):
"""从事件更新bot实例(用于手动命令)"""
if hasattr(event, 'bot') and event.bot:
if hasattr(event, "bot") and event.bot:
self.set_bot_instance(event.bot)
# 每次都尝试从bot实例提取QQ号
bot_qq_id = self._extract_bot_qq_id(event.bot)
@@ -113,18 +112,18 @@ class BotManager:
def _extract_bot_qq_id(self, bot_instance):
"""从bot实例中提取QQ号"""
# 尝试多种方式获取bot QQ号
if hasattr(bot_instance, 'self_id') and bot_instance.self_id:
if hasattr(bot_instance, "self_id") and bot_instance.self_id:
return str(bot_instance.self_id)
elif hasattr(bot_instance, 'qq') and bot_instance.qq:
elif hasattr(bot_instance, "qq") and bot_instance.qq:
return str(bot_instance.qq)
elif hasattr(bot_instance, 'user_id') and bot_instance.user_id:
elif hasattr(bot_instance, "user_id") and bot_instance.user_id:
return str(bot_instance.user_id)
return None
def validate_for_message_fetching(self, group_id: str) -> bool:
"""验证是否可以进行消息获取"""
return self.has_bot_instance() and bool(group_id)
def should_filter_bot_message(self, sender_id: str) -> bool:
"""判断是否应该过滤bot自己的消息"""
if not self._bot_qq_id:
+38 -22
View File
@@ -4,8 +4,6 @@
"""
import sys
import importlib
from pathlib import Path
from typing import Optional, List
from astrbot.api import logger, AstrBotConfig
@@ -110,25 +108,30 @@ class ConfigManager:
def get_custom_model_name(self) -> str:
"""获取自定义 LLM 服务的模型名称"""
return self.config.get("custom_model_name", "")
def get_pdf_output_dir(self) -> str:
"""获取PDF输出目录"""
return self.config.get("pdf_output_dir", "data/plugins/astrbot-qq-group-daily-analysis/reports")
return self.config.get(
"pdf_output_dir", "data/plugins/astrbot-qq-group-daily-analysis/reports"
)
def get_bot_qq_id(self) -> str:
"""获取bot QQ号"""
return str(self.config.get("bot_qq_id", ""))
def get_pdf_filename_format(self) -> str:
"""获取PDF文件名格式"""
return self.config.get("pdf_filename_format", "群聊分析报告_{group_id}_{date}.pdf")
return self.config.get(
"pdf_filename_format", "群聊分析报告_{group_id}_{date}.pdf"
)
def get_topic_analysis_prompt(self, style: str = "topic_prompt") -> str:
"""
获取话题分析提示词模板
Args:
style: 提示词风格,默认为 "topic_prompt"
Returns:
提示词模板字符串
"""
@@ -144,10 +147,10 @@ class ConfigManager:
def get_user_title_analysis_prompt(self, style: str = "user_title_prompt") -> str:
"""
获取用户称号分析提示词模板
Args:
style: 提示词风格,默认为 "user_title_prompt"
Returns:
提示词模板字符串
"""
@@ -160,13 +163,15 @@ class ConfigManager:
# 兼容旧配置
return self.config.get("user_title_analysis_prompt", "")
def get_golden_quote_analysis_prompt(self, style: str = "golden_quote_prompt") -> str:
def get_golden_quote_analysis_prompt(
self, style: str = "golden_quote_prompt"
) -> str:
"""
获取金句分析提示词模板
Args:
style: 提示词风格,默认为 "golden_quote_prompt"
Returns:
提示词模板字符串
"""
@@ -309,6 +314,7 @@ class ConfigManager:
try:
import pyppeteer
from pyppeteer import launch
self._pyppeteer_available = True
# 检查版本
@@ -322,7 +328,9 @@ class ConfigManager:
except ImportError:
self._pyppeteer_available = False
self._pyppeteer_version = None
logger.warning("pyppeteer 未安装,PDF 功能将不可用。请使用 /安装PDF 命令安装 pyppeteer==1.0.2")
logger.warning(
"pyppeteer 未安装,PDF 功能将不可用。请使用 /安装PDF 命令安装 pyppeteer==1.0.2"
)
def reload_pyppeteer(self) -> bool:
"""重新加载 pyppeteer 模块"""
@@ -330,7 +338,9 @@ class ConfigManager:
logger.info("开始重新加载 pyppeteer 模块...")
# 移除所有 pyppeteer 相关模块
modules_to_remove = [mod for mod in sys.modules.keys() if mod.startswith('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]
@@ -344,22 +354,28 @@ class ConfigManager:
self._pyppeteer_available = True
try:
self._pyppeteer_version = pyppeteer.__version__
logger.info(f"重新加载成功,pyppeteer 版本: {self._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 功能")
except ImportError:
logger.info("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 功能")
except Exception:
logger.info("pyppeteer 重新导入需要重启 AstrBot 才能生效")
logger.info(
"💡 提示:pyppeteer 安装成功,但需要重启 AstrBot 后才能使用 PDF 功能"
)
self._pyppeteer_available = False
self._pyppeteer_version = None
return False
@@ -384,4 +400,4 @@ class ConfigManager:
# 配置会自动从self.config中重新读取
logger.info("配置重载完成")
except Exception as e:
logger.error(f"重新加载配置失败: {e}")
logger.error(f"重新加载配置失败: {e}")
+77 -34
View File
@@ -3,12 +3,11 @@
负责群聊消息的获取、过滤和预处理
"""
import asyncio
from datetime import datetime, timedelta
from typing import List, Dict, Optional
from typing import List, Dict
from collections import defaultdict
from astrbot.api import logger
from ...src.models.data_models import GroupStatistics, TokenUsage, EmojiStatistics, ActivityVisualization
from ...src.models.data_models import GroupStatistics, TokenUsage, EmojiStatistics
from ...src.visualization.activity_charts import ActivityVisualizer
@@ -35,15 +34,17 @@ class MessageHandler:
def _extract_bot_qq_id_from_instance(self, bot_instance):
"""从bot实例中提取QQ号"""
if hasattr(bot_instance, 'self_id') and bot_instance.self_id:
if hasattr(bot_instance, "self_id") and bot_instance.self_id:
return str(bot_instance.self_id)
elif hasattr(bot_instance, 'qq') and bot_instance.qq:
elif hasattr(bot_instance, "qq") and bot_instance.qq:
return str(bot_instance.qq)
elif hasattr(bot_instance, 'user_id') and bot_instance.user_id:
elif hasattr(bot_instance, "user_id") and bot_instance.user_id:
return str(bot_instance.user_id)
return None
async def fetch_group_messages(self, bot_instance, group_id: str, days: int) -> List[Dict]:
async def fetch_group_messages(
self, bot_instance, group_id: str, days: int
) -> List[Dict]:
"""获取群聊消息记录"""
try:
# 验证参数
@@ -68,7 +69,9 @@ class MessageHandler:
query_rounds = 0
logger.info(f"开始获取群 {group_id}{days} 天的消息记录")
logger.info(f"时间范围: {start_time.strftime('%Y-%m-%d %H:%M:%S')}{end_time.strftime('%Y-%m-%d %H:%M:%S')}")
logger.info(
f"时间范围: {start_time.strftime('%Y-%m-%d %H:%M:%S')}{end_time.strftime('%Y-%m-%d %H:%M:%S')}"
)
# 单次请求,按配置的 max_messages 作为 count
try:
@@ -78,21 +81,29 @@ class MessageHandler:
}
result = None
if hasattr(bot_instance, 'call_action'):
if hasattr(bot_instance, "call_action"):
try:
# aiocqhttp (CQHttp) 方式
result = await bot_instance.call_action("get_group_msg_history", **payloads)
result = await bot_instance.call_action(
"get_group_msg_history", **payloads
)
query_rounds = 1
except Exception as api_err:
logger.error(f"{group_id} API 调用失败: {api_err}")
logger.error(f"{group_id} 当前 OneBot 实现可能不支持 get_group_msg_history API")
logger.error(
f"{group_id} 当前 OneBot 实现可能不支持 get_group_msg_history API"
)
return []
elif hasattr(bot_instance, 'api'):
elif hasattr(bot_instance, "api"):
# QQ 官方 bot (botClient) 不支持历史消息
logger.error(f"{group_id} 检测到 QQ 官方 Bot,官方 API 不支持获取历史消息")
logger.error(
f"{group_id} 检测到 QQ 官方 Bot,官方 API 不支持获取历史消息"
)
return []
else:
logger.error(f"{group_id} 未知的 bot_instance 类型,无法调用 API,类型: {type(bot_instance)}")
logger.error(
f"{group_id} 未知的 bot_instance 类型,无法调用 API,类型: {type(bot_instance)}"
)
return []
if not result or "messages" not in result:
@@ -102,7 +113,7 @@ class MessageHandler:
round_messages = result.get("messages", [])
if not round_messages:
logger.info(f"{group_id} 未获取到消息")
# 过滤时间范围内的消息并过滤机器人自身消息
for msg in round_messages:
try:
@@ -110,7 +121,10 @@ class MessageHandler:
if not (start_time <= msg_time <= end_time):
continue
sender_id = str(msg.get("sender", {}).get("user_id", ""))
if self.bot_manager and self.bot_manager.should_filter_bot_message(sender_id):
if (
self.bot_manager
and self.bot_manager.should_filter_bot_message(sender_id)
):
continue
messages.append(msg)
except Exception as msg_error:
@@ -122,25 +136,32 @@ class MessageHandler:
# ========== 最终清理步骤:严格过滤和限制 ==========
original_count = len(messages)
# 1. 严格过滤时间范围外的消息
messages = [
msg for msg in messages
msg
for msg in messages
if start_time <= datetime.fromtimestamp(msg.get("time", 0)) <= end_time
]
time_filtered_count = len(messages)
# 2. 严格限制消息数量
if len(messages) > max_messages:
# 保留最新的消息(假设messages已按时间排序,从新到旧)
messages = messages[:max_messages]
logger.info(f"{group_id} 消息数量超过限制,已截断: {time_filtered_count} -> {max_messages}")
logger.info(
f"{group_id} 消息数量超过限制,已截断: {time_filtered_count} -> {max_messages}"
)
# 记录清理结果
if original_count != len(messages):
logger.info(f"{group_id} 最终清理: 原始 {original_count} 条 -> 时间过滤 {time_filtered_count} 条 -> 最终 {len(messages)}")
logger.info(f"{group_id} 消息获取完成,共获取到 {len(messages)} 条有效消息(时间范围: 近{days}天),查询轮数: {query_rounds}")
logger.info(
f"{group_id} 最终清理: 原始 {original_count} 条 -> 时间过滤 {time_filtered_count} 条 -> 最终 {len(messages)}"
)
logger.info(
f"{group_id} 消息获取完成,共获取到 {len(messages)} 条有效消息(时间范围: 近{days}天),查询轮数: {query_rounds}"
)
return messages
except Exception as e:
@@ -171,22 +192,30 @@ class MessageHandler:
# QQ基础表情
emoji_statistics.face_count += 1
face_id = content.get("data", {}).get("id", "unknown")
emoji_statistics.face_details[f"face_{face_id}"] = emoji_statistics.face_details.get(f"face_{face_id}", 0) + 1
emoji_statistics.face_details[f"face_{face_id}"] = (
emoji_statistics.face_details.get(f"face_{face_id}", 0) + 1
)
elif content.get("type") == "mface":
# 动画表情/魔法表情
emoji_statistics.mface_count += 1
emoji_id = content.get("data", {}).get("emoji_id", "unknown")
emoji_statistics.face_details[f"mface_{emoji_id}"] = emoji_statistics.face_details.get(f"mface_{emoji_id}", 0) + 1
emoji_statistics.face_details[f"mface_{emoji_id}"] = (
emoji_statistics.face_details.get(f"mface_{emoji_id}", 0) + 1
)
elif content.get("type") == "bface":
# 超级表情
emoji_statistics.bface_count += 1
emoji_id = content.get("data", {}).get("p", "unknown")
emoji_statistics.face_details[f"bface_{emoji_id}"] = emoji_statistics.face_details.get(f"bface_{emoji_id}", 0) + 1
emoji_statistics.face_details[f"bface_{emoji_id}"] = (
emoji_statistics.face_details.get(f"bface_{emoji_id}", 0) + 1
)
elif content.get("type") == "sface":
# 小表情
emoji_statistics.sface_count += 1
emoji_id = content.get("data", {}).get("id", "unknown")
emoji_statistics.face_details[f"sface_{emoji_id}"] = emoji_statistics.face_details.get(f"sface_{emoji_id}", 0) + 1
emoji_statistics.face_details[f"sface_{emoji_id}"] = (
emoji_statistics.face_details.get(f"sface_{emoji_id}", 0) + 1
)
elif content.get("type") == "image":
# 检查是否是动画表情(通过summary字段判断)
data = content.get("data", {})
@@ -195,20 +224,34 @@ class MessageHandler:
# 动画表情(以image形式发送)
emoji_statistics.mface_count += 1
file_name = data.get("file", "unknown")
emoji_statistics.face_details[f"animated_{file_name}"] = emoji_statistics.face_details.get(f"animated_{file_name}", 0) + 1
emoji_statistics.face_details[f"animated_{file_name}"] = (
emoji_statistics.face_details.get(
f"animated_{file_name}", 0
)
+ 1
)
else:
# 普通图片,不计入表情统计
pass
elif content.get("type") in ["record", "video"] and "emoji" in str(content.get("data", {})).lower():
elif (
content.get("type") in ["record", "video"]
and "emoji" in str(content.get("data", {})).lower()
):
# 其他可能的表情类型
emoji_statistics.other_emoji_count += 1
# 找出最活跃时段
most_active_hour = max(hour_counts.items(), key=lambda x: x[1])[0] if hour_counts else 0
most_active_period = f"{most_active_hour:02d}:00-{(most_active_hour+1)%24:02d}:00"
most_active_hour = (
max(hour_counts.items(), key=lambda x: x[1])[0] if hour_counts else 0
)
most_active_period = (
f"{most_active_hour:02d}:00-{(most_active_hour + 1) % 24:02d}:00"
)
# 生成活跃度可视化数据
activity_visualization = self.activity_visualizer.generate_activity_visualization(messages)
activity_visualization = (
self.activity_visualizer.generate_activity_visualization(messages)
)
return GroupStatistics(
message_count=len(messages),
@@ -219,5 +262,5 @@ class MessageHandler:
emoji_count=emoji_statistics.total_emoji_count, # 保持向后兼容
emoji_statistics=emoji_statistics,
activity_visualization=activity_visualization,
token_usage=TokenUsage()
token_usage=TokenUsage(),
)
+2 -8
View File
@@ -7,13 +7,7 @@ from .data_models import (
UserTitle,
GoldenQuote,
TokenUsage,
GroupStatistics
GroupStatistics,
)
__all__ = [
'SummaryTopic',
'UserTitle',
'GoldenQuote',
'TokenUsage',
'GroupStatistics'
]
__all__ = ["SummaryTopic", "UserTitle", "GoldenQuote", "TokenUsage", "GroupStatistics"]
+19 -4
View File
@@ -10,6 +10,7 @@ from typing import List
@dataclass
class SummaryTopic:
"""话题总结数据结构"""
topic: str
contributors: List[str]
detail: str
@@ -18,6 +19,7 @@ class SummaryTopic:
@dataclass
class UserTitle:
"""用户称号数据结构"""
name: str
qq: int
title: str
@@ -28,6 +30,7 @@ class UserTitle:
@dataclass
class GoldenQuote:
"""群聊金句数据结构"""
content: str
sender: str
reason: str
@@ -36,6 +39,7 @@ class GoldenQuote:
@dataclass
class TokenUsage:
"""Token使用统计"""
prompt_tokens: int = 0
completion_tokens: int = 0
total_tokens: int = 0
@@ -44,6 +48,7 @@ class TokenUsage:
@dataclass
class EmojiStatistics:
"""表情统计数据结构"""
face_count: int = 0 # QQ基础表情数量
mface_count: int = 0 # 动画表情数量
bface_count: int = 0 # 超级表情数量
@@ -54,14 +59,21 @@ class EmojiStatistics:
@property
def total_emoji_count(self) -> int:
"""总表情数量"""
return self.face_count + self.mface_count + self.bface_count + self.sface_count + self.other_emoji_count
return (
self.face_count
+ self.mface_count
+ self.bface_count
+ self.sface_count
+ self.other_emoji_count
)
@dataclass
class ActivityVisualization:
"""活跃度可视化数据结构"""
hourly_activity: dict = field(default_factory=dict) # {hour: count}
daily_activity: dict = field(default_factory=dict) # {date: count}
daily_activity: dict = field(default_factory=dict) # {date: count}
user_activity_ranking: list = field(default_factory=list) # 用户活跃度排行
peak_hours: list = field(default_factory=list) # 高峰时段
activity_heatmap_data: dict = field(default_factory=dict) # 热力图数据
@@ -70,6 +82,7 @@ class ActivityVisualization:
@dataclass
class GroupStatistics:
"""群聊统计数据结构"""
message_count: int
total_characters: int
participant_count: int
@@ -77,5 +90,7 @@ class GroupStatistics:
golden_quotes: List[GoldenQuote]
emoji_count: int # 保持向后兼容
emoji_statistics: EmojiStatistics = field(default_factory=EmojiStatistics)
activity_visualization: ActivityVisualization = field(default_factory=ActivityVisualization)
token_usage: TokenUsage = field(default_factory=TokenUsage)
activity_visualization: ActivityVisualization = field(
default_factory=ActivityVisualization
)
token_usage: TokenUsage = field(default_factory=TokenUsage)
+1 -4
View File
@@ -6,7 +6,4 @@
from .generators import ReportGenerator
from .templates import HTMLTemplates
__all__ = [
'ReportGenerator',
'HTMLTemplates'
]
__all__ = ["ReportGenerator", "HTMLTemplates"]
+141 -109
View File
@@ -21,7 +21,9 @@ class ReportGenerator:
self.config_manager = config_manager
self.activity_visualizer = ActivityVisualizer()
async def generate_image_report(self, analysis_result: Dict, group_id: str, html_render_func) -> Optional[str]:
async def generate_image_report(
self, analysis_result: Dict, group_id: str, html_render_func
) -> Optional[str]:
"""生成图片格式的分析报告"""
try:
# 准备渲染数据
@@ -31,13 +33,13 @@ class ReportGenerator:
image_options = {
"full_page": True,
"type": "jpeg", # 使用默认的jpeg格式提高兼容性
"quality": 95, # 设置合理的质量
"quality": 95, # 设置合理的质量
}
image_url = await html_render_func(
HTMLTemplates.get_image_template(),
render_payload,
True, # return_url=True,返回URL而不是下载文件
image_options
image_options,
)
logger.info(f"图片生成成功: {image_url}")
@@ -51,13 +53,13 @@ class ReportGenerator:
simple_options = {
"full_page": True,
"type": "jpeg",
"quality": 70 # 降低质量以提高兼容性
"quality": 70, # 降低质量以提高兼容性
}
image_url = await html_render_func(
HTMLTemplates.get_image_template(),
render_payload,
True,
simple_options
simple_options,
)
logger.info(f"使用低质量选项生成成功: {image_url}")
return image_url
@@ -65,9 +67,9 @@ class ReportGenerator:
logger.error(f"后备低质量方案也失败: {fallback_e}")
return None
async def generate_pdf_report(self, analysis_result: Dict, group_id: str) -> Optional[str]:
async def generate_pdf_report(
self, analysis_result: Dict, group_id: str
) -> Optional[str]:
"""生成PDF格式的分析报告"""
try:
# 确保输出目录存在
@@ -75,10 +77,9 @@ class ReportGenerator:
output_dir.mkdir(parents=True, exist_ok=True)
# 生成文件名
current_date = datetime.now().strftime('%Y%m%d')
current_date = datetime.now().strftime("%Y%m%d")
filename = self.config_manager.get_pdf_filename_format().format(
group_id=group_id,
date=current_date
group_id=group_id, date=current_date
)
pdf_path = output_dir / filename
@@ -87,7 +88,9 @@ class ReportGenerator:
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)
html_content = self._render_html_template(
HTMLTemplates.get_pdf_template(), render_data, use_jinja_style=False
)
logger.info(f"HTML 内容生成完成,长度: {len(html_content)} 字符")
# 转换为 PDF
@@ -110,7 +113,7 @@ class ReportGenerator:
report = f"""
🎯 群聊日常分析报告
📅 {datetime.now().strftime('%Y年%m月%d')}
📅 {datetime.now().strftime("%Y年%m月%d")}
📊 基础统计
• 消息总数: {stats.message_count}
@@ -138,7 +141,7 @@ class ReportGenerator:
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'{i}. "{quote.content}" —— {quote.sender}\n'
report += f" {quote.reason}\n\n"
return report
@@ -172,7 +175,11 @@ class ReportGenerator:
for title in user_titles[:max_user_titles]:
# 获取用户头像
avatar_data = await self._get_user_avatar(str(title.qq))
avatar_html = f'<img src="{avatar_data}" class="user-avatar" alt="头像">' if avatar_data else '<div class="user-avatar-placeholder">👤</div>'
avatar_html = (
f'<img src="{avatar_data}" class="user-avatar" alt="头像">'
if avatar_data
else '<div class="user-avatar-placeholder">👤</div>'
)
titles_html += f"""
<div class="user-title">
@@ -209,8 +216,8 @@ class ReportGenerator:
# 返回扁平化的渲染数据
return {
"current_date": datetime.now().strftime('%Y年%m月%d'),
"current_datetime": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
"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,
@@ -220,15 +227,20 @@ class ReportGenerator:
"titles_html": titles_html,
"quotes_html": quotes_html,
"hourly_chart_html": hourly_chart_html,
"total_tokens": stats.token_usage.total_tokens if stats.token_usage.total_tokens else 0,
"prompt_tokens": stats.token_usage.prompt_tokens if stats.token_usage.prompt_tokens else 0,
"completion_tokens": stats.token_usage.completion_tokens if stats.token_usage.completion_tokens else 0
"total_tokens": stats.token_usage.total_tokens
if stats.token_usage.total_tokens
else 0,
"prompt_tokens": stats.token_usage.prompt_tokens
if stats.token_usage.prompt_tokens
else 0,
"completion_tokens": stats.token_usage.completion_tokens
if stats.token_usage.completion_tokens
else 0,
}
def _render_html_template(self, template: str, data: Dict, use_jinja_style: bool = False) -> str:
def _render_html_template(
self, template: str, data: Dict, use_jinja_style: bool = False
) -> str:
"""HTML模板渲染,支持两种占位符格式
Args:
@@ -239,7 +251,9 @@ class ReportGenerator:
result = template
# 调试:记录渲染数据
logger.info(f"渲染数据键: {list(data.keys())}, 使用Jinja风格: {use_jinja_style}")
logger.info(
f"渲染数据键: {list(data.keys())}, 使用Jinja风格: {use_jinja_style}"
)
for key, value in data.items():
if use_jinja_style:
@@ -256,10 +270,11 @@ class ReportGenerator:
# 检查是否还有未替换的占位符
import re
if use_jinja_style:
remaining_placeholders = re.findall(r'\{\{[^}]+\}\}', result)
remaining_placeholders = re.findall(r"\{\{[^}]+\}\}", result)
else:
remaining_placeholders = re.findall(r'\{[^}]+\}', result)
remaining_placeholders = re.findall(r"\{[^}]+\}", result)
if remaining_placeholders:
logger.warning(f"未替换的占位符: {remaining_placeholders[:10]}")
@@ -275,7 +290,7 @@ class ReportGenerator:
response.raise_for_status()
avatar_data = await response.read()
# 转换为base64编码
avatar_base64 = base64.b64encode(avatar_data).decode('utf-8')
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}")
@@ -300,91 +315,98 @@ class ReportGenerator:
# 配置浏览器启动参数,解决Docker环境中的沙盒问题
launch_options = {
'headless': True,
'args': [
'--no-sandbox', # Docker环境必需 - 禁用沙盒
'--disable-setuid-sandbox', # Docker环境必需 - 禁用setuid沙盒
'--disable-dev-shm-usage', # 避免共享内存问题
'--disable-gpu', # 禁用GPU加速
'--no-first-run',
'--disable-extensions',
'--disable-default-apps',
'--disable-background-timer-throttling',
'--disable-backgrounding-occluded-windows',
'--disable-renderer-backgrounding',
'--disable-features=TranslateUI',
'--disable-ipc-flooding-protection',
'--disable-background-networking',
'--enable-features=NetworkService,NetworkServiceInProcess',
'--force-color-profile=srgb',
'--metrics-recording-only',
'--disable-breakpad',
'--disable-component-extensions-with-background-pages',
'--disable-features=Translate,BackForwardCache,AcceptCHFrame,AvoidUnnecessaryBeforeUnloadCheckSync',
'--enable-automation',
'--password-store=basic',
'--use-mock-keychain',
'--export-tagged-pdf',
'--disable-web-security',
'--disable-features=VizDisplayCompositor',
'--disable-blink-features=AutomationControlled', # 隐藏自动化特征
]
"headless": True,
"args": [
"--no-sandbox", # Docker环境必需 - 禁用沙盒
"--disable-setuid-sandbox", # Docker环境必需 - 禁用setuid沙盒
"--disable-dev-shm-usage", # 避免共享内存问题
"--disable-gpu", # 禁用GPU加速
"--no-first-run",
"--disable-extensions",
"--disable-default-apps",
"--disable-background-timer-throttling",
"--disable-backgrounding-occluded-windows",
"--disable-renderer-backgrounding",
"--disable-features=TranslateUI",
"--disable-ipc-flooding-protection",
"--disable-background-networking",
"--enable-features=NetworkService,NetworkServiceInProcess",
"--force-color-profile=srgb",
"--metrics-recording-only",
"--disable-breakpad",
"--disable-component-extensions-with-background-pages",
"--disable-features=Translate,BackForwardCache,AcceptCHFrame,AvoidUnnecessaryBeforeUnloadCheckSync",
"--enable-automation",
"--password-store=basic",
"--use-mock-keychain",
"--export-tagged-pdf",
"--disable-web-security",
"--disable-features=VizDisplayCompositor",
"--disable-blink-features=AutomationControlled", # 隐藏自动化特征
],
}
# 检测系统 Chrome/Chromium 路径
chrome_paths = []
if sys.platform.startswith('win'):
if sys.platform.startswith("win"):
# Windows 系统 Chrome 安装路径
username = os.environ.get('USERNAME', '')
username = os.environ.get("USERNAME", "")
chrome_paths = [
r"C:\Program Files\Google\Chrome\Application\chrome.exe",
r"C:\Program Files (x86)\Google\Chrome\Application\chrome.exe",
rf"C:\Users\{username}\AppData\Local\Google\Chrome\Application\chrome.exe",
r"C:\Program Files\Chromium\Application\chrome.exe",
]
elif sys.platform.startswith('linux'):
elif sys.platform.startswith("linux"):
# Linux 系统 Chrome/Chromium 路径
chrome_paths = [
'/usr/bin/google-chrome',
'/usr/bin/google-chrome-stable',
'/usr/bin/chromium',
'/usr/bin/chromium-browser',
'/snap/bin/chromium',
'/usr/bin/chromium-freeworld',
"/usr/bin/google-chrome",
"/usr/bin/google-chrome-stable",
"/usr/bin/chromium",
"/usr/bin/chromium-browser",
"/snap/bin/chromium",
"/usr/bin/chromium-freeworld",
]
elif sys.platform.startswith('darwin'):
elif sys.platform.startswith("darwin"):
# macOS 系统 Chrome 路径
chrome_paths = [
'/Applications/Google Chrome.app/Contents/MacOS/Google Chrome',
'/Applications/Chromium.app/Contents/MacOS/Chromium',
"/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
"/Applications/Chromium.app/Contents/MacOS/Chromium",
]
# 查找可用的浏览器
found_browser = False
for chrome_path in chrome_paths:
if Path(chrome_path).exists():
launch_options['executablePath'] = chrome_path
launch_options["executablePath"] = chrome_path
logger.info(f"使用系统浏览器: {chrome_path}")
found_browser = True
break
if not found_browser:
logger.info("未找到系统浏览器,将使用 pyppeteer 默认下载的 Chromium")
# 先尝试确保 Chromium 已下载
try:
from pyppeteer import connection, browser, launcher
from pyppeteer import browser, launcher
launcher_instance = launcher.Launcher(
headless=True,
args=['--no-sandbox', '--disable-setuid-sandbox', '--disable-dev-shm-usage']
args=[
"--no-sandbox",
"--disable-setuid-sandbox",
"--disable-dev-shm-usage",
],
)
await launcher_instance._get_chromium_revision()
await launcher_instance._download_chromium()
chromium_path = pyppeteer.executablePath()
launch_options['executablePath'] = chromium_path
launch_options["executablePath"] = chromium_path
logger.info(f"使用 pyppeteer 下载的 Chromium: {chromium_path}")
except Exception as pre_download_err:
logger.warning(f"预下载 Chromium 失败,继续尝试直接启动: {pre_download_err}")
logger.warning(
f"预下载 Chromium 失败,继续尝试直接启动: {pre_download_err}"
)
# 尝试启动浏览器
try:
@@ -398,49 +420,53 @@ class ReportGenerator:
try:
# 创建新页面,设置更合理的超时时间
page = await browser.newPage()
# 设置页面视口,减少内存占用
await page.setViewport({
'width': 1024,
'height': 768,
'deviceScaleFactor': 1,
'isMobile': False,
'hasTouch': False,
'isLandscape': False
})
await page.setViewport(
{
"width": 1024,
"height": 768,
"deviceScaleFactor": 1,
"isMobile": False,
"hasTouch": False,
"isLandscape": False,
}
)
# 设置页面内容,使用更安全的加载方式
logger.info("开始设置页面内容...")
await page.setContent(html_content, {'waitUntil': 'domcontentloaded', 'timeout': 30000})
await page.setContent(
html_content, {"waitUntil": "domcontentloaded", "timeout": 30000}
)
# 等待页面基本加载完成,但不要太长时间
try:
await page.waitForSelector('body', {'timeout': 5000})
await page.waitForSelector("body", {"timeout": 5000})
logger.info("页面基本加载完成")
except Exception:
logger.warning("等待页面加载超时,继续执行")
# 减少等待时间,避免内存累积
await asyncio.sleep(1)
# 导出 PDF,使用更保守的设置
logger.info("开始生成PDF...")
pdf_options = {
'path': output_path,
'format': 'A4',
'printBackground': True,
'margin': {
'top': '10mm',
'right': '10mm',
'bottom': '10mm',
'left': '10mm'
"path": output_path,
"format": "A4",
"printBackground": True,
"margin": {
"top": "10mm",
"right": "10mm",
"bottom": "10mm",
"left": "10mm",
},
'scale': 0.8,
'displayHeaderFooter': False,
'preferCSSPageSize': True,
'timeout': 60000 # 增加PDF生成超时时间到60秒
"scale": 0.8,
"displayHeaderFooter": False,
"preferCSSPageSize": True,
"timeout": 60000, # 增加PDF生成超时时间到60秒
}
await page.pdf(pdf_options)
logger.info(f"PDF 生成成功: {output_path}")
return True
@@ -448,7 +474,7 @@ class ReportGenerator:
except Exception as e:
logger.error(f"PDF生成过程中出错: {e}")
return False
finally:
# 确保浏览器被正确关闭
if browser:
@@ -461,10 +487,10 @@ class ReportGenerator:
await page.close()
except:
pass
# 等待一小段时间让资源释放
await asyncio.sleep(0.5)
# 关闭浏览器
await browser.close()
logger.info("浏览器已关闭")
@@ -480,14 +506,18 @@ class ReportGenerator:
error_msg = str(e)
if "Chromium downloadable not found" in error_msg:
logger.error("Chromium 下载失败,建议安装系统 Chrome/Chromium")
logger.info("💡 Linux 系统建议: sudo apt-get install chromium-browser 或 sudo yum install chromium")
logger.info(
"💡 Linux 系统建议: sudo apt-get install chromium-browser 或 sudo yum install chromium"
)
elif "No usable sandbox" in error_msg:
logger.error("沙盒权限问题,已尝试禁用沙盒")
elif "Connection refused" in error_msg or "connect" in error_msg.lower():
logger.error("浏览器连接失败,请检查系统资源或尝试重启")
elif "executablePath" in error_msg and "not found" in error_msg:
logger.error("未找到系统浏览器,请安装 Chrome 或 Chromium")
logger.info("💡 安装建议: sudo apt-get install chromium-browser (Ubuntu/Debian) 或 sudo yum install chromium (CentOS/RHEL)")
logger.info(
"💡 安装建议: sudo apt-get install chromium-browser (Ubuntu/Debian) 或 sudo yum install chromium (CentOS/RHEL)"
)
elif "Browser closed unexpectedly" in error_msg:
logger.error("浏览器意外关闭,可能是由于内存不足或系统资源限制")
logger.info("💡 建议: 检查系统内存,或重启 AstrBot 后重试")
@@ -497,5 +527,7 @@ class ReportGenerator:
logger.info(" 3. 考虑使用其他 PDF 生成方案")
else:
logger.error(f"HTML 转 PDF 失败: {e}")
logger.info("💡 可以尝试使用 /安装PDF 命令重新安装依赖,或检查系统日志获取更多信息")
return False
logger.info(
"💡 可以尝试使用 /安装PDF 命令重新安装依赖,或检查系统日志获取更多信息"
)
return False
+2 -2
View File
@@ -6,7 +6,7 @@ HTML模板模块
class HTMLTemplates:
"""HTML模板管理类"""
@staticmethod
def get_image_template() -> str:
"""获取图片报告的HTML模板(使用{{ }}占位符)"""
@@ -594,7 +594,7 @@ class HTMLTemplates:
</div>
</body>
</html>"""
@staticmethod
def get_pdf_template() -> str:
"""获取PDF报告的HTML模板(使用{}占位符)"""
+1 -3
View File
@@ -5,6 +5,4 @@
from .auto_scheduler import AutoScheduler
__all__ = [
'AutoScheduler'
]
__all__ = ["AutoScheduler"]
+139 -74
View File
@@ -5,14 +5,21 @@
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, bot_manager, html_render_func=None):
def __init__(
self,
config_manager,
message_handler,
analyzer,
report_generator,
bot_manager,
html_render_func=None,
):
self.config_manager = config_manager
self.message_handler = message_handler
self.analyzer = analyzer
@@ -33,16 +40,20 @@ class AutoScheduler:
def _get_platform_id(self):
"""获取平台ID"""
try:
if hasattr(self.bot_manager, '_context') and self.bot_manager._context:
if hasattr(self.bot_manager, "_context") and self.bot_manager._context:
context = self.bot_manager._context
if hasattr(context, 'platform_manager') and hasattr(context.platform_manager, 'platform_insts'):
if hasattr(context, "platform_manager") and hasattr(
context.platform_manager, "platform_insts"
):
platforms = context.platform_manager.platform_insts
for platform in platforms:
if hasattr(platform, 'metadata') and hasattr(platform.metadata, 'id'):
if hasattr(platform, "metadata") and hasattr(
platform.metadata, "id"
):
platform_id = platform.metadata.id
return platform_id
return "aiocqhttp" # 默认值
except Exception as e:
except Exception:
return "aiocqhttp" # 默认值
async def start_scheduler(self):
@@ -54,7 +65,9 @@ class AutoScheduler:
# 延迟启动,给系统时间初始化
await asyncio.sleep(10)
logger.info(f"启动定时任务调度器,自动分析时间: {self.config_manager.get_auto_analysis_time()}")
logger.info(
f"启动定时任务调度器,自动分析时间: {self.config_manager.get_auto_analysis_time()}"
)
self.scheduler_task = asyncio.create_task(self._scheduler_loop())
@@ -75,9 +88,9 @@ class AutoScheduler:
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
)
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:
@@ -85,7 +98,9 @@ class AutoScheduler:
# 计算等待时间
wait_seconds = (target_time - now).total_seconds()
logger.info(f"定时分析将在 {target_time.strftime('%Y-%m-%d %H:%M:%S')} 执行,等待 {wait_seconds:.0f}")
logger.info(
f"定时分析将在 {target_time.strftime('%Y-%m-%d %H:%M:%S')} 执行,等待 {wait_seconds:.0f}"
)
# 等待到目标时间
await asyncio.sleep(wait_seconds)
@@ -94,7 +109,9 @@ class AutoScheduler:
if self.config_manager.get_enable_auto_analysis():
# 检查今天是否已经执行过,防止重复执行
if self.last_execution_date == target_time.date():
logger.info(f"今天 {target_time.date()} 已经执行过自动分析,跳过执行")
logger.info(
f"今天 {target_time.date()} 已经执行过自动分析,跳过执行"
)
# 等待到明天再检查
await asyncio.sleep(3600) # 等待1小时后再检查
continue
@@ -102,7 +119,9 @@ class AutoScheduler:
logger.info("开始执行定时分析")
await self._run_auto_analysis()
self.last_execution_date = target_time.date() # 记录执行日期
logger.info(f"定时分析执行完成,记录执行日期: {self.last_execution_date}")
logger.info(
f"定时分析执行完成,记录执行日期: {self.last_execution_date}"
)
else:
logger.info("自动分析已禁用,跳过执行")
break
@@ -122,24 +141,26 @@ class AutoScheduler:
logger.info("没有启用的群聊需要分析")
return
logger.info(f"将为 {len(enabled_groups)} 个群聊并发执行分析: {enabled_groups}")
logger.info(
f"将为 {len(enabled_groups)} 个群聊并发执行分析: {enabled_groups}"
)
# 创建并发任务 - 为每个群聊创建独立的分析任务
analysis_tasks = []
for group_id in enabled_groups:
task = asyncio.create_task(
self._perform_auto_analysis_for_group_with_timeout(group_id),
name=f"analysis_group_{group_id}"
name=f"analysis_group_{group_id}",
)
analysis_tasks.append(task)
# 并发执行所有分析任务,使用 return_exceptions=True 确保单个任务失败不影响其他任务
results = await asyncio.gather(*analysis_tasks, return_exceptions=True)
# 统计执行结果
success_count = 0
error_count = 0
for i, result in enumerate(results):
group_id = enabled_groups[i]
if isinstance(result, Exception):
@@ -147,8 +168,10 @@ class AutoScheduler:
error_count += 1
else:
success_count += 1
logger.info(f"并发分析完成 - 成功: {success_count}, 失败: {error_count}, 总计: {len(enabled_groups)}")
logger.info(
f"并发分析完成 - 成功: {success_count}, 失败: {error_count}, 总计: {len(enabled_groups)}"
)
except Exception as e:
logger.error(f"自动分析执行失败: {e}", exc_info=True)
@@ -157,30 +180,34 @@ class AutoScheduler:
"""为指定群执行自动分析(带超时控制)"""
try:
# 为每个群聊设置独立的超时时间(20分钟)- 使用 asyncio.wait_for 兼容所有 Python 版本
await asyncio.wait_for(self._perform_auto_analysis_for_group(group_id), timeout=1200)
await asyncio.wait_for(
self._perform_auto_analysis_for_group(group_id), timeout=1200
)
except asyncio.TimeoutError:
logger.error(f"{group_id} 分析超时(20分钟),跳过该群分析")
except Exception as e:
logger.error(f"{group_id} 分析任务执行失败: {e}")
async def _perform_auto_analysis_for_group(self, group_id: str):
"""为指定群执行自动分析(核心逻辑)"""
# 为每个群聊使用独立的锁,避免全局锁导致串行化
group_lock_key = f"analysis_{group_id}"
if not hasattr(self, '_group_locks'):
if not hasattr(self, "_group_locks"):
self._group_locks = {}
if group_lock_key not in self._group_locks:
self._group_locks[group_lock_key] = asyncio.Lock()
async with self._group_locks[group_lock_key]:
try:
start_time = asyncio.get_event_loop().time()
# 检查bot管理器状态
if not self.bot_manager.is_ready_for_auto_analysis():
status = self.bot_manager.get_status_info()
logger.warning(f"{group_id} 自动分析跳过:bot管理器未就绪 - {status}")
logger.warning(
f"{group_id} 自动分析跳过:bot管理器未就绪 - {status}"
)
return
logger.info(f"开始为群 {group_id} 执行自动分析(并发任务)")
@@ -189,8 +216,10 @@ class AutoScheduler:
analysis_days = self.config_manager.get_analysis_days()
bot_instance = self.bot_manager.get_bot_instance()
messages = await self.message_handler.fetch_group_messages(bot_instance, group_id, analysis_days)
messages = await self.message_handler.fetch_group_messages(
bot_instance, group_id, analysis_days
)
if not messages:
logger.warning(f"{group_id} 未获取到足够的消息记录")
return
@@ -198,7 +227,9 @@ class AutoScheduler:
# 检查消息数量
min_threshold = self.config_manager.get_min_messages_threshold()
if len(messages) < min_threshold:
logger.warning(f"{group_id} 消息数量不足({len(messages)}条),跳过分析")
logger.warning(
f"{group_id} 消息数量不足({len(messages)}条),跳过分析"
)
return
logger.info(f"{group_id} 获取到 {len(messages)} 条消息,开始分析")
@@ -206,14 +237,16 @@ class AutoScheduler:
# 进行分析 - 构造正确的 unified_msg_origin
platform_id = self._get_platform_id()
umo = f"{platform_id}:GroupMessage:{group_id}" if platform_id else None
analysis_result = await self.analyzer.analyze_messages(messages, group_id, umo)
analysis_result = await self.analyzer.analyze_messages(
messages, group_id, umo
)
if not analysis_result:
logger.error(f"{group_id} 分析失败")
return
# 生成并发送报告
await self._send_analysis_report(group_id, analysis_result)
# 记录执行时间
end_time = asyncio.get_event_loop().time()
execution_time = end_time - start_time
@@ -221,10 +254,10 @@ class AutoScheduler:
except Exception as e:
logger.error(f"{group_id} 自动分析执行失败: {e}", exc_info=True)
finally:
# 清理群聊锁资源(可选,防止内存泄漏)
if hasattr(self, '_group_locks') and len(self._group_locks) > 50:
if hasattr(self, "_group_locks") and len(self._group_locks) > 50:
old_locks = list(self._group_locks.keys())[:10]
for lock_key in old_locks:
if not self._group_locks[lock_key].locked():
@@ -240,47 +273,87 @@ class AutoScheduler:
# 使用图片格式
logger.info(f"{group_id} 自动分析使用图片报告格式")
try:
image_url = await self.report_generator.generate_image_report(analysis_result, group_id, self.html_render_func)
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} 图片报告生成失败(返回None),回退到文本报告")
text_report = self.report_generator.generate_text_report(analysis_result)
await self._send_text_message(group_id, f"📊 每日群聊分析报告:\n\n{text_report}")
logger.warning(
f"{group_id} 图片报告生成失败(返回None),回退到文本报告"
)
text_report = self.report_generator.generate_text_report(
analysis_result
)
await self._send_text_message(
group_id, f"📊 每日群聊分析报告:\n\n{text_report}"
)
except Exception as img_e:
logger.error(f"{group_id} 图片报告生成异常: {img_e},回退到文本报告")
text_report = self.report_generator.generate_text_report(analysis_result)
await self._send_text_message(group_id, f"📊 每日群聊分析报告:\n\n{text_report}")
logger.error(
f"{group_id} 图片报告生成异常: {img_e},回退到文本报告"
)
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}")
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}")
text_report = self.report_generator.generate_text_report(
analysis_result
)
await self._send_text_message(
group_id, f"📊 每日群聊分析报告:\n\n{text_report}"
)
else:
try:
pdf_path = await self.report_generator.generate_pdf_report(analysis_result, group_id)
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报告生成失败(返回None),回退到文本报告")
text_report = self.report_generator.generate_text_report(analysis_result)
await self._send_text_message(group_id, f"📊 每日群聊分析报告:\n\n{text_report}")
logger.error(
f"{group_id} PDF报告生成失败(返回None),回退到文本报告"
)
text_report = self.report_generator.generate_text_report(
analysis_result
)
await self._send_text_message(
group_id, f"📊 每日群聊分析报告:\n\n{text_report}"
)
except Exception as pdf_e:
logger.error(f"{group_id} PDF报告生成异常: {pdf_e},回退到文本报告")
text_report = self.report_generator.generate_text_report(analysis_result)
await self._send_text_message(group_id, f"📊 每日群聊分析报告:\n\n{text_report}")
logger.error(
f"{group_id} PDF报告生成异常: {pdf_e},回退到文本报告"
)
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}")
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} 自动分析完成,已发送报告")
@@ -299,13 +372,10 @@ class AutoScheduler:
await bot_instance.api.call_action(
"send_group_msg",
group_id=group_id,
message=[{
"type": "text",
"data": {"text": "📊 每日群聊分析报告已生成:"}
}, {
"type": "image",
"data": {"url": image_url}
}]
message=[
{"type": "text", "data": {"text": "📊 每日群聊分析报告已生成:"}},
{"type": "image", "data": {"url": image_url}},
],
)
logger.info(f"{group_id} 图片消息发送成功")
@@ -322,9 +392,7 @@ class AutoScheduler:
# 发送文本消息到群
await bot_instance.api.call_action(
"send_group_msg",
group_id=group_id,
message=text_content
"send_group_msg", group_id=group_id, message=text_content
)
logger.info(f"{group_id} 文本消息发送成功")
@@ -343,13 +411,10 @@ class AutoScheduler:
await bot_instance.api.call_action(
"send_group_msg",
group_id=group_id,
message=[{
"type": "text",
"data": {"text": "📊 每日群聊分析报告已生成:"}
}, {
"type": "file",
"data": {"file": pdf_path}
}]
message=[
{"type": "text", "data": {"text": "📊 每日群聊分析报告已生成:"}},
{"type": "file", "data": {"file": pdf_path}},
],
)
logger.info(f"{group_id} PDF文件发送成功")
@@ -360,7 +425,7 @@ class AutoScheduler:
await bot_instance.api.call_action(
"send_group_msg",
group_id=group_id,
message=f"📊 每日群聊分析报告已生成,但发送PDF文件失败。PDF文件路径:{pdf_path}"
message=f"📊 每日群聊分析报告已生成,但发送PDF文件失败。PDF文件路径:{pdf_path}",
)
except Exception as e2:
logger.error(f"发送PDF失败提示到群 {group_id} 也失败: {e2}")
logger.error(f"发送PDF失败提示到群 {group_id} 也失败: {e2}")
+1 -4
View File
@@ -6,7 +6,4 @@
from .pdf_utils import PDFInstaller
from .helpers import MessageAnalyzer
__all__ = [
'PDFInstaller',
'MessageAnalyzer'
]
__all__ = ["PDFInstaller", "MessageAnalyzer"]
+49 -16
View File
@@ -4,12 +4,13 @@
"""
from typing import List, Dict
from ...src.models.data_models import GroupStatistics, SummaryTopic, UserTitle, GoldenQuote, TokenUsage
from ...src.models.data_models import TokenUsage
from ...src.core.message_handler import MessageHandler
from ...src.analysis.llm_analyzer import LLMAnalyzer
from ...src.analysis.statistics import UserAnalyzer
from astrbot.api import logger
class MessageAnalyzer:
"""消息分析器 - 整合所有分析功能"""
@@ -28,7 +29,9 @@ class MessageAnalyzer:
else:
await self.message_handler.set_bot_qq_id(bot_instance)
async def analyze_messages(self, messages: List[Dict], group_id: str, unified_msg_origin: str = None) -> Dict:
async def analyze_messages(
self, messages: List[Dict], group_id: str, unified_msg_origin: str = None
) -> Dict:
"""完整的消息分析流程"""
try:
# 基础统计
@@ -36,11 +39,15 @@ class MessageAnalyzer:
# 用户分析
user_analysis = self.user_analyzer.analyze_users(messages)
# 获取活跃用户列表 - 使用get_top_users方法,limit从配置中读取
max_user_titles = self.config_manager.get_max_user_titles()
top_users = self.user_analyzer.get_top_users(user_analysis, limit=max_user_titles)
logger.info(f"获取到 {len(top_users)} 个活跃用户用于称号分析(配置上限: {max_user_titles})")
top_users = self.user_analyzer.get_top_users(
user_analysis, limit=max_user_titles
)
logger.info(
f"获取到 {len(top_users)} 个活跃用户用于称号分析(配置上限: {max_user_titles})"
)
# LLM分析 - 使用并发方式
topics = []
@@ -51,35 +58,61 @@ class MessageAnalyzer:
# 检查各个分析功能是否启用
topic_enabled = self.config_manager.get_topic_analysis_enabled()
user_title_enabled = self.config_manager.get_user_title_analysis_enabled()
golden_quote_enabled = self.config_manager.get_golden_quote_analysis_enabled()
golden_quote_enabled = (
self.config_manager.get_golden_quote_analysis_enabled()
)
# 如果三个分析都启用,使用并发执行
if topic_enabled and user_title_enabled and golden_quote_enabled:
# 并发执行所有三个分析任务,传入活跃用户列表
topics, user_titles, golden_quotes, total_token_usage = await self.llm_analyzer.analyze_all_concurrent(
(
topics,
user_titles,
golden_quotes,
total_token_usage,
) = await self.llm_analyzer.analyze_all_concurrent(
messages, user_analysis, umo=unified_msg_origin, top_users=top_users
)
else:
# 如果只启用部分分析,则按需执行
if topic_enabled:
topics, topic_tokens = await self.llm_analyzer.analyze_topics(messages, umo=unified_msg_origin)
topics, topic_tokens = await self.llm_analyzer.analyze_topics(
messages, umo=unified_msg_origin
)
total_token_usage.prompt_tokens += topic_tokens.prompt_tokens
total_token_usage.completion_tokens += topic_tokens.completion_tokens
total_token_usage.completion_tokens += (
topic_tokens.completion_tokens
)
total_token_usage.total_tokens += topic_tokens.total_tokens
if user_title_enabled:
# 传入活跃用户列表
user_titles, title_tokens = await self.llm_analyzer.analyze_user_titles(
messages, user_analysis, umo=unified_msg_origin, top_users=top_users
(
user_titles,
title_tokens,
) = await self.llm_analyzer.analyze_user_titles(
messages,
user_analysis,
umo=unified_msg_origin,
top_users=top_users,
)
total_token_usage.prompt_tokens += title_tokens.prompt_tokens
total_token_usage.completion_tokens += title_tokens.completion_tokens
total_token_usage.completion_tokens += (
title_tokens.completion_tokens
)
total_token_usage.total_tokens += title_tokens.total_tokens
if golden_quote_enabled:
golden_quotes, quote_tokens = await self.llm_analyzer.analyze_golden_quotes(messages, umo=unified_msg_origin)
(
golden_quotes,
quote_tokens,
) = await self.llm_analyzer.analyze_golden_quotes(
messages, umo=unified_msg_origin
)
total_token_usage.prompt_tokens += quote_tokens.prompt_tokens
total_token_usage.completion_tokens += quote_tokens.completion_tokens
total_token_usage.completion_tokens += (
quote_tokens.completion_tokens
)
total_token_usage.total_tokens += quote_tokens.total_tokens
# 更新统计数据
@@ -90,7 +123,7 @@ class MessageAnalyzer:
"statistics": statistics,
"topics": topics,
"user_titles": user_titles,
"user_analysis": user_analysis
"user_analysis": user_analysis,
}
except Exception as e:
+84 -66
View File
@@ -5,21 +5,22 @@ PDF工具模块
import sys
import asyncio
import threading
from concurrent.futures import ThreadPoolExecutor
from astrbot.api import logger
class PDFInstaller:
"""PDF功能安装器"""
# 类级别的线程池,用于异步下载任务
_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="chromium_download")
_executor = ThreadPoolExecutor(
max_workers=1, thread_name_prefix="chromium_download"
)
_download_status = {
"in_progress": False,
"completed": False,
"failed": False,
"error_message": None
"error_message": None,
}
@staticmethod
@@ -31,10 +32,14 @@ class PDFInstaller:
# 使用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",
sys.executable,
"-m",
"pip",
"install",
"pyppeteer==1.0.2",
"websockets==10.4",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await process.communicate()
@@ -63,20 +68,20 @@ class PDFInstaller:
"""通过 pyppeteer 自动安装 Chromium(异步非阻塞方式)"""
try:
logger.info("正在通过 pyppeteer 自动安装 Chromium...")
# 检查是否已经在下载中
if PDFInstaller._download_status["in_progress"]:
return "⏳ Chromium 正在后台下载中,请稍候..."
# 启动异步下载任务
PDFInstaller._download_status["in_progress"] = True
PDFInstaller._download_status["completed"] = False
PDFInstaller._download_status["failed"] = False
PDFInstaller._download_status["error_message"] = None
# 在后台线程中启动下载
asyncio.create_task(PDFInstaller._background_chromium_download())
return """⏳ Chromium 下载已在后台启动
这可能需要几分钟时间,请稍候...
@@ -96,31 +101,35 @@ class PDFInstaller:
"""后台下载 Chromium,带超时控制"""
try:
logger.info("后台 Chromium 下载任务开始")
# 设置10分钟超时
timeout_seconds = 600
try:
# 使用 asyncio.wait_for 实现超时控制
success = await asyncio.wait_for(
PDFInstaller._download_chromium_via_pyppeteer(),
timeout=timeout_seconds
timeout=timeout_seconds,
)
if success:
PDFInstaller._download_status["completed"] = True
PDFInstaller._download_status["failed"] = False
logger.info("✅ Chromium 后台下载完成!")
else:
PDFInstaller._download_status["failed"] = True
PDFInstaller._download_status["error_message"] = "下载失败,请检查网络连接"
PDFInstaller._download_status["error_message"] = (
"下载失败,请检查网络连接"
)
logger.error("❌ Chromium 下载失败")
except asyncio.TimeoutError:
PDFInstaller._download_status["failed"] = True
PDFInstaller._download_status["error_message"] = f"下载超时({timeout_seconds}秒)"
PDFInstaller._download_status["error_message"] = (
f"下载超时({timeout_seconds}秒)"
)
logger.error(f"❌ Chromium 下载超时({timeout_seconds}秒)")
except Exception as e:
PDFInstaller._download_status["failed"] = True
PDFInstaller._download_status["error_message"] = str(e)
@@ -133,96 +142,101 @@ class PDFInstaller:
"""通过 pyppeteer 自动下载 Chromium(带重试机制)"""
max_retries = 2
retry_count = 0
while retry_count <= max_retries:
try:
if retry_count > 0:
logger.info(f"正在重试下载 Chromium(第 {retry_count}/{max_retries} 次)...")
logger.info(
f"正在重试下载 Chromium(第 {retry_count}/{max_retries} 次)..."
)
else:
logger.info("通过 pyppeteer 自动下载 Chromium...")
# 导入 pyppeteer 并尝试下载
try:
import pyppeteer
from pyppeteer import launch
from pyppeteer.errors import BrowserError
# 方法1: 尝试直接下载 Chromium 而不启动浏览器
logger.info("尝试直接下载 Chromium...")
try:
from pyppeteer.launcher import Launcher
# 创建 Launcher 实例但不启动浏览器
launcher = Launcher(
headless=True,
args=['--no-sandbox', '--disable-setuid-sandbox']
args=["--no-sandbox", "--disable-setuid-sandbox"],
)
# 只下载 Chromium
chromium_revision = launcher._get_chromium_revision()
await launcher._download_chromium()
logger.info("✅ Chromium 下载完成")
return True
except Exception as download_error:
logger.warning(f"直接下载 Chromium 失败: {download_error}")
logger.info("尝试通过启动浏览器来下载...")
# 方法2: 通过启动浏览器触发自动下载
import platform
system = platform.system().lower()
if system == "linux":
browser_args = [
'--no-sandbox',
'--disable-setuid-sandbox',
'--disable-dev-shm-usage',
'--disable-accelerated-2d-canvas',
'--no-first-run',
'--no-zygote',
'--disable-gpu',
'--disable-background-timer-throttling',
'--disable-backgrounding-occluded-windows',
'--disable-renderer-backgrounding',
'--disable-features=TranslateUI',
'--disable-ipc-flooding-protection'
"--no-sandbox",
"--disable-setuid-sandbox",
"--disable-dev-shm-usage",
"--disable-accelerated-2d-canvas",
"--no-first-run",
"--no-zygote",
"--disable-gpu",
"--disable-background-timer-throttling",
"--disable-backgrounding-occluded-windows",
"--disable-renderer-backgrounding",
"--disable-features=TranslateUI",
"--disable-ipc-flooding-protection",
]
else:
browser_args = [
'--no-sandbox',
'--disable-setuid-sandbox',
'--disable-gpu'
"--no-sandbox",
"--disable-setuid-sandbox",
"--disable-gpu",
]
logger.info("启动 pyppeteer 浏览器以触发 Chromium 自动下载...")
browser = await launch(
headless=True,
args=browser_args,
ignoreHTTPSErrors=True,
dumpio=False # 关闭浏览器日志输出以减少干扰
dumpio=False, # 关闭浏览器日志输出以减少干扰
)
# 获取 Chromium 路径
chromium_path = pyppeteer.executablePath()
logger.info(f"✅ Chromium 自动下载完成,路径: {chromium_path}")
await browser.close()
return True
except BrowserError as e:
logger.error(f"浏览器错误: {e}")
# 方法3: 使用子进程命令行触发下载
try:
logger.info("尝试使用命令行触发 Chromium 自动下载...")
import platform
system = platform.system().lower()
if system == "linux":
cmd = [
sys.executable, "-c",
sys.executable,
"-c",
"""
import pyppeteer
import asyncio
@@ -246,33 +260,34 @@ async def download_chrome():
raise
asyncio.run(download_chrome())
"""
""",
]
else:
cmd = [
sys.executable, "-c",
"import pyppeteer; import asyncio; asyncio.run(pyppeteer.launch())"
sys.executable,
"-c",
"import pyppeteer; import asyncio; asyncio.run(pyppeteer.launch())",
]
process = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=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()}")
raise Exception(f"命令行下载失败: {stderr.decode()}")
except Exception as e2:
logger.error(f"命令行触发自动下载失败: {e2}")
raise
except Exception as e:
retry_count += 1
if retry_count <= max_retries:
@@ -280,9 +295,12 @@ asyncio.run(download_chrome())
logger.warning(f"下载失败,{wait_time}秒后重试... 错误: {e}")
await asyncio.sleep(wait_time)
else:
logger.error(f"通过 pyppeteer 自动下载 Chromium 失败(已重试{max_retries}次): {e}", exc_info=True)
logger.error(
f"通过 pyppeteer 自动下载 Chromium 失败(已重试{max_retries}次): {e}",
exc_info=True,
)
return False
return False
@staticmethod
@@ -292,4 +310,4 @@ asyncio.run(download_chrome())
version = config_manager.pyppeteer_version or "未知版本"
return f"✅ PDF 功能可用 (pyppeteer {version})"
else:
return "❌ PDF 功能不可用 - 需要安装 pyppeteer"
return "❌ PDF 功能不可用 - 需要安装 pyppeteer"
+1 -1
View File
@@ -4,4 +4,4 @@
from .activity_charts import ActivityVisualizer
__all__ = ['ActivityVisualizer']
__all__ = ["ActivityVisualizer"]
+29 -21
View File
@@ -3,19 +3,21 @@
参考 astrbot_plugin_github_analyzer 的实现方式
"""
from datetime import datetime, timedelta
from datetime import datetime
from collections import defaultdict
from typing import Dict, List, Any
from typing import Dict, List
from ..models.data_models import ActivityVisualization
class ActivityVisualizer:
"""活跃度可视化器"""
def __init__(self):
pass
def generate_activity_visualization(self, messages: List[Dict]) -> ActivityVisualization:
def generate_activity_visualization(
self, messages: List[Dict]
) -> ActivityVisualization:
"""生成活跃度可视化数据 - 专注于小时级别分析"""
hourly_activity = defaultdict(int)
user_activity = defaultdict(int)
@@ -54,15 +56,19 @@ class ActivityVisualizer:
# 生成用户活跃度排行
user_ranking = []
for user_id, data in user_activity.items():
user_ranking.append({
"user_id": user_id,
"nickname": data["nickname"],
"message_count": data["count"]
})
user_ranking.append(
{
"user_id": user_id,
"nickname": data["nickname"],
"message_count": data["count"],
}
)
user_ranking.sort(key=lambda x: x["message_count"], reverse=True)
# 找出高峰时段(活跃度最高的3个小时)
peak_hours = sorted(hourly_activity.items(), key=lambda x: x[1], reverse=True)[:3]
peak_hours = sorted(hourly_activity.items(), key=lambda x: x[1], reverse=True)[
:3
]
peak_hours = [{"hour": hour, "count": count} for hour, count in peak_hours]
return ActivityVisualization(
@@ -70,10 +76,14 @@ class ActivityVisualizer:
daily_activity={}, # 不使用日期分析
user_activity_ranking=user_ranking[:10], # 前10名
peak_hours=peak_hours,
activity_heatmap_data=self._generate_hourly_heatmap_data(hourly_activity, emoji_activity)
activity_heatmap_data=self._generate_hourly_heatmap_data(
hourly_activity, emoji_activity
),
)
def _generate_hourly_heatmap_data(self, hourly_activity: dict, emoji_activity: dict) -> dict:
def _generate_hourly_heatmap_data(
self, hourly_activity: dict, emoji_activity: dict
) -> dict:
"""生成小时级热力图数据"""
# 计算活跃度等级
max_hourly = max(hourly_activity.values()) if hourly_activity else 1
@@ -90,7 +100,7 @@ class ActivityVisualizer:
hour: (emoji_activity.get(hour, 0) / max_emoji) * 100
for hour in range(24)
},
"activity_levels": self._calculate_activity_levels(hourly_activity)
"activity_levels": self._calculate_activity_levels(hourly_activity),
}
def _calculate_activity_levels(self, hourly_activity: dict) -> dict:
@@ -114,17 +124,17 @@ class ActivityVisualizer:
levels[hour] = level
return levels
def generate_hourly_chart_html(self, hourly_activity: dict) -> str:
"""生成每小时活动分布的HTML图表"""
html_parts = []
max_activity = max(hourly_activity.values()) if hourly_activity else 1
threshold_percentage = 20 # 数值显示阈值
for hour in range(24):
count = hourly_activity.get(hour, 0)
percentage = (count / max_activity) * 100 if max_activity > 0 else 0
if count > 0 and percentage >= threshold_percentage:
# 活动数较多,数值显示在条形图内部
html_segment = f"""
@@ -159,7 +169,5 @@ class ActivityVisualizer:
</div>
"""
html_parts.append(html_segment)
return "".join(html_parts)