liuwei_x_x/abot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

扩展 value_rank 周报能力并新增周报命令

Browse Source 
- 新增命令 身价周报，并接入配置与帮助文案

- 新增每周定时动作 value_rank_weekly_report_push（周一09:35）自动推送周报

- 周报内容包含：综合排行Top5、上升榜Top5、下行榜Top5（对比7天前）

- 扩展 ValueRankDB：新增按日期读取快照分数字典能力，支持周报对比计算

- 调度执行中支持周报推送并补充重算保障，确保周报数据为当天最新
This commit is contained in:
liuwei
2026-04-21 14:00:57 +08:00
parent 78adab65b2
commit d60d496bc3
2 changed files with 134 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
plugins/value_rank/config.toml
View File

@@ -1,14 +1,15 @@
[ValueRank]
enable = true
command = ["我的身价", "身价排行", "社交热度榜", "搭子榜", "身价说明", "重算身价"]
command = ["我的身价", "身价排行", "社交热度榜", "搭子榜", "身价周报", "身价说明", "重算身价"]
command-format = """
📊 身价系统命令：
1. 我的身价
2. 身价排行 [名次]
3. 社交热度榜 [名次]
4. 搭子榜 [名次]
5. 身价说明
6. 重算身价（管理员）
5. 身价周报
6. 身价说明
7. 重算身价（管理员）
"""
# 统计窗口（天）

136
plugins/value_rank/main.py
View File

@@ -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 (