hhhge/AwesomeXiangQi
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: hhhge/AwesomeXiangQi:7c55096a4e2a9956e7dbd462e0276467770b469f
hhhge/AwesomeXiangQi:master
...
compare: hhhge/AwesomeXiangQi:master
hhhge/AwesomeXiangQi:master
1 Commits
7c55096a4e ... master
Author SHA1 Message Date
AwesomeXiangQi ee55ee9c8b TUI 界面(Textual): 棋盘实时渲染+状态面板+快捷键(p暂停 c云库 n换局 q退出); bot 收敛为纯逻辑核心 2026-09-13 16:50:33 +08:00
6 changed files with 419 additions and 77 deletions
Expand all files Collapse all files

No files matched your search

README.md
+3
View File
@@ -3,3 +3,6 @@
Xiangqi.com 在线对弈自动机器人: 云库(chessdb.cn)优先 + 本地皮卡鱼兜底。
使用: uv sync 后 uv run online_bot.py, 参数见 uv run online_bot.py --help
## TUI
默认启动 TUI 界面: 棋盘实时渲染/状态面板/日志, 快捷键 p 暂停 c 云库 n 换一局 q 退出。
online_bot.py
+9 -2
View File
@@ -1,6 +1,13 @@
# -*- coding: utf-8 -*-
"""Xiangqi.com 在线对弈 + 本地皮卡鱼 + 云库。"""
from xiangqi_bot.bot import main
"""Xiangqi.com 在线对弈机器人入口: 默认 TUI, --no-tui 纯日志。"""
from xiangqi_bot.bot import build_parser
from xiangqi_bot.tui import run_tui
def main():
args = build_parser().parse_args()
raise SystemExit(run_tui(args))
if __name__ == '__main__':
main()
pyproject.toml
+1
View File
@@ -5,6 +5,7 @@ description = "Xiangqi.com 在线对弈 + 本地皮卡鱼引擎 + chessdb 云库
requires-python = ">=3.11"
dependencies = [
"aiohttp>=3.10",
"textual>=8.2.8",
]
[project.scripts]
uv.lock
Generated
+100 -1
View File
@@ -157,10 +157,14 @@ version = "1.0.0"
source = { virtual = "." }
dependencies = [
{ name = "aiohttp" },
{ name = "textual" },
]
[package.metadata]
requires-dist = [{ name = "aiohttp", specifier = ">=3.10" }]
requires-dist = [
{ name = "aiohttp", specifier = ">=3.10" },
{ name = "textual", specifier = ">=8.2.8" },
]
[[package]]
name = "frozenlist"
@@ -276,6 +280,53 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/57/b0/0e52c878c53f245edd3a11020f20979b3f490f245af532c7cae3027754b5/idna-3.19-py3-none-any.whl", hash = "sha256:815e7be7a7806d54abb586dc943addc79e8b2ee16915059658cbeff4b1b43bf4", size = 68550, upload-time = "2026-08-18T05:14:22.343Z" },
]
[[package]]
name = "linkify-it-py"
version = "2.2.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/45/98/7a1a5f31fd5c7ba93e963b168e244b8e3dd705b3d2a718e3c3307583bf57/linkify_it_py-2.2.0.tar.gz", hash = "sha256:907acd2d17ac1fbb9ddb62c8957ccbd6158cac602231a15c3b0cd1e215f03cee", size = 32939, upload-time = "2026-08-29T07:07:08.305Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/13/d4/1152d1c7ab42d8b908be64fd200ddc870dc9d4925e951198702084aa1a7d/linkify_it_py-2.2.0-py3-none-any.whl", hash = "sha256:3adc40eb5af300b2605fcfdb968c24e1d780a90f1f2221af7c15e5111e94d443", size = 21971, upload-time = "2026-08-29T07:07:07.164Z" },
]
[[package]]
name = "markdown-it-py"
version = "4.2.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "mdurl" },
]
sdist = { url = "https://files.pythonhosted.org/packages/06/ff/7841249c247aa650a76b9ee4bbaeae59370dc8bfd2f6c01f3630c35eb134/markdown_it_py-4.2.0.tar.gz", hash = "sha256:04a21681d6fbb623de53f6f364d352309d4094dd4194040a10fd51833e418d49", size = 82454, upload-time = "2026-05-07T12:08:28.36Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/b3/81/4da04ced5a082363ecfa159c010d200ecbd959ae410c10c0264a38cac0f5/markdown_it_py-4.2.0-py3-none-any.whl", hash = "sha256:9f7ebbcd14fe59494226453aed97c1070d83f8d24b6fc3a3bcf9a38092641c4a", size = 91687, upload-time = "2026-05-07T12:08:27.182Z" },
]
[package.optional-dependencies]
linkify = [
{ name = "linkify-it-py" },
]
[[package]]
name = "mdit-py-plugins"
version = "0.6.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "markdown-it-py" },
]
sdist = { url = "https://files.pythonhosted.org/packages/59/fc/f8d0863f8862f25602c0404d75568e89fb6b4109804645e5cdfb1be5cf56/mdit_py_plugins-0.6.1.tar.gz", hash = "sha256:a2bca0f039f39dbd35fb74ae1b5f998608c437463371f0ff7f49a19a17a114d0", size = 56114, upload-time = "2026-05-13T09:03:38.91Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/a5/69/6da5581c6a7fede7dc261bf4e67d6adca4196f176b43288b55b3db395b6e/mdit_py_plugins-0.6.1-py3-none-any.whl", hash = "sha256:214c82fb2ac524472ab6a5bcab1de80f73b50443e187f401bfd77efbc7c6481d", size = 66663, upload-time = "2026-05-13T09:03:37.76Z" },
]
[[package]]
name = "mdurl"
version = "0.1.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/d6/54/cfe61301667036ec958cb99bd3efefba235e65cdeb9c84d24a8293ba1d90/mdurl-0.1.2.tar.gz", hash = "sha256:bb413d29f5eea38f31dd4754dd7377d4465116fb207585f97bf925588687c1ba", size = 8729, upload-time = "2022-08-14T12:40:10.846Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" },
]
[[package]]
name = "multidict"
version = "6.8.0"
@@ -434,6 +485,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/b1/ee/be4e1a4b7a2b27f4fb6936510d4bebcb41b0562c946930ad26916e069cf9/multidict-6.8.0-py3-none-any.whl", hash = "sha256:75daa15ca16d6285eb2e104b2f05ee6f8d9836c68da3ce5c85f615a0450eed0e", size = 16297, upload-time = "2026-09-09T13:57:56.106Z" },
]
[[package]]
name = "platformdirs"
version = "4.11.8"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/53/18/f3bb8ef0d3b930692343da8aa4d3cbcd6749477c053959395ac81965a6e9/platformdirs-4.11.8.tar.gz", hash = "sha256:f23abafea7dd4276d1f29104b83598d7dcc567cafd07c9c951e66665645437fc", size = 37182, upload-time = "2026-09-08T22:20:42.866Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/f4/e1/5b7b8bbb55084d1425bcb9bc823ff519e1b2be05f6ebb0089e2eacc38413/platformdirs-4.11.8-py3-none-any.whl", hash = "sha256:52f2f181bbfde907966932cc8312d967d02976422d66d537ea16092b8e291081", size = 24027, upload-time = "2026-09-08T22:20:41.537Z" },
]
[[package]]
name = "propcache"
version = "0.5.2"
@@ -545,6 +605,45 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/3a/ed/1cdcab6ba3d6ab7feca11fc14f0eeea80755bb53ef4e892079f31b10a25f/propcache-0.5.2-py3-none-any.whl", hash = "sha256:be1ddfcbb376e3de5d2e2db1d58d6d67463e6b4f9f040c000de8e300295465fe", size = 14036, upload-time = "2026-05-08T21:02:10.673Z" },
]
[[package]]
name = "pygments"
version = "2.21.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/49/2e/ced460408999b33da6b31b0021b0f37d329e202d4169aeb164493778f25b/pygments-2.21.0.tar.gz", hash = "sha256:610ca751c9bc2492b38eb9a38a7fbc93edbbb2d7182edaf34e66ae493dee5c8c", size = 5005329, upload-time = "2026-08-17T08:02:48.824Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/71/46/17f022dd3e953bf20a04a028a21ec746d942f8d2af30fa0f124fa0e6a684/pygments-2.21.0-py3-none-any.whl", hash = "sha256:2363c69b61c4a97c838da3b130dcd6468f4848992b21a82f2a63ec34377137d9", size = 1250147, upload-time = "2026-08-17T08:02:44.912Z" },
]
[[package]]
name = "rich"
version = "15.0.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "markdown-it-py" },
{ name = "pygments" },
]
sdist = { url = "https://files.pythonhosted.org/packages/c0/8f/0722ca900cc807c13a6a0c696dacf35430f72e0ec571c4275d2371fca3e9/rich-15.0.0.tar.gz", hash = "sha256:edd07a4824c6b40189fb7ac9bc4c52536e9780fbbfbddf6f1e2502c31b068c36", size = 230680, upload-time = "2026-04-12T08:24:00.75Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/82/3b/64d4899d73f91ba49a8c18a8ff3f0ea8f1c1d75481760df8c68ef5235bf5/rich-15.0.0-py3-none-any.whl", hash = "sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb", size = 310654, upload-time = "2026-04-12T08:24:02.83Z" },
]
[[package]]
name = "textual"
version = "8.2.8"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "markdown-it-py", extra = ["linkify"] },
{ name = "mdit-py-plugins" },
{ name = "platformdirs" },
{ name = "pygments" },
{ name = "rich" },
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/00/21/39a76b01bd5eea82a04baaca7580e105d8c59450df03998345bb2cfb307b/textual-8.2.8.tar.gz", hash = "sha256:3f106a9fbc73e39dd266c9712432087de78a6d644084c7c241d6a25c3169115b", size = 1860502, upload-time = "2026-06-30T06:51:24.495Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/fb/be/35261223d9416a0751cdff1c7b4a6f881387218a12d439fe22fefebc8c04/textual-8.2.8-py3-none-any.whl", hash = "sha256:267375fd402dc8d981457212efa71f0e3365fd17bba144ba9bb3ed7563cb374a", size = 731418, upload-time = "2026-06-30T06:51:26.364Z" },
]
[[package]]
name = "typing-extensions"
version = "4.16.0"
xiangqi_bot/bot.py
+158 -74
View File
@@ -1,13 +1,10 @@
# -*- coding: utf-8 -*-
"""Xiangqi.com 在线对弈自动机器人(aiohttp 全异步版)。
"""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
- 匹配成功后: lobby.left -> g_v2.get 入座
- 走子: g_v2.move {game_id, move, thinking_time}; 回执 game.move.success
"""
import argparse
@@ -24,7 +21,6 @@ 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'
@@ -36,25 +32,133 @@ END_REASONS = {100: '绝杀', 101: '困毙', 200: '超时', 300: '认输',
400: '和棋', 401: '不吃子和', 402: '重复和', 500: '断线', 600: '弃局'}
def build_parser():
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('--move-delay', type=float, default=1.0, help='每步最小间隔秒')
p.add_argument('--cloud-min-score', type=int, default=15, help='云库采纳阈值(厘兵)')
p.add_argument('--no-cloud', action='store_true', help='禁用云库')
p.add_argument('--no-tui', action='store_true', help='不启用 TUI, 仅日志输出')
return p
class OnlineBot:
"""对局机器人。状态字段(status/moves/fen/score 等)供 TUI 直接读取。"""
def __init__(self, args):
self.args = args
self.session = aiohttp.ClientSession()
self.session = None # 惰性: 首个协程调用时创建(需事件循环)
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.my_side = None # 'red' | 'black'
self.moves = [] # UCCI
self.site_moves = [] # 站点坐标(展示用)
self.fen = None
self.score = None
self.source = None # 'cloud' | 'engine' | 'cloud(n候选)'
self.status = '初始化'
self.end_reason = None
self.moves_count_opponent = 0
self.paused = False
self.game_over = False
self.game_started = False
self.thinking = False
self._stop = False
self.first_deadline = None
self.last_move_at = 0.0
self._ack = 0
self._send_lock = asyncio.Lock()
# ---------- 基础 ----------
# ---------- 状态/控制接口(TUI 用) ----------
@property
def use_cloud(self):
return not self.args.no_cloud
def toggle_cloud(self):
self.args.no_cloud = self.use_cloud
def toggle_pause(self):
self.paused = not self.paused
log.info('[bot] %s', '已暂停' if self.paused else '继续')
return self.paused
async def new_match(self):
"""弃掉当前对局并重新匹配。"""
if self.game_id and not self.game_over:
try:
await self.emit('g_v2.abandon', {'game_id': self.game_id,
'abandonmentSource': 'timeout'})
except Exception:
pass
self.game_over = True
await asyncio.sleep(1)
self.game_over = False
self.game_started = False
self.game_id = None
self.my_side = None
self.moves.clear()
self.site_moves.clear()
self.fen = None
self.score = self.source = None
self.end_reason = None
self.first_deadline = None
self.status = '重新匹配中…'
await self.quick_challenge()
# ---------- 生命周期 ----------
async def _get_session(self):
if self.session is None or self.session.closed:
self.session = aiohttp.ClientSession()
return self.session
async def run(self, on_ready=None):
session = await self._get_session()
async with 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,
autoping=True, heartbeat=20) as ws:
self.ws = ws
log.info('[ws] connected')
self.status = '已连接, 匹配中…'
if on_ready:
on_ready(self)
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._stop:
break
self.status = '连接断开'
async def close(self):
self._stop = True
self.game_over = True
await self.cloud.close()
if self.engine:
await self.engine.quit()
await self.session.close()
# ---------- 协议 ----------
async def emit(self, event, data, ack=False):
self._ack += 1
n = self._ack if ack else ''
@@ -69,43 +173,8 @@ class OnlineBot:
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': # 心跳
if raw == '2':
await self.ws.send_str('3')
return
if raw.startswith('0'):
@@ -116,7 +185,7 @@ class OnlineBot:
await self.emit('lobby.join', {'room': 'lobby'})
if not self.args.game:
await asyncio.sleep(1)
asyncio.create_task(self.quick_challenge())
await self.quick_challenge()
return
if not raw.startswith(f'42{NS},'):
return
@@ -132,6 +201,7 @@ class OnlineBot:
async def on_event(self, event, data):
if event == 'lobby.my.game.new':
self.game_id = data.get('game_id')
self.status = f'已匹配 {self.game_id}'
log.info('[game] matched %s', self.game_id)
await self.emit('lobby.left', {})
await asyncio.sleep(0.3)
@@ -145,33 +215,48 @@ class OnlineBot:
break
self.moves = [site_to_ucci(str(x.get('move', x)).split()[0])
for x in d.get('move_history', [])]
self.site_moves = [ucci_to_site(m) for m in self.moves]
self.fen = d.get('fen') or fen_from_moves(self.moves)
self.game_started = True
self.status = f'对局中({len(self.moves)} 步)'
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.game_started = True
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]
d = data.get('data') or {}
mv = str(d.get('move', '')).split()[0]
if d.get('fen'):
self.fen = d['fen']
if mv:
ucci = site_to_ucci(mv)
if not self.moves or self.moves[-1] != ucci:
self.moves.append(ucci)
self.site_moves.append(mv)
self.fen = d.get('fen') or fen_from_moves(self.moves)
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),
d = data.get('data') or {}
self.end_reason = d.get('end_reason')
res = d.get('end_result') or {}
self.status = f'对局结束: {END_REASONS.get(self.end_reason, self.end_reason)}'
log.info('[game] 结束: %s (%s)', self.status,
json.dumps(res, ensure_ascii=False)[:120])
self.game_over = True
self.thinking = False
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(
self.status = '匹配中…'
session = await self._get_session()
async with session.post(
f'{API}/v2/games/quick-challenge',
headers={'Authorization': f'Bearer {self.jwt}'},
json={'config_str': self.args.config}) as r:
@@ -179,23 +264,24 @@ class OnlineBot:
# ---------- 走子 ----------
def is_my_turn(self):
if not self.my_side or self.game_over:
if not self.my_side or self.game_over or self.paused:
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:
if (self.thinking or self.game_over or not self.game_id
or not self.my_side or self.paused):
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
self.status = f'红方首步延迟 {wait}s'
log.info('[game] 红方首步延迟 %ds', wait)
await asyncio.sleep(wait)
if self.game_over:
if self.game_over or not self.is_my_turn():
return
fen = fen_from_moves(self.moves)
@@ -204,28 +290,34 @@ class OnlineBot:
if self.args.move_delay:
gap = self.args.move_delay - (t0 - self.last_move_at)
if gap > 0:
self.status = f'行棋节奏延迟 {gap:.0f}s'
log.info('[game] 行棋节奏延迟 %.1fs', gap)
await asyncio.sleep(gap)
if self.game_over or not self.is_my_turn():
return
mv = score = None
source = 'engine'
# 云库优先(深度模式同样生效)
if not self.args.no_cloud:
if self.use_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:
self.status = '皮卡鱼思考中…'
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)
self.site_moves.append(site_mv)
self.fen = fen_from_moves(self.moves)
self.last_move_at = asyncio.get_event_loop().time()
think_ms = int((asyncio.get_event_loop().time() - t0) * 1000)
self.score, self.source = score, source
self.status = f'已走 {site_mv}'
think_ms = int((self.last_move_at - t0) * 1000)
log.info('[game] %s: %s → %s %s %dms', source, mv, site_mv, score, think_ms)
await self.emit('g_v2.move', {'game_id': self.game_id, 'move': site_mv,
'thinking_time': think_ms})
@@ -246,18 +338,10 @@ async def async_main(args):
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('--move-delay', type=float, default=1.0, help='每步最小间隔秒(深度/云库秒回时保持节奏)')
p.add_argument('--cloud-min-score', type=int, default=15, help='云库采纳阈值(厘兵)')
p.add_argument('--no-cloud', action='store_true', help='禁用云库')
args = p.parse_args()
args = build_parser().parse_args()
if not args.no_tui:
from .tui import run_tui
raise SystemExit(run_tui(args))
try:
asyncio.run(async_main(args))
except KeyboardInterrupt:
xiangqi_bot/tui.py
+148
View File
@@ -0,0 +1,148 @@
# -*- coding: utf-8 -*-
"""Textual TUI: 棋盘实时渲染 + 对局状态面板 + 快捷键控制。"""
import asyncio
import logging
from textual.app import App, ComposeResult
from textual.containers import Horizontal, Vertical
from textual.widgets import Footer, RichLog, Static
from .bot import OnlineBot, build_parser
RED_NAMES = {'R': '车', 'N': '马', 'B': '相', 'A': '仕', 'K': '帅', 'C': '炮', 'P': '兵'}
BLACK_NAMES = {'r': '車', 'n': '馬', 'b': '象', 'a': '士', 'k': '將', 'c': '砲', 'p': '卒'}
NAMES = {**RED_NAMES, **BLACK_NAMES}
ROW_LABELS_FLIP = ['十', '九', '八', '七', '六', '五', '四', '三', '二', '一'] # 我方红(黑在上)
ROW_LABELS_NORM = ['一', '二', '三', '四', '五', '六', '七', '八', '九', '十'] # 我方黑(红在上)
def board_text(fen, flip):
"""FEN -> 多行棋盘文本(红字黑字用不同样式标记)。"""
if not fen:
return '(等待对局…)'
rows = fen.split()[0].split('/')
cells = []
for row in rows:
line = []
for ch in row:
if ch.isdigit():
line.extend(['・'] * int(ch))
else:
line.append(NAMES.get(ch, '?'))
cells.append(line)
if flip:
cells.reverse()
row_labels = ROW_LABELS_FLIP
else:
row_labels = ROW_LABELS_NORM
lines = [' ' + ' '.join('１２３４５６７８９' if flip else '１２３４５６７８９')]
out = []
for i, line in enumerate(cells):
out.append(f'{row_labels[i]} ' + ' '.join(line))
if i == 4:
out.append(' ～楚 河 汉 界～')
return '\n'.join(out)
class XiangqiTUI(App):
CSS = '''
Screen { layout: horizontal; }
#left { width: 1fr; }
#right { width: 44; }
#board { border: round #884444; padding: 0 1; content-align: left middle; }
#board:focus { border: round #cc6666; }
#panel { height: auto; border: round #444488; padding: 0 1; }
#log { border: round #448844; height: 1fr; }
'''
BINDINGS = [
('q', 'quit', '退出'),
('p', 'pause', '暂停/继续'),
('c', 'cloud', '云库开/关'),
('n', 'new_match', '换一局'),
]
TITLE = '🐴 皮卡鱼在线对弈'
def __init__(self, args):
super().__init__()
self.args = args
self.bot = OnlineBot(args)
self._task = None
def compose(self) -> ComposeResult:
with Horizontal():
with Vertical(id='left'):
yield Static('', id='board')
yield Static('', id='panel')
yield RichLog(id='log', markup=False, wrap=True, max_lines=400)
yield Footer()
def on_mount(self):
log_handler = TextualLogHandler(self.query_one('#log', RichLog))
logging.getLogger().addHandler(log_handler)
logging.getLogger().setLevel(logging.INFO)
self.set_interval(0.3, self.refresh_state)
self._task = asyncio.create_task(self._run_bot())
async def _run_bot(self):
try:
await self.bot.run()
except Exception as e:
self.bot.status = f'异常退出: {e}'
log.error('[bot] %s: %s', type(e).__name__, e)
def refresh_state(self):
b = self.bot
flip = b.my_side != 'black'
self.query_one('#board', Static).update(board_text(b.fen, flip))
turn = {'red': '红方', 'black': '黑方', None: '-'}.get(b.my_side)
my_turn = '← 轮到我' if b.is_my_turn() else ''
panel = '\n'.join([
f'状态 : {b.status}',
f'对局 : {b.game_id or "-"} 我方: {turn or "-"} {my_turn}',
f'暂停 : {"是" if b.paused else "否"} 云库: {"开" if b.use_cloud else "关"}',
f'步数 : {len(b.moves)}',
f'评分 : {b.score or "-"} 来源: {b.source or "-"}',
f'终局 : {b.end_reason or "-"}',
'',
'着法: ' + ' '.join(b.site_moves[-40:]),
])
self.query_one('#panel', Static).update(panel)
def action_pause(self):
self.bot.toggle_pause()
def action_cloud(self):
self.bot.toggle_cloud()
log.info('[bot] 云库 %s', '开' if self.bot.use_cloud else '关')
async def action_new_match(self):
log.info('[bot] 弃局重匹配…')
await self.bot.new_match()
async def action_quit(self):
await self.bot.close()
self.exit()
class TextualLogHandler(logging.Handler):
def __init__(self, widget):
super().__init__()
self.widget = widget
def emit(self, record):
try:
msg = self.format(record)
self.widget.write(msg)
except Exception:
pass
def run_tui(args):
import sys
logging.basicConfig(level=logging.INFO,
format='%(asctime)s %(name)s %(message)s',
datefmt='%H:%M:%S')
app = XiangqiTUI(args)
# 无终端时(重定向/后台)以 headless 运行, 方便观察日志
app.run(headless=not (sys.stdout.isatty() and sys.stdin.isatty()))
return 0