from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey, Boolean from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, relationship from datetime import datetime from contextlib import contextmanager import threading Base = declarative_base() # Thread-safe engine creation engine = create_engine( 'sqlite:///codenames_games.db', connect_args={'check_same_thread': False}, pool_pre_ping=True, pool_size=10, max_overflow=20 ) # Thread-local session SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False) session_lock = threading.Lock() @contextmanager def get_db_session(): """Thread-safe database session context manager""" session = SessionLocal() try: yield session session.commit() except Exception as e: session.rollback() raise e finally: session.close() class Game(Base): __tablename__ = 'games' id = Column(Integer, primary_key=True, autoincrement=True) timestamp = Column(DateTime, default=datetime.utcnow, nullable=False) winner_team = Column(String(10), nullable=False) # 'red' or 'blue' red_score = Column(Integer, nullable=False) # remaining words blue_score = Column(Integer, nullable=False) # remaining words reason = Column(String(50), nullable=True) # None or 'killer' red_team_preset = Column(String(50), nullable=True) # 'openai', 'google', etc. blue_team_preset = Column(String(50), nullable=True) # Relationships players = relationship("GamePlayer", back_populates="game", cascade="all, delete-orphan") class GamePlayer(Base): __tablename__ = 'game_players' id = Column(Integer, primary_key=True, autoincrement=True) game_id = Column(Integer, ForeignKey('games.id'), nullable=False) player_name = Column(String(100), nullable=False) team = Column(String(10), nullable=False) # 'red' or 'blue' role = Column(String(20), nullable=False) # 'boss', 'captain', 'player' model_name = Column(String(100), nullable=False) is_human = Column(Boolean, default=False) # Relationship game = relationship("Game", back_populates="players") def init_database(): """Initialize the database tables""" Base.metadata.create_all(bind=engine) def save_game_to_db(game, winner_and_score): """ Save a completed game to the database Args: game: Game object with players and board info winner_and_score: tuple (winner_team, scores, reason) """ winner_team = winner_and_score[0] scores = winner_and_score[1] reason = winner_and_score[2] if len(winner_and_score) > 2 else None with get_db_session() as session: # Create game record new_game = Game( winner_team=winner_team, red_score=scores[0], blue_score=scores[1], reason=reason, red_team_preset=game.red_team_choice, blue_team_preset=game.blue_team_choice ) session.add(new_game) session.flush() # Get the game ID # Create player records for player in game.players: is_human = player.model_name.lower() == "human brain" game_player = GamePlayer( game_id=new_game.id, player_name=player.name, team=player.team, role=player.role, model_name=player.model_name, is_human=is_human ) session.add(game_player) session.commit() return new_game.id def get_stats(): """ Retrieve comprehensive statistics from the database Returns: dict: Contains total_games, game_history, model_stats, provider_stats """ with get_db_session() as session: # Total games total_games = session.query(Game).count() # Game history games = session.query(Game).order_by(Game.timestamp.desc()).all() game_history = [] for game in games: game_info = { 'id': game.id, 'timestamp': game.timestamp.strftime('%Y-%m-%d %H:%M:%S'), 'winner': game.winner_team, 'red_score': game.red_score, 'blue_score': game.blue_score, 'reason': game.reason, 'red_preset': game.red_team_preset, 'blue_preset': game.blue_team_preset, 'players': [] } for player in game.players: game_info['players'].append({ 'name': player.player_name, 'team': player.team, 'role': player.role, 'model': player.model_name, 'is_human': player.is_human }) game_history.append(game_info) # Model stats (individual) model_stats = {} all_players = session.query(GamePlayer).all() for player in all_players: model = player.model_name if model not in model_stats: model_stats[model] = {'wins': 0, 'losses': 0, 'games': 0} model_stats[model]['games'] += 1 if player.team == player.game.winner_team: model_stats[model]['wins'] += 1 else: model_stats[model]['losses'] += 1 # Provider stats (grouped) provider_stats = { 'openai': {'wins': 0, 'losses': 0, 'games': 0}, 'google': {'wins': 0, 'losses': 0, 'games': 0}, 'anthropic': {'wins': 0, 'losses': 0, 'games': 0}, 'opensource': {'wins': 0, 'losses': 0, 'games': 0}, 'human': {'wins': 0, 'losses': 0, 'games': 0} } # Map models to providers model_to_provider = { 'gpt-5': 'openai', 'gpt-5-mini': 'openai', 'gpt-5-nano': 'openai', 'gpt-4.1-nano': 'openai', 'gemini-2.5-pro': 'google', 'gemini-2.5-flash': 'google', 'gemini-2.0-flash-001': 'google', 'gemini-2.0-flash-lite-001': 'google', 'claude-sonnet-4-5-20250929':'anthropic', 'claude-3-7-sonnet-20250219': 'anthropic', 'claude-3-5-haiku-20241022': 'anthropic', 'claude-3-haiku-20240307': 'anthropic', 'deepseek-ai/DeepSeek-V3.1': 'opensource', 'deepseek-ai/DeepSeek-R1': 'opensource', 'moonshotai/Kimi-K2-Thinking': 'opensource', 'Qwen/Qwen3-235B-A22B-Instruct-2507': 'opensource', 'openai/gpt-oss-120b': 'opensource', 'openai/gpt-oss-20b': 'opensource', 'human brain': 'human' } for player in all_players: provider = model_to_provider.get(player.model_name.lower(), 'unknown') if provider in provider_stats: provider_stats[provider]['games'] += 1 if player.team == player.game.winner_team: provider_stats[provider]['wins'] += 1 else: provider_stats[provider]['losses'] += 1 return { 'total_games': total_games, 'game_history': game_history, 'model_stats': model_stats, 'provider_stats': provider_stats } # Initialize database on import init_database()