alee/golfgame
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Add PostgreSQL game logging system

Browse Source 
- Add GameLogger service for move logging to PostgreSQL
- Add moves table to event_store.py for AI decision analysis
- Update main.py to initialize GameLogger in lifespan
- Update game_analyzer.py to query PostgreSQL instead of SQLite
- Add VDD documentation V2_08_GAME_LOGGING.md

Replaces SQLite game_log.py with unified PostgreSQL backend.
See docs/v2/V2_08_GAME_LOGGING.md for architecture and API.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
adlee-was-taken
2026-02-14 08:42:49 -05:00
parent 7d28e83a49
commit 49b2490c25
5 changed files with 1003 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
server/game_analyzer.py
View File

@@ -3,10 +3,20 @@ Game Analyzer for 6-Card Golf AI decisions.
Evaluates AI decisions against optimal play baselines and generates
reports on decision quality, mistake rates, and areas for improvement.
NOTE: This analyzer has been updated to use PostgreSQL. It requires
POSTGRES_URL to be configured. For quick analysis during simulations,
use the SimulationStats class in simulate.py instead.
Usage:
python game_analyzer.py blunders [limit]
python game_analyzer.py recent [limit]
"""
import asyncio
import json
import sqlite3
import os
import sqlite3 # For legacy GameAnalyzer class (deprecated)
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
@@ -339,7 +349,12 @@ class DecisionEvaluator:
# =============================================================================
class GameAnalyzer:
"""Analyzes logged games for decision quality."""
"""Analyzes logged games for decision quality.
DEPRECATED: This class uses SQLite which has been replaced by PostgreSQL.
Use the CLI commands (blunders, recent) instead, or query the moves table
in PostgreSQL directly.
"""
def __init__(self, db_path: str = "games.db"):
self.db_path = Path(db_path)
@@ -579,59 +594,76 @@ def print_blunder_report(blunders: list[dict]):
# =============================================================================
# CLI Interface
# CLI Interface (PostgreSQL version)
# =============================================================================
if __name__ == "__main__":
async def run_cli():
"""Async CLI entry point."""
import sys
if len(sys.argv) < 2:
print("Usage:")
print(" python game_analyzer.py blunders [limit]")
print(" python game_analyzer.py game <game_id> <player_name>")
print(" python game_analyzer.py summary")
print(" python game_analyzer.py recent [limit]")
print("")
print("Requires POSTGRES_URL environment variable.")
sys.exit(1)
postgres_url = os.environ.get("POSTGRES_URL")
if not postgres_url:
print("Error: POSTGRES_URL environment variable not set.")
print("")
print("Set it like: export POSTGRES_URL=postgresql://golf:devpassword@localhost:5432/golf")
print("")
print("For simulation analysis without PostgreSQL, use:")
print(" python simulate.py 100 --preset baseline")
sys.exit(1)
from stores.event_store import EventStore
try:
event_store = await EventStore.create(postgres_url)
except Exception as e:
print(f"Error connecting to PostgreSQL: {e}")
sys.exit(1)
command = sys.argv[1]
try:
analyzer = GameAnalyzer()
except FileNotFoundError:
print("No games.db found. Play some games first!")
sys.exit(1)
if command == "blunders":
limit = int(sys.argv[2]) if len(sys.argv) > 2 else 20
blunders = await event_store.find_suspicious_discards(limit)
if command == "blunders":
limit = int(sys.argv[2]) if len(sys.argv) > 2 else 20
blunders = analyzer.find_blunders(limit)
print_blunder_report(blunders)
print(f"\n=== Suspicious Discards ({len(blunders)} found) ===\n")
for b in blunders:
print(f"Player: {b.get('player_name', 'Unknown')}")
print(f"Action: discard {b.get('card_rank', '?')}")
print(f"Room: {b.get('room_code', 'N/A')}")
print(f"Reason: {b.get('decision_reason', 'N/A')}")
print("-" * 40)
elif command == "game" and len(sys.argv) >= 4:
game_id = sys.argv[2]
player_name = sys.argv[3]
summary = analyzer.analyze_player_game(game_id, player_name)
print(generate_player_report(summary))
elif command == "summary":
# Quick summary of recent games
with sqlite3.connect("games.db") as conn:
conn.row_factory = sqlite3.Row
cursor = conn.execute("""
SELECT g.id, g.room_code, g.started_at, g.num_players,
COUNT(m.id) as move_count
FROM games g
LEFT JOIN moves m ON g.id = m.game_id
GROUP BY g.id
ORDER BY g.started_at DESC
LIMIT 10
""")
elif command == "recent":
limit = int(sys.argv[2]) if len(sys.argv) > 2 else 10
games = await event_store.get_recent_games_with_stats(limit)
print("\n=== Recent Games ===\n")
for row in cursor:
print(f"Game: {row['id'][:8]}... Room: {row['room_code']}")
print(f" Players: {row['num_players']}, Moves: {row['move_count']}")
print(f" Started: {row['started_at']}")
print()
for game in games:
game_id = str(game.get('id', ''))[:8]
room_code = game.get('room_code', 'N/A')
status = game.get('status', 'unknown')
moves = game.get('total_moves', 0)
print(f"{game_id}... | Room: {room_code} | Status: {status} | Moves: {moves}")
else:
print(f"Unknown command: {command}")
sys.exit(1)
else:
print(f"Unknown command: {command}")
print("Available: blunders, recent")
finally:
await event_store.close()
if __name__ == "__main__":
# Note: The detailed analysis (GameAnalyzer class) still uses the old SQLite
# schema format. For now, use the CLI commands above for PostgreSQL queries.
# Full migration of the analysis logic is TODO.
asyncio.run(run_cli())

56
server/main.py
View File

@@ -16,7 +16,7 @@ from config import config
from room import RoomManager, Room
from game import GamePhase, GameOptions
from ai import GolfAI, process_cpu_turn, get_all_profiles, reset_all_profiles, cleanup_room_profiles
from game_log import get_logger
from services.game_logger import GameLogger, get_logger, set_logger
# Import production components
from logging_config import setup_logging
@@ -142,6 +142,12 @@ async def lifespan(app: FastAPI):
set_stats_auth_service(_auth_service)
logger.info("Stats services initialized successfully")
# Initialize game logger (uses event_store for move logging)
logger.info("Initializing game logger...")
_game_logger = GameLogger(_event_store)
set_logger(_game_logger)
logger.info("Game logger initialized with PostgreSQL backend")
# Initialize replay service
logger.info("Initializing replay services...")
from services.replay_service import get_replay_service, set_replay_service
@@ -343,9 +349,8 @@ app.add_middleware(LazyRateLimitMiddleware)
room_manager = RoomManager()
# Initialize game logger database at startup
_game_logger = get_logger()
logger.info(f"Game analytics database initialized at: {_game_logger.db_path}")
# Game logger is initialized in lifespan after event_store is available
# The get_logger() function returns None until set_logger() is called
# =============================================================================
@@ -692,11 +697,12 @@ async def websocket_endpoint(websocket: WebSocket):
# Log game start for AI analysis
game_logger = get_logger()
current_room.game_log_id = game_logger.log_game_start(
room_code=current_room.code,
num_players=len(current_room.players),
options=options,
)
if game_logger:
current_room.game_log_id = game_logger.log_game_start(
room_code=current_room.code,
num_players=len(current_room.players),
options=options,
)
# CPU players do their initial flips immediately (if required)
if options.initial_flips > 0:
@@ -740,8 +746,8 @@ async def websocket_endpoint(websocket: WebSocket):
if card:
# Log draw decision for human player
if current_room.game_log_id:
game_logger = get_logger()
game_logger = get_logger()
if game_logger and current_room.game_log_id:
player = current_room.game.get_player(player_id)
if player:
reason = f"took {discard_before_draw.rank.value} from discard" if source == "discard" else "drew from deck"
@@ -779,8 +785,8 @@ async def websocket_endpoint(websocket: WebSocket):
if discarded:
# Log swap decision for human player
if current_room.game_log_id and drawn_card and player:
game_logger = get_logger()
game_logger = get_logger()
if game_logger and current_room.game_log_id and drawn_card and player:
old_rank = old_card.rank.value if old_card else "?"
game_logger.log_move(
game_id=current_room.game_log_id,
@@ -810,8 +816,8 @@ async def websocket_endpoint(websocket: WebSocket):
if current_room.game.discard_drawn(player_id):
# Log discard decision for human player
if current_room.game_log_id and drawn_card and player:
game_logger = get_logger()
game_logger = get_logger()
if game_logger and current_room.game_log_id and drawn_card and player:
game_logger.log_move(
game_id=current_room.game_log_id,
player=player,
@@ -864,8 +870,8 @@ async def websocket_endpoint(websocket: WebSocket):
current_room.game.flip_and_end_turn(player_id, position)
# Log flip decision for human player
if current_room.game_log_id and player and 0 <= position < len(player.cards):
game_logger = get_logger()
game_logger = get_logger()
if game_logger and current_room.game_log_id and player and 0 <= position < len(player.cards):
flipped_card = player.cards[position]
game_logger.log_move(
game_id=current_room.game_log_id,
@@ -889,8 +895,8 @@ async def websocket_endpoint(websocket: WebSocket):
player = current_room.game.get_player(player_id)
if current_room.game.skip_flip_and_end_turn(player_id):
# Log skip flip decision for human player
if current_room.game_log_id and player:
game_logger = get_logger()
game_logger = get_logger()
if game_logger and current_room.game_log_id and player:
game_logger.log_move(
game_id=current_room.game_log_id,
player=player,
@@ -913,8 +919,8 @@ async def websocket_endpoint(websocket: WebSocket):
player = current_room.game.get_player(player_id)
if current_room.game.flip_card_as_action(player_id, position):
# Log flip-as-action for human player
if current_room.game_log_id and player and 0 <= position < len(player.cards):
game_logger = get_logger()
game_logger = get_logger()
if game_logger and current_room.game_log_id and player and 0 <= position < len(player.cards):
flipped_card = player.cards[position]
game_logger.log_move(
game_id=current_room.game_log_id,
@@ -938,8 +944,8 @@ async def websocket_endpoint(websocket: WebSocket):
player = current_room.game.get_player(player_id)
if current_room.game.knock_early(player_id):
# Log knock early for human player
if current_room.game_log_id and player:
game_logger = get_logger()
game_logger = get_logger()
if game_logger and current_room.game_log_id and player:
face_down_count = sum(1 for c in player.cards if not c.face_up)
game_logger.log_move(
game_id=current_room.game_log_id,
@@ -1098,8 +1104,8 @@ async def broadcast_game_state(room: Room):
# Check for game over
elif room.game.phase == GamePhase.GAME_OVER:
# Log game end
if room.game_log_id:
game_logger = get_logger()
game_logger = get_logger()
if game_logger and room.game_log_id:
game_logger.log_game_end(room.game_log_id)
room.game_log_id = None # Clear to avoid duplicate logging

327
server/services/game_logger.py Normal file
View File

@@ -0,0 +1,327 @@
"""
PostgreSQL-backed game logging for AI decision analysis.
Replaces SQLite game_log.py with unified event store integration.
Provides sync-compatible interface for existing callers (main.py, ai.py).
Usage:
# Initialize in main.py lifespan
from services.game_logger import GameLogger, set_logger
game_logger = GameLogger(event_store)
set_logger(game_logger)
# Use in handlers
from services.game_logger import get_logger
logger = get_logger()
if logger:
logger.log_move(game_id, player, is_cpu=False, action="swap", ...)
"""
from typing import Optional, TYPE_CHECKING
import asyncio
import uuid
import logging
if TYPE_CHECKING:
from stores.event_store import EventStore
from game import Card, Player, Game, GameOptions
log = logging.getLogger(__name__)
class GameLogger:
"""
Logs game events and moves to PostgreSQL.
Provides sync wrappers for compatibility with existing callers.
Uses fire-and-forget async tasks to avoid blocking game logic.
"""
def __init__(self, event_store: "EventStore"):
"""
Initialize the game logger.
Args:
event_store: PostgreSQL event store instance.
"""
self.event_store = event_store
# -------------------------------------------------------------------------
# Game Lifecycle
# -------------------------------------------------------------------------
async def log_game_start_async(
self,
room_code: str,
num_players: int,
options: "GameOptions",
) -> str:
"""
Log game start, return game_id.
Creates a game record in games_v2 table.
Args:
room_code: Room code for the game.
num_players: Number of players.
options: Game options/house rules.
Returns:
Generated game UUID.
"""
game_id = str(uuid.uuid4())
options_dict = {
"flip_mode": options.flip_mode,
"initial_flips": options.initial_flips,
"knock_penalty": options.knock_penalty,
"use_jokers": options.use_jokers,
"lucky_swing": options.lucky_swing,
"super_kings": options.super_kings,
"ten_penny": options.ten_penny,
"knock_bonus": options.knock_bonus,
"underdog_bonus": options.underdog_bonus,
"tied_shame": options.tied_shame,
"blackjack": options.blackjack,
"eagle_eye": options.eagle_eye,
"negative_pairs_keep_value": getattr(options, "negative_pairs_keep_value", False),
"four_of_a_kind": getattr(options, "four_of_a_kind", False),
"wolfpack": getattr(options, "wolfpack", False),
}
try:
await self.event_store.create_game(
game_id=game_id,
room_code=room_code,
host_id="system",
options=options_dict,
)
log.debug(f"Logged game start: {game_id} room={room_code}")
except Exception as e:
log.error(f"Failed to log game start: {e}")
return game_id
def log_game_start(
self,
room_code: str,
num_players: int,
options: "GameOptions",
) -> str:
"""
Sync wrapper for log_game_start_async.
In async context: fires task and returns generated ID immediately.
In sync context: runs synchronously.
"""
game_id = str(uuid.uuid4())
try:
loop = asyncio.get_running_loop()
# Already in async context - fire task, return ID immediately
asyncio.create_task(self._log_game_start_with_id(game_id, room_code, num_players, options))
return game_id
except RuntimeError:
# Not in async context - run synchronously
return asyncio.run(self.log_game_start_async(room_code, num_players, options))
async def _log_game_start_with_id(
self,
game_id: str,
room_code: str,
num_players: int,
options: "GameOptions",
) -> None:
"""Helper to log game start with pre-generated ID."""
options_dict = {
"flip_mode": options.flip_mode,
"initial_flips": options.initial_flips,
"knock_penalty": options.knock_penalty,
"use_jokers": options.use_jokers,
"lucky_swing": options.lucky_swing,
"super_kings": options.super_kings,
"ten_penny": options.ten_penny,
"knock_bonus": options.knock_bonus,
"underdog_bonus": options.underdog_bonus,
"tied_shame": options.tied_shame,
"blackjack": options.blackjack,
"eagle_eye": options.eagle_eye,
"negative_pairs_keep_value": getattr(options, "negative_pairs_keep_value", False),
"four_of_a_kind": getattr(options, "four_of_a_kind", False),
"wolfpack": getattr(options, "wolfpack", False),
}
try:
await self.event_store.create_game(
game_id=game_id,
room_code=room_code,
host_id="system",
options=options_dict,
)
log.debug(f"Logged game start: {game_id} room={room_code}")
except Exception as e:
log.error(f"Failed to log game start: {e}")
async def log_game_end_async(self, game_id: str) -> None:
"""
Mark game as ended.
Args:
game_id: Game UUID.
"""
try:
await self.event_store.update_game_completed(game_id)
log.debug(f"Logged game end: {game_id}")
except Exception as e:
log.error(f"Failed to log game end: {e}")
def log_game_end(self, game_id: str) -> None:
"""
Sync wrapper for log_game_end_async.
Fires async task in async context, skips in sync context.
"""
if not game_id:
return
try:
loop = asyncio.get_running_loop()
asyncio.create_task(self.log_game_end_async(game_id))
except RuntimeError:
# Not in async context - skip (simulations don't need this)
pass
# -------------------------------------------------------------------------
# Move Logging
# -------------------------------------------------------------------------
async def log_move_async(
self,
game_id: str,
player: "Player",
is_cpu: bool,
action: str,
card: Optional["Card"] = None,
position: Optional[int] = None,
game: Optional["Game"] = None,
decision_reason: Optional[str] = None,
) -> None:
"""
Log a move with AI context to PostgreSQL.
Args:
game_id: Game UUID.
player: Player who made the move.
is_cpu: Whether this is a CPU player.
action: Action type (draw_deck, take_discard, swap, discard, flip, etc.).
card: Card involved in the action.
position: Hand position (0-5) for swaps/flips.
game: Game instance for context capture.
decision_reason: AI reasoning for the decision.
"""
# Build AI context from game state
hand_state = None
discard_top = None
visible_opponents = None
if game:
# Serialize player's hand
hand_state = [
{"rank": c.rank.value, "suit": c.suit.value, "face_up": c.face_up}
for c in player.cards
]
# Serialize discard top
dt = game.discard_top()
if dt:
discard_top = {"rank": dt.rank.value, "suit": dt.suit.value}
# Serialize visible opponent cards
visible_opponents = {}
for p in game.players:
if p.id != player.id:
visible = [
{"rank": c.rank.value, "suit": c.suit.value}
for c in p.cards if c.face_up
]
visible_opponents[p.name] = visible
try:
await self.event_store.append_move(
game_id=game_id,
player_id=player.id,
player_name=player.name,
is_cpu=is_cpu,
action=action,
card_rank=card.rank.value if card else None,
card_suit=card.suit.value if card else None,
position=position,
hand_state=hand_state,
discard_top=discard_top,
visible_opponents=visible_opponents,
decision_reason=decision_reason,
)
except Exception as e:
log.error(f"Failed to log move: {e}")
def log_move(
self,
game_id: str,
player: "Player",
is_cpu: bool,
action: str,
card: Optional["Card"] = None,
position: Optional[int] = None,
game: Optional["Game"] = None,
decision_reason: Optional[str] = None,
) -> None:
"""
Sync wrapper for log_move_async.
Fires async task in async context. Does nothing if no game_id or not in async context.
"""
if not game_id:
return
try:
loop = asyncio.get_running_loop()
asyncio.create_task(
self.log_move_async(
game_id, player, is_cpu, action,
card=card, position=position, game=game, decision_reason=decision_reason
)
)
except RuntimeError:
# Not in async context - skip logging (simulations)
pass
# -------------------------------------------------------------------------
# Global Instance Management
# -------------------------------------------------------------------------
_game_logger: Optional[GameLogger] = None
def get_logger() -> Optional[GameLogger]:
"""
Get the global game logger instance.
Returns:
GameLogger if initialized, None otherwise.
"""
return _game_logger
def set_logger(logger: GameLogger) -> None:
"""
Set the global game logger instance.
Called during application startup in main.py lifespan.
Args:
logger: GameLogger instance to use globally.
"""
global _game_logger
_game_logger = logger
log.info("Game logger initialized with PostgreSQL backend")

255
server/stores/event_store.py
View File

@@ -62,6 +62,32 @@ CREATE TABLE IF NOT EXISTS games_v2 (
player_ids VARCHAR(50)[] DEFAULT '{}'
);
-- Moves table (denormalized for AI decision analysis)
-- Replaces SQLite game_log.py - provides efficient queries for move-level analysis
CREATE TABLE IF NOT EXISTS moves (
id BIGSERIAL PRIMARY KEY,
game_id UUID NOT NULL,
sequence_num INT NOT NULL,
timestamp TIMESTAMPTZ DEFAULT NOW(),
player_id VARCHAR(50) NOT NULL,
player_name VARCHAR(100),
is_cpu BOOLEAN DEFAULT FALSE,
-- Action details
action VARCHAR(30) NOT NULL, -- draw_deck, take_discard, swap, discard, flip, etc.
card_rank VARCHAR(5),
card_suit VARCHAR(10),
position INT,
-- AI context (JSONB for flexibility)
hand_state JSONB, -- Player's hand at decision time
discard_top JSONB, -- Top of discard pile
visible_opponents JSONB, -- Face-up cards of opponents
decision_reason TEXT, -- AI reasoning
UNIQUE(game_id, sequence_num)
);
-- Indexes for common queries
CREATE INDEX IF NOT EXISTS idx_events_game_seq ON events(game_id, sequence_num);
CREATE INDEX IF NOT EXISTS idx_events_type ON events(event_type);
@@ -72,6 +98,11 @@ CREATE INDEX IF NOT EXISTS idx_games_status ON games_v2(status);
CREATE INDEX IF NOT EXISTS idx_games_room ON games_v2(room_code) WHERE status = 'active';
CREATE INDEX IF NOT EXISTS idx_games_players ON games_v2 USING GIN(player_ids);
CREATE INDEX IF NOT EXISTS idx_games_completed ON games_v2(completed_at) WHERE status = 'completed';
CREATE INDEX IF NOT EXISTS idx_moves_game ON moves(game_id);
CREATE INDEX IF NOT EXISTS idx_moves_action ON moves(action);
CREATE INDEX IF NOT EXISTS idx_moves_is_cpu ON moves(is_cpu);
CREATE INDEX IF NOT EXISTS idx_moves_player ON moves(player_id);
"""
@@ -441,6 +472,230 @@ class EventStore:
)
return dict(row) if row else None
# -------------------------------------------------------------------------
# Move Logging (for AI decision analysis)
# -------------------------------------------------------------------------
async def append_move(
self,
game_id: str,
player_id: str,
player_name: str,
is_cpu: bool,
action: str,
card_rank: Optional[str] = None,
card_suit: Optional[str] = None,
position: Optional[int] = None,
hand_state: Optional[list] = None,
discard_top: Optional[dict] = None,
visible_opponents: Optional[dict] = None,
decision_reason: Optional[str] = None,
) -> int:
"""
Append a move to the moves table for AI decision analysis.
Args:
game_id: Game UUID.
player_id: Player who made the move.
player_name: Display name of the player.
is_cpu: Whether this is a CPU player.
action: Action type (draw_deck, take_discard, swap, discard, flip, etc.).
card_rank: Rank of the card involved.
card_suit: Suit of the card involved.
position: Hand position (0-5) for swaps/flips.
hand_state: Player's hand at decision time.
discard_top: Top of discard pile at decision time.
visible_opponents: Face-up cards of opponents.
decision_reason: AI reasoning for the decision.
Returns:
The database ID of the inserted move.
"""
async with self.pool.acquire() as conn:
# Get next sequence number for this game
seq_row = await conn.fetchrow(
"SELECT COALESCE(MAX(sequence_num), 0) + 1 as seq FROM moves WHERE game_id = $1",
game_id,
)
sequence_num = seq_row["seq"]
row = await conn.fetchrow(
"""
INSERT INTO moves (
game_id, sequence_num, player_id, player_name, is_cpu,
action, card_rank, card_suit, position,
hand_state, discard_top, visible_opponents, decision_reason
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
RETURNING id
""",
game_id,
sequence_num,
player_id,
player_name,
is_cpu,
action,
card_rank,
card_suit,
position,
json.dumps(hand_state) if hand_state else None,
json.dumps(discard_top) if discard_top else None,
json.dumps(visible_opponents) if visible_opponents else None,
decision_reason,
)
return row["id"]
async def get_moves(
self,
game_id: str,
player_id: Optional[str] = None,
is_cpu: Optional[bool] = None,
action: Optional[str] = None,
limit: int = 100,
) -> list[dict]:
"""
Get moves for a game with optional filters.
Args:
game_id: Game UUID.
player_id: Filter by player ID.
is_cpu: Filter by CPU status.
action: Filter by action type.
limit: Maximum number of moves to return.
Returns:
List of move dicts.
"""
conditions = ["game_id = $1"]
params = [game_id]
param_idx = 2
if player_id is not None:
conditions.append(f"player_id = ${param_idx}")
params.append(player_id)
param_idx += 1
if is_cpu is not None:
conditions.append(f"is_cpu = ${param_idx}")
params.append(is_cpu)
param_idx += 1
if action is not None:
conditions.append(f"action = ${param_idx}")
params.append(action)
param_idx += 1
params.append(limit)
where_clause = " AND ".join(conditions)
async with self.pool.acquire() as conn:
rows = await conn.fetch(
f"""
SELECT id, game_id, sequence_num, timestamp, player_id, player_name, is_cpu,
action, card_rank, card_suit, position,
hand_state, discard_top, visible_opponents, decision_reason
FROM moves
WHERE {where_clause}
ORDER BY sequence_num
LIMIT ${param_idx}
""",
*params,
)
return [self._row_to_move(row) for row in rows]
async def get_player_decisions(
self,
game_id: str,
player_name: str,
) -> list[dict]:
"""
Get all decisions made by a specific player in a game.
Args:
game_id: Game UUID.
player_name: Display name of the player.
Returns:
List of move dicts.
"""
async with self.pool.acquire() as conn:
rows = await conn.fetch(
"""
SELECT id, game_id, sequence_num, timestamp, player_id, player_name, is_cpu,
action, card_rank, card_suit, position,
hand_state, discard_top, visible_opponents, decision_reason
FROM moves
WHERE game_id = $1 AND player_name = $2
ORDER BY sequence_num
""",
game_id,
player_name,
)
return [self._row_to_move(row) for row in rows]
async def find_suspicious_discards(self, limit: int = 50) -> list[dict]:
"""
Find cases where CPU discarded good cards (Ace, 2, King).
Used for AI decision quality analysis.
Args:
limit: Maximum number of results.
Returns:
List of suspicious move dicts with game room_code.
"""
async with self.pool.acquire() as conn:
rows = await conn.fetch(
"""
SELECT m.*, g.room_code
FROM moves m
JOIN games_v2 g ON m.game_id = g.id
WHERE m.action = 'discard'
AND m.card_rank IN ('A', '2', 'K')
AND m.is_cpu = TRUE
ORDER BY m.timestamp DESC
LIMIT $1
""",
limit,
)
return [self._row_to_move(row) for row in rows]
async def get_recent_games_with_stats(self, limit: int = 10) -> list[dict]:
"""
Get recent games with move counts.
Args:
limit: Maximum number of games.
Returns:
List of game dicts with total_moves count.
"""
async with self.pool.acquire() as conn:
rows = await conn.fetch(
"""
SELECT g.*, COUNT(m.id) as total_moves
FROM games_v2 g
LEFT JOIN moves m ON g.id = m.game_id
GROUP BY g.id
ORDER BY g.created_at DESC
LIMIT $1
""",
limit,
)
return [dict(row) for row in rows]
def _row_to_move(self, row: asyncpg.Record) -> dict:
"""Convert a database row to a move dict."""
result = dict(row)
# Parse JSON fields
if result.get("hand_state"):
result["hand_state"] = json.loads(result["hand_state"])
if result.get("discard_top"):
result["discard_top"] = json.loads(result["discard_top"])
if result.get("visible_opponents"):
result["visible_opponents"] = json.loads(result["visible_opponents"])
return result
# -------------------------------------------------------------------------
# Helpers
# -------------------------------------------------------------------------