如何将Python脚本（如pytest）的输出实时展示到Flask网页？

一、异步运行脚本并实时捕获输出 #

1. pytest --capture=tee-sys 参数用法 #

pytest默认会捕获标准输出/错误，仅在测试结束后统一展示。--capture=tee-sys的作用是同时将输出定向到终端和标准输出流，这样用subprocess调用时，能实时拿到pytest的输出内容。直接追加到pytest命令后即可：

pytest --capture=tee-sys your_test_module.py

2. 安全的subprocess实时读取方案（避免死锁） #

subprocess文档提到的readline()死锁，本质是因为子进程的stdout或stderr缓冲区填满后，会阻塞等待父进程读取。解决方式是合并stderr到stdout，结合select模块实现非阻塞读取：

import subprocess
import select

def capture_pytest_output():
    # 启动pytest，合并stderr到stdout，行缓冲保证实时输出
    proc = subprocess.Popen(
        ["pytest", "--capture=tee-sys", "test_sample.py"],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1
    )

    # 实时读取输出，直到进程结束
    while proc.poll() is None:
        # 用select监听stdout，超时0.1秒避免阻塞
        ready, _, _ = select.select([proc.stdout], [], [], 0.1)
        if ready:
            line = proc.stdout.readline()
            if line:  # 过滤空行
                yield line.strip()
    
    # 读取进程结束后剩余的输出
    for line in proc.stdout.readlines():
        if line:
            yield line.strip()

• bufsize=1：启用行缓冲，子进程输出一行就立即刷新到管道，保证实时性。
• stderr=subprocess.STDOUT：将错误输出合并到标准输出，减少流监听逻辑，简化代码。
• select.select()：非阻塞监听流状态，避免单一流阻塞导致子进程挂起。

二、无需前端请求的实时页面更新最佳实践 #

如果不想用WebSocket/socketio、轮询，**服务器发送事件（SSE）**是轻量化的标准方案，基于HTTP长连接，后端主动推送数据，前端无需额外依赖：

后端实现（以Flask为例） #

from flask import Flask, Response

app = Flask(__name__)

@app.route('/pytest-stream')
def pytest_stream():
    def generate():
        for line in capture_pytest_output():
            # SSE格式要求：data: 内容\n\n
            yield f"data: {line}\n\n"
        # 发送结束标记
        yield "data: __TEST_FINISHED__\n\n"
    
    # 返回text/event-stream类型响应，保持连接
    return Response(generate(), mimetype='text/event-stream')

@app.route('/')
def index():
    return """
    <html>
    <body>
        <h2>Pytest 实时测试进度</h2>
        <div id="output-container" style="height: 400px; overflow-y: auto; border: 1px solid #eee; padding: 10px;"></div>
        <script>
            const outputContainer = document.getElementById('output-container');
            const eventSource = new EventSource('/pytest-stream');
            
            eventSource.onmessage = (e) => {
                if (e.data === '__TEST_FINISHED__') {
                    eventSource.close();
                    const finishMsg = document.createElement('p');
                    finishMsg.style.color = '#2ecc71';
                    finishMsg.textContent = '测试执行完成！';
                    outputContainer.appendChild(finishMsg);
                    return;
                }
                const line = document.createElement('p');
                // 解析ANSI颜色码，还原终端彩色输出
                line.innerHTML = e.data.replace(/\\x1b\\[31m/g, '<span style="color: #e74c3c;">')
                                      .replace(/\\x1b\\[32m/g, '<span style="color: #2ecc71;">')
                                      .replace(/\\x1b\\[0m/g, '</span>');
                outputContainer.appendChild(line);
                // 自动滚动到底部
                outputContainer.scrollTop = outputContainer.scrollHeight;
            };
            
            eventSource.onerror = () => {
                eventSource.close();
                const errorMsg = document.createElement('p');
                errorMsg.style.color = '#e74c3c';
                errorMsg.textContent = '连接中断，请刷新页面重试';
                outputContainer.appendChild(errorMsg);
            };
        </script>
    </body>
    </html>
    """

if __name__ == '__main__':
    app.run(debug=True)

SSE的优势 #

• 无需额外安装依赖，基于标准HTML5 API和HTTP协议。
• 长连接推送，比轮询更高效，减少不必要的请求。
• 前端实现简单，仅需EventSource对象即可监听推送。

三、后续优化方向 #

1. 彻底解决死锁问题：

   • 始终用select或线程分别处理stdout/stderr，避免单一流阻塞。
   • 进程结束后务必读取剩余输出，避免缓冲区残留数据导致资源泄漏。

2. 过滤空行：

   • 在读取每一行时增加if line判断，过滤进程结束后readline()返回的空字符串，避免页面出现大量空行。

3. 增强用户体验：

   • 添加测试终止按钮，后端接收请求后调用proc.terminate()终止子进程。
   • 限制并发测试请求，避免服务器资源耗尽，可通过锁或任务队列实现。
   • 前端做日志分页或滚动限制，避免DOM节点过多导致页面卡顿。

4. 稳定性优化：

   • 处理子进程异常退出的情况，捕获OSError等异常，返回友好提示。
   • 给SSE连接添加心跳机制，避免代理或防火墙主动断开长连接。

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