83 lines
3.0 KiB
Python
83 lines
3.0 KiB
Python
"""Leaderboard queries over the full collection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from .. import pb_client
|
|
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]
|