4cd13503a7
Adds economy.get_economy_stats() (single-scan total coins, house balance, player-held coins, player count) and a "Rahavaru" field to /status, so admins can see whether the sinks are keeping pace with minted income. Also removed a duplicate bot_admin_check import. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013VbAVsrZuYesea99mPMmPT
140 lines
5.4 KiB
Python
140 lines
5.4 KiB
Python
"""Leaderboard queries over the full collection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from .. import pb_client
|
|
from . import house
|
|
from .levels import get_level
|
|
|
|
|
|
async def get_leaderboard(top_n: int | None = 10) -> list[tuple[str, int]]:
|
|
"""Return top_n (user_id_str, balance) pairs sorted descending."""
|
|
records = await pb_client.list_all_records()
|
|
result = sorted(
|
|
((r["user_id"], r.get("balance", 0)) for r in records if r.get("user_id")),
|
|
key=lambda x: x[1],
|
|
reverse=True,
|
|
)
|
|
return result if top_n is None else result[:top_n]
|
|
|
|
|
|
async def get_leaderboard_exp(top_n: int | None = 10) -> list[tuple[str, int, int]]:
|
|
"""Return top_n (user_id_str, exp, level) sorted by EXP descending."""
|
|
records = await pb_client.list_all_records()
|
|
result = sorted(
|
|
((r["user_id"], r.get("exp", 0)) for r in records if r.get("user_id")),
|
|
key=lambda x: x[1],
|
|
reverse=True,
|
|
)
|
|
entries = [(uid, exp, get_level(exp)) for uid, exp in result]
|
|
return entries if top_n is None else entries[:top_n]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Extended leaderboards
|
|
# ---------------------------------------------------------------------------
|
|
async def get_leaderboard_season_exp(top_n: int | None = 10) -> list[tuple[str, int, int]]:
|
|
"""Return (user_id, season_total_exp, prestige_level) sorted by season EXP."""
|
|
records = await pb_client.list_all_records()
|
|
result = sorted(
|
|
(
|
|
(r["user_id"], r.get("season_total_exp", 0), r.get("prestige_level", 0))
|
|
for r in records if r.get("user_id")
|
|
),
|
|
key=lambda x: x[1],
|
|
reverse=True,
|
|
)
|
|
return result if top_n is None else result[:top_n]
|
|
|
|
|
|
async def get_leaderboard_prestige(top_n: int | None = 10) -> list[tuple[str, int, int]]:
|
|
"""Return (user_id, prestige_level, prestige_points) sorted by prestige_level then PP."""
|
|
records = await pb_client.list_all_records()
|
|
result = sorted(
|
|
(
|
|
(r["user_id"], r.get("prestige_level", 0), r.get("prestige_points", 0))
|
|
for r in records if r.get("user_id")
|
|
),
|
|
key=lambda x: (x[1], x[2]),
|
|
reverse=True,
|
|
)
|
|
return result if top_n is None else result[:top_n]
|
|
|
|
|
|
async def get_leaderboard_wagered(top_n: int | None = 10) -> list[tuple[str, int]]:
|
|
"""Return (user_id, total_wagered) sorted descending."""
|
|
records = await pb_client.list_all_records()
|
|
result = sorted(
|
|
((r["user_id"], r.get("total_wagered", 0)) for r in records if r.get("user_id")),
|
|
key=lambda x: x[1],
|
|
reverse=True,
|
|
)
|
|
return result if top_n is None else result[:top_n]
|
|
|
|
|
|
async def get_leaderboard_fish(top_n: int | None = 10) -> list[tuple[str, int]]:
|
|
"""Return (user_id, total_fish_caught) sorted descending."""
|
|
records = await pb_client.list_all_records()
|
|
result = sorted(
|
|
((r["user_id"], r.get("total_fish_caught", 0)) for r in records if r.get("user_id")),
|
|
key=lambda x: x[1],
|
|
reverse=True,
|
|
)
|
|
return result if top_n is None else result[:top_n]
|
|
|
|
|
|
async def get_all_leaderboards() -> dict[str, list[tuple]]:
|
|
"""Build every leaderboard view from a SINGLE collection scan.
|
|
|
|
/leaderboard shows six tabs; calling each get_leaderboard_* separately would
|
|
read the whole collection six times. This reads once and sorts in memory,
|
|
returning the same tuple shapes the individual functions produce (unbounded -
|
|
the command paginates)."""
|
|
records = await pb_client.list_all_records()
|
|
users = [r for r in records if r.get("user_id")]
|
|
|
|
def desc(keyfn) -> list[dict]:
|
|
return sorted(users, key=keyfn, reverse=True)
|
|
|
|
return {
|
|
"coins": [(r["user_id"], r.get("balance", 0))
|
|
for r in desc(lambda r: r.get("balance", 0))],
|
|
"exp": [(r["user_id"], r.get("exp", 0), get_level(r.get("exp", 0)))
|
|
for r in desc(lambda r: r.get("exp", 0))],
|
|
"season": [(r["user_id"], r.get("season_total_exp", 0), r.get("prestige_level", 0))
|
|
for r in desc(lambda r: r.get("season_total_exp", 0))],
|
|
"prestige": [(r["user_id"], r.get("prestige_level", 0), r.get("prestige_points", 0))
|
|
for r in desc(lambda r: (r.get("prestige_level", 0), r.get("prestige_points", 0)))],
|
|
"wagered": [(r["user_id"], r.get("total_wagered", 0))
|
|
for r in desc(lambda r: r.get("total_wagered", 0))],
|
|
"fish": [(r["user_id"], r.get("total_fish_caught", 0))
|
|
for r in desc(lambda r: r.get("total_fish_caught", 0))],
|
|
}
|
|
|
|
|
|
async def get_economy_stats() -> dict[str, int]:
|
|
"""Money-supply snapshot from a single scan: total coins in circulation, how
|
|
much players hold vs. the house. Lets /status show whether the sinks (house,
|
|
vanity burn, bail, fines) are keeping pace with minted income."""
|
|
records = await pb_client.list_all_records()
|
|
house_id = str(house.HOUSE_ID) if house.HOUSE_ID is not None else None
|
|
total = 0
|
|
house_balance = 0
|
|
player_count = 0
|
|
for r in records:
|
|
uid = r.get("user_id")
|
|
if not uid:
|
|
continue
|
|
bal = r.get("balance", 0) or 0
|
|
total += bal
|
|
if uid == house_id:
|
|
house_balance = bal
|
|
else:
|
|
player_count += 1
|
|
return {
|
|
"total_coins": total,
|
|
"house_balance": house_balance,
|
|
"player_coins": total - house_balance,
|
|
"player_count": player_count,
|
|
}
|