""" Miku UNO Player - Browser automation and AI strategy Handles joining games via Playwright and making LLM-powered decisions """ import asyncio import json import requests from typing import Optional, Dict, Any, List from playwright.async_api import async_playwright, Page, Browser from utils.llm import query_llama from utils.logger import get_logger import globals logger = get_logger('uno') # Configuration # Use host.docker.internal to reach host machine from inside container # Fallback to 192.168.1.2 if host.docker.internal doesn't work UNO_SERVER_URL = "http://192.168.1.2:5000" UNO_CLIENT_URL = "http://192.168.1.2:3002" POLL_INTERVAL = 2 # seconds between checking for turn class MikuUnoPlayer: """Miku's UNO player with browser automation and AI strategy""" def __init__(self, room_code: str, discord_channel, cleanup_callback=None): self.room_code = room_code self.discord_channel = discord_channel self.browser: Optional[Browser] = None self.page: Optional[Page] = None self.playwright = None self.is_playing = False self.game_started = False self.last_card_count = 7 self.last_turn_processed = None # Track last turn we processed to avoid duplicate moves self.cleanup_callback = cleanup_callback # Callback to remove from active_uno_games async def join_game(self) -> bool: """Join an existing UNO game as Player 2 via browser automation""" try: logger.info(f"[UNO] Joining game: {self.room_code}") # Launch browser self.playwright = await async_playwright().start() self.browser = await self.playwright.chromium.launch(headless=True) self.page = await self.browser.new_page() # Enable console logging to debug (filter out verbose game state logs) def log_console(msg): text = msg.text # Skip verbose game state logs but keep important ones if "FULL GAME STATE" in text or "JSON for Bot API" in text: return logger.debug(f"[Browser] {text[:150]}...") # Truncate to 150 chars self.page.on("console", log_console) self.page.on("pageerror", lambda err: logger.error(f"[Browser Error] {err}")) # Navigate to homepage logger.info(f"[UNO] Navigating to: {UNO_CLIENT_URL}") await self.page.goto(UNO_CLIENT_URL) await asyncio.sleep(2) # Find and fill the room code input try: # Look for input field and fill with room code input_field = await self.page.query_selector('input[type="text"]') if not input_field: logger.error("[UNO] Could not find input field") return False await input_field.fill(self.room_code) logger.info(f"[UNO] Filled room code: {self.room_code}") await asyncio.sleep(0.5) # Click the "Join Room" button buttons = await self.page.query_selector_all('button') join_clicked = False for button in buttons: text = await button.inner_text() if 'JOIN' in text.upper(): logger.info(f"[UNO] Found join button, clicking...") await button.click() join_clicked = True break if not join_clicked: logger.error("[UNO] Could not find join button") return False # Wait for navigation to /play logger.info("[UNO] Waiting for navigation to game page...") await asyncio.sleep(3) # Verify we're on the play page current_url = self.page.url logger.info(f"[UNO] Current URL after click: {current_url}") if '/play' not in current_url: logger.error(f"[UNO] Did not navigate to game page, still on: {current_url}") return False # Wait longer for Socket.IO connection and game setup logger.info("[UNO] Waiting for Socket.IO connection and game initialization...") await asyncio.sleep(5) # Take a screenshot for debugging try: screenshot_path = f"/app/memory/uno_debug_{self.room_code}.png" await self.page.screenshot(path=screenshot_path) logger.info(f"[UNO] Screenshot saved to {screenshot_path}") except Exception as e: logger.error(f"[UNO] Could not save screenshot: {e}") # Get page content for debugging content = await self.page.content() logger.debug(f"[UNO] Page content length: {len(content)} chars") # Check current URL current_url = self.page.url logger.info(f"[UNO] Current URL: {current_url}") # Check if we're actually in the game by looking for game elements game_element = await self.page.query_selector('.game-screen, .player-deck, .uno-card') if game_element: logger.info(f"[UNO] Successfully joined room {self.room_code} as Player 2 - game elements found") else: logger.warning(f"[UNO] Joined room {self.room_code} but game elements not found yet") return True except Exception as e: logger.error(f"[UNO] Error during join process: {e}", exc_info=True) return False except Exception as e: logger.error(f"[UNO] Error joining game: {e}", exc_info=True) await self.cleanup() return False async def play_game(self): """Main game loop - poll for turns and make moves""" self.is_playing = True logger.info(f"Starting game loop for room {self.room_code}") try: while self.is_playing: # Get current game state game_state = await self.get_game_state() if not game_state: await asyncio.sleep(POLL_INTERVAL) continue # Check if game started if not self.game_started and game_state['game'].get('currentTurn'): self.game_started = True await self.discord_channel.send("🎮 Game started! Let's do this! 🎤✨") # Check if game over if is_over: # Game has ended winner = game_state.get('game', {}).get('winner') if winner == 2: await self.discord_channel.send(f"🎉 **I WON!** That was too easy! GG! 🎤✨") else: await self.discord_channel.send(f"😤 You got lucky this time... I'll win next time! 💢") logger.info(f"[UNO] Game over in room {self.room_code}. Winner: Player {winner}") # Call cleanup callback to remove from active_uno_games if self.cleanup_callback: await self.cleanup_callback(self.room_code) break # Check if it's Miku's turn if game_state['game']['currentTurn'] == 'Player 2': # Create a unique turn identifier combining multiple factors # This handles cases where bot's turn comes twice in a row (after Skip, etc) turn_id = f"{game_state['game']['turnNumber']}_{game_state['player2']['cardCount']}_{len(game_state['currentCard'])}" if turn_id != self.last_turn_processed: logger.info("It's Miku's turn!") self.last_turn_processed = turn_id await self.make_move(game_state) else: # Same turn state, but check if it's been more than 5 seconds (might be stuck) # For now just skip to avoid duplicate moves pass # Wait before next check await asyncio.sleep(POLL_INTERVAL) except Exception as e: logger.error(f"Error in game loop: {e}", exc_info=True) await self.discord_channel.send(f"❌ Oops! Something went wrong in the game: {str(e)}") finally: await self.cleanup() async def get_game_state(self) -> Optional[Dict[str, Any]]: """Get current game state from server""" try: response = requests.get( f"{UNO_SERVER_URL}/api/game/{self.room_code}/state", timeout=5 ) if response.status_code == 200: data = response.json() if data.get('success'): return data['gameState'] return None except Exception as e: logger.error(f"Error getting game state: {e}") return None async def make_move(self, game_state: Dict[str, Any]): """Use LLM to decide and execute a move""" try: # Check if bot can play any cards can_play = len(game_state['player2']['playableCards']) > 0 # Get Miku's decision from LLM action = await self.get_miku_decision(game_state) if not action: logger.warning("No action from LLM, drawing card") action = {"action": "draw"} logger.info(f"🎮 Miku's decision: {json.dumps(action)}") # Send trash talk before move await self.send_trash_talk(game_state, action) # Execute the action success = await self.send_action(action) if success: # Check for UNO situation current_cards = game_state['player2']['cardCount'] if action['action'] == 'play' and current_cards == 2: await self.discord_channel.send("🔥 **UNO!!** One more card and I win! 🎤") logger.info(f"✅ Action executed successfully") # Reset turn tracker after successful action so we can process next turn self.last_turn_processed = None # Brief wait for socket sync (now that useEffect dependencies are fixed, this can be much shorter) await asyncio.sleep(0.5) else: logger.warning(f"⚠️ Action failed (invalid move), will try different action next turn") # Don't reset turn tracker - let it skip this turn state # The game state will update and we'll try again with updated info except Exception as e: logger.error(f"Error making move: {e}", exc_info=True) async def get_miku_decision(self, game_state: Dict[str, Any]) -> Optional[Dict[str, Any]]: """Use Miku's LLM to decide the best move""" try: # Build strategic prompt prompt = self.build_strategy_prompt(game_state) # Query LLM with required parameters (query_llama is already async) guild_id = self.discord_channel.guild.id if hasattr(self.discord_channel, 'guild') and self.discord_channel.guild else None response = await query_llama( user_prompt=prompt, user_id="uno_bot", guild_id=guild_id, response_type="uno_strategy", author_name="Miku UNO Bot" ) # Extract JSON from response action = self.parse_llm_response(response) return action except Exception as e: logger.error(f"Error getting LLM decision: {e}", exc_info=True) return None def build_strategy_prompt(self, game_state: Dict[str, Any]) -> str: """Build a prompt for Miku to make strategic decisions""" current_card = game_state['currentCard'] my_cards = game_state['player2']['cards'] playable_cards = game_state['player2']['playableCards'] opponent_cards = game_state['player1']['cardCount'] my_card_count = game_state['player2']['cardCount'] # Build card list my_cards_str = ", ".join([f"{c['displayName']} ({c['code']})" for c in my_cards]) playable_str = ", ".join([f"{c['displayName']} ({c['code']})" for c in playable_cards]) prompt = f"""You are Hatsune Miku, the cheerful virtual idol! You're playing UNO and it's your turn. GAME STATE: - Current card on table: {current_card['displayName']} ({current_card['code']}) - Your cards ({my_card_count}): {my_cards_str} - Playable cards: {playable_str if playable_str else "NONE - must draw"} - Opponent has {opponent_cards} cards STRATEGY: - If opponent has 1-2 cards, play attack cards (Draw 2, Draw 4, Skip) to stop them! - Play Draw 2/Draw 4 aggressively to disrupt opponent - Save Wild cards for when you have no other options - When playing Wild cards, choose the color you have most of - Call UNO when you have 2 cards and are about to play one YOUR TASK: Respond with ONLY a valid JSON action. No explanation, just the JSON. ACTION FORMAT: 1. To play a card: {{"action": "play", "card": "CODE"}} 2. To play a Wild: {{"action": "play", "card": "W", "color": "R/G/B/Y"}} 3. To play Wild Draw 4: {{"action": "play", "card": "D4W", "color": "R/G/B/Y"}} 4. To draw a card: {{"action": "draw"}} 5. To play + call UNO: {{"action": "play", "card": "CODE", "callUno": true}} VALID CARD CODES: {playable_str if playable_str else "No playable cards - must draw"} Choose wisely! What's your move? RESPONSE (JSON only):""" return prompt def parse_llm_response(self, response: str) -> Optional[Dict[str, Any]]: """Parse LLM response to extract JSON action""" try: # Try to find JSON in response import re # Look for JSON object json_match = re.search(r'\{[^}]+\}', response) if json_match: json_str = json_match.group(0) action = json.loads(json_str) # Validate action format if 'action' in action: return action logger.warning(f"Could not parse LLM response: {response}") return None except Exception as e: logger.error(f"Error parsing LLM response: {e}") return None async def send_trash_talk(self, game_state: Dict[str, Any], action: Dict[str, Any]): """Send personality-driven trash talk before moves""" try: opponent_cards = game_state['player1']['cardCount'] my_cards = game_state['player2']['cardCount'] # Special trash talk for different situations if action['action'] == 'play': card_code = action.get('card', '') if 'D4W' in card_code: messages = [ "Wild Draw 4! Take that! 😈", "Draw 4 cards! Ahahaha! 🌈💥", "This is what happens when you challenge me! +4! 💫" ] elif 'D2' in card_code: messages = [ "Draw 2! Better luck next time~ 🎵", "Here, have some extra cards! 📥", "+2 for you! Hope you like drawing! 😊" ] elif 'skip' in card_code: messages = [ "Skip! You lose your turn! ⏭️", "Not so fast! Skipped! 🎤", "Your turn? Nope! Skipped! ✨" ] elif 'W' in card_code: color_names = {'R': 'Red', 'G': 'Green', 'B': 'Blue', 'Y': 'Yellow'} chosen_color = color_names.get(action.get('color', 'R'), 'Red') messages = [ f"Wild card! Changing to {chosen_color}! 🌈", f"Let's go {chosen_color}! Time to mix things up! 💫" ] else: if my_cards == 2: messages = ["Almost there... one more card! 🎯"] elif opponent_cards <= 2: messages = ["Not gonna let you win! 😤", "I see you getting close... not on my watch! 💢"] else: messages = ["Hehe, perfect card! ✨", "This is too easy~ 🎤", "Watch and learn! 🎶"] import random await self.discord_channel.send(random.choice(messages)) except Exception as e: logger.error(f"Error sending trash talk: {e}") async def send_action(self, action: Dict[str, Any]) -> bool: """Send action to game server""" try: response = requests.post( f"{UNO_SERVER_URL}/api/game/{self.room_code}/action", json=action, headers={'Content-Type': 'application/json'}, timeout=5 ) if response.status_code == 200: data = response.json() return data.get('success', False) return False except Exception as e: logger.error(f"Error sending action: {e}") return False def is_game_active(self) -> bool: """Check if game is currently active""" return self.is_playing async def quit_game(self): """Quit the game and cleanup""" self.is_playing = False await self.cleanup() async def cleanup(self): """Cleanup browser resources""" try: if self.page: await self.page.close() if self.browser: await self.browser.close() if self.playwright: await self.playwright.stop() logger.info(f"Cleaned up resources for room {self.room_code}") except Exception as e: logger.error(f"Error during cleanup: {e}")