258 lines
10 KiB
Python
258 lines
10 KiB
Python
# -*- 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()
|