Ubuntu
270 lines
11 KiB
Python
270 lines
11 KiB
Python
#!/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())
|