支持 input 函数的在线 python 运行环境 - 基于队列

2024-01-08 11:59:52

支持 input 函数的在线 python 运行环境 - 基于队列

本文提供了一种方式来实现支持 input 函数，即支持用户输的在线 python 运行环境。效果如下图所示：

在这里插入图片描述

思路

前端使用一个数组 input_queue 记录用户从点击运行按钮到现在的所有输入
点击运行按钮时将 code、input_queue 传给后端
后端将参数传给执行 python 代码的子进程
子进程重写 input() 函数，假设新的实现为 input_wrapper，代码如下，到用户代码运行到 input() 函数时，会执行重写的 input_wrapper()，在 input_wrapper 中获取到 input_queue，然后使用 input_queue.pop(0) 弹出用户输入最早的信息，如果 input_queue 为空，则说明需要用户输入，通过抛出 InputRequestException 异常的方式通知后端
```
def input_wrapper(prompt=''):
    if input_queue:
        input_str = input_queue.pop(0)
        sys.stdout.write(str(prompt) + input_str + "\n")
        return input_str
    raise InputRequestException(str(prompt))
```
后端通过处理子进程的标准输出、标准错误，知晓需要用户输入，然后向前端返回以下 json，event.type 为 input_request 代表需要用户输入，prompt 是提示信息
```
{
    "is_timeout": false,
    "done": false,
    "event": {
        "type": "input_request",
        "prompt": "请输入姓名："
    }
}
```
前端弹出弹框提示用户输入，用户输入并点击继续执行按钮时，会将本次的输入追加到 input_queue 的末尾，然后再次调用运行接口，这样循环往复直到程序结束

在执行以下代码时，可能需要两次用户输入，也可能需要三次。

name = input("请输入姓名：")
print("姓名：", name)

if name == "tom":
    age = input("请输入年龄：")
    print("年龄：", age)

gender = input("请输入性别：")
print("性别：", gender)

两次用户输入

点击运行按钮

请求参数中的 input_queue 为 []

{
    "code": "name = input(\"请输入姓名：\")\nprint(\"姓名：\", name)\n\nif name == \"tom\":\n    age = input(\"请输入年龄：\")\n    print(\"年龄：\", age)\n\ngender = input(\"请输入性别：\")\nprint(\"性别：\", gender)",
    "input_queue": []
}

返回值
{
    "is_timeout": false,
    "done": false,
    "event": {
        "type": "input_request",
        "prompt": "请输入姓名："
    }
}

输入 jack

请求参数中的 input_queue 为 ["jack"]

{
    "code": "name = input(\"请输入姓名：\")\nprint(\"姓名：\", name)\n\nif name == \"tom\":\n    age = input(\"请输入年龄：\")\n    print(\"年龄：\", age)\n\ngender = input(\"请输入性别：\")\nprint(\"性别：\", gender)",
    
    "input_queue": [
        "jack"
    ]
}

返回值
{
    "is_timeout": false,
    "done": false,
    "event": {
        "type": "input_request",
        "prompt": "请输入性别："
    }
}

输入男

请求参数中的 input_queue 为 ["jack", "男"]
{
    "code": "name = input(\"请输入姓名：\")\nprint(\"姓名：\", name)\n\nif name == \"tom\":\n    age = input(\"请输入年龄：\")\n    print(\"年龄：\", age)\n\ngender = input(\"请输入性别：\")\nprint(\"性别：\", gender)",
    "input_queue": [
        "jack",
        "男"
    ]
}

返回值
{
    "is_timeout": false,
    "done": true,
    "output": "请输入姓名：jack\r\n姓名： jack\r\n请输入性别：男\r\n性别： 男\r\n"
}

三次用户输入

点击运行按钮

请求参数中的 input_queue 为 []
{
    "code": "name = input(\"请输入姓名：\")\nprint(\"姓名：\", name)\n\nif name == \"tom\":\n    age = input(\"请输入年龄：\")\n    print(\"年龄：\", age)\n\ngender = input(\"请输入性别：\")\nprint(\"性别：\", gender)",
    "input_queue": []
}

返回值
{
    "is_timeout": false,
    "done": false,
    "event": {
        "type": "input_request",
        "prompt": "请输入姓名："
    }
}

输入 tom

请求参数中的 input_queue 为 ["tom"]
{
    "code": "name = input(\"请输入姓名：\")\nprint(\"姓名：\", name)\n\nif name == \"tom\":\n    age = input(\"请输入年龄：\")\n    print(\"年龄：\", age)\n\ngender = input(\"请输入性别：\")\nprint(\"性别：\", gender)",
    "input_queue": [
        "tom"
    ]
}

返回值
{
    "is_timeout": false,
    "done": false,
    "event": {
        "type": "input_request",
        "prompt": "请输入年龄："
    }
}

输入 18

请求参数中的 input_queue 为 ["tom", "18"]
{
    "code": "name = input(\"请输入姓名：\")\nprint(\"姓名：\", name)\n\nif name == \"tom\":\n    age = input(\"请输入年龄：\")\n    print(\"年龄：\", age)\n\ngender = input(\"请输入性别：\")\nprint(\"性别：\", gender)",
    "input_queue": [
        "tom",
        "18"
    ]
}

返回值
{
    "is_timeout": false,
    "done": false,
    "event": {
        "type": "input_request",
        "prompt": "请输入性别："
    }
}

输入男

请求参数中的 input_queue 为 ["tom", "18", "男"]
{
    "code": "name = input(\"请输入姓名：\")\nprint(\"姓名：\", name)\n\nif name == \"tom\":\n    age = input(\"请输入年龄：\")\n    print(\"年龄：\", age)\n\ngender = input(\"请输入性别：\")\nprint(\"性别：\", gender)",
    "input_queue": [
        "tom",
        "18",
        "男"
    ]
}

返回值
{
    "is_timeout": false,
    "done": true,
    "output": "请输入姓名：tom\r\n姓名： tom\r\n请输入年龄：18\r\n年龄： 18\r\n请输入性别：男\r\n性别： 男\r\n"
}

实现

前端使用 vue + element ui

<!DOCTYPE html>
<html lang="" style="height: 100%;">
<head>
    <meta charset="UTF-8">
    <link rel="stylesheet" href="./element-ui/index.css">
    <title>在线 python 执行</title>
</head>
<body style="height: 100%;margin: 0;">
<div id="app" style="height: 98%;width: 98%;padding: 5px">
    <el-input
            type="textarea"
            :autosize="{ minRows: 10, maxRows: 100}"
            placeholder="请输入代码"
            v-model="code">
    </el-input>
    <el-button type="primary" style="margin-top: 5px;margin-bottom: 5px" @click="exec()">运行</el-button>
    <el-input
            type="textarea"
            :autosize="{ minRows: 10, maxRows: 100}"
            placeholder="运行结果"
            v-model="result">
    </el-input>
</div>
</body>
<script src="./axios.min.js"></script>
<script src="./vue.js"></script>
<script src="./element-ui/index.js"></script>
<script>

    new Vue({
        el: '#app',
        mounted() {
        },
        methods: {
            exec() {
                const params = {
                    code: this.code,
                    input_queue: this.input_queue
                }
                axios.post('http://localhost:8080/exec', params).then(res => {
                    console.log("exec", res)

                    if (res.data.done) {
                        // 执行结束了，需要清空队列
                        this.clearQueue()

                        if (res.data.is_timeout) {
                            // 执行超时
                            this.$message("执行超时");
                        } else {
                            // 正常执行结束
                            this.result = res.data.output
                        }
                    } else {
                        // 执行中，需要用户输入
                        const event = res.data.event
                        if (event.type === 'input_request') {
                            // 弹框提示用户输入
                            this.$prompt(event.prompt, '输入', {
                                confirmButtonText: '继续执行',
                                cancelButtonText: '终止执行',
                                showClose: false,
                                closeOnClickModal: false,
                                closeOnPressEscape: false
                            }).then(({value}) => {
                                // 继续执行，将本次输入的信息追加进队列，然后再次执行
                                this.input_queue.push(value)
                                this.exec()
                            }).catch((action) => {
                                // 终止执行，需要清空队列
                                console.log("action ", action)
                                this.clearQueue()
                                this.$message("终止执行")
                            });
                        }
                    }
                })
            },
            clearQueue() {
                this.input_queue = []
            }
        },
        data() {
            return {
                code:
`name = input("请输入姓名：")
print("姓名：", name)

if name == "tom":
    age = input("请输入年龄：")
    print("年龄：", age)

gender = input("请输入性别：")
print("性别：", gender)
`,
                input_queue: [],
                result: null,
            }
        }
    })

</script>
</html>

Windows 环境的执行器

import json
import os
import subprocess
import threading
from threading import Timer

import psutil


class AbstractExecutor:

    def __init__(self, param):
        # param 包括 code、input_queue
        self.param = param
        # 用于保护 is_timeout 的锁
        self.lock = threading.Lock()
        # 是否执行超时了
        self.is_timeout = None

    def timeout_callback(self, p: subprocess.Popen):
        """
        执行超时时的回调，会终止执行 python 代码的进程组
        :param p: 执行 python 代码的进程
        """
        with self.lock:
            if self.is_timeout is None:
                self.is_timeout = True

        if self.is_timeout:
            try:
                # 终止执行 python 代码的进程组
                self.terminating_process_group(p)
            except Exception as e:
                print("超时回调异常, error: %s", e)

    def terminating_process_group(self, p: subprocess.Popen):
        """
        终止进程 p 及其子进程
        :param p: 要终止的进程
        """
        raise NotImplementedError()

    def create_popen(self) -> subprocess.Popen:
        """
        创建 subprocess.Popen，必须将 stderr 重定向到 stdout
        """
        raise NotImplementedError()

    def output(self, stdout):
        if stdout is not None:
            return stdout.decode("utf-8")
        else:
            return ""

    def execute(self):
        p = self.create_popen()
        timer = Timer(3, self.timeout_callback, [p])
        timer.start()
        try:
            # 从标准输入传入 json 参数：code、input_queue
            p.stdin.write(json.dumps(self.param).encode(encoding="utf-8"))

            stdout, stderr = p.communicate()

            with self.lock:
                if self.is_timeout is None:
                    self.is_timeout = False

        finally:
            timer.cancel()
        return self.is_timeout, self.output(stdout)


class WindowsExecutor(AbstractExecutor):

    __output_prefix = "Active code page: 65001\r\n"

    def create_popen(self) -> subprocess.Popen:
        filename = r"D:\project\python\online-python-code-executor\queue-base\exec_py.py"
        cmd = 'chcp 65001 & set PYTHONIOENCODING=utf-8 & python ' + filename

        # 将 stderr 重定向到了 stdout
        return subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                                shell=True)

    def terminating_process_group(self, p: subprocess.Popen):
        proc_pid = p.pid
        parent_proc = psutil.Process(proc_pid)
        for child_proc in parent_proc.children(recursive=True):
            print(child_proc.pid)
            child_proc.kill()
        parent_proc.kill()
        print(parent_proc.pid)

    def output(self, stdout):
        output = super().output(stdout)
        if output.startswith(self.__output_prefix):
            return output.removeprefix(self.__output_prefix)
        else:
            return output


if os.name == "nt":
    executor_cls = WindowsExecutor


def execute(param):

    # 执行用户代码
    is_timeout, stdout = executor_cls(param).execute()

    if is_timeout:
        # 执行超时了
        return {
            "is_timeout": is_timeout,
            "done": True,
            "output": stdout,
        }
    else:
        arr = stdout.split("InputRequestException")
        if len(arr) > 1:
            # 需要用户输入
            return {
                "is_timeout": is_timeout,
                "done": False,
                "event": {
                    "type": "input_request",
                    "prompt": arr[-1]
                }
            }
        else:
            # 正常执行结束
            return {
                "is_timeout": is_timeout,
                "done": True,
                "output": stdout,
            }

子进程需要执行的代码

import json
import sys


input_queue = []


class InputRequestException(Exception):
    """
    抛出此异常表示需要用户输入
    """
    pass


def execute(param):
    # 重写 input 函数
    __builtins__.input = input_wrapper

    # input_queue
    global input_queue
    input_queue = param["input_queue"]
    try:
        # 执行代码
        exec(param["code"])
    except InputRequestException as e:
        # 如果需要用户输入，则直接退出
        sys.stdout.write("\n" + "InputRequestException" + e.args[0])
        exit()


def input_wrapper(prompt=''):
    # 从 input_queue 中弹出
    if input_queue:
        input_str = input_queue.pop(0)
        sys.stdout.write(str(prompt) + input_str + "\n")
        return input_str
    # 需要用户输入
    raise InputRequestException(str(prompt))


if __name__ == '__main__':
    # 从标准输入读取 json 参数：code、input_queue
    arg = sys.stdin.read()
    # 执行
    execute(json.loads(arg))

代码仓库

online-python-code-executor/queue-base (github.com)

参考

https://pythontutor.com
https://github.com/seamile/PyTutor

文章来源:https://blog.csdn.net/qq_20919883/article/details/135402325
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系我的编程经验分享网邮箱：veading@qq.com进行投诉反馈，一经查实，立即删除！