如何在Flask中运行Python脚本

Question

我有一个使用 Flask 创建网站并动态打印一些数据的脚本。- 它要打印的数据应该来自另一个Python脚本。

目前我遇到的问题是，如果在执行运行Flask应用程序的代码行之前放置执行Python脚本的代码行，则会在不运行Flask的情况下运行Python脚本；反之亦然。

Python脚本：

import websocket
from bitmex_websocket import Instrument
from bitmex_websocket.constants import InstrumentChannels
from bitmex_websocket.constants import Channels
import json

websocket.enableTrace(True)

sells = 0
buys = 0


channels = [
    InstrumentChannels.trade,
]


XBTUSD = Instrument(symbol='XBTUSD',
                    channels=channels)
XBTUSD.on('action', lambda msg: test(msg))


def test(msg):
    parsed = json.loads(json.dumps(msg))


    print(parsed)

XBTUSD.run_forever()

Flask脚本（注意：价格应该是来自其他脚本的变量“parsed”）：

(Flask脚本（NB: 价格应为其他脚本中的变量“parsed”）:)

# Start with a basic flask app webpage.
from flask_socketio import SocketIO, emit
from flask import Flask, render_template, url_for, copy_current_request_context
from random import random
from time import sleep
from threading import Thread, Event
import requests, json
import time

__author__ = 'slynn'

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
app.config['DEBUG'] = True

#turn the flask app into a socketio app
socketio = SocketIO(app)

#random number Generator Thread
thread = Thread()
thread_stop_event = Event()

class RandomThread(Thread):
    def __init__(self):
        self.delay = 1
        super(RandomThread, self).__init__()

    def randomNumberGenerator(self):
        while not thread_stop_event.isSet():
            socketio.emit('newnumber', {'number': parsed}, namespace='/test')
            sleep(self.delay)

    def run(self):
        self.randomNumberGenerator()


@app.route('/')
def index():
    #only by sending this page first will the client be connected to the socketio instance
    return render_template('index.html')

@socketio.on('connect', namespace='/test')
def test_connect():
    # need visibility of the global thread object
    global thread
    print('Client connected')

    #Start the random number generator thread only if the thread has not been started before.
    if not thread.isAlive():
        print("Starting Thread")
        thread = RandomThread()
        thread.start()

@socketio.on('disconnect', namespace='/test')
def test_disconnect():
    print('Client disconnected')


if __name__ == '__main__':
    socketio.run(app)

- Jack022

---

请介绍一下你尝试过的一些代码。此外，您可以使用subprocess.call(["python","scriptname.py"], shell=true)在后台运行一个脚本，同时运行另一个脚本。 - Arkistarvh Kltzuonstev

我添加了代码，以便更具体！ - Jack022

Answer 1

使用 import：

• 将Python脚本（例如website_generator.py）生成的内容封装成一个函数。
• 将其放置在与您的app.py或flask.py相同的目录中。
• 在flask.py中使用from website_generator import function_name
• 使用function_name()来运行它

您可以使用其他函数，如subprocess.call等;尽管它们可能不会给出响应。

使用import的示例：

from flask import Flask
import your_module # this will be your file name; minus the `.py`

app = Flask(__name__)

@app.route('/')
def dynamic_page():
    return your_module.your_function_in_the_module()

if __name__ == '__main__':
    app.run(host='0.0.0.0', port='8000', debug=True)

- Julian Camilleri

---

我也尝试了，但由于某种原因，它仍然会先运行其中的一部分，我不知道为什么，我现在正在再次尝试！我添加了我的代码。 - Jack022

如果你真的在调用 test 函数，那么你必须使用 return 而不是 print。 - Julian Camilleri

你看不到这个导入是因为在它没有工作之后，我改变了它，以测试其他方法。预期输出是将XBTUSD.run_forever的数据实时打印在网页上，并进行异步更新。 - Jack022

抱歉没有说明清楚！我想要实现的是将从测试中解析出来的数据打印在我的网页上。我的两个主要问题是我在问题中指定的一个，以及我无法从不同的函数中调用变量“parsed”。 - Jack022

为什么不使用“队列”而不是调用“Lambda”？你可以简单地共享队列。 - Julian Camilleri

Answer 2

尝试这个：

from flask import Flask

app = Flask(__name__)

@app.route('/')
def run_script():
    file = open(r'/path/to/your/file.py', 'r').read()
    return exec(file)

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

- sujay simha