如何重定向QThread中运行函数的标准输出到QTextEdit并支持input交互？

问题根因 #

1. 你当前仅重定向了标准输出stdout，且使用redirect_stdout的实现是等目标函数完全执行结束后才一次性读取缓冲内容，input()执行时函数处于阻塞等待输入的状态，还没走到emit输出的逻辑，所以提示文字不会实时显示。
2. 完全没有处理标准输入stdin的重定向逻辑，input()需要从stdin读取内容，没有对应实现的话函数会一直卡在内核等待终端输入，自然不会有任何响应。
3. 原有更新文本的逻辑用setText会覆盖所有历史内容，不符合终端追加输出的使用习惯。

修复后的实现代码 #

from contextlib import redirect_stdout, redirect_stdin
from io import StringIO, IOBase
import sys

from PyQt5.QtWidgets import QTextEdit
from PyQt5.QtCore import QObject, QThread, pyqtSignal, Qt, pyqtSlot

# 自定义可发信号的输出流类，实现实时输出
class EmittingStream(IOBase):
    def __init__(self, signal):
        self._signal = signal
        self._buffer = []

    def write(self, text):
        self._buffer.append(text)
        # 遇到换行符或常见提示后缀就触发一次输出，兼容input提示不带换行的场景
        if "\n" in text or ":" in text or "?" in text:
            self._signal.emit("".join(self._buffer))
            self._buffer.clear()
        return len(text)

    def flush(self):
        if self._buffer:
            self._signal.emit("".join(self._buffer))
            self._buffer.clear()

# 自定义输入流类，接收前端输入喂给input
class RecvStream(IOBase):
    def __init__(self):
        self._buffer = StringIO()
    
    def write(self, text):
        # 把用户输入写到缓冲里
        pos = self._buffer.tell()
        self._buffer.seek(0, 2)
        self._buffer.write(text)
        self._buffer.seek(pos)
    
    def readline(self, size=-1):
        # input默认调用readline获取输入
        return self._buffer.readline(size)

class Terminal(QTextEdit):
    inputReceived = pyqtSignal(str)
    def __init__(self, parent):
        super(Terminal, self).__init__(parent)
        self.setPlaceholderText("终端输入将在此处显示，输入完成按回车提交")
        self._stdin_buffer = RecvStream()
        # 允许用户编辑输入
        self.setReadOnly(False)
        self._last_output_pos = 0

    def keyPressEvent(self, event):
        # 拦截回车事件，提交输入内容
        if event.key() == Qt.Key_Return or event.key() == Qt.Key_Enter:
            # 只取最后一次输出之后的用户输入内容，避免历史输出被当作输入提交
            input_text = self.toPlainText()[self._last_output_pos:] + "\n"
            self.inputReceived.emit(input_text)
            self.append("")
            self._last_output_pos = self.textCursor().position()
            return
        # 禁止光标移动到历史输出区域修改内容
        cursor = self.textCursor()
        if event.key() in (Qt.Key_Left, Qt.Key_Up, Qt.Key_Backspace) and cursor.position() <= self._last_output_pos:
            return
        super().keyPressEvent(event)

    def run_func(self, func, *args, **kwargs):
        self.thread = QThread()
        self.worker = TerminalWorker(func, args, kwargs, self._stdin_buffer)
        self.worker.moveToThread(self.thread)
        self.thread.started.connect(self.worker.run)
        self.worker.finished.connect(self.thread.quit)
        self.worker.finished.connect(self.worker.deleteLater)
        self.thread.finished.connect(self.thread.deleteLater)
        self.worker.outputChanged.connect(self.update_terminal)
        self.inputReceived.connect(self._stdin_buffer.write)
        self.thread.start()

    @pyqtSlot(str)
    def update_terminal(self, text):
        # 追加输出而不是覆盖原有内容
        self.insertPlainText(text)
        # 自动滚动到底部
        self.verticalScrollBar().setValue(self.verticalScrollBar().maximum())
        self._last_output_pos = self.textCursor().position()


class TerminalWorker(QObject):
    finished = pyqtSignal()
    outputChanged = pyqtSignal(str)

    def __init__(self, func, args, kwargs, stdin_stream):
        QObject.__init__(self)
        self.func = func
        self.args = args
        self.kwargs = kwargs
        self.stdin_stream = stdin_stream

    def run(self):
        out_stream = EmittingStream(self.outputChanged)
        with redirect_stdout(out_stream), redirect_stdin(self.stdin_stream):
            self.func(*self.args, **self.kwargs)
        out_stream.flush()
        self.finished.emit()

补充说明 #

上述实现保留了原有线程逻辑，目标函数执行完全在子线程不会卡住UI，输入输出的跨线程通信全部通过Qt原生信号槽实现，不需要额外加锁，可以直接适配多轮input()调用的交互场景。

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