From edc63fad0a25354cacd9117f7fd9c089f64d6b3d Mon Sep 17 00:00:00 2001 From: liuwei Date: Fri, 9 Jan 2026 14:51:27 +0800 Subject: [PATCH] =?UTF-8?q?=E8=B0=83=E6=95=B4=E6=80=BB=E7=BB=93=E4=B8=9A?= =?UTF-8?q?=E5=8A=A1=E3=80=82=E6=94=AF=E6=8C=81=E6=AF=8F=E5=A4=A9=E6=97=A9?= =?UTF-8?q?=E4=B8=8A9=E7=82=B9=E6=80=BB=E7=BB=93=E6=98=A8=E5=A4=A9?= =?UTF-8?q?=E7=9A=84=E6=B6=88=E6=81=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- db/message_storage.py | 56 ++++++ plugins/message_summary/README_SCHEDULED.md | 201 ++++++++++++++++++++ plugins/message_summary/main.py | 109 +++++++++++ utils/wechat/message_to_db.py | 49 +++++ 4 files changed, 415 insertions(+) create mode 100644 plugins/message_summary/README_SCHEDULED.md diff --git a/db/message_storage.py b/db/message_storage.py index 80a7e6f..ad58ca0 100644 --- a/db/message_storage.py +++ b/db/message_storage.py @@ -342,3 +342,59 @@ class MessageStorageDB(BaseDBOperator): # 限制最大返回数量（5000条足以覆盖1-2天的活跃群聊） return messages[:max_results] if messages else [] + + def get_messages_by_date_range(self, group_id: str, start_time: datetime, end_time: datetime) -> List[Dict]: + """获取指定时间范围内的消息 + + Args: + group_id: 群组ID + start_time: 开始时间 + end_time: 结束时间 + + Returns: + 消息列表 + """ + sql = """ + SELECT timestamp, sender, content, message_type + FROM messages + WHERE timestamp >= %s + AND timestamp <= %s + AND message_type in (1, 49) + AND group_id = %s + AND length(content) > 6 + AND CHAR_LENGTH(content) < 300 + AND content NOT LIKE '/%' + ORDER BY timestamp ASC + """ + params = (start_time.strftime('%Y-%m-%d %H:%M:%S'), + end_time.strftime('%Y-%m-%d %H:%M:%S'), + group_id) + return self.execute_query(sql, params) or [] + + def count_messages_by_date_range(self, group_id: str, start_time: datetime, end_time: datetime) -> int: + """统计指定时间范围内的消息数量 + + Args: + group_id: 群组ID + start_time: 开始时间 + end_time: 结束时间 + + Returns: + 消息数量 + """ + sql = """ + SELECT COUNT(*) as count + FROM messages + WHERE timestamp >= %s + AND timestamp <= %s + AND message_type in (1, 49) + AND group_id = %s + AND length(content) > 6 + AND CHAR_LENGTH(content) < 300 + AND content NOT LIKE '/%' + """ + params = (start_time.strftime('%Y-%m-%d %H:%M:%S'), + end_time.strftime('%Y-%m-%d %H:%M:%S'), + group_id) + result = self.execute_query(sql, params) + return result[0]['count'] if result else 0 diff --git a/plugins/message_summary/README_SCHEDULED.md b/plugins/message_summary/README_SCHEDULED.md new file mode 100644 index 0000000..d456e5e --- /dev/null +++ b/plugins/message_summary/README_SCHEDULED.md @@ -0,0 +1,201 @@ +# 消息总结定时任务使用说明 + +## 功能概述 + +已为 `message_summary` 插件添加了定时任务功能，每天早上 9:00 自动总结昨天的群聊信息并发送到群聊中。 + +## 实现原理 + +参考了 `game_task` 插件的定时任务策略，在 `message_summary` 插件初始化时注册定时任务： + +```python +async_job.at_times(["09:00"])(self.daily_summary_job) +``` + +## 核心修改 + +### 1. plugins/message_summary/main.py + +#### 添加导入 +```python +from datetime import datetime, timedelta +from utils.decorator.async_job import async_job +import asyncio +``` + +#### 在 `__init__` 中注册定时任务 +```python +def __init__(self): + super().__init__() + self.message_storage = None + self.revoke = None + self.feature = self.register_feature() + # 注册定时任务：每天早上9点总结昨天的聊天信息 + async_job.at_times(["09:00"])(self.daily_summary_job) +``` + +#### 添加定时任务方法 +```python +async def daily_summary_job(self): + """定时任务：每天早上9点总结昨天的聊天信息""" + # 计算昨天的时间范围 + yesterday = datetime.now() - timedelta(days=1) + yesterday_start = yesterday.replace(hour=0, minute=0, second=0, microsecond=0) + yesterday_end = yesterday.replace(hour=23, minute=59, second=59, microsecond=999999) + + # 获取所有启用了群机器人的群聊 + all_groups = GroupBotManager.get_group_list() + + # 筛选出开启了总结功能的群聊 + enabled_groups = [] + for group_id in all_groups: + if GroupBotManager.get_group_permission(group_id, self.feature) == PermissionStatus.ENABLED: + enabled_groups.append(group_id) + + # 为每个群生成总结 + for group_id in enabled_groups: + # 获取昨天的聊天记录 + chat_content = self.message_storage.get_messages_by_date_range( + group_id, all_contacts, yesterday_start, yesterday_end + ) + + # 生成并发送总结 + summary, image_path = await self._generate_summary(chat_content, group_name) + # 发送到群聊... +``` + +### 2. db/message_storage.py + +添加了按时间范围查询消息的方法： + +```python +def get_messages_by_date_range(self, group_id: str, start_time: datetime, end_time: datetime) -> List[Dict]: + """获取指定时间范围内的消息""" + sql = """ + SELECT timestamp, sender, content, message_type + FROM messages + WHERE timestamp >= %s AND timestamp <= %s + AND message_type in (1, 49) + AND group_id = %s + AND length(content) > 6 + AND CHAR_LENGTH(content) < 300 + AND content NOT LIKE '/%' + ORDER BY timestamp ASC + """ + params = (start_time.strftime('%Y-%m-%d %H:%M:%S'), + end_time.strftime('%Y-%m-%d %H:%M:%S'), + group_id) + return self.execute_query(sql, params) or [] +``` + +### 3. utils/wechat/message_to_db.py + +添加了对应的包装方法： + +```python +def get_messages_by_date_range(self, group_id, all_contacts: dict, start_time: datetime, end_time: datetime): + """获取指定时间范围内的消息""" + messages = self.message_db.get_messages_by_date_range(group_id, start_time, end_time) + result_str = self._format_messages_optimized(messages, all_contacts) + return result_str +``` + +## 使用方法 + +### 1. 启用插件 + +确保 `message_summary` 插件已启用，并且已在插件配置中开启。 + +### 2. 配置群权限 + +对于需要自动总结的群聊，需要开启"群总结能力"功能： + +- 通过机器人群管理命令设置群权限为 `ENABLED` +- 功能键：`SUMMARY_CAPABILITY` + +### 3. 自动运行 + +- 定时任务会在每天早上 9:00 自动执行 +- 只会总结昨天（00:00:00 - 23:59:59）的聊天记录 +- **只有消息数量超过 50 条的群才会生成总结**（按消息条数统计，不是字符数） + +### 4. 手动触发 + +除了定时任务外，原有的手动总结功能仍然保留： + +- 在群聊中发送 `#总结` 或 `#summary` +- 会总结最近 8 小时的聊天记录 + +## 特性说明 + +1. **智能过滤** + - 只总结文本消息（message_type 1 和 49） + - 过滤掉命令消息（以 `/` 开头） + - 过滤掉过短（<6字）和过长（>300字）的消息 + +2. **权限控制** + - 只有开启了总结功能的群才会自动总结 + - 使用 `GroupBotManager` 管理群权限 + +3. **消息数量过滤** + - **只有消息数量 >= 50 条的群才会生成总结** + - 先统计消息数量，不足 50 条直接跳过，避免浪费资源 + - 按消息条数统计，不是按字符数 + +4. **错误处理** + - 如果某个群的总结失败，不会影响其他群 + - 详细的日志记录便于排查问题 + +5. **性能优化** + - 群之间间隔 2 秒，避免 API 请求过快 + - 使用现有的消息格式化方法，保持一致性 + +## 日志示例 + +``` +2026-01-09 09:00:00 | INFO | 开始执行每日聊天总结任务 +2026-01-09 09:00:00 | INFO | 总结时间范围: 2026-01-08 00:00:00 至 2026-01-08 23:59:59 +2026-01-09 09:00:00 | INFO | 找到 3 个开启定时总结的群聊 +2026-01-09 09:00:01 | INFO | 群 test_group_1 昨天有 234 条消息，开始获取内容 +2026-01-09 09:00:01 | INFO | 群 test_group_2 昨天只有 35 条消息，不足50条，跳过总结 +2026-01-09 09:00:01 | INFO | 群 test_group_3 昨天有 156 条消息，开始获取内容 +2026-01-09 09:00:02 | INFO | 获取到 234 条消息（时间范围：2026-01-08 00:00:00 至 2026-01-08 23:59:59），格式化后长度: 12580 +2026-01-09 09:00:02 | INFO | 开始为群 测试群1 生成总结，消息数量: 234，内容长度: 12580 +2026-01-09 09:00:15 | INFO | 成功发送群 测试群 的昨日总结图片 +2026-01-09 09:00:17 | INFO | 每日聊天总结任务执行完成 +``` + +## 注意事项 + +1. **时间精度**：定时任务使用系统时间，确保服务器时间准确 +2. **数据库时区**：确保数据库时区设置正确，存储的 timestamp 使用的是本地时间 +3. **API 限流**：如果总结的群很多，注意调整 `await asyncio.sleep(2)` 的间隔时间 +4. **消息量**：如果某个群昨天的消息量很大（>5000条），可能需要调整 `max_results` 参数 + +## 自定义配置 + +如需修改执行时间，可以修改 `message_summary/main.py` 中的定时任务注册： + +```python +# 改为每天早上 8:00 +async_job.at_times(["08:00"])(self.daily_summary_job) + +# 改为每天晚上 22:00 +async_job.at_times(["22:00"])(self.daily_summary_job) + +# 改为每天多个时间点 +async_job.at_times(["09:00", "21:00"])(self.daily_summary_job) +``` + +## 测试方法 + +为了测试定时任务功能，可以临时修改时间： + +```python +# 临时修改为当前时间后 1 分钟（用于测试） +import datetime +next_minute = (datetime.datetime.now() + datetime.timedelta(minutes=1)).strftime("%H:%M") +async_job.at_times([next_minute])(self.daily_summary_job) +``` + +测试完成后记得改回 `["09:00"]`。 diff --git a/plugins/message_summary/main.py b/plugins/message_summary/main.py index c42b333..f31d869 100644 --- a/plugins/message_summary/main.py +++ b/plugins/message_summary/main.py @@ -1,6 +1,8 @@ +import asyncio import json import re import time +from datetime import datetime, timedelta from pathlib import Path from typing import Dict, Any, Tuple, Optional, List @@ -10,6 +12,7 @@ from loguru import logger from base.plugin_common.message_plugin_interface import MessagePluginInterface from base.plugin_common.plugin_interface import PluginStatus from utils.compress_chat_data import compress_chat_data +from utils.decorator.async_job import async_job from utils.decorator.plugin_decorators import plugin_stats_decorator from utils.decorator.points_decorator import plugin_points_cost from utils.decorator.rate_limit_decorator import group_feature_rate_limit @@ -67,6 +70,8 @@ class MessageSummaryPlugin(MessagePluginInterface): self.revoke = None # 注册功能权限 self.feature = self.register_feature() + # 注册定时任务：每天早上9点总结昨天的聊天信息 + async_job.at_times(["09:00"])(self.daily_summary_job) def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" @@ -288,3 +293,107 @@ class MessageSummaryPlugin(MessagePluginInterface): except Exception as e: self.LOG.error(f"处理总结时出现未知错误: {e}") return f"生成总结时出现未知错误: {str(e)}", None + + async def daily_summary_job(self): + """定时任务：每天早上9点总结昨天的聊天信息""" + try: + self.LOG.info("开始执行每日聊天总结任务") + + # 计算昨天的时间范围 + yesterday = datetime.now() - timedelta(days=1) + yesterday_start = yesterday.replace(hour=0, minute=0, second=0, microsecond=0) + yesterday_end = yesterday.replace(hour=23, minute=59, second=59, microsecond=999999) + + self.LOG.info(f"总结时间范围: {yesterday_start.strftime('%Y-%m-%d %H:%M:%S')} 至 {yesterday_end.strftime('%Y-%m-%d %H:%M:%S')}") + + # 获取所有启用了群机器人的群聊 + all_groups = GroupBotManager.get_group_list() + + if not all_groups: + self.LOG.info("没有群聊启用群机器人，跳过定时总结") + return + + # 筛选出开启了总结功能的群聊 + enabled_groups = [] + for group_id in all_groups: + if GroupBotManager.get_group_permission(group_id, self.feature) == PermissionStatus.ENABLED: + enabled_groups.append(group_id) + + if not enabled_groups: + self.LOG.info("没有群聊开启定时总结功能，跳过") + return + + self.LOG.info(f"找到 {len(enabled_groups)} 个开启定时总结的群聊") + + # 获取所有联系人 + all_contacts = ContactManager.get_instance().get_all_contacts() + + # 为每个群生成总结 + for group_id in enabled_groups: + try: + # 先统计昨天的消息数量 + message_count = self.message_storage.count_messages_by_date_range( + group_id, + yesterday_start, + yesterday_end + ) + + # 消息少于50条，跳过总结 + if message_count < 50: + self.LOG.info(f"群 {group_id} 昨天只有 {message_count} 条消息，不足50条，跳过总结") + continue + + self.LOG.info(f"群 {group_id} 昨天有 {message_count} 条消息，开始获取内容") + + # 获取群名 + group_name = all_contacts.get(group_id, group_id) + group_name = self._sanitize_group_name(group_name) + + # 获取昨天的聊天记录 + chat_content = self.message_storage.get_messages_by_date_range( + group_id, + all_contacts, + yesterday_start, + yesterday_end + ) + + if not chat_content: + self.LOG.info(f"群 {group_id} 昨天聊天记录为空，跳过总结") + continue + + self.LOG.info(f"开始为群 {group_name} 生成总结，消息数量: {message_count}，内容长度: {len(chat_content)}") + + # 发送提示消息 + await self.bot.send_text_message( + group_id, + f"⏳ 正在生成昨日聊天总结 ({yesterday.strftime('%Y-%m-%d')})… 😊" + ) + + # 生成总结 + summary, image_path = await self._generate_summary(chat_content, group_name) + + if image_path: + # 图片生成成功，发送图片 + await self.bot.send_image_message(group_id, Path(image_path)) + self.LOG.info(f"成功发送群 {group_name} 的昨日总结图片") + else: + # 图片生成失败，发送文本消息 + if summary and len(summary.strip()) > 0: + max_length = 2000 + if len(summary) > max_length: + summary = summary[:max_length] + "\n\n... (内容过长，已截断)" + + await self.bot.send_text_message(group_id, summary) + self.LOG.info(f"成功发送群 {group_name} 的昨日总结文本") + + # 避免请求过快 + await asyncio.sleep(2) + + except Exception as e: + self.LOG.error(f"为群 {group_id} 生成昨日总结失败: {e}", exc_info=True) + continue + + self.LOG.info("每日聊天总结任务执行完成") + + except Exception as e: + self.LOG.error(f"每日聊天总结任务执行失败: {e}", exc_info=True) diff --git a/utils/wechat/message_to_db.py b/utils/wechat/message_to_db.py index deeaa0c..5a5cef9 100644 --- a/utils/wechat/message_to_db.py +++ b/utils/wechat/message_to_db.py @@ -496,6 +496,55 @@ class MessageStorage: logger.error(f"获取消息出错: {e}") return "" + def get_messages_by_date_range(self, group_id, all_contacts: dict, start_time: datetime, end_time: datetime): + """获取指定时间范围内的消息 + + Args: + group_id: 群组ID + all_contacts: 联系人字典 + start_time: 开始时间 + end_time: 结束时间 + + Returns: + 格式化后的消息字符串 + """ + try: + # 使用新的数据库方法获取指定时间范围的消息 + messages = self.message_db.get_messages_by_date_range( + group_id, + start_time, + end_time + ) + + # 使用优化后的格式化方法 + result_str = self._format_messages_optimized(messages, all_contacts) + logger.info(f"获取到 {len(messages)} 条消息（时间范围：{start_time} 至 {end_time}），格式化后长度: {len(result_str)}") + + return result_str + + except Exception as e: + logger.error(f"按时间范围获取消息出错: {e}") + return "" + + def count_messages_by_date_range(self, group_id, start_time: datetime, end_time: datetime) -> int: + """统计指定时间范围内的消息数量 + + Args: + group_id: 群组ID + start_time: 开始时间 + end_time: 结束时间 + + Returns: + 消息数量 + """ + try: + count = self.message_db.count_messages_by_date_range(group_id, start_time, end_time) + logger.info(f"群 {group_id} 在 {start_time} 至 {end_time} 之间有 {count} 条消息") + return count + except Exception as e: + logger.error(f"统计消息数量出错: {e}") + return 0 + def _format_messages_optimized(self, messages: list, all_contacts: dict) -> str: """优化的消息格式化方法，减少冗余