Rob*_*t16 9 html python django subprocess
我需要在我的 Django 应用程序中使用一个 HTML 网页来加载并在可滚动框中显示脚本的连续输出。这可能吗？
我目前正在使用子进程来运行 Python 脚本，但 HTML 页面在脚本完成后才会加载（这可能需要大约 5 分钟）。我希望用户看到正在发生的事情，而不仅仅是一个旋转的圆圈。
我已经用文本中的“\n”卸载了脚本的完整输出；如果可能，我希望它输出每个新行。
我的代码如下：
视图.py：
def projectprogress(request):
GenerateProjectConfig(request)
home = os.getcwd()
project_id = request.session['projectname']
staging_folder = home + "/staging/" + project_id + "/"
output = ""
os.chdir(staging_folder)
script = home + '/webscripts/terraformdeploy.py'
try:
output = subprocess.check_output(['python', script], shell=True)
except subprocess.CalledProcessError:
exit_code, error_msg = output.returncode, output.output
os.chdir(home)
return render(request, 'projectprogress.html', locals())
项目进度.html：
<style>
div.ex1 {
background-color: black;
width: 900px;
height: 500px;
overflow: scroll;
margin: 50px;
}
</style>
<body style="background-color: #565c60; font-family: Georgia, 'Times New Roman', Times, serif; color: white; margin:0"></body>
<div class="ex1">
{% if output %}<h3>{{ output }}</h3>{% endif %}
{% if exit_code %}<h3> The command returned an error: {{ error_msg }}</h3>{% endif %}
</div>
<div class="container">
<a class="button button--wide button--white" href="home.html" title="Home" style="color: white; margin: 60px;">
<span class="button__inner">
Home
</span>
</a>
</div>
</body>
</html>
您可以使用StreamingHttpResponse和Popen简化您的任务：
def test_iterator():
from subprocess import Popen, PIPE, CalledProcessError
with Popen(['ping', 'localhost'], stdout=PIPE, bufsize=1, universal_newlines=True) as p:
for line in p.stdout:
yield(line + '<br>') # process line here
if p.returncode != 0:
raise CalledProcessError(p.returncode, p.args)
def busy_view(request):
from django.http import StreamingHttpResponse
return StreamingHttpResponse(test_iterator())
StreamingHttpResponse需要一个迭代器作为其参数。迭代器函数是具有yield表达式（或生成器表达式）的函数，其返回值是生成器对象（迭代器）。
在此示例中，我只是回显 ping 命令来证明它有效。
替换['ping', 'localhost']为列表（如果将参数传递给命令，则它必须是列表 - 在本例中为localhost）。你原来的['python', script]应该可以用。
如果你想了解更多关于生成器的知识，我推荐 Trey Hunner 的演讲，并且强烈建议你阅读Fluent Python书的第 14 章。两者都是令人惊奇的来源。
免责声明：
性能考虑
Django 是为短期请求而设计的。流式响应将在整个响应期间绑定一个工作进程。这可能会导致性能不佳。
一般来说，您应该在请求-响应周期之外执行昂贵的任务，而不是诉诸流式响应。
您需要的是 websockets，或者 Django 中的 Channels。
https://channels.readthedocs.io/en/latest/
这允许您从后端向前端发送消息，而无需在前端拉取消息或重新加载页面。
值得一提的是，您还可以将输出流式传输到多个客户端，并将命令发送回后端。
适合您的代码的方法
请注意，这是未经测试的，因为我无权访问您的代码，因此您可能需要一些小的调整，但我相信提供的代码应该说明这个概念。
设置.py
INSTALLED_APPS = (
#Other installed Apps
'Channels',
)
CHANNEL_LAYERS = {
"default": {
"BACKEND": "asgiref.inmemory.ChannelLayer",
"ROUTING": "django_channels.routing.channel_routing",
},
}
routing.py（将文件添加到与 settings.py 相同的文件夹中）
from django_channels_app.consumers import message_ws, listener_add, listener_discconect
channel_routing = [
route("websocket.receive", message_ws),
route("websocket.disconnect", listener_discconect),
route("websocket.connect", listener_add),
]
在你的模块中：
import threading
from channels import Group
class PreserializeThread(threading.Thread):
def __init__(self, request, *args, **kwargs):
self.request = request
super(PreserializeThread, self).__init__(*args, **kwargs)
def run(self):
GenerateProjectConfig(request)
home = os.getcwd()
project_id = request.session['projectname']
staging_folder = home + "/staging/" + project_id + "/"
output = ""
os.chdir(staging_folder)
script = home + '/webscripts/terraformdeploy.py'
try:
output = subprocess.check_output(['python', script], shell=True)
Group("django_channels_group").send({
"text": output,
})
# NOTICE THIS WILL BLOCK;
# You could try the following, untested snippet
# proc = subprocess.Popen(['python', script], shell=True, #stdout=subprocess.PIPE)
#
# line = proc.stdout.readline()
# while line:
# line = proc.stdout.readline()
# Group("django_channels_group").send({
# "text": line,
# })
# Group("django_channels_group").send({
# "text": "Finished",
# })
except subprocess.CalledProcessError:
exit_code, error_msg = (
output.returncode,output.output)
os.chdir(home)
def listener_add(message):
Group("django_channels_group").add(
message.reply_channel)
def listener_discconect(message):
Group("django_channels_group").discard(
message.reply_channel)
def message_ws(message):
Group("django_channels_group").send({
"text": "My group message",
})
def projectprogress(request):
ProgressThread(request).start()
return render(request, 'projectprogress.html', locals())
html
<style>
div.ex1 {
background-color: black;
width: 900px;
height: 500px;
overflow: scroll;
margin: 50px;
}
</style>
<body style="background-color: #565c60; font-family: Georgia, 'Times New Roman', Times, serif; color: white; margin:0"></body>
<div id="output">
</div>
<div class="container">
<a class="button button--wide button--white" href="home.html" title="Home" style="color: white; margin: 60px;">
<span class="button__inner">
Home
</span>
</a>
</div>
</body>
</html>
<script>
socket = new WebSocket("ws://127.0.0.1:8000/"); #Or your server IP address
socket.onmessage = function(e) {
const data = JSON.parse(e.data);
document.querySelector('#ouput').value += (data.message + '\n');
}
socket.onopen = function() {
socket.send("Test message");
}
</script>
更通用的答案
后端：
聊天/消费者.py：
import json
from asgiref.sync import async_to_sync
from channels.generic.websocket import WebsocketConsumer
class ChatConsumer(WebsocketConsumer):
def connect(self):
self.room_name = self.scope['url_route']['kwargs']['room_name']
self.room_group_name = 'chat_%s' % self.room_name
# Join room group
async_to_sync(self.channel_layer.group_add)(
self.room_group_name,
self.channel_name
)
self.accept()
def disconnect(self, close_code):
# Leave room group
async_to_sync(self.channel_layer.group_discard)(
self.room_group_name,
selfreturn render(request, 'projectprogress.html', locals()).channel_name
)
def send_message(self, event):
message = event['message']
# Send message to WebSocket
self.send(text_data=json.dumps({
'message': message
}))
我的网站/settings.py：
# Channels
ASGI_APPLICATION = 'mysite.routing.application'
CHANNEL_LAYERS = {
'default': {
'BACKEND': 'channels_redis.core.RedisChannelLayer',
'CONFIG': {
"hosts": [('127.0.0.1', 6379)],
},
},
}
我的网站/路由.py：
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
import chat.routing
application = ProtocolTypeRouter({
# (http->django views is added by default)
'websocket': AuthMiddlewareStack(
URLRouter(
chat.routing.websocket_urlpatterns
)
),
})
聊天/路由.py：
from django.urls import re_path
from . import consumers
websocket_urlpatterns = [
re_path(r'ws/chat/(?P<room_name>\w+)/$', consumers.ChatConsumer),
]
前端：
<script>
const chatSocket = new WebSocket(
'ws://'
+ window.location.host
+ '/ws/chat/'
+ roomName
+ '/'
);
chatSocket.onmessage = function(e) {
const data = JSON.parse(e.data);
document.querySelector('#chat-log').value += (data.message + '\n');
};
</script>
| 归档时间： |
|
| 查看次数： |
656 次 |
| 最近记录： |