如何让guilded.py扩展库的控制台提示符始终位于stdout底部？

核心问题是异步输入过程中，其他stdout输出会打乱提示符和输入内容的位置。通过ANSI控制序列管理光标位置，结合输入状态跟踪，可以完美解决这个问题。

解决方案实现 #

1. 核心思路 #

• 跟踪控制台的输入状态（是否正在等待用户输入）
• 使用ANSI控制序列保存/恢复光标位置、清除当前行
• 自定义输入处理逻辑，实时跟踪用户已输入的内容
• 包装stdout，在有输出时自动调整光标和提示符位置

2. 完整代码实现 #

import sys
import asyncio
import traceback
import logging

logger = logging.getLogger(__name__)

class ConsoleManager:
    def __init__(self, prompt="> "):
        self.prompt = prompt
        self.console_stop_flag = False
        # 输入状态跟踪变量
        self._is_waiting_input = False
        self._current_input = ""
        # 替换系统stdout为自定义包装类
        self._original_stdout = sys.stdout
        sys.stdout = self._StdoutWrapper(self)

    class _StdoutWrapper:
        def __init__(self, manager):
            self.manager = manager
            self.original = sys.__stdout__

        def write(self, message):
            # 非输入状态下直接输出内容
            if not self.manager._is_waiting_input:
                self.original.write(message)
                self.original.flush()
                return

            # 输入状态下的输出处理流程
            self.original.write('\x1b[s')          # 保存当前光标位置
            self.original.write('\r\x1b[K')        # 回到行首并清除当前输入行
            self.original.write(message)            # 输出目标内容
            # 重新打印提示符和用户已输入的内容
            self.original.write('\n' + self.manager.prompt + self.manager._current_input)
            self.original.write('\x1b[u')          # 恢复光标到之前的输入位置
            self.original.flush()

        def flush(self):
            self.original.flush()

    async def _on_console(self):
        """异步监听控制台输入"""
        logger.info("Console is ready and is listening for commands\n")
        
        while True:
            if self.console_stop_flag:
                self.console_stop_flag = False
                logger.info("Console was stopped and is no longer listening for commands\n")
                break
            try:
                self._is_waiting_input = True
                # 使用自定义跟踪输入的方法替代原生ainput
                console_in = await self._tracked_ainput()
                console_in = console_in.strip()
                self._is_waiting_input = False
                self._current_input = ""

                if len(console_in) == 0:
                    # 空输入时重置提示符位置
                    sys.stdout.write('\r\x1b[K' + self.prompt)
                    sys.stdout.flush()
                    continue
                self.dispatch("console_message", console_in)
            except Exception:
                self._is_waiting_input = False
                self._current_input = ""
                traceback.print_exc()
                # 异常后重置提示符位置
                sys.stdout.write('\r\x1b[K' + self.prompt)
                sys.stdout.flush()

    async def _tracked_ainput(self):
        """跟踪用户输入内容的异步输入方法"""
        loop = asyncio.get_event_loop()
        reader = asyncio.StreamReader()
        protocol = asyncio.StreamReaderProtocol(reader)
        await loop.connect_read_pipe(lambda: protocol, sys.stdin)

        input_data = []
        # 先打印初始提示符
        sys.stdout.write(self.prompt)
        sys.stdout.flush()

        while True:
            char = await reader.read(1)
            # 处理换行或输入结束
            if not char or char == b'\n':
                sys.stdout.write('\n')
                sys.stdout.flush()
                break
            # 处理退格键
            if char == b'\x7f':
                if input_data:
                    input_data.pop()
                    sys.stdout.write('\b \b')
                    sys.stdout.flush()
            # 处理普通字符输入
            else:
                char_str = char.decode('utf-8')
                input_data.append(char_str)
                sys.stdout.write(char_str)
                sys.stdout.flush()
            # 更新当前输入内容快照
            self._current_input = ''.join(input_data)
        
        return ''.join(input_data)

    def dispatch(self, event_name, *args):
        """自定义事件分发逻辑，替换为你的业务代码"""
        logger.info(f"Received console message: {args[0]}")

# 测试用例
if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    manager = ConsoleManager()

    # 模拟异步任务定时输出内容
    async def simulate_external_output():
        for i in range(3):
            await asyncio.sleep(2)
            print(f"[External Output] Test message {i+1}")

    async def main():
        await asyncio.gather(manager._on_console(), simulate_external_output())

    asyncio.run(main())

3. 代码说明 #

• _StdoutWrapper：拦截所有stdout输出，根据输入状态切换输出逻辑。输入状态下会临时隐藏当前输入，输出内容后再恢复输入场景。
• _tracked_ainput：替代原生ainput，通过StreamReader实时跟踪每一个输入字符，确保外部输出时能精准恢复输入内容和光标位置。
• 状态管理：通过_is_waiting_input和_current_input变量，精准控制输出时的界面展示逻辑。

效果验证 #

• 正常输入命令：输出内容显示在提示符上方，提示符自动回到底部
• 输入过程中收到外部输出：已输入内容保留在底部，输出内容出现在上方，光标自动回到输入位置，无内容错乱

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