bridzik/api/leaderboard.py

"""Rebricky (leaderboard) nad `db/`.

Tri rebricky, vsetky len z naplno dohranych hier (rovnako ako `get_player_history`):
  - najviac bodov za jednu hru -- hry len s ludmi,
  - najviac bodov za jednu hru -- hry, v ktorych hral aspon jeden bot,
  - hodiny straveny v hrach (`ended_at - created_at`, aj hry s botmi).
Boti sa v rebricku nikdy nezobrazuju. Pri bodoch ma kazdy hrac najviac jeden
riadok (jeho najlepsia hra). Kazdy rebricek vrati `top` (TOP_N riadkov) a `me`
(riadok prihlaseneho hraca s jeho poradim, alebo None ak v rebricku nie je).

Citanie z DB (`get_leaderboard`) je oddelene od radenia (`build_leaderboard`),
aby sa pravidla dali testovat bez DB.
"""

from datetime import timedelta

from sqlalchemy import DateTime, cast, func, select

from api.bots import is_bot_username
from api.history import FULL_GAME_ROUNDS, _usernames_for
from db.db import async_session, engine
from db.models import Game, Guess

TOP_N = 10

# Do hodin sa pocita len hra, ktora trvala najviac tolko. Plynula hra cez polnoc
# sa tym zapocita; obnovena hra dohravana o hodiny ci dni neskor (ended_at sa pri
# obnove znovu nastavi) by do casu pridala celu pauzu, tak sa vynecha.
MAX_GAME_DURATION = timedelta(hours=3)


def _ranked(entries: list[dict], player_id: int) -> dict:
    """Ocisluje uz zoradeny zoznam, vrati {top, me}."""
    for i, entry in enumerate(entries):
        entry["rank"] = i + 1
    me = next((e for e in entries if e["player_id"] == player_id), None)
    return {"top": entries[:TOP_N], "me": me}


def _points_board(results, games, usernames, player_id: int, with_bots: bool) -> dict:
    """Najlepsia hra kazdeho cloveka; `with_bots` vyberie hry s botom / bez bota."""
    # vyssie body vyhrava; pri rovnosti skorsi koniec hry, potom player_id
    rows = sorted(
        (
            (points, games[gid]["ended_at"], pid)
            for gid, pid, points in results
            if games[gid]["has_bot"] == with_bots
        ),
        key=lambda r: (-r[0], r[1], r[2]),
    )
    entries, seen = [], set()
    for points, ended_at, pid in rows:
        if pid in seen:
            continue  # hrac uz ma riadok so svojou najlepsou hrou
        seen.add(pid)
        entries.append(
            {
                "player_id": pid,
                "username": usernames.get(pid, "?"),
                "points": points,
                "ended_at": ended_at.isoformat(),
            }
        )
    return _ranked(entries, player_id)


def _hours_board(results, games, usernames, player_id: int) -> dict:
    """Sucet dlzky hier hraca (len hry do MAX_GAME_DURATION) a ich pocet."""
    totals: dict[int, list[int]] = {}  # player_id -> [sekundy, pocet hier]
    for gid, pid, _points in results:
        duration = games[gid]["ended_at"] - games[gid]["created_at"]
        if duration < timedelta(0) or duration > MAX_GAME_DURATION:
            continue
        agg = totals.setdefault(pid, [0, 0])
        agg[0] += int(duration.total_seconds())
        agg[1] += 1

    ordered = sorted(totals.items(), key=lambda kv: (-kv[1][0], kv[0]))
    entries = [
        {
            "player_id": pid,
            "username": usernames.get(pid, "?"),
            "seconds": seconds,
            "games": count,
        }
        for pid, (seconds, count) in ordered
    ]
    return _ranked(entries, player_id)


def build_leaderboard(results, games, usernames, player_id: int) -> dict:
    """Poskladaj vsetky rebricky.

    `results` = zoznam (game_id, player_id, sucet bodov hraca v hre) len za dohrane
    hry; `games[gid]` = {created_at, ended_at, has_bot}; `usernames` = player_id -> meno.
    """
    # boti hraju, ale v rebricku sa nezobrazuju
    results = [r for r in results if not is_bot_username(usernames.get(r[1], ""))]
    return {
        "points": {
            "humans": _points_board(results, games, usernames, player_id, False),
            "bots": _points_board(results, games, usernames, player_id, True),
        },
        "hours": _hours_board(results, games, usernames, player_id),
    }


def _created_at_utc():
    """`Game.created_at` ako naivny UTC, porovnatelny s `ended_at`.

    `created_at` zapisuje DB (`server_default=func.now()`) v casovej zone DB
    session -- v produkcii PostgreSQL s TZ Europe/Bratislava -- kym `ended_at`
    zapisuje Python ako naivny UTC (history._utcnow_naive). Na PostgreSQL preto
    cas prevedieme cez zonu session do UTC (spravne aj cez letny cas). SQLite
    CURRENT_TIMESTAMP je uz UTC.
    """
    if engine.dialect.name == "postgresql":
        return func.timezone("UTC", cast(Game.created_at, DateTime(timezone=True)))
    return Game.created_at


async def get_leaderboard(player_id: int) -> dict:
    """Rebricky z DB + poradie hraca `player_id` (aj ked je mimo top)."""
    created_at = _created_at_utc().label("created_at")
    async with async_session() as session:
        # Kazdy hrac ma jeden Guess na dohrate kolo (unikat v Guess), takze pocet
        # riadkov (hra, hrac) = pocet dohratych kol -- rovnake pravidlo "dohrana
        # hra" ako history._completed_rounds_per_game, bez druheho dotazu.
        rows = (
            await session.execute(
                select(
                    Guess.game_id,
                    Guess.player_id,
                    func.sum(Guess.points),
                    created_at,
                    Game.ended_at,
                )
                .join(Game, Game.id == Guess.game_id)
                .where(Game.ended_at.is_not(None))
                .group_by(Guess.game_id, Guess.player_id, Game.created_at, Game.ended_at)
                .having(func.count() >= FULL_GAME_ROUNDS)
            )
        ).all()
        usernames = await _usernames_for(session, [r[1] for r in rows])

    games: dict[str, dict] = {}
    for gid, pid, _points, created, ended in rows:
        game = games.setdefault(gid, {"created_at": created, "ended_at": ended, "has_bot": False})
        # dohrana hra ma Guess pre vsetkych 4 hracov, takze vidime kazde sedadlo
        if is_bot_username(usernames.get(pid, "")):
            game["has_bot"] = True
    results = [(gid, pid, int(points or 0)) for gid, pid, points, _c, _e in rows]
    return build_leaderboard(results, games, usernames, player_id)