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

优化排行榜

Browse Source
This commit is contained in:
liuwei
2025-11-25 13:21:44 +08:00
parent 45d16fbd51
commit c91d9811d4
1 changed files with 16 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
plugins/xiuxian/main.py
View File

@@ -148,41 +148,41 @@ class XiuxianRedisDB:
logger.error(f"检查限流失败: {e}")
return False
def leaderboard_add(self, user_id: str, score: float):
def leaderboard_add(self, group_id: str, user_id: str, score: float):
try:
if not self.leaderboard_key:
return
with self.get_redis() as r:
r.zadd(self.leaderboard_key, {user_id: score})
r.zadd(f"{self.leaderboard_key}:{group_id}", {user_id: score})
except Exception as e:
logger.error(f"更新排行榜失败: {e}")
def leaderboard_top(self, top_n: int = 10) -> List[Tuple[str, float]]:
def leaderboard_top(self, group_id: str, top_n: int = 10) -> List[Tuple[str, float]]:
try:
if not self.leaderboard_key:
return []
with self.get_redis() as r:
res = r.zrevrange(self.leaderboard_key, 0, top_n - 1, withscores=True)
res = r.zrevrange(f"{self.leaderboard_key}:{group_id}", 0, top_n - 1, withscores=True)
return [(uid if isinstance(uid, str) else uid.decode("utf-8"), score) for uid, score in res]
except Exception as e:
logger.error(f"读取排行榜失败: {e}")
return []
def leaderboard_realm_add(self, user_id: str, score: float):
def leaderboard_realm_add(self, group_id: str, user_id: str, score: float):
try:
if not self.leaderboard_realm_key:
return
with self.get_redis() as r:
r.zadd(self.leaderboard_realm_key, {user_id: score})
r.zadd(f"{self.leaderboard_realm_key}:{group_id}", {user_id: score})
except Exception as e:
logger.error(f"更新境界排行榜失败: {e}")
def leaderboard_realm_top(self, top_n: int = 10) -> List[Tuple[str, float]]:
def leaderboard_realm_top(self, group_id: str, top_n: int = 10) -> List[Tuple[str, float]]:
try:
if not self.leaderboard_realm_key:
return []
with self.get_redis() as r:
res = r.zrevrange(self.leaderboard_realm_key, 0, top_n - 1, withscores=True)
res = r.zrevrange(f"{self.leaderboard_realm_key}:{group_id}", 0, top_n - 1, withscores=True)
return [(uid if isinstance(uid, str) else uid.decode("utf-8"), score) for uid, score in res]
except Exception as e:
logger.error(f"读取境界排行榜失败: {e}")
@@ -639,7 +639,7 @@ class XiuxianPlugin(MessagePluginInterface):
self._save_player(player)
self._rate_set(sender, roomid or "", "注册修仙")
# 初始化境界排行榜分值
self.redis_db.leaderboard_realm_add(sender, float(self._realm_score(player["realm"])))
self.redis_db.leaderboard_realm_add(roomid or "", sender, float(self._realm_score(player["realm"])))
client_msg_id, create_time, new_msg_id = await bot.send_text_message((roomid if roomid else sender),
f"✅ 入道已定，道号：{dao_name}\n灵根：{root_name}",
sender)
@@ -852,7 +852,7 @@ class XiuxianPlugin(MessagePluginInterface):
self._save_player(player)
# 自动层级提升（不跨瓶颈）
self._auto_layer_up(sender, player)
self.redis_db.leaderboard_add(sender, float(player["cultivation_points"]))
self.redis_db.leaderboard_add(player.get("group_id", ""), sender, float(player["cultivation_points"]))
self._rate_set(sender, roomid or "", "出关")
await self._send_text_with_status(bot, sender, roomid,
f"✅ 出关一朝，修为进益：{gain}（{duration_hours:.1f}小时）\n现有修为：{player['cultivation_points']}\n状态：气息未平 {self.unstable_qi_minutes}分钟",
@@ -907,13 +907,13 @@ class XiuxianPlugin(MessagePluginInterface):
self._save_player(player)
# 自动层级提升（不跨瓶颈）
self._auto_layer_up(sender, player)
self.redis_db.leaderboard_add(sender, float(player["cultivation_points"]))
self.redis_db.leaderboard_add(player.get("group_id", ""), sender, float(player["cultivation_points"]))
self._rate_set(sender, roomid or "", "聚灵")
await self._send_text_with_status(bot, sender, roomid, f"✅ 灵气入体，消耗灵石{qty}，修为涨{gain}", 10)
return True, "聚灵成功"
async def _cmd_leaderboard(self, bot: WechatAPIClient, sender: str, roomid: str) -> Tuple[bool, str]:
top = self.redis_db.leaderboard_top(50)
top = self.redis_db.leaderboard_top(roomid or "", 50)
items = []
for uid, score in top:
player = self._get_player_with_cache(uid, roomid or "")
@@ -1016,7 +1016,7 @@ class XiuxianPlugin(MessagePluginInterface):
pass
self._save_player(player)
# 更新境界排行榜
self.redis_db.leaderboard_realm_add(user_id, float(self._realm_score(new_realm)))
self.redis_db.leaderboard_realm_add(player.get("group_id", ""), user_id, float(self._realm_score(new_realm)))
def _auto_layer_up(self, user_id: str, player: Dict[str, Any]):
"""自动层级提升：根据修为自动提升层数（不跨瓶颈）
@@ -1329,7 +1329,7 @@ class XiuxianPlugin(MessagePluginInterface):
player["spirit_stone"] = stones
if cult_gain > 0:
player["cultivation_points"] = int(player.get("cultivation_points", 0)) + cult_gain
self.redis_db.leaderboard_add(sender, float(player["cultivation_points"]))
self.redis_db.leaderboard_add(player.get("group_id", ""), sender, float(player["cultivation_points"]))
inv = player.get("inventory") or {}
for k, v in mats_gain.items():
inv[k] = int(inv.get(k, 0)) + v
@@ -1588,7 +1588,7 @@ class XiuxianPlugin(MessagePluginInterface):
if roll < pill_conf["rate"]:
# 成功，更新境界并排行榜
self._set_realm(sender, player, pill_conf["target"])
self.redis_db.leaderboard_add(sender, float(player["cultivation_points"]))
self.redis_db.leaderboard_add(player.get("group_id", ""), sender, float(player["cultivation_points"]))
self._rate_set(sender, roomid or "", "突破")
await self._send_text_with_status(bot, sender, roomid, f"✅ 雷霆破关，晋至{pill_conf['target']}", 10)
@@ -1645,7 +1645,7 @@ class XiuxianPlugin(MessagePluginInterface):
roll = random.random()
if roll < hard_conf["rate"]:
self._set_realm(sender, player, hard_conf["target"])
self.redis_db.leaderboard_add(sender, float(player["cultivation_points"]))
self.redis_db.leaderboard_add(player.get("group_id", ""), sender, float(player["cultivation_points"]))
self._rate_set(sender, roomid or "", "强行突破")
await self._send_text_with_status(bot, sender, roomid, f"✅ 硬撼瓶颈，勉强晋至{hard_conf['target']}", 10)