diff --git a/bots/psi_bot.py b/bots/psi_bot.py new file mode 100644 index 0000000..e17b7ec --- /dev/null +++ b/bots/psi_bot.py @@ -0,0 +1,263 @@ +#!/usr/bin/env python3 +""" +PsiBot - Private Set Intersection Bot +""" + +import hashlib +import os +import random +import re +from typing import Dict, Any, List + +class PsiBot: + """ + A bot for Private Set Intersection Verification games. + """ + + def __init__(self, ctx, params): + self.ctx = ctx + self.params = params + + # Game parameters + self.range = params.get('range', [1, 100]) + self.timeout_s = params.get('timeout_s', 600) + self.intersection_size = params.get('intersection_size', 3) + self.set_size = params.get('set_size', 10) + + # Initialize state + state = self.ctx.get_state() + if not state: + # Ensure at least intersection_size numbers in the intersection + intersection = random.sample(range(self.range[0], self.range[1] + 1), self.intersection_size) + + # Create set1: include all intersection numbers, then fill to set_size + remaining1 = [x for x in range(self.range[0], self.range[1] + 1) if x not in intersection] + additional1 = random.sample(remaining1, self.set_size - len(intersection)) + set1 = sorted(intersection + additional1) + + # Create set2: include all intersection numbers, then fill to set_size + remaining2 = [x for x in range(self.range[0], self.range[1] + 1) if x not in intersection] + additional2 = random.sample(remaining2, self.set_size - len(intersection)) + set2 = sorted(intersection + additional2) + + # Target is the intersection (should be at least intersection_size numbers) + target = set(set1) & set(set2) + + self.target = params.get('target', target) # intersection of set 1 and 2 + self.set1 = params.get('set1', set1) + self.set2 = params.get('set2', set2) + + self.players = [] + self.game_started = False + self.game_ended = False + self.guesses = dict() + self.sets = dict() + + # Save initial state + self._save_state() + else: + # Load existing state + self._load_state(state) + + def _save_state(self): + """Save current state.""" + state = { + 'set1': self.set1, + 'set2': self.set2, + 'sets': self.sets, + 'target': self.target, + 'players': self.players, + 'guesses': self.guesses, + 'game_started': self.game_started, + 'game_ended': self.game_ended, + 'range': self.range + } + self.ctx.set_state(state) + + def _load_state(self, state: Dict[str, Any]): + """Load state from storage.""" + self.set1 = state.get('set1') + self.set2 = state.get('set2') + self.sets = state.get('sets', dict()) + self.target = state.get('target') + self.players = state.get('players', []) + self.game_started = state.get('game_started', False) + self.game_ended = state.get('game_ended', False) + self.guesses = state.get('guesses', dict()) + self.range = state.get('range', [1, 100]) + + def _extract_numbers(self, text): + # Find all numbers in the string + if not text: + return None + numbers = re.findall(r'\d+', text) + if numbers: + return [int(n) for n in numbers] # Get the last number (or numbers[0] for first) + return None + + def on_init(self): + """Initialize the game when bot is attached.""" + # Post game prompt + self.ctx.post("bot", { + "type": "prompt", + "text": f"Private Set Intersection Verification game!\n Guess the intersection (with range {self.range[0]} and {self.range[1]}) between your set and your opponent's set!", + "range": self.range + }) + + # Post initial state + self._post_public_state() + + def on_join(self, slot_id: str): + """Handle a player joining the channel.""" + if slot_id not in self.players and not self.game_ended: + self.players.append(slot_id) + + self._save_state() + + self.ctx.post("bot", { + "type": "player_joined", + "player": slot_id, + "player_count": len(self.players) + }) + + # Start game when we have 2 players + if len(self.players) >= 2 and not self.game_started: + self._start_game() + + def on_message(self, msg: Dict[str, Any]): + """Handle incoming messages.""" + if msg.get('kind') != 'user': + return + + if self.game_ended: + return + + body = msg.get('body', {}) + sender = msg.get('sender') + + # Check if message text contains "guess" or process all user messages + text = body.get('text', '') + if text: + self._handle_guess_move(sender, body) + + def on_private_message(self, slot_id: str, msg: Dict[str, Any]): + """Handle private messages from players (via dm_bot).""" + text = msg.get('text', '').strip() + + self.ctx.dm(slot_id, "private", { + "type": "error", + "text": f"Unknown command: {text}. Try !help" + }) + + def _start_game(self): + """Start the game when enough players have joined.""" + self.game_started = True + + self._save_state() + + self.ctx.post("bot", { + "type": "game_start", + "players": self.players, + }) + + self.sets[self.players[0]] = self.set1 + + msgId = self.ctx.dm("s1", "private", { + "text": f"Your set: {self.set1}" + }) + + self.sets[self.players[1]] = self.set2 + msgId = self.ctx.dm("s2", "private", { + "text": f"Your set: {self.set2}" + }) + + def _handle_guess_move(self, sender: str, body: Dict[str, Any]): + """Handle a guess move from a player.""" + if not self.game_started: + self.ctx.post("control", { + "type": "violation", + "reason": "GAME_NOT_STARTED", + "details": "Game hasn't started yet" + }) + return + + if sender in self.guesses and self.guesses[sender] is not None: + self.ctx.post("control", { + "type": "violation", + "reason": "GUESS_ALREADY_MADE", + "details": "You have already made a guess" + }) + return + + # Get guess value + guess_text = body.get('text', '') + + if not isinstance(guess_text, list): + guess = self._extract_numbers(guess_text) + else: + guess = guess_text + + if guess is None: + self.ctx.post("control", { + "type": "violation", + "reason": "BAD_MOVE", + "details": "Missing guess value" + }) + return + + # Check if intersection is correct + self.guesses[sender] = guess + result = "correct" if (set(guess) & set(self.target)) == set(self.target) else "incorrect" + self.ctx.post("bot", { + "type": "judge", + "result": result, + "player": sender, + "guess": guess, + }) + + if len(self.guesses) == len(self.players): + self._end_game() + + self._save_state() + + + def _end_game(self, winner: str = None, reason: str = "correct"): + """End the game and reveal the target.""" + self.game_ended = True + self._save_state() + + + self.ctx.post("control", { + "type": "bot:reveal", + "target": self.target, + }) + + # Announce game end + self.ctx.post("bot", { + "type": "game_end", + "winner": winner, + "reason": reason, + "target": self.target, + "players": self.players + }) + + self.ctx.post("system", { + "type": "end" + }) + + def _post_public_state(self): + """Post the current public game state.""" + public_state = { + "range": self.range, + "players": self.players, + "game_started": self.game_started, + "game_ended": self.game_ended, + } + + self.ctx.post("control", { + "type": "bot:state", + "public_state": public_state, + "state_version": self.ctx.bot_manager.get_bot_state_version( + self.ctx.channel_id, self.ctx.bot_id + ) + }) \ No newline at end of file