diff --git a/main.py b/main.py
index 9099e34..6ee69fa 100644
--- a/main.py
+++ b/main.py
@@ -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""")
-
-
diff --git a/src/__init__.py b/src/__init__.py
index 33c66e7..5cf2d5f 100644
--- a/src/__init__.py
+++ b/src/__init__.py
@@ -2,4 +2,4 @@
QQ群日常分析插件 - 源代码包
"""
-__author__ = "SXP-Simon"
\ No newline at end of file
+__author__ = "SXP-Simon"
diff --git a/src/analysis/__init__.py b/src/analysis/__init__.py
index d1d48ee..8a9ab59 100644
--- a/src/analysis/__init__.py
+++ b/src/analysis/__init__.py
@@ -6,7 +6,4 @@
from .llm_analyzer import LLMAnalyzer
from .statistics import UserAnalyzer
-__all__ = [
- 'LLMAnalyzer',
- 'UserAnalyzer'
-]
\ No newline at end of file
+__all__ = ["LLMAnalyzer", "UserAnalyzer"]
diff --git a/src/analysis/analyzers/__init__.py b/src/analysis/analyzers/__init__.py
index e4d49df..fe370dd 100644
--- a/src/analysis/analyzers/__init__.py
+++ b/src/analysis/analyzers/__init__.py
@@ -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'
-]
\ No newline at end of file
+__all__ = ["BaseAnalyzer", "TopicAnalyzer", "UserTitleAnalyzer", "GoldenQuoteAnalyzer"]
diff --git a/src/analysis/analyzers/base_analyzer.py b/src/analysis/analyzers/base_analyzer.py
index cf11bcd..261cb09 100644
--- a/src/analysis/analyzers/base_analyzer.py
+++ b/src/analysis/analyzers/base_analyzer.py
@@ -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
-
-
\ No newline at end of file
diff --git a/src/analysis/analyzers/golden_quote_analyzer.py b/src/analysis/analyzers/golden_quote_analyzer.py
index 5417860..5921547 100644
--- a/src/analysis/analyzers/golden_quote_analyzer.py
+++ b/src/analysis/analyzers/golden_quote_analyzer.py
@@ -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()
\ No newline at end of file
+ return [], TokenUsage()
diff --git a/src/analysis/analyzers/topic_analyzer.py b/src/analysis/analyzers/topic_analyzer.py
index 8399224..27ac694 100644
--- a/src/analysis/analyzers/topic_analyzer.py
+++ b/src/analysis/analyzers/topic_analyzer.py
@@ -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()
\ No newline at end of file
+ return [], TokenUsage()
diff --git a/src/analysis/analyzers/user_title_analyzer.py b/src/analysis/analyzers/user_title_analyzer.py
index 127c24d..ed8e11c 100644
--- a/src/analysis/analyzers/user_title_analyzer.py
+++ b/src/analysis/analyzers/user_title_analyzer.py
@@ -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()
\ No newline at end of file
+ return [], TokenUsage()
diff --git a/src/analysis/llm_analyzer.py b/src/analysis/llm_analyzer.py
index ebd11f9..c5a3583 100644
--- a/src/analysis/llm_analyzer.py
+++ b/src/analysis/llm_analyzer.py
@@ -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)
-
\ No newline at end of file
diff --git a/src/analysis/statistics.py b/src/analysis/statistics.py
index 473ac5d..937478c 100644
--- a/src/analysis/statistics.py
+++ b/src/analysis/statistics.py
@@ -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),
}
-
diff --git a/src/analysis/utils/__init__.py b/src/analysis/utils/__init__.py
index b935e13..8388263 100644
--- a/src/analysis/utils/__init__.py
+++ b/src/analysis/utils/__init__.py
@@ -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'
-]
\ No newline at end of file
+ "InfoUtils",
+]
diff --git a/src/analysis/utils/json_utils.py b/src/analysis/utils/json_utils.py
index 3af948b..d17fcd5 100644
--- a/src/analysis/utils/json_utils.py
+++ b/src/analysis/utils/json_utils.py
@@ -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 []
diff --git a/src/analysis/utils/llm_utils.py b/src/analysis/utils/llm_utils.py
index 002e9fc..aba04e2 100644
--- a/src/analysis/utils/llm_utils.py
+++ b/src/analysis/utils/llm_utils.py
@@ -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 ""
\ No newline at end of file
+ return ""
diff --git a/src/core/__init__.py b/src/core/__init__.py
index 9e7dda8..0999975 100644
--- a/src/core/__init__.py
+++ b/src/core/__init__.py
@@ -5,7 +5,4 @@
from .config import ConfigManager
from .message_handler import MessageHandler
-__all__ = [
- 'ConfigManager',
- 'MessageHandler'
-]
\ No newline at end of file
+__all__ = ["ConfigManager", "MessageHandler"]
diff --git a/src/core/bot_manager.py b/src/core/bot_manager.py
index 505fcec..5340121 100644
--- a/src/core/bot_manager.py
+++ b/src/core/bot_manager.py
@@ -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:
diff --git a/src/core/config.py b/src/core/config.py
index 82fcb9e..6f89fb5 100644
--- a/src/core/config.py
+++ b/src/core/config.py
@@ -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}")
\ No newline at end of file
+ logger.error(f"重新加载配置失败: {e}")
diff --git a/src/core/message_handler.py b/src/core/message_handler.py
index 2aa07b9..5f6e3c1 100644
--- a/src/core/message_handler.py
+++ b/src/core/message_handler.py
@@ -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(),
)
diff --git a/src/models/__init__.py b/src/models/__init__.py
index 5a42962..1630360 100644
--- a/src/models/__init__.py
+++ b/src/models/__init__.py
@@ -7,13 +7,7 @@ from .data_models import (
UserTitle,
GoldenQuote,
TokenUsage,
- GroupStatistics
+ GroupStatistics,
)
-__all__ = [
- 'SummaryTopic',
- 'UserTitle',
- 'GoldenQuote',
- 'TokenUsage',
- 'GroupStatistics'
-]
\ No newline at end of file
+__all__ = ["SummaryTopic", "UserTitle", "GoldenQuote", "TokenUsage", "GroupStatistics"]
diff --git a/src/models/data_models.py b/src/models/data_models.py
index c6210c3..688c25d 100644
--- a/src/models/data_models.py
+++ b/src/models/data_models.py
@@ -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)
\ No newline at end of file
+ activity_visualization: ActivityVisualization = field(
+ default_factory=ActivityVisualization
+ )
+ token_usage: TokenUsage = field(default_factory=TokenUsage)
diff --git a/src/reports/__init__.py b/src/reports/__init__.py
index 267df1e..6ff4113 100644
--- a/src/reports/__init__.py
+++ b/src/reports/__init__.py
@@ -6,7 +6,4 @@
from .generators import ReportGenerator
from .templates import HTMLTemplates
-__all__ = [
- 'ReportGenerator',
- 'HTMLTemplates'
-]
\ No newline at end of file
+__all__ = ["ReportGenerator", "HTMLTemplates"]
diff --git a/src/reports/generators.py b/src/reports/generators.py
index fb3c04e..8ea212d 100644
--- a/src/reports/generators.py
+++ b/src/reports/generators.py
@@ -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'' if avatar_data else '