From d60d496bc35b4aa5c2ea1d0161646de790157fdb Mon Sep 17 00:00:00 2001 From: liuwei Date: Tue, 21 Apr 2026 14:00:57 +0800 Subject: [PATCH] =?UTF-8?q?=E6=89=A9=E5=B1=95=20value=5Frank=20=E5=91=A8?= =?UTF-8?q?=E6=8A=A5=E8=83=BD=E5=8A=9B=E5=B9=B6=E6=96=B0=E5=A2=9E=E5=91=A8?= =?UTF-8?q?=E6=8A=A5=E5=91=BD=E4=BB=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增命令 身价周报，并接入配置与帮助文案 - 新增每周定时动作 value_rank_weekly_report_push（周一09:35）自动推送周报 - 周报内容包含：综合排行Top5、上升榜Top5、下行榜Top5（对比7天前） - 扩展 ValueRankDB：新增按日期读取快照分数字典能力，支持周报对比计算 - 调度执行中支持周报推送并补充重算保障，确保周报数据为当天最新 --- plugins/value_rank/config.toml | 7 +- plugins/value_rank/main.py | 136 +++++++++++++++++++++++++++++++-- 2 files changed, 134 insertions(+), 9 deletions(-) diff --git a/plugins/value_rank/config.toml b/plugins/value_rank/config.toml index 9221c10..497a56f 100644 --- a/plugins/value_rank/config.toml +++ b/plugins/value_rank/config.toml @@ -1,14 +1,15 @@ [ValueRank] enable = true -command = ["我的身价", "身价排行", "社交热度榜", "搭子榜", "身价说明", "重算身价"] +command = ["我的身价", "身价排行", "社交热度榜", "搭子榜", "身价周报", "身价说明", "重算身价"] command-format = """ 📊 身价系统命令： 1. 我的身价 2. 身价排行 [名次] 3. 社交热度榜 [名次] 4. 搭子榜 [名次] -5. 身价说明 -6. 重算身价（管理员） +5. 身价周报 +6. 身价说明 +7. 重算身价（管理员） """ # 统计窗口（天） diff --git a/plugins/value_rank/main.py b/plugins/value_rank/main.py index 79aebee..79ce737 100644 --- a/plugins/value_rank/main.py +++ b/plugins/value_rank/main.py @@ -1,7 +1,7 @@ # -*- coding: utf-8 -*- import json import math -from datetime import datetime +from datetime import datetime, timedelta from typing import Any, Dict, List, Optional, Tuple from loguru import logger @@ -251,6 +251,23 @@ class ValueRankDB(BaseDBOperator): """ return self.execute_query(sql, (group_id, social_window_days, limit)) or [] + def get_snapshot_score_map(self, stat_date: str, group_id: str) -> Dict[str, float]: + """读取某天群内所有用户分数字典：{user_id: score}。""" + sql = """ + SELECT user_id, score + FROM t_value_rank_snapshot + WHERE stat_date = %s + AND group_id = %s + """ + rows = self.execute_query(sql, (stat_date, group_id)) or [] + result: Dict[str, float] = {} + for row in rows: + user_id = str(row.get("user_id") or "").strip() + if not user_id: + continue + result[user_id] = float(row.get("score") or 0.0) + return result + class ValueRankPlugin(MessagePluginInterface): """群成员身价排行插件。 @@ -303,7 +320,7 @@ class ValueRankPlugin(MessagePluginInterface): # 配置默认值：即使未配置 config.toml，也能以保守参数运行。 self.enable = True - self._commands = ["我的身价", "身价排行", "社交热度榜", "搭子榜", "身价说明", "重算身价"] + self._commands = ["我的身价", "身价排行", "社交热度榜", "搭子榜", "身价周报", "身价说明", "重算身价"] self.command_format = "我的身价 | 身价排行 [名次] | 身价说明 | 重算身价" self.message_window_days = 7 @@ -416,6 +433,11 @@ class ValueRankPlugin(MessagePluginInterface): await bot.send_text_message(roomid, text, sender) return True, "查询成功" + if command == "身价周报": + text = await self._build_weekly_report_text(roomid) + await bot.send_text_message(roomid, text, sender) + return True, "查询成功" + if command == "身价说明": await bot.send_text_message(roomid, self._build_explain_text(), sender) return True, "查询成功" @@ -435,7 +457,7 @@ class ValueRankPlugin(MessagePluginInterface): return True, "命令错误" def get_schedule_actions(self) -> List[Dict[str, Any]]: - """声明可调度动作：每日凌晨全量重算。""" + """声明可调度动作：每日重算 + 每周周报。""" return [ { "action_key": "value_rank_daily_recompute", @@ -447,15 +469,26 @@ class ValueRankPlugin(MessagePluginInterface): "target_config": {}, "payload": {}, "default_enabled": True, - } + }, + { + "action_key": "value_rank_weekly_report_push", + "name": "身价周报推送", + "description": "每周推送身价周报（上升榜/下降榜/综合排行）", + "trigger_type": "every_week_time", + # weekday 取值与 datetime.weekday() 一致：周一=0 ... 周日=6。 + "trigger_config": {"weekday": 0, "time_str": "09:35"}, + "target_scope": "all_enabled_groups", + "target_config": {}, + "payload": {}, + "default_enabled": True, + }, ] async def run_scheduled_action(self, action_key: str, context: Dict[str, Any]) -> Dict[str, Any]: """执行调度动作。""" - if action_key != "value_rank_daily_recompute": + if action_key not in {"value_rank_daily_recompute", "value_rank_weekly_report_push"}: return {"success": False, "summary": f"不支持动作: {action_key}", "detail": {}} - stat_date = datetime.now().strftime("%Y-%m-%d") target_groups = [str(g).strip() for g in (context.get("target_groups") or []) if str(g).strip()] if not target_groups: target_groups = [ @@ -466,14 +499,39 @@ class ValueRankPlugin(MessagePluginInterface): success_groups: List[str] = [] failed_groups: Dict[str, str] = {} updated_users = 0 + pushed_groups: List[str] = [] + + stat_date = datetime.now().strftime("%Y-%m-%d") + bot = context.get("bot") or getattr(self, "bot", None) + + # 周报任务先确保当日快照存在，再执行推送，避免“有报表命令但无数据”。 + if action_key == "value_rank_weekly_report_push" and not bot: + return {"success": False, "summary": "周报推送失败：bot 未注入", "detail": {}} for gid in target_groups: try: + # 每个群都先重算一次，确保报告与排行数据是“当天最新口径”。 updated_users += self._recompute_group_snapshot(gid, stat_date) + if action_key == "value_rank_weekly_report_push": + weekly_text = await self._build_weekly_report_text(gid) + await bot.send_text_message(gid, weekly_text, "") + pushed_groups.append(gid) success_groups.append(gid) except Exception as e: failed_groups[gid] = str(e) + if action_key == "value_rank_weekly_report_push": + return { + "success": len(failed_groups) == 0, + "summary": f"身价周报完成：推送{len(pushed_groups)}群，失败{len(failed_groups)}群", + "detail": { + "stat_date": stat_date, + "updated_users": updated_users, + "pushed_groups": pushed_groups, + "failed_groups": failed_groups, + }, + } + return { "success": len(failed_groups) == 0, "summary": f"身价重算完成：成功{len(success_groups)}群，失败{len(failed_groups)}群", @@ -740,6 +798,72 @@ class ValueRankPlugin(MessagePluginInterface): lines.append(f"{medal} {nick_a} × {nick_b} | 互动{score:.1f} | @次数{mention_count}") return "\n".join(lines) + async def _build_weekly_report_text(self, group_id: str) -> str: + """构建“身价周报”文本。 + + 周报口径： + 1. 综合排行：取今日 Top5； + 2. 上升榜/下降榜：对比“今日分数 - 7天前分数”； + 3. 若7天前无数据则按 0 处理，保证新群也可输出。 + """ + if not self.db: + return "❌ 身价模块未初始化" + + today = datetime.now().strftime("%Y-%m-%d") + # 使用 SQL DATE_SUB 也可计算，但这里使用 Python 日期便于阅读与调试。 + base_date = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d") + + # 确保今天有快照，避免周报命令调用时出现空数据。 + today_rank_rows = self.db.get_today_rankings(today, group_id, 50) + if not today_rank_rows: + self._recompute_group_snapshot(group_id, today) + today_rank_rows = self.db.get_today_rankings(today, group_id, 50) + if not today_rank_rows: + return "📊 本周暂无可用身价数据。" + + today_score_map = self.db.get_snapshot_score_map(today, group_id) + base_score_map = self.db.get_snapshot_score_map(base_date, group_id) + + # 计算用户分数变化，并做升降序榜单。 + delta_rows: List[Tuple[str, float, float, float]] = [] + for user_id, today_score in today_score_map.items(): + old_score = float(base_score_map.get(user_id, 0.0)) + delta = round(float(today_score) - old_score, 2) + delta_rows.append((user_id, float(today_score), old_score, delta)) + + delta_rows_sorted_up = sorted(delta_rows, key=lambda x: x[3], reverse=True) + delta_rows_sorted_down = sorted(delta_rows, key=lambda x: x[3]) + + cm = ContactManager.get_instance() + lines: List[str] = [ + f"📈 身价周报（{today}）", + f"对比基线：{base_date}", + "", + "🏆 本周综合排行 Top5", + ] + for idx, row in enumerate(today_rank_rows[:5], start=1): + uid = str(row.get("user_id") or "") + nick = cm.get_group_name(group_id, uid) or uid + score = float(row.get("score") or 0.0) + title = str(row.get("title") or "") + lines.append(f"{idx}. {nick} | {score:.2f} | {title}") + + lines.append("") + lines.append("🚀 本周上升榜 Top5") + for idx, (uid, today_score, old_score, delta) in enumerate(delta_rows_sorted_up[:5], start=1): + nick = cm.get_group_name(group_id, uid) or uid + lines.append(f"{idx}. {nick} | {old_score:.2f} -> {today_score:.2f} | {delta:+.2f}") + + lines.append("") + lines.append("🧊 本周波动下行 Top5") + for idx, (uid, today_score, old_score, delta) in enumerate(delta_rows_sorted_down[:5], start=1): + nick = cm.get_group_name(group_id, uid) or uid + lines.append(f"{idx}. {nick} | {old_score:.2f} -> {today_score:.2f} | {delta:+.2f}") + + lines.append("") + lines.append("提示：分数由积分/发言/活跃/社交影响力综合计算。") + return "\n".join(lines) + def _build_explain_text(self) -> str: """输出算法说明文本。""" return (