tipibot/core/economy/leaderboards.py

"""Leaderboard queries over the full collection."""

from __future__ import annotations

from .. import pb_client
from . import house
from .levels import get_level


def _net_worth(r: dict) -> int:
    """Coins that count toward wealth: liquid balance + banked vault."""
    return (r.get("balance", 0) or 0) + (r.get("bank_balance", 0) or 0)


async def get_leaderboard(top_n: int | None = 10) -> list[tuple[str, int]]:
    """Return top_n (user_id_str, net_worth) pairs sorted descending.
    Net worth = balance + bank_balance, so banking coins does not hide them."""
    records = await pb_client.list_all_records()
    result = sorted(
        ((r["user_id"], _net_worth(r)) 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"], _net_worth(r))
                     for r in desc(_net_worth)],
        "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
        # Banked coins are still part of the money supply.
        worth = (r.get("balance", 0) or 0) + (r.get("bank_balance", 0) or 0)
        total += worth
        if uid == house_id:
            house_balance = worth
        else:
            player_count += 1
    return {
        "total_coins": total,
        "house_balance": house_balance,
        "player_coins": total - house_balance,
        "player_count": player_count,
    }