我想捕获长时间运行但冗长的脚本（目前是 pytest）的输出，并在网页上打印进度。我在这里面临两个问题
首先：我需要异步运行脚本并在脚本仍在运行时捕获输出。我发现 pytest 有
--capture=tee-syssubprocess.PopenPopen.communicat()TimeoutExpired第二：我需要在没有收到请求的情况下更新前端。起初，我什至认为拥有终端样式输出会很酷，并发现了像 ttyd 这样的东西，但这似乎有点过分了，如果我没有记错的话，将允许用户输入到终端，这是我不想要的。现在我还发现了这个答案，建议使用 iframe 和
flask 文档中的
stream_template_context()，但是如果我在这两种情况下都没有弄错的话，似乎我需要在使用从子流程收集的数据运行，感觉很容易出错。我还发现了 flask-socketio，这是我想避免的，因为 falsk 服务器应该在自定义 Linux 上运行，我需要首先添加这个模块。这最终应该是可行的，但如果有更简单的方法，我会更喜欢。我考虑过从前端 JavaScript 进行轮询的一个选项，但这看起来又是一个解决方法。这里最好的做法是什么？
对于这两部分，我很高兴有更多资源来阅读最佳实践等。经过更多研究，我得到了一个像这样的最小工作示例：
from flask import Flask, Response, render_template
from subprocess import Popen, PIPE
app = Flask(__name__)
@app.route('/content')
def content():
# start subprocess
def inner():
# proc = Popen(['/usr/local/bin/pytest', 'test1.py'], stdout=PIPE)
with Popen(['/usr/local/bin/pytest', 'test1.py'], stdout=PIPE) as proc:
while proc.poll() is None:
line = proc.stdout.readline()
# yield line.decode() + '<br/>\n'
yield str(line) + '<br/>\n'
return Response(inner(), mimetype='text/html')
@app.route('/')
def index():
return render_template('test.html')
if __name__ == '__main__':
app.run(debug=True)
其中test.py
只是定义了一系列测试函数，每个函数都休眠 一秒钟和
assert True只是为了获得所需的样本输出， 和一个
test.html：
<!doctype html>
<head>
<title>Title</title>
</head>
<body>
<div>
<iframe frameborder="1"
width="52%"
height="500px"
style='background: transparent;'
src="{{ url_for('content')}}">
</iframe>
</div>
</body>
但是，子进程文档指出，不鼓励像这样使用proc.stdout.readline()
，因为它可能会导致子进程死锁。 此外，在子进程终止后，我有时会得到一些，有时是很多空行。我使用
str(line)使它们可见，因为字节字符串中的
b''就在那里。 关于从这里去哪里有什么想法吗？
但首先：我想我误解了
子流程文档
中关于不使用
Popen.stdout.read()的警告。我只使用 readline我应该没问题，因为无论如何呼叫都应该返回。 （如果我在这里错了，请纠正我，因为整个答案是基于这样的假设，即
readline不会在
read的情况下陷入僵局。）第一个解决方案：修改
这个答案。
实际上并不需要 <iframe>
，Flask 可以将数据流式传输到
<div>。 html 看起来像这样：
<p>Test Output</p>
<div id="output">
{% if data %}
{% for item in data %}
{{ item }}<br />
{% endfor %}
{% endif %}
</div>
{% if data %}
是可选的。我用它来加载带有空
<div>的页面，然后按按钮重新加载网站并提供
data来填充
<div>
Python部分如下所示：
@app.route('/')
def stream() -> Response:
"""stream data to template"""
def update(task: str | list[str]):
"""update the stream data"""
with Popen(task, stdout=PIPE, stderr=PIPE) as proc:
while proc.poll() is None:
line = proc.stdout.readline()
log.info(line.decode())
yield line.decode()
while (line := proc.stdout.readline()) != b'':
log.info(line.decode())
yield line.decode()
err = proc.stderr.read() # risk of deadlock
# could be replaced by while readline as above
if err != b'':
log.warning("additionional stderr produced while running tests: %s", err.decode())
return Response(
stream_template('index.html.jinja',
data=update(['/usr/local/bin/pytest', 'dummytest.py'])))
我必须在 proc.poll()
返回后添加第二个 while 循环，因为我一直缺少最后一行数据。但我宁愿不单独使用第二个 while 循环，因为如果一行为空，这将有中断的风险（不应该发生，因为换行符应该始终存在）。第二个解决方案基本上是
这个答案利用EventStream
。我很犹豫要不要尝试这个，因为我对 javascript 不太流利，但最终这就像一个魅力。我在按钮上添加了一个事件处理程序，阻止表单
post发送到服务器，但手动打开 EventStream。 html 看起来像这样：
<div id="output"></div>
<p><form action="/action" method="post" role="form" id="testsForm">
<button class="btn btn-primary" type="submit">Run Test</button>
</form></p>
<script type="text/javascript">
(() => {
'use strict'
const form = document.getElementById("testsForm")
form.addEventListener('submit', event => {
event.preventDefault()
var target = document.getElementById("output")
var update = new EventSource("/action/stream")
update.onmessage = function(e) {
if (e.data == "close") {
update.close();
} else {
target.innerHTML += (e.data + '<br/>');
}
};
}, false)
})()
</script>
表单中的 action
和
method可能是完全可选的，因为表单永远不会传播到后端。 （这是使用bootstrap 类，如果有人想重新创建它） 这个Python看起来像这样：
@app.route("/action/stream")
def stream() -> Response:
"""open event stream to client"""
def update(task: str | list[str]):
"""update the event stream data"""
with Popen(task, stdout=PIPE, stderr=PIPE) as proc:
log.info("opened subprocess for async task communication")
while proc.poll() is None:
line = proc.stdout.readline()
log.info(line.decode().rstrip())
# the '\n\n' is needed for termination in th frontend
# the stream handling (especially closing) will break without it
yield 'data: ' + line.decode() + "\n\n"
while (line := proc.stdout.readline()) != b'':
log.info(line.decode())
yield 'data: ' + line.decode() + "\n\n"
err = proc.stderr.read() # risk of deadlock
if err != b'':
log.warning("additionional stderr produced while running script: %s", err.decode())
yield "data: close\n\n"
return Response(update(['/usr/local/bin/pytest', 'dummytest.py']), mimetype="text/event-stream")
此解决方案的一大优点是，网站无需重新加载即可传输数据。我有一个空框，按下按钮时，脚本的输出将显示在屏幕上。 （我什至找到了一些样式，让它看起来像一个老式学校显示器）
在更长的通信方面：我在 pytest 脚本中的一次测试中以 10 分钟超时测试了这两个版本，并且都没有问题。甚至可能有