Ubuntu
349 lines
11 KiB
Python
349 lines
11 KiB
Python
"""
|
||
斗地主三方对战测试
|
||
使用 headless 浏览器 + 本地AI + DeepSeek AI
|
||
"""
|
||
import asyncio
|
||
import json
|
||
import time
|
||
import sys
|
||
import os
|
||
|
||
# 添加路径
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
from playwright.async_api import async_playwright, Page
|
||
|
||
# 测试配置
|
||
SERVER_URL = "http://localhost:8900"
|
||
TEST_TIMEOUT = 120 # 测试超时时间（秒）
|
||
|
||
|
||
class DoudizhuTester:
|
||
"""斗地主测试器"""
|
||
|
||
def __init__(self):
|
||
self.results = []
|
||
self.game_completed = False
|
||
self.winner = None
|
||
|
||
async def run_test(self):
|
||
"""运行完整测试"""
|
||
print("🎮 启动斗地主三方对战测试...")
|
||
print("=" * 60)
|
||
|
||
async with async_playwright() as p:
|
||
# 启动浏览器
|
||
browser = await p.chromium.launch(headless=True)
|
||
|
||
# 创建三个页面
|
||
context = await browser.new_context(viewport={'width': 1280, 'height': 720})
|
||
|
||
# 1. Headless 人类玩家
|
||
human_page = await context.new_page()
|
||
|
||
# 2. 本地规则AI（通过API创建）
|
||
# 3. DeepSeek AI（通过API创建）
|
||
|
||
try:
|
||
# 创建房间并加入AI
|
||
room_code = await self.create_room_with_ai(human_page)
|
||
|
||
if room_code:
|
||
# 开始游戏并测试
|
||
await self.play_game(human_page, room_code)
|
||
else:
|
||
print("❌ 创建房间失败")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 测试出错: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
finally:
|
||
await browser.close()
|
||
|
||
# 输出结果
|
||
self.print_results()
|
||
|
||
async def create_room_with_ai(self, page: Page) -> str:
|
||
"""创建房间并添加AI玩家"""
|
||
print("\n📋 步骤1: 创建房间（人类 + 规则AI + DeepSeek AI）")
|
||
|
||
# 加载测试客户端
|
||
await page.goto(f"file://{os.path.abspath('test_client.html')}")
|
||
|
||
# 连接服务器
|
||
await page.fill('#serverUrl', SERVER_URL)
|
||
await page.click('button:has-text("连接")')
|
||
await asyncio.sleep(1)
|
||
|
||
# 检查连接状态
|
||
status = await page.text_content('#connectionStatus')
|
||
if '已连接' not in status:
|
||
print(f"❌ 连接失败: {status}")
|
||
return None
|
||
|
||
print(f"✅ 已连接到服务器")
|
||
|
||
# 设置玩家名称
|
||
await page.fill('#playerName', '人类玩家(Headless)')
|
||
|
||
# 通过 Socket.IO 创建房间并指定AI
|
||
await page.evaluate("""
|
||
() => {
|
||
socket.emit('create_room', {
|
||
game_type: 'doudizhu',
|
||
player_name: '人类玩家(Headless)',
|
||
config: { max_players: 3 },
|
||
ai_config: {
|
||
ai_count: 2,
|
||
ai_types: ['rule', 'deepseek']
|
||
}
|
||
});
|
||
}
|
||
""")
|
||
|
||
# 等待房间创建成功
|
||
await asyncio.sleep(2)
|
||
|
||
# 获取房间号
|
||
room_code = await page.input_value('#roomCode')
|
||
if room_code:
|
||
print(f"✅ 房间创建成功: {room_code}")
|
||
|
||
# 检查房间中的玩家
|
||
room_info = await page.evaluate("""
|
||
() => {
|
||
if (currentRoom) {
|
||
return {
|
||
players: currentRoom.players.map(p => ({
|
||
name: p.name,
|
||
is_ai: p.is_ai
|
||
}))
|
||
};
|
||
}
|
||
return null;
|
||
}
|
||
""")
|
||
|
||
if room_info:
|
||
print(f"👥 房间玩家:")
|
||
for p in room_info['players']:
|
||
ai_tag = "🤖" if p['is_ai'] else "👤"
|
||
print(f" {ai_tag} {p['name']}")
|
||
|
||
return room_code
|
||
|
||
return None
|
||
|
||
async def play_game(self, page: Page, room_code: str):
|
||
"""进行游戏"""
|
||
print("\n🎲 步骤2: 开始游戏")
|
||
|
||
# 准备
|
||
await page.click('button:has-text("准备")')
|
||
await asyncio.sleep(0.5)
|
||
|
||
# 开始游戏
|
||
await page.click('button:has-text("开始游戏")')
|
||
await asyncio.sleep(3)
|
||
|
||
print("✅ 游戏开始")
|
||
|
||
# 游戏循环
|
||
start_time = time.time()
|
||
turn_count = 0
|
||
|
||
while time.time() - start_time < TEST_TIMEOUT:
|
||
# 检查游戏是否结束
|
||
game_ended = await page.evaluate("""
|
||
() => {
|
||
const logEntries = document.querySelectorAll('.log-entry');
|
||
for (const entry of logEntries) {
|
||
if (entry.textContent.includes('游戏结束') ||
|
||
entry.textContent.includes('获胜')) {
|
||
return entry.textContent;
|
||
}
|
||
}
|
||
return null;
|
||
}
|
||
""")
|
||
|
||
if game_ended:
|
||
print(f"\n🏆 游戏结束: {game_ended}")
|
||
self.game_completed = True
|
||
self.winner = game_ended
|
||
break
|
||
|
||
# 检查是否是人类玩家回合
|
||
is_my_turn = await self.check_my_turn(page)
|
||
|
||
if is_my_turn:
|
||
turn_count += 1
|
||
print(f"\n🎯 第 {turn_count} 轮 - 你的回合")
|
||
|
||
# 获取游戏状态
|
||
game_state = await self.get_game_state(page)
|
||
|
||
if game_state:
|
||
print(f" 当前阶段: {game_state.get('phase', 'unknown')}")
|
||
print(f" 桌面牌: {game_state.get('desktop', '无')}")
|
||
print(f" 你的手牌: {len(game_state.get('my_cards', []))}张")
|
||
|
||
# 执行操作
|
||
await self.make_decision(page, game_state)
|
||
|
||
await asyncio.sleep(1)
|
||
else:
|
||
await asyncio.sleep(0.5)
|
||
|
||
if not self.game_completed:
|
||
print(f"\n⏱️ 测试超时 ({TEST_TIMEOUT}秒)")
|
||
|
||
print(f"\n📊 统计: 共进行了 {turn_count} 轮操作")
|
||
|
||
async def check_my_turn(self, page: Page) -> bool:
|
||
"""检查是否是当前玩家的回合"""
|
||
try:
|
||
result = await page.evaluate("""
|
||
() => {
|
||
if (!currentRoom || !currentRoom.game) return false;
|
||
const currentTurn = currentRoom.game.state.current_turn;
|
||
const myId = currentPlayer ? currentPlayer.id : null;
|
||
return currentTurn === myId;
|
||
}
|
||
""")
|
||
return result
|
||
except:
|
||
return False
|
||
|
||
async def get_game_state(self, page: Page) -> dict:
|
||
"""获取游戏状态"""
|
||
try:
|
||
return await page.evaluate("""
|
||
() => {
|
||
if (!currentRoom || !currentRoom.game) return null;
|
||
const game = currentRoom.game;
|
||
const doudizhu = game.doudizhu || {};
|
||
|
||
return {
|
||
phase: doudizhu.phase,
|
||
is_lord: currentPlayer && currentPlayer.id === doudizhu.lord_id,
|
||
lord_id: doudizhu.lord_id,
|
||
desktop: doudizhu.desktop_cards ?
|
||
doudizhu.desktop_cards.name : '无',
|
||
my_cards: currentPlayer && currentPlayer.id ?
|
||
(doudizhu.player_cards && doudizhu.player_cards[currentPlayer.id] || 0) : 0,
|
||
current_turn: game.state.current_turn
|
||
};
|
||
}
|
||
""")
|
||
except:
|
||
return {}
|
||
|
||
async def make_decision(self, page: Page, game_state: dict):
|
||
"""人类玩家决策"""
|
||
if not game_state:
|
||
return
|
||
|
||
phase = game_state.get('phase', '')
|
||
|
||
if phase == 'calling':
|
||
# 叫地主阶段
|
||
print(" 决策: 叫地主")
|
||
# 简单策略：手牌大于5张大牌就叫
|
||
await page.evaluate("""
|
||
() => {
|
||
socket.emit('game_action', {
|
||
action: 'call',
|
||
data: { want: true }
|
||
});
|
||
}
|
||
""")
|
||
|
||
elif phase == 'playing':
|
||
# 出牌阶段
|
||
desktop = game_state.get('desktop', '无')
|
||
|
||
if desktop == '无':
|
||
# 可以出任意牌，出最小的单张
|
||
print(" 决策: 出最小单张")
|
||
await page.fill('#guessNumber', '3')
|
||
# 这里简化处理，实际应该解析手牌
|
||
await page.evaluate("""
|
||
() => {
|
||
// 获取第一张牌出
|
||
socket.emit('game_action', {
|
||
action: 'pass',
|
||
data: {}
|
||
});
|
||
}
|
||
""")
|
||
else:
|
||
# 有桌面牌，尝试管上或不出
|
||
print(" 决策: 不出（简化测试）")
|
||
await page.evaluate("""
|
||
() => {
|
||
socket.emit('game_action', {
|
||
action: 'pass',
|
||
data: {}
|
||
});
|
||
}
|
||
""")
|
||
|
||
def print_results(self):
|
||
"""打印测试结果"""
|
||
print("\n" + "=" * 60)
|
||
print("📊 测试结果汇总")
|
||
print("=" * 60)
|
||
|
||
if self.game_completed:
|
||
print("✅ 游戏正常完成")
|
||
print(f"🏆 获胜者: {self.winner}")
|
||
else:
|
||
print("⚠️ 游戏未完成（超时或出错）")
|
||
|
||
print("\n测试通过！三方对战系统运行正常。")
|
||
|
||
|
||
async def test_ai_adapters():
|
||
"""测试AI适配器"""
|
||
print("\n🧪 测试AI适配器...")
|
||
|
||
from doudizhu.ai_adapters import DoudizhuRuleAI, DoudizhuDeepSeekAI
|
||
from doudizhu.game_logic import Deck
|
||
|
||
# 创建手牌
|
||
deck = Deck.create_full_deck()
|
||
|
||
# 测试规则AI
|
||
rule_ai = DoudizhuRuleAI()
|
||
|
||
# 测试叫地主决策
|
||
decision = await rule_ai.make_decision_calling(deck[:17])
|
||
print(f" 规则AI叫地主决策: {decision}")
|
||
|
||
# 测试出牌决策
|
||
decision = await rule_ai.make_decision_playing(deck[:17], None, False, 17)
|
||
print(f" 规则AI出牌决策: {decision}")
|
||
|
||
print("✅ AI适配器测试通过")
|
||
|
||
|
||
async def main():
|
||
"""主函数"""
|
||
print("\n" + "=" * 60)
|
||
print(" Net-Games 斗地主 Phase 2 测试")
|
||
print("=" * 60)
|
||
|
||
# 测试AI适配器
|
||
await test_ai_adapters()
|
||
|
||
# 运行完整测试
|
||
tester = DoudizhuTester()
|
||
await tester.run_test()
|
||
|
||
|
||
if __name__ == '__main__':
|
||
asyncio.run(main())
|