lichess-bot中stdin被重定向后如何读取用户输入？

核心原因 #

lichess-bot为实现与引擎的UCI通信，会重定向标准输入流，导致input()无法读取终端的用户输入，从而抛出EOFError。

解决方案 #

1. 直接读取终端设备（推荐） #

绕过被重定向的stdin，直接读取终端的原始输入设备，适配Linux/macOS和Windows系统：

import sys
from typing import PlayResult
from chess import Board
from lichess_bot.engine_wrapper import MinimalEngine

class HumanCLI(MinimalEngine):
    def search(self, board: Board, *args) -> PlayResult:
        # 适配不同系统的终端输入设备
        if sys.platform.startswith('win'):
            terminal_input = open('con', 'r')
        else:
            terminal_input = open('/dev/tty', 'r')
        
        while True:
            user_input = terminal_input.readline().strip()
            # 验证走法合法性
            try:
                move = board.parse_uci(user_input)
                if move in board.legal_moves:
                    terminal_input.close()
                    return PlayResult(move, None)
            except ValueError:
                pass
            print("非法走法，请输入UCI格式的合法走法（例如e2e4）")

2. 提前保存原始stdin引用 #

在lichess-bot重定向stdin之前，保存原始标准输入的文件描述符，后续通过该描述符读取输入：

import sys
import os
from typing import PlayResult
from chess import Board
from lichess_bot.engine_wrapper import MinimalEngine

# 在引擎初始化前保存原始stdin
original_stdin_fd = os.dup(sys.stdin.fileno())
original_stdin = os.fdopen(original_stdin_fd, 'r')

class HumanCLI(MinimalEngine):
    def search(self, board: Board, *args) -> PlayResult:
        while True:
            user_input = original_stdin.readline().strip()
            try:
                move = board.parse_uci(user_input)
                if move in board.legal_moves:
                    return PlayResult(move, None)
            except ValueError:
                pass
            print("非法走法，请输入UCI格式的合法走法（例如e2e4）")

3. 线程后台监听输入（适合复杂交互） #

启动独立线程持续监听终端输入，缓存合法走法，search方法直接读取缓存内容：

import sys
import threading
from typing import PlayResult
from chess import Board
from lichess_bot.engine_wrapper import MinimalEngine

class HumanCLI(MinimalEngine):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.cached_move = None
        self.lock = threading.Lock()
        # 后台启动输入监听线程
        input_thread = threading.Thread(target=self._listen_terminal, daemon=True)
        input_thread.start()

    def _listen_terminal(self):
        if sys.platform.startswith('win'):
            terminal_input = open('con', 'r')
        else:
            terminal_input = open('/dev/tty', 'r')
        
        while True:
            user_input = terminal_input.readline().strip()
            with self.lock:
                self.cached_move = user_input

    def search(self, board: Board, *args) -> PlayResult:
        while True:
            with self.lock:
                move_str = self.cached_move
                self.cached_move = None
            if move_str:
                try:
                    move = board.parse_uci(move_str)
                    if move in board.legal_moves:
                        return PlayResult(move, None)
                except ValueError:
                    pass
                print("非法走法，请输入UCI格式的合法走法（例如e2e4）")

内容的提问来源于stack exchange，提问作者ebcoder