Game logic and callback handling optimizations

bot/db/requests.py

@@ -23,21 +23,21 @@ async def get_games_by_id(session: AsyncSession, user_id: int) -> List[GameHisto
     return game_data_request.scalars().all()
 
 
-async def log_game(session: AsyncSession, data: Dict, telegram_id: int, status: str):
+async def log_game(session: AsyncSession, data: Dict, telegram_id: int, is_win: bool):
     """
     Send end game event to database
 
     :param session: SQLAlchemy DB session
     :param data: game data dictionary (only size is taken for now)
     :param telegram_id: Player's Telegram ID
-    :param status: "win" or "lose"
+    :param is_win: True if player has won the current game
     """
     entry = GameHistoryEntry()
     entry.game_id = data["game_id"]
     entry.played_at = datetime.utcnow()
     entry.telegram_id = telegram_id
     entry.field_size = data["game_data"]["size"]
-    entry.victory = status == "win"
+    entry.victory = is_win
     session.add(entry)
     # If a user is quick enough, there might be 2 events with the same UUID.
     # There's not much we can do, so simply ignore it until we come up with a better solution

bot/handlers/callbacks.py

@@ -1,7 +1,8 @@
 from contextlib import suppress
+from typing import Dict, List
 from uuid import uuid4
 
-from aiogram import types, Router
+from aiogram import types, Router, F
 from aiogram.dispatcher.fsm.context import FSMContext
 from aiogram.exceptions import TelegramBadRequest
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -11,9 +12,9 @@
 from bot.db.requests import log_game
 from bot.keyboards.kb_minefield import make_keyboard_from_minefield
 from bot.keyboards.kb_newgame import make_replay_keyboard
-from bot.minesweeper.game import (get_fake_newgame_data, untouched_cells_count, all_flags_match_bombs,
-                                  all_free_cells_are_open, make_text_table, get_real_game_data, gather_open_cells)
-from bot.minesweeper.states import ClickMode, CellMask
+from bot.minesweeper.game import (get_fake_newgame_data, make_text_table, update_game_field,
+                                  ensure_real_game_field, analyze_game_field)
+from bot.minesweeper.states import ClickMode, CellMask, GameState
 
 router = Router()
 
@@ -36,12 +37,43 @@ async def callback_newgame(call: types.CallbackQuery, state: FSMContext, callbac
     await call.answer()
 
 
+async def update_player_keyboard(
+        callback: types.CallbackQuery,
+        cells: List[List[Dict]],
+        game_id: str,
+        click_mode: int
+):
+    """
+    Updates player's keyboard
+    """
+    with suppress(TelegramBadRequest):
+        await callback.message.edit_reply_markup(
+            make_keyboard_from_minefield(cells, game_id, click_mode)
+        )
+
+
 @router.callback_query(ClickCallbackFactory.filter(), flags={"need_check_game": True})
 async def callback_open_square(call: types.CallbackQuery, state: FSMContext,
                                callback_data: ClickCallbackFactory, session: AsyncSession):
     """
     Called when player clicks a HIDDEN cell (without any flags or numbers)
     """
+
+    async def finish_game(is_win: bool):
+        """
+        Finishes the game with either win or lose
+        :param is_win: True if user won the game
+        """
+        added_text = "<b>You won!</b> 🎉" if is_win else "<b>You lost</b> 😞"
+
+        with suppress(TelegramBadRequest):
+            await call.message.edit_text(
+                call.message.html_text + f"\n\n{make_text_table(cells)}\n\n{added_text}",
+                reply_markup=make_replay_keyboard(field_size, bombs)
+            )
+        await log_game(session, fsm_data, call.from_user.id, is_win)
+        await call.answer()
+
     fsm_data = await state.get_data()
     game_id = fsm_data.get("game_id")
     game_data = fsm_data.get("game_data", {})
@@ -51,62 +83,33 @@ async def callback_open_square(call: types.CallbackQuery, state: FSMContext,
     x: int = callback_data.x
     y: int = callback_data.y
 
-    # If this is the first click, it's time to generate the real game field
-    if game_data["initial"] is True:
-        cells = get_real_game_data(game_data["size"], game_data["bombs"], (x, y))
-        game_data["cells"] = cells
-        game_data["initial"] = False
-    else:
-        cells = game_data.get("cells")
+    # Get the real game field in case of the first click
+    ensure_real_game_field(game_data, (x, y))
+    cells = game_data.get("cells")
+
+    # Update game field (in memory, not for user yet)
+    update_game_field(cells, x, y)
 
-    # This cell contained a bomb
+    # Clicked cell contained a bomb
     if cells[x][y]["value"] == "*":
-        cells[x][y]["mask"] = CellMask.BOMB
-        with suppress(TelegramBadRequest):
-            await call.message.edit_text(
-                call.message.html_text + f"\n\n{make_text_table(cells)}\n\n<b>You lost</b> 😞",
-                reply_markup=make_replay_keyboard(field_size, bombs)
-            )
-        await log_game(session, fsm_data, call.from_user.id, "lose")
-    # This cell contained a number
-    else:
-        # If cell is empty (0), open all adjacent squares
-        if cells[x][y]["value"] == 0:
-            for item in gather_open_cells(cells, (x, y)):
-                cells[item[0]][item[1]]["mask"] = CellMask.OPEN
-        # ... or just the current one
-        else:
-            cells[x][y]["mask"] = CellMask.OPEN
-
-        if all_free_cells_are_open(cells):
-            with suppress(TelegramBadRequest):
-                await call.message.edit_text(
-                    call.message.html_text + f"\n\n{make_text_table(cells)}\n\n<b>You won!</b> 🎉",
-                    reply_markup=make_replay_keyboard(field_size, bombs)
-                )
-            await log_game(session, fsm_data, call.from_user.id, "win")
-            await call.answer()
-            return
-        # There are more flags than there should be
-        elif untouched_cells_count(cells) == 0 and not all_flags_match_bombs(cells):
-            await state.update_data(game_data=game_data)
-            with suppress(TelegramBadRequest):
-                await call.message.edit_reply_markup(
-                    make_keyboard_from_minefield(cells, game_id, game_data["current_mode"])
-                )
-            await call.answer(
-                show_alert=True,
-                text="Looks like you've placed more flags than there are bombs on field. Please check them again."
-            )
-            return
-        # If this is not the last cell to open
-        else:
-            await state.update_data(game_data=game_data)
-            with suppress(TelegramBadRequest):
-                await call.message.edit_reply_markup(
-                    make_keyboard_from_minefield(cells, game_id, game_data["current_mode"])
-                )
-    await call.answer(cache_time=2)
+        await finish_game(is_win=False)
+        return
+
+    game_state = analyze_game_field(cells)
+    if game_state == GameState.HAS_HIDDEN_NUMBERS:
+        await state.update_data(game_data=game_data)
+        await update_player_keyboard(call, cells, game_id, game_data["current_mode"])
+        await call.answer()
+    elif game_state == GameState.MORE_FLAGS_THAN_BOMBS:
+        await state.update_data(game_data=game_data)
+        await update_player_keyboard(call, cells, game_id, game_data["current_mode"])
+        await call.answer(
+            show_alert=True,
+            text="Looks like you've placed more flags than there are bombs on the field. "
+                 "Please check them again."
+        )
+    else:  # == GameState.Victory
+        await finish_game(is_win=True)
 
 
 @router.callback_query(SwitchModeCallbackFactory.filter(), flags={"need_check_game": True})
@@ -126,67 +129,58 @@ async def switch_click_mode(call: types.CallbackQuery, state: FSMContext, callba
     game_data["current_mode"] = callback_data.new_mode
     await state.update_data(game_data=game_data)
 
-    with suppress(TelegramBadRequest):
-        await call.message.edit_reply_markup(
-            make_keyboard_from_minefield(cells, game_id, game_data["current_mode"])
-        )
+    await update_player_keyboard(call, cells, game_id, game_data["current_mode"])
     await call.answer()
 
 
-@router.callback_query(SwitchFlagCallbackFactory.filter(), flags={"need_check_game": True})
-async def add_or_remove_flag(call: types.CallbackQuery, state: FSMContext,
-                             callback_data: SwitchFlagCallbackFactory, session: AsyncSession):
-    """
-    Called when player puts a flag on HIDDEN cell or clicks a flag to remove it
-    """
+@router.callback_query(
+    SwitchFlagCallbackFactory.filter(F.action == "remove"),
+    flags={"need_check_game": True}
+)
+async def cb_remove_flag(
+        call: types.CallbackQuery,
+        state: FSMContext,
+        callback_data: SwitchFlagCallbackFactory
+):
     fsm_data = await state.get_data()
     game_id = fsm_data.get("game_id")
-    game_data = fsm_data.get("game_data", {})
+    game_data = fsm_data.get("game_data")
     cells = game_data.get("cells")
-    field_size = int(game_data.get("size"))
-    bombs = int(game_data.get("bombs"))
 
-    action = callback_data.action
-    flag_x = callback_data.x
-    flag_y = callback_data.y
+    cells[callback_data.x][callback_data.y].update(mask=CellMask.HIDDEN)
+    await state.update_data(game_data=game_data)
+    await update_player_keyboard(call, cells, game_id, game_data["current_mode"])
+    await call.answer()
+
 
-    if action == "remove":
-        cells[flag_x][flag_y].update(mask=CellMask.HIDDEN)
+@router.callback_query(
+    SwitchFlagCallbackFactory.filter(F.action == "add"),
+    flags={"need_check_game": True}
+)
+async def cb_add_flag(
+        call: types.CallbackQuery,
+        state: FSMContext,
+        callback_data: SwitchFlagCallbackFactory
+):
+    fsm_data = await state.get_data()
+    game_id = fsm_data.get("game_id")
+    game_data = fsm_data.get("game_data")
+    cells = game_data.get("cells")
+
+    cells[callback_data.x][callback_data.y].update(mask=CellMask.FLAG)
+    game_state = analyze_game_field(cells)
+    if game_state == GameState.HAS_HIDDEN_NUMBERS:
         await state.update_data(game_data=game_data)
-        with suppress(TelegramBadRequest):
-            await call.message.edit_reply_markup(
-                make_keyboard_from_minefield(cells, game_id, game_data["current_mode"])
-            )
-    elif action == "add":
-        cells[flag_x][flag_y].update(mask=CellMask.FLAG)
-        # See callback_open_square() for explanation
-        if untouched_cells_count(cells) == 0:
-            if all_flags_match_bombs(cells):
-                with suppress(TelegramBadRequest):
-                    await call.message.edit_text(
-                        call.message.html_text + f"\n\n{make_text_table(cells)}\n\n<b>You won!</b> 🎉",
-                        reply_markup=make_replay_keyboard(field_size, bombs)
-                    )
-                await log_game(session, fsm_data, call.from_user.id, "win")
-            else:
-                await state.update_data(game_data=game_data)
-                with suppress(TelegramBadRequest):
-                    await call.message.edit_reply_markup(
-                        make_keyboard_from_minefield(cells, game_id, game_data["current_mode"])
-                    )
-                await call.answer(
-                    show_alert=True,
-                    text="Looks like you've placed more flags than there are bombs on field. "
-                         "Please check them again."
-                )
-                return
-        else:
-            await state.update_data(game_data=game_data)
-            with suppress(TelegramBadRequest):
-                await call.message.edit_reply_markup(
-                    make_keyboard_from_minefield(cells, game_id, game_data["current_mode"])
-                )
-    await call.answer(cache_time=1)
+        await update_player_keyboard(call, cells, game_id, game_data["current_mode"])
+        await call.answer()
+    elif game_state == GameState.MORE_FLAGS_THAN_BOMBS:
+        await state.update_data(game_data=game_data)
+        await update_player_keyboard(call, cells, game_id, game_data["current_mode"])
+        await call.answer(
+            show_alert=True,
+            text="Looks like you've placed more flags than there are bombs on the field. "
+                 "Please check them again."
+        )
 
 
 @router.callback_query(IgnoreCallbackFactory.filter())