# -*- coding: utf-8 -*- """Xiangqi.com 在线对弈自动机器人(aiohttp 全异步版)。 uv run online_bot.py --strength 500 uv run online_bot.py --depth 20 --hash 512 --first-delay 10 协议要点(实测): - REST 与 WS 均不带 Cookie; WS 握手必须带 Origin, 否则立即被关 - 匹配: POST /api/v2/games/quick-challenge {"config_str":"10/2"} - 匹配成功后: lobby.left -> g_v2.get - 走子: g_v2.move {game_id, move, thinking_time}; 回执 game.move.success """ import argparse import asyncio import base64 import json import logging import re import ssl import aiohttp from .cloud import CloudBook from .coords import fen_from_moves, site_to_ucci, ucci_to_site from .pikafish import Pikafish logging.basicConfig(level=logging.INFO, format='%(asctime)s %(message)s') log = logging.getLogger('xiangqi.bot') API = 'https://api.xiangqi.com/api' WS_URL = 'wss://api.xiangqi.com/socket.io/?jwt={jwt}&EIO=4&transport=websocket' NS = '/xiangqi' FIRST_MOVE_EXTRA = 5 # 红方首步额外延迟秒数 END_REASONS = {100: '绝杀', 101: '困毙', 200: '超时', 300: '认输', 400: '和棋', 401: '不吃子和', 402: '重复和', 500: '断线', 600: '弃局'} class OnlineBot: def __init__(self, args): self.args = args self.session = aiohttp.ClientSession() self.jwt = None self.engine = None self.cloud = CloudBook(min_score=args.cloud_min_score) self.ws = None self.game_id = None self.my_side = None self.moves = [] # UCCI self.thinking = False self.game_over = False self.first_deadline = None self._ack = 0 self._send_lock = asyncio.Lock() # ---------- 基础 ---------- def emit(self, event, data, ack=False): self._ack += 1 n = self._ack if ack else '' payload = json.dumps([event, data], ensure_ascii=False) return self.ws.send_str(f'42{NS},{n}{payload}') def my_name(self): try: payload = self.jwt.split('.')[1] payload += '=' * (4 - len(payload) % 4) return json.loads(base64.urlsafe_b64decode(payload)).get('sub') except Exception: return None # ---------- 生命周期 ---------- async def run(self): async with self.session.post(f'{API}/users/guest_jwt', timeout=aiohttp.ClientTimeout(10)) as r: r.raise_for_status() self.jwt = (await r.json()).get('jwt') log.info('[auth] guest JWT %s...', self.jwt[:24]) self.engine = await Pikafish( threads=self.args.threads, hash_mb=self.args.hash, movetime=self.args.strength).start() log.info('[engine] threads=%s hash=%sMB', self.engine.threads, self.args.hash) origin = 'https://play.xiangqi.com' async with self.session.ws_connect( WS_URL.format(jwt=self.jwt), origin=origin, ssl=ssl.SSLContext() if False else None, # aiohttp 默认校验; 站点证书正常 autoping=True, heartbeat=20) as ws: self.ws = ws log.info('[ws] connected') async for msg in ws: if msg.type == aiohttp.WSMsgType.TEXT: await self.on_frame(msg.data) elif msg.type in (aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR): break if self.game_over: break async def close(self): self.game_over = True await self.cloud.close() if self.engine: await self.engine.quit() await self.session.close() # ---------- 协议 ---------- async def on_frame(self, raw): if raw == '2': # 心跳 await self.ws.send_str('3') return if raw.startswith('0'): await self.ws.send_str(f'40{NS},' + json.dumps({'token': self.jwt})) return if raw.startswith('40'): log.info('[ws] joined /xiangqi') self.emit('lobby.join', {'room': 'lobby'}) if not self.args.game: await asyncio.sleep(1) asyncio.create_task(self.quick_challenge()) return if not raw.startswith(f'42{NS},'): return m = re.match(rf'^42{NS},(\d+)?(\[.*\])$', raw, re.DOTALL) if not m: return try: arr = json.loads(m.group(2)) except json.JSONDecodeError: return await self.on_event(arr[0], arr[1] if len(arr) > 1 else {}) async def on_event(self, event, data): if event == 'lobby.my.game.new': self.game_id = data.get('game_id') log.info('[game] matched %s', self.game_id) self.emit('lobby.left', {}) await asyncio.sleep(0.3) self.emit('g_v2.get', {'game_id': self.game_id}, ack=True) elif event == 'game.get.success': d = data.get('data', {}) me = self.my_name() for p, side in ((d.get('rplayer', {}), 'red'), (d.get('bplayer', {}), 'black')): if p.get('username') == me or p.get('identity') == me: self.my_side = side break self.moves = [site_to_ucci(str(x.get('move', x)).split()[0]) for x in d.get('move_history', [])] log.info('[game] %s 我方=%s, 已 %d 步', self.game_id, self.my_side, len(self.moves)) asyncio.create_task(self.maybe_move()) elif event == 'game.start.success': self.game_over = False self.first_deadline = None asyncio.create_task(self.maybe_move()) elif event == 'game.move.opponent': mv = str((data.get('data') or {}).get('move', '')).split()[0] if mv: ucci = site_to_ucci(mv) if not self.moves or self.moves[-1] != ucci: self.moves.append(ucci) log.info('[game] 对手 %s (%d 步)', mv, len(self.moves)) asyncio.create_task(self.maybe_move()) elif event == 'game.move.success': self.thinking = False elif event == 'game.server.end': reason = (data.get('data') or {}).get('end_reason') res = (data.get('data') or {}).get('end_result') or {} log.info('[game] 结束: %s (%s)', END_REASONS.get(reason, reason), json.dumps(res, ensure_ascii=False)[:120]) self.game_over = True elif event == 'game.error': log.warning('[game.error] %s', json.dumps(data, ensure_ascii=False)[:200]) async def quick_challenge(self): async with self.session.post( f'{API}/v2/games/quick-challenge', headers={'Authorization': f'Bearer {self.jwt}'}, json={'config_str': self.args.config}) as r: log.info('[match] %s %s', r.status, (await r.text())[:80]) # ---------- 走子 ---------- def is_my_turn(self): if not self.my_side or self.game_over: return False return (len(self.moves) % 2 == 0) == (self.my_side == 'red') async def maybe_move(self): if self.thinking or self.game_over or not self.game_id or not self.my_side: return if not self.is_my_turn(): return self.thinking = True try: # 红方首步延迟 if not self.moves and self.my_side == 'red' and self.args.first_delay: wait = self.args.first_delay + FIRST_MOVE_EXTRA log.info('[game] 红方首步延迟 %ds', wait) await asyncio.sleep(wait) if self.game_over: return fen = fen_from_moves(self.moves) t0 = asyncio.get_event_loop().time() mv = score = None source = 'engine' # 云库优先(深度模式同样生效) if not self.args.no_cloud: cloud = await self.cloud.best(fen, moves_count=len(self.moves)) if cloud: mv, score, source = cloud[0], f'cp {cloud[1]}', f'cloud({cloud[2]})' if mv is None: mv, score = await self.engine.bestmove( moves=self.moves, fen=fen, movetime=self.args.strength, depth=self.args.depth) site_mv = ucci_to_site(mv) self.moves.append(mv) think_ms = int((asyncio.get_event_loop().time() - t0) * 1000) log.info('[game] %s: %s → %s %s %dms', source, mv, site_mv, score, think_ms) self.emit('g_v2.move', {'game_id': self.game_id, 'move': site_mv, 'thinking_time': think_ms}) except Exception as e: log.error('[move] %s: %s', type(e).__name__, e) if self.engine._proc.returncode is not None: await self.engine.restart() finally: self.thinking = False async def async_main(args): bot = OnlineBot(args) try: await bot.run() finally: await bot.close() def main(): p = argparse.ArgumentParser(description='Xiangqi.com 在线对弈 + 皮卡鱼 + 云库') p.add_argument('--game', help='接管已有对局 ID') p.add_argument('--strength', type=int, default=1000, help='每步思考毫秒') p.add_argument('--depth', type=int, default=None, help='固定深度(优先于 strength)') p.add_argument('--threads', type=int, default=None, help='线程数(默认=CPU核心数)') p.add_argument('--hash', type=int, default=256, help='哈希表 MB') p.add_argument('--config', default='10/2', help='匹配时制 总分钟/加秒') p.add_argument('--first-delay', type=int, default=0, help='红方首步延迟秒(实际=此值+5)') p.add_argument('--cloud-min-score', type=int, default=15, help='云库采纳阈值(厘兵)') p.add_argument('--no-cloud', action='store_true', help='禁用云库') args = p.parse_args() try: asyncio.run(async_main(args)) except KeyboardInterrupt: pass if __name__ == '__main__': main()