在Django中运行Python脚本并在网页展示实时输出缓冲区

1. 基础准备 #

• 确保你的Django项目已正常启动，新建一个专用app（比如命名为script_runner），并将其添加到settings.py的INSTALLED_APPS列表中。
• 安全提示：仅运行你完全信任的固定脚本，绝对禁止直接执行用户上传或输入的未验证代码。如果需要支持自定义脚本，后续必须做沙箱隔离（比如Docker容器），先从固定脚本场景入手。

2. 后端实现：用WebSocket实时捕获脚本输出 #

HTTP是无状态协议，无法直接实现实时输出推送，这里用Django Channels实现WebSocket双向通信，实时传递脚本输出。

安装并配置Channels #

首先安装依赖：

pip install channels channels-redis

然后修改settings.py：

INSTALLED_APPS = [
    # ... 其他已有app
    'channels',
    'script_runner',
]

# 配置ASGI应用
ASGI_APPLICATION = '你的项目名.asgi.application'

# 配置通道层（生产环境用Redis，本地开发也可临时用InMemoryChannelLayer）
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [("127.0.0.1", 6379)],
        },
    },
}

更新项目根目录的asgi.py：

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
import script_runner.routing

os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目名.settings')

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AuthMiddlewareStack(
        URLRouter(
            script_runner.routing.websocket_urlpatterns
        )
    ),
})

编写WebSocket消费者 #

在script_runner目录下创建consumers.py：

import subprocess
import asyncio
import os
from channels.generic.websocket import AsyncWebsocketConsumer

class ScriptRunnerConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        await self.accept()

    async def disconnect(self, close_code):
        pass

    async def receive(self, text_data):
        # 替换为你的目标脚本路径
        script_path = os.path.join(os.path.dirname(__file__), 'test_script.py')
        
        # 启动子进程运行脚本，合并stdout和stderr
        process = await asyncio.create_subprocess_exec(
            'python', script_path,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True
        )

        # 实时读取输出并推送到前端
        while True:
            line = await process.stdout.readline()
            if not line:
                break
            await self.send(text_data=line.strip())
        
        # 等待进程结束，发送完成提示
        await process.wait()
        await self.send(text_data='=== 脚本执行完成 ===')

再创建routing.py定义WebSocket路由：

from django.urls import re_path
from . import consumers

websocket_urlpatterns = [
    re_path(r'ws/run-script/$', consumers.ScriptRunnerConsumer.as_asgi()),
]

3. 前端页面实现 #

在script_runner/templates目录下创建run_script.html：

<!DOCTYPE html>
<html>
<head>
    <title>实时运行脚本</title>
    <style>
        #output {
            width: 800px;
            height: 400px;
            border: 1px solid #ccc;
            padding: 10px;
            overflow-y: auto;
            margin-top: 10px;
        }
    </style>
</head>
<body>
    <button id="runBtn">点击运行脚本</button>
    <div id="output"></div>

    <script>
        const runBtn = document.getElementById('runBtn');
        const outputDiv = document.getElementById('output');
        let socket;

        runBtn.addEventListener('click', () => {
            // 初始化WebSocket连接
            socket = new WebSocket(`ws://${window.location.host}/ws/run-script/`);
            
            // 接收后端推送的输出内容
            socket.onmessage = function(e) {
                const line = document.createElement('p');
                line.textContent = e.data;
                outputDiv.appendChild(line);
                // 自动滚动到最新内容
                outputDiv.scrollTop = outputDiv.scrollHeight;
            };

            socket.onclose = function() {
                console.log('脚本执行连接已关闭');
            };

            // 发送消息触发脚本运行
            socket.send('start');
        });
    </script>
</body>
</html>

在script_runner/views.py中添加页面视图：

from django.shortcuts import render

def run_script_page(request):
    return render(request, 'run_script.html')

最后在script_runner/urls.py中配置页面路由：

from django.urls import path
from . import views

urlpatterns = [
    path('run-script/', views.run_script_page, name='run_script_page'),
]

4. 测试运行 #

1. 启动Redis服务器（如果用Redis通道层）
2. 运行Django服务：python manage.py runserver
3. 访问http://localhost:8000/run-script/，点击按钮即可看到脚本的实时输出。

关键注意事项 #

• 安全加固：如果要支持用户自定义脚本，必须用Docker等沙箱工具隔离运行环境，禁止脚本访问敏感资源。
• 异常处理：可在消费者中添加try-except块，捕获脚本运行错误并推送到前端，避免连接异常断开。
• 资源控制：限制同时运行的脚本进程数，防止服务器资源耗尽。

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