"""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)