101 lines
3.4 KiB
Python
101 lines
3.4 KiB
Python
import json
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
class GameplayBackendTemplate:
|
|
"""Abstract base for gameplay database backends.
|
|
|
|
Subclasses must override every method that raises NotImplementedError.
|
|
Shared pure-Python helpers (_utc_now, _to_json, etc.) live here so they
|
|
are available to both SQLite and PostgreSQL implementations.
|
|
"""
|
|
|
|
# ── public async interface ─────────────────────────────────────────────────
|
|
|
|
async def initialize(self) -> None:
|
|
"""Called once on server startup. Backends that need eager connection
|
|
(pool creation, schema init, migration) should override this."""
|
|
return None
|
|
|
|
async def record_game_start(self, game_state:dict, snake_type:str|None=None, snake_version:str|None=None) -> None:
|
|
raise NotImplementedError
|
|
|
|
async def record_turn(self, game_state:dict, my_move:str|None, my_thinking:dict|None=None) -> None:
|
|
raise NotImplementedError
|
|
|
|
async def record_game_end(self, game_state:dict) -> None:
|
|
raise NotImplementedError
|
|
|
|
async def get_summary(self, recent_limit:int=15) -> dict:
|
|
raise NotImplementedError
|
|
|
|
async def list_games(self, limit:int=50) -> list[dict]:
|
|
raise NotImplementedError
|
|
|
|
async def finalize_stale_running_games(self, stale_after_seconds:int=600) -> int:
|
|
raise NotImplementedError
|
|
|
|
async def get_game_replay(self, game_id:str) -> dict|None:
|
|
raise NotImplementedError
|
|
|
|
async def close(self) -> None:
|
|
return None
|
|
|
|
# ── shared pure-python helpers ─────────────────────────────────────────────
|
|
|
|
def _utc_now(self) -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
def _parse_utc_timestamp(self, value:str|None) -> datetime|None:
|
|
if not value:
|
|
return None
|
|
normalized = value.strip()
|
|
if normalized.endswith("Z"):
|
|
normalized = normalized[:-1] + "+00:00"
|
|
try:
|
|
parsed = datetime.fromisoformat(normalized)
|
|
except ValueError:
|
|
return None
|
|
if parsed.tzinfo is None:
|
|
return parsed.replace(tzinfo=timezone.utc)
|
|
return parsed.astimezone(timezone.utc)
|
|
|
|
def _to_json(self, payload:object) -> str:
|
|
return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
|
|
|
|
def _from_json(self, payload:str|None) -> Any:
|
|
if payload is None or payload == "":
|
|
return None
|
|
try:
|
|
return json.loads(payload)
|
|
except (json.JSONDecodeError, TypeError):
|
|
return None
|
|
|
|
def _extract_snakes(self, game_state:dict) -> list[dict]:
|
|
return list(game_state.get("board", {}).get("snakes", []))
|
|
|
|
def _extract_you(self, game_state:dict) -> dict:
|
|
return dict(game_state.get("you", {}))
|
|
|
|
def _infer_direction(self, old_head:tuple[int, int]|None, new_head:tuple[int, int]|None) -> str|None:
|
|
if old_head is None or new_head is None:
|
|
return None
|
|
|
|
dx = new_head[0] - old_head[0]
|
|
dy = new_head[1] - old_head[1]
|
|
if dx == 1 and dy == 0:
|
|
return "right"
|
|
if dx == -1 and dy == 0:
|
|
return "left"
|
|
if dx == 0 and dy == 1:
|
|
return "up"
|
|
if dx == 0 and dy == -1:
|
|
return "down"
|
|
return None
|
|
|
|
def _derive_game_type(self, board:dict, ruleset:dict) -> str:
|
|
if len(board.get("snakes", [])) == 2:
|
|
return "duel"
|
|
|
|
return ruleset.get("name") or "standard"
|