TUI 界面(Textual): 棋盘实时渲染+状态面板+快捷键(p暂停 c云库 n换局 q退出); bot 收敛为纯逻辑核心

2026-09-13 16:50:33 +08:00
1 parent 7c55096a4e
commit ee55ee9c8b
@@ -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