#!/usr/bin/env python3 """ 斗地主 Headless 浏览器测试 V2 智能轮询版本 - 自动处理人类玩家回合 """ import asyncio import sys import os sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from playwright.async_api import async_playwright # 服务器地址 SERVER_URL = "http://129.211.173.87:8900" async def run_headless_test(): """运行 Headless 浏览器测试""" print("=" * 70) print(" 🎮 斗地主 Headless 浏览器测试 V2") print("=" * 70) print(f"\n🌐 服务器: {SERVER_URL}") print("\n👥 三方对战:") print(" 1️⃣ Headless人类 (浏览器自动化)") print(" 2️⃣ 规则AI (本地算法)") print(" 3️⃣ DeepSeek AI (大模型API)") async with async_playwright() as p: browser = await p.chromium.launch(headless=True) page = await browser.new_page(viewport={'width': 1280, 'height': 900}) try: # 访问测试客户端 await page.goto(f"{SERVER_URL}/test_client.html") await asyncio.sleep(1) # 连接到服务器 print("\n📡 连接到游戏服务器...") await page.fill('input[placeholder*="服务器地址"]', SERVER_URL) await page.click('button:has-text("连接")') await asyncio.sleep(1) # 设置玩家名称 await page.fill('input[placeholder*="玩家名称"]', 'Headless玩家') # 创建房间（带AI） print("🏠 创建房间（含2个AI）...") await page.evaluate(""" () => { socket.emit('create_room', { game_type: 'doudizhu', player_name: 'Headless玩家', ai_config: {ai_count: 2, ai_types: ['rule', 'deepseek']} }); } """) await asyncio.sleep(1.5) # 获取房间信息 room_info = await page.evaluate(""" () => { const state = window.roomState || {}; return { code: state.code, players: state.players || [] }; } """) room_code = room_info.get('code', 'UNKNOWN') players = room_info.get('players', []) print(f"✅ 房间创建成功: {room_code}") print(f"\n👥 房间玩家 ({len(players)}/3):") for p in players: ai_tag = "🤖" if p.get('is_ai') else "👤" print(f" {ai_tag} {p.get('name', 'Unknown')} ({p.get('id', 'N/A')[:8]})") # 准备 print("\n✅ 准备游戏...") await page.click('button:has-text("准备")') await asyncio.sleep(0.5) # 开始游戏 print("🚀 开始游戏...") await page.click('button:has-text("开始游戏")') await asyncio.sleep(2) print("\n🎲 游戏进行中...") print("-" * 70) # 游戏主循环 turn_count = 0 my_player_id = None game_ended = False for i in range(120): # 最多120秒 # 获取当前游戏状态 game_state = await page.evaluate(""" () => { const state = window.gameState || {}; return { phase: state.phase, current_turn: state.current_turn, doudizhu: state.doudizhu || {}, players: state.players || {} }; } """) # 获取自己的ID if not my_player_id: my_info = await page.evaluate(""" () => window.playerInfo || {} """) my_player_id = my_info.get('id', '') current_turn = game_state.get('current_turn', '') phase = game_state.get('phase', '') doudizhu = game_state.get('doudizhu', {}) doudizhu_phase = doudizhu.get('phase', '') # 获取玩家名称映射 players_map = game_state.get('players', {}) # 检查是否是当前玩家的回合 if current_turn == my_player_id and not game_ended: turn_count += 1 my_name = "Headless玩家" if doudizhu_phase == 'calling': # 叫地主阶段 print(f"{turn_count:2d}. [{my_name:10}] 轮到叫地主...") await page.evaluate(""" () => { socket.emit('game_action', { action: 'call', data: { want: true } }); } """) print(f" -> 叫地主") await asyncio.sleep(1) elif doudizhu_phase == 'playing': # 出牌阶段 - 简化策略：出最小单张或不出 player_cards = await page.evaluate(""" () => window.playerCards || [] """) if player_cards: # 出第一张牌（单张） card_id = player_cards[0].get('id') await page.evaluate(f""" () => {{ socket.emit('game_action', {{ action: 'play', data: {{ card_ids: ['{card_id}'] }} }}); }} """) print(f"{turn_count:2d}. [{my_name:10}] 出牌: 单张") else: # 没牌了，不出 await page.evaluate(""" () => { socket.emit('game_action', { action: 'pass', data: {} }); } """) print(f"{turn_count:2d}. [{my_name:10}] 不出") await asyncio.sleep(1) # 监听其他玩家动作 events = await page.evaluate(""" () => { const e = window.gameEvents || []; window.gameEvents = []; return e; } """) for event in events: if event.get('type') == 'update': data = event.get('data', {}) action = data.get('action', 'unknown') player_id = data.get('player_id', '') result = data.get('result', {}) # 获取玩家名称 player_name = "Unknown" player_info = players_map.get(player_id, {}) if player_info: player_name = player_info.get('name', 'Unknown')[:10] else: # 从初始玩家列表查找 for p in players: if p.get('id') == player_id: player_name = p.get('name', 'Unknown')[:10] break if player_id != my_player_id and result.get('success'): turn_count += 1 if action == 'play': card_type = result.get('card_type', {}) remaining = result.get('remaining_cards', 0) print(f"{turn_count:2d}. [{player_name:10}] 出牌: {card_type.get('name', '?'):12} 剩余: {remaining:2d}张") elif action == 'call': want = result.get('data', {}).get('want', False) if result.get('lord_id'): print(f"{turn_count:2d}. [{player_name:10}] 🎩 叫地主!") else: print(f"{turn_count:2d}. [{player_name:10}] 叫地主: {'是' if want else '否'}") elif action == 'pass': print(f"{turn_count:2d}. [{player_name:10}] 不出") elif event.get('type') == 'ended': print("\n" + "=" * 70) winner = event.get('data', {}).get('winner', 'N/A') print(f"🏆 游戏结束! 获胜者: {winner[:8] if winner else 'N/A'}") game_ended = True break if game_ended: break # 检查游戏是否结束（通过检查winner） winner_check = await page.evaluate(""" () => { const state = window.gameState || {}; return state.winner; } """) if winner_check: print("\n" + "=" * 70) print(f"🏆 游戏结束! 获胜者: {winner_check[:8]}") game_ended = True break await asyncio.sleep(0.5) if not game_ended: print("\n" + "=" * 70) print(f"⏱️ 测试超时 (120秒)") print(f"📊 共进行了 {turn_count} 步") # 截图 screenshot_path = '/tmp/doudizhu_test_result_v2.png' await page.screenshot(path=screenshot_path, full_page=True) print(f"\n📸 截图已保存: {screenshot_path}") await browser.close() print("\n" + "=" * 70) print("✅ Headless测试完成!") print("=" * 70) except Exception as e: print(f"\n❌ 测试出错: {e}") import traceback traceback.print_exc() await browser.close() raise # 运行测试 if __name__ == '__main__': asyncio.run(run_headless_test())