如何在另一Python脚本中访问textrecive.py函数内的body变量

首先修正你的textrecive.py中的几个问题，这些问题会导致服务无法正常运行：

1. 线程启动方式错误：Thread(target = app.run()).start()会直接执行阻塞的app.run()，线程无法正确创建，应改为Thread(target=app.run).start()；同时while True循环会重复创建线程导致端口占用，需要移除。
2. 视图函数返回值错误：Flask视图只能返回合法响应格式，多余的body会引发报错，应只返回响应对象。

修正后的textrecive.py基础版本：

from threading import Thread, Lock
import mimetypes
from flask import Flask, Response, request
from twilio.twiml.messaging_response import MessagingResponse

app = Flask(__name__)
# 用锁保证多线程下全局变量的线程安全
body_lock = Lock()
latest_body = ""

@app.route("/sms", methods=['GET', 'POST'])
def sms_reply():
    global latest_body
    body = request.values.get('Body', '').lower()
    who = request.values.get('From', '').lower()
    print(who)
    print(body)
    
    # 线程安全地更新最新短信内容
    with body_lock:
        latest_body = body
    
    resp = MessagingResponse()
    resp.message(input('what would you like to say'))
    return Response(str(resp), mimetype="application/xml")

if __name__ == '__main__':
    Thread(target=app.run, kwargs={'port':5000}).start()

下面提供三种跨脚本获取latest_body的方案：

方案一：新增Flask接口暴露短信内容 #

在textrecive.py中添加一个新路由，供其他脚本通过HTTP请求获取最新短信：

@app.route("/get-latest-sms", methods=['GET'])
def get_latest_sms():
    with body_lock:
        return latest_body

然后在另一个脚本中用requests库请求该接口：

import requests

def get_sms_content():
    try:
        resp = requests.get("http://localhost:5000/get-latest-sms")
        resp.raise_for_status()
        return resp.text
    except requests.exceptions.RequestException as e:
        print(f"获取短信失败: {e}")
        return ""

# 调用示例
sms_content = get_sms_content()
print(sms_content)

方案二：文件存储共享内容 #

每次收到短信时将内容写入本地文件，另一个脚本读取该文件：
修改textrecive.py的sms_reply函数：

@app.route("/sms", methods=['GET', 'POST'])
def sms_reply():
    body = request.values.get('Body', '').lower()
    who = request.values.get('From', '').lower()
    print(who)
    print(body)
    
    # 写入文件
    with open("latest_sms.txt", "w", encoding="utf-8") as f:
        f.write(body)
    
    resp = MessagingResponse()
    resp.message(input('what would you like to say'))
    return Response(str(resp), mimetype="application/xml")

另一个脚本读取文件：

def get_sms_content():
    try:
        with open("latest_sms.txt", "r", encoding="utf-8") as f:
            return f.read().strip()
    except FileNotFoundError:
        print("暂无短信内容")
        return ""

# 调用示例
sms_content = get_sms_content()
print(sms_content)

方案三：使用线程/进程安全的队列 #

同进程线程场景 #

修改textrecive.py：

from queue import Queue
# 初始化队列，存储最新短信（只保留最新一条）
sms_queue = Queue(maxsize=1)

@app.route("/sms", methods=['GET', 'POST'])
def sms_reply():
    body = request.values.get('Body', '').lower()
    who = request.values.get('From', '').lower()
    print(who)
    print(body)
    
    # 队列满时移除旧内容，存入新内容
    if sms_queue.full():
        sms_queue.get()
    sms_queue.put(body)
    
    resp = MessagingResponse()
    resp.message(input('what would you like to say'))
    return Response(str(resp), mimetype="application/xml")

同进程的另一个线程脚本中访问：

from textrecive import sms_queue

def get_sms_content():
    if not sms_queue.empty():
        return sms_queue.get()
    return ""

跨进程场景（用multiprocessing） #

需要将Flask服务作为子进程启动，用multiprocessing.Manager创建共享队列：

# textrecive.py
from multiprocessing import Manager
from threading import Thread
import mimetypes
from flask import Flask, Response, request
from twilio.twiml.messaging_response import MessagingResponse

def run_flask(sms_queue):
    app = Flask(__name__)
    
    @app.route("/sms", methods=['GET', 'POST'])
    def sms_reply():
        body = request.values.get('Body', '').lower()
        who = request.values.get('From', '').lower()
        print(who)
        print(body)
        
        if sms_queue.full():
            sms_queue.get()
        sms_queue.put(body)
        
        resp = MessagingResponse()
        resp.message(input('what would you like to say'))
        return Response(str(resp), mimetype="application/xml")
    
    app.run(port=5000)

if __name__ == '__main__':
    with Manager() as manager:
        sms_queue = manager.Queue(maxsize=1)
        flask_thread = Thread(target=run_flask, args=(sms_queue,))
        flask_thread.start()

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