""" 斗地主 API 测试 测试三方对战：人类 + 规则AI + DeepSeek AI """ import asyncio import json import time import sys import os import socketio # 添加路径 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from config import DEEPSEEK_API_KEY SERVER_URL = "http://localhost:8900" class DoudizhuAPITest: """斗地主API测试""" def __init__(self): self.sio = socketio.Client() self.room_code = None self.player_id = None self.game_ended = False self.turn_count = 0 self.results = [] # 注册事件处理器 self.sio.on('connected', self.on_connected) self.sio.on('room_created', self.on_room_created) self.sio.on('game_started', self.on_game_started) self.sio.on('state_update', self.on_state_update) self.sio.on('game_ended', self.on_game_ended) self.sio.on('error', self.on_error) def on_connected(self, data): print(f"✅ 连接到服务器: {data.get('message')}") def on_room_created(self, data): if data.get('success'): room = data.get('room', {}) self.room_code = room.get('code') player = data.get('player', {}) self.player_id = player.get('id') print(f"✅ 房间创建成功: {self.room_code}") print(f"👤 玩家ID: {self.player_id[:8] if self.player_id else 'N/A'}") # 显示房间玩家 players = room.get('players', []) print(f"\n👥 房间玩家 ({len(players)}/3):") for p in players: ai_tag = "🤖 AI" if p.get('is_ai') else "👤 人类" print(f" {ai_tag}: {p.get('name', 'Unknown')}") else: print(f"❌ 创建房间失败: {data}") def on_game_started(self, data): print(f"\n🎲 游戏开始!") room = data.get('room', {}) game = room.get('game', {}) doudizhu = game.get('doudizhu', {}) print(f" 底牌: {len(doudizhu.get('bottom_cards', []))}张") print(f" 当前回合: {doudizhu.get('current_turn', 'N/A')[:8] if doudizhu.get('current_turn') else 'N/A'}") # 检查是否是叫地主阶段 phase = doudizhu.get('phase') if phase == 'calling': print(f" 阶段: 叫地主") def on_state_update(self, data): self.turn_count += 1 action = data.get('action') player_id = data.get('player_id', '') result = data.get('result', {}) game_state = data.get('game_state', {}) doudizhu = game_state.get('doudizhu', {}) player_name = player_id[:8] if player_id else 'N/A' ai_reasoning = data.get('ai_reasoning', '') print(f"\n📍 第 {self.turn_count} 步") print(f" 玩家: {player_name}") print(f" 操作: {action}") if result.get('success'): if action == 'play': card_type = result.get('card_type', {}) print(f" 出牌: {card_type.get('name', 'Unknown')}") remaining = result.get('remaining_cards', 0) print(f" 剩余: {remaining}张") elif action == 'call': want = result.get('data', {}).get('want', False) print(f" 叫地主: {'是' if want else '否'}") if result.get('lord_id'): print(f" 🎩 地主确定!") elif action == 'pass': print(f" 不出") if ai_reasoning: print(f" 🤖 AI思考: {ai_reasoning[:60]}...") next_player = doudizhu.get('current_turn', '') if next_player: print(f" 下家: {next_player[:8]}") else: print(f" ❌ 失败: {result.get('message', 'Unknown')}") def on_game_ended(self, data): self.game_ended = True winner = data.get('winner', '') room = data.get('room', {}) game = room.get('game', {}) print(f"\n" + "=" * 50) print(f"🏆 游戏结束!") print(f" 获胜者: {winner[:8] if winner else 'N/A'}") print(f" 总步数: {self.turn_count}") print("=" * 50) def on_error(self, data): print(f"❌ 错误: {data.get('message', 'Unknown')}") def run(self): """运行测试""" print("=" * 60) print(" 斗地主三方对战 API 测试") print("=" * 60) print(f"🤖 AI配置: 规则AI + DeepSeek AI") print(f"🔑 DeepSeek Key: {DEEPSEEK_API_KEY[:10]}..." if DEEPSEEK_API_KEY else "❌ 未配置") try: # 连接服务器 print(f"\n📡 连接服务器: {SERVER_URL}") self.sio.connect(SERVER_URL) # 等待连接 time.sleep(1) # 创建房间 print("\n📝 创建房间...") self.sio.emit('create_room', { 'game_type': 'doudizhu', 'player_name': '人类玩家', 'config': {'max_players': 3}, 'ai_config': { 'ai_count': 2, 'ai_types': ['rule', 'deepseek'] } }) # 等待房间创建 for _ in range(10): if self.room_code: break time.sleep(0.5) if not self.room_code: print("❌ 房间创建超时") return # 准备并开始 time.sleep(1) print("\n✅ 准备...") self.sio.emit('set_ready', {'ready': True}) time.sleep(0.5) print("🚀 开始游戏...") self.sio.emit('start_game') # 等待游戏结束或超时 print("\n⏳ 等待游戏完成...") start_time = time.time() timeout = 90 # 90秒超时 while not self.game_ended and time.time() - start_time < timeout: time.sleep(0.5) if not self.game_ended: print(f"\n⏱️ 测试超时 ({timeout}秒)") # 断开连接 self.sio.disconnect() except Exception as e: print(f"❌ 测试出错: {e}") import traceback traceback.print_exc() def print_summary(self): """打印总结""" print("\n" + "=" * 60) print("📊 测试总结") print("=" * 60) if self.game_ended: print("✅ 游戏正常完成") else: print("⚠️ 游戏未完成") print(f"📈 总回合数: {self.turn_count}") print(f"\n🎮 三方配置:") print(" 👤 人类玩家") print(" 🤖 规则AI (本地算法)") print(" 🧠 DeepSeek AI (大模型)") if self.game_ended: print("\n✅ Phase 2 测试通过!") else: print("\n⚠️ 部分功能正常，游戏可能因超时而未结束") def test_ai_only(): """仅测试AI适配器""" print("=" * 60) print(" AI适配器单元测试") print("=" * 60) from doudizhu.ai_adapters import DoudizhuRuleAI, DoudizhuDeepSeekAI from doudizhu.game_logic import Deck async def test_rule_ai(): print("\n🧪 测试规则AI...") # 创建手牌 deck = Deck.shuffle(Deck.create_full_deck()) hand = deck[:17] rule_ai = DoudizhuRuleAI() # 测试叫地主 decision = await rule_ai.make_decision_calling(hand) print(f" 叫地主: {decision['action']} (want={decision['data'].get('want')})") print(f" 理由: {decision['reasoning']}") # 测试出牌 decision = await rule_ai.make_decision_playing(hand, None, False, 17) print(f" 出牌决策: {decision['action']}") print(f" 理由: {decision['reasoning']}") print("✅ 规则AI测试通过") async def test_deepseek_ai(): if not DEEPSEEK_API_KEY: print("\n⏭️ 跳过DeepSeek AI测试 (未配置API Key)") return print("\n🧠 测试DeepSeek AI...") deck = Deck.shuffle(Deck.create_full_deck()) hand = deck[:17] deepseek_ai = DoudizhuDeepSeekAI(api_key=DEEPSEEK_API_KEY) # 测试叫地主 try: decision = await deepseek_ai.make_decision_calling(hand) print(f" 叫地主: {decision['action']}") print(f" 理由: {decision['reasoning'][:80]}...") print("✅ DeepSeek AI测试通过") except Exception as e: print(f" ⚠️ DeepSeek API调用失败: {e}") print(" (这是正常的，API可能暂时不可用)") # 运行异步测试 asyncio.run(test_rule_ai()) asyncio.run(test_deepseek_ai()) if __name__ == '__main__': # 先测试AI适配器 test_ai_only() # 然后运行完整测试 print("\n") test = DoudizhuAPITest() test.run() test.print_summary()