"""Zapis a citanie historie hier nad `db/`. Cita hodnoty z ciste-Python enginu (bridzik.Bridzik) a uklada ich do DB. Engine sa neupravuje -- pouzivame len jeho existujuce metody. """ from datetime import datetime, timezone from random import shuffle from sqlalchemy import func, or_, select from bridzik import Bridzik, ROUNDS_PER_SERIES, Round, SERIES_PER_GAME, Series from db.db import async_session from db.models import Game, Guess, Player # Naplno dohrana hra ma zapisanych SERIES_PER_GAME * ROUNDS_PER_SERIES # dokoncenych kol -- tvar hry je definovany v bridzik.py, tu sa len cita. FULL_GAME_ROUNDS = SERIES_PER_GAME * ROUNDS_PER_SERIES def _utcnow_naive() -> datetime: """Naive UTC `datetime` na zapis do `ended_at`. Stlpec je TIMESTAMP WITHOUT TIME ZONE (ako `created_at`), tz-aware hodnotu by asyncpg/Postgres odmietol.""" return datetime.now(timezone.utc).replace(tzinfo=None) async def record_game_started(gid: str, name: str, player_ids: list[int]) -> None: """Zapise riadok Game so 4 ID hracov (podla sedadla). Idempotentne.""" async with async_session() as session: if await session.get(Game, gid) is not None: return session.add( Game( id=gid, name=name, player0_id=player_ids[0], player1_id=player_ids[1], player2_id=player_ids[2], player3_id=player_ids[3], ) ) await session.commit() async def record_completed_rounds(gid: str, core) -> None: """Zapise 4 Guess-y za kazde nove dohrate kolo; po dohrani hry vyplni ended_at. Idempotentne: kola uz zapisane v DB sa preskakuju. """ async with async_session() as session: game = await session.get(Game, gid) if game is None: return rows = await session.execute( select(Guess.series_number, Guess.round_number) .where(Guess.game_id == gid) .distinct() ) already = {(r.series_number, r.round_number) for r in rows} for series in core.series: for rnd in series.rounds: if not rnd.is_completed(): continue key = (series.series_number, rnd.round_number) if key in already: continue points = rnd.get_points_summary() # list[4], body za kolo for seat in range(4): session.add( Guess( game_id=gid, player_id=game.player_id_for_seat(seat), series_number=series.series_number, round_number=rnd.round_number, guess=rnd.guesses[seat], points=points[seat], ) ) # Priebezne uloz aktualnu poziciu hry (na restore). last_series = core.series[-1] game.series = last_series.series_number game.round = last_series.get_last_round().round_number if core.is_completed() and game.ended_at is None: game.ended_at = _utcnow_naive() await session.commit() async def get_standings(gid: str) -> tuple[list[list[list[int]]], list[list[list[int]]]]: """Body aj tipy po seriach/kolach z DB, oboje v tvare ktory caka frontend: `[serie][kolo][sedadlo 0..3]`. Vracia dvojicu `(points, guesses)` -- tipy su tam, aby frontend pri 0 bodoch ukazal preskrtnuty tip namiesto nuly. Citaju sa z tych istych `Guess` riadkov, takze jeden dotaz staci. """ async with async_session() as session: game = await session.get(Game, gid) if game is None: return [], [] seat_of = { game.player0_id: 0, game.player1_id: 1, game.player2_id: 2, game.player3_id: 3, } rows = ( await session.scalars( select(Guess) .where(Guess.game_id == gid) .order_by(Guess.series_number, Guess.round_number) ) ).all() # series_map[serie][kolo] = ([body sedadiel], [tipy sedadiel]) series_map: dict[int, dict[int, tuple[list[int], list[int]]]] = {} for gz in rows: rounds = series_map.setdefault(gz.series_number, {}) points, tips = rounds.setdefault(gz.round_number, ([0, 0, 0, 0], [0, 0, 0, 0])) seat = seat_of.get(gz.player_id) if seat is not None: points[seat] = gz.points tips[seat] = gz.guess points_table: list[list[list[int]]] = [] guesses_table: list[list[list[int]]] = [] for s in sorted(series_map): round_nums = sorted(series_map[s]) points_table.append([series_map[s][r][0] for r in round_nums]) guesses_table.append([series_map[s][r][1] for r in round_nums]) return points_table, guesses_table async def get_player_history(player_id: int) -> list[dict]: """Zoznam hier daneho hraca (najnovsie prve) so sumarom jeho bodov.""" async with async_session() as session: stmt = ( select(Game) .where( Game.ended_at.is_not(None), # iba ukoncene hry or_( Game.player0_id == player_id, Game.player1_id == player_id, Game.player2_id == player_id, Game.player3_id == player_id, ), ) .order_by(Game.created_at.desc()) ) games = (await session.scalars(stmt)).all() # Pocet dokoncenych kol na hru (na rozlisenie naplno dohranej hry od # predcasne ukoncenej) -- jeden batch dotaz pre vsetky hry hraca. completed_rounds = await _completed_rounds_per_game( session, [g.id for g in games] ) result = [] for g in games: seat_ids = [g.player0_id, g.player1_id, g.player2_id, g.player3_id] usernames = await _usernames_for(session, seat_ids) total = await session.scalar( select(func.coalesce(func.sum(Guess.points), 0)).where( Guess.game_id == g.id, Guess.player_id == player_id ) ) result.append( { "gid": g.id, "name": g.name, "created_at": g.created_at.isoformat() if g.created_at else None, "ended_at": g.ended_at.isoformat() if g.ended_at else None, "players": [usernames[pid] for pid in seat_ids], "my_points": int(total or 0), # True = dohrana naplno; False = predcasne ukoncena (da sa obnovit). "completed": completed_rounds.get(g.id, 0) >= FULL_GAME_ROUNDS, } ) return result async def get_game_detail(gid: str) -> dict | None: """Detail hry: tipy a body po kolach (won = points > 0).""" async with async_session() as session: game = await session.get(Game, gid) if game is None: return None seat_ids = [game.player0_id, game.player1_id, game.player2_id, game.player3_id] usernames = await _usernames_for(session, seat_ids) guesses = ( await session.scalars( select(Guess) .where(Guess.game_id == gid) .order_by(Guess.series_number, Guess.round_number, Guess.player_id) ) ).all() return { "gid": game.id, "name": game.name, "created_at": game.created_at.isoformat() if game.created_at else None, "ended_at": game.ended_at.isoformat() if game.ended_at else None, "players": [ {"player_id": pid, "username": usernames[pid]} for pid in seat_ids ], "rounds": [ { "series_number": gz.series_number, "round_number": gz.round_number, "player_id": gz.player_id, "username": usernames.get(gz.player_id), "guess": gz.guess, "points": gz.points, "won": gz.points > 0, } for gz in guesses ], } async def _usernames_for(session, player_ids: list[int]) -> dict[int, str]: rows = await session.scalars( select(Player).where(Player.id.in_(set(player_ids))) ) return {p.id: p.username for p in rows} async def _completed_rounds_per_game(session, gids: list[str]) -> dict[str, int]: """Pocet dokoncenych (series, round) kol na hru. Guess sa zapisuje len za dohrate kola, takze pocet unikatnych dvojic = pocet dokoncenych kol.""" if not gids: return {} rows = await session.execute( select(Guess.game_id, Guess.series_number, Guess.round_number) .where(Guess.game_id.in_(gids)) .distinct() ) counts: dict[str, int] = {} for r in rows: counts[r.game_id] = counts.get(r.game_id, 0) + 1 return counts async def _restore_info(session, game: Game) -> dict: """Postavi restore-payload pre jednu hru: gid, name, sedadla (player_id + username podla poradia 0..3) a uz postaveny Bridzik na ulozenej pozicii. Spolocny tvar pre `reopen_game` aj `get_unfinished_games`.""" seat_ids = [game.player0_id, game.player1_id, game.player2_id, game.player3_id] usernames = await _usernames_for(session, seat_ids) return { "gid": game.id, "name": game.name, "seats": [(pid, usernames.get(pid, "?")) for pid in seat_ids], "core": rebuild_core(game.series, game.round), } async def reopen_game(gid: str, player_id: int) -> dict | None: """Znovu otvori predcasne ukoncenu hru: vymaze `ended_at` a vrati info na obnovu do pamate (rovnaky tvar ako polozka z `get_unfinished_games`). Vrati None, ak hra neexistuje, hrac v nej nie je, alebo uz bola dohrana naplno (vtedy nie je co pokracovat). """ async with async_session() as session: game = await session.get(Game, gid) if game is None: return None seat_ids = [game.player0_id, game.player1_id, game.player2_id, game.player3_id] if player_id not in seat_ids: return None counts = await _completed_rounds_per_game(session, [gid]) if counts.get(gid, 0) >= FULL_GAME_ROUNDS: return None # naplno dohrana hra sa neobnovuje game.ended_at = None info = await _restore_info(session, game) await session.commit() return info # --- restore --------------------------------------------------------------- def rebuild_core(series_number: int, round_number: int, shuffler=shuffle) -> Bridzik: """Postavi Bridzik na danu poziciu (zaciatok kola), karty rozda nanovo. Vsetko ostatne (rotacia first_player, pocet kariet) je deterministicke z dvojice (series_number, round_number), takze tieto dve cisla staci na obnovu hracej kostry. Tipy a rozohrate kopky aktualneho kola sa NEobnovuju -- kolo sa zacne odznova; historicke body si vola get_game_detail z tabulky Guess. """ core = Bridzik(shuffler=shuffler) # Doplnaj serie az po cielovu (kazda nova zacne svojim kolom 0). while core.series[-1].series_number < series_number: core.series.append(Series(len(core.series), shuffler=shuffler)) # V poslednej serii doplnaj kola az po cielove (Round rozda karty nanovo). last = core.series[-1] while last.get_last_round().round_number < round_number: rn = len(last.rounds) last.rounds.append(Round(rn, (last.first_player + rn) % 4, shuffler=shuffler)) return core async def restore_game_core(gid: str, shuffler=shuffle) -> Bridzik | None: """Nacita poziciu hry z DB a vrati obnoveny Bridzik (alebo None).""" async with async_session() as session: game = await session.get(Game, gid) if game is None: return None return rebuild_core(game.series, game.round, shuffler=shuffler) async def mark_game_ended(gid: str) -> None: """Natrvalo ukonci hru (host ju zrusil, ked sa nedohra). Uz sa neobnovi.""" async with async_session() as session: game = await session.get(Game, gid) if game is not None and game.ended_at is None: game.ended_at = _utcnow_naive() await session.commit() async def get_unfinished_games() -> list[dict]: """Nedohrate hry (ended_at IS NULL) aj s obnovenym jadrom -- na obnovu pri starte.""" async with async_session() as session: games = ( await session.scalars(select(Game).where(Game.ended_at.is_(None))) ).all() return [await _restore_info(session, g) for g in games]