Flask中如何从Python后端views.py调用前端JavaScript函数

首先明确核心逻辑：普通HTTP请求是单向的，前端不发请求后端没法主动给前端推指令，你之前用setInterval轮询能跑通，本质就是前端主动拉取数据。要实现任务执行过程中触发前端更新，按改造成本从低到高选下面的方案就行。

先把你现有代码里的两个显性bug修了，不然换什么方案都容易出问题：

• 前端AJAX的id选择器写法错误：$(update_progress) 要改成 $('#update_progress')，jQuery的id选择器必须加#且传入字符串
• 后端updateprogress接口没必要套jsonify，你返回的是渲染好的HTML片段，直接返回模板渲染结果即可，套jsonify会让前端拿到结构异常的响应，替换DOM时出错

方案1：优化版短轮询（改造成本最低，无额外依赖） #

不用追求“后端主动触发”，把轮询逻辑绑定到任务生命周期上就行，实现只需要两步：

1. 用全局变量（生产环境换Redis等缓存）存储当前任务进度，doTasks每执行完一个步骤就更新这个进度值，/update_progress接口直接读取当前进度返回
2. 前端只在任务执行期间启动短间隔轮询，进度拉取到100%就自动停止轮询

代码示例：
views.py 相关代码：

# 全局存储当前任务进度，多用户场景请替换为绑定用户session的缓存存储
current_task_progress = 0

@views.route('/start_task', methods=['POST'])
def doTasks():
    global current_task_progress
    current_task_progress = 0

    # 第一阶段处理逻辑
    # ...... 你的业务代码
    current_task_progress = 20

    # 第二阶段处理逻辑
    # ...... 你的业务代码
    current_task_progress = 50

    # 第三阶段处理逻辑
    # ...... 你的业务代码
    current_task_progress = 80

    # 最终收尾逻辑
    # ...... 你的业务代码
    current_task_progress = 100
    return "任务执行完成"

@views.route('/update_progress', methods=['GET'])
def updateprogress():
    global current_task_progress
    return render_template('update_progress.html', x=current_task_progress)

home.html 相关JS代码：

let progressPollTimer = null;

function loadValue() {
    $.ajax({
        url: "/update_progress",
        type: "GET",
        success: function(data) {
            $('#update_progress').replaceWith(data);
            // 进度到100%自动停止轮询
            if (data.includes('100')) {
                clearInterval(progressPollTimer);
            }
        }
    });
}

// 绑定任务启动按钮，点击后启动任务同时开轮询
$('#taskStartBtn').on('click', function() {
    $.post('/start_task');
    // 每500ms拉一次进度，延迟完全感知不到
    progressPollTimer = setInterval(loadValue, 500);
});

这个方案10分钟就能改完，兼容所有浏览器，90%的内部系统、普通业务场景用这个完全够用，唯一的小缺点是进度更新最多有几百毫秒的延迟，对用户体验几乎无影响。

方案2：SSE服务器推送事件（无轮询延迟，后端可精确控制更新时机） #

如果接受不了轮询的固定间隔，就用SSE，后端可以在任务执行的任意节点主动向前端推送信号，前端收到信号就拉取最新进度，没有多余请求。

代码示例：
views.py 新增SSE监听接口：

from flask import Response
import queue

# 消息队列，多用户场景请替换为绑定用户session的独立队列
task_msg_queue = queue.Queue()

@views.route('/start_task', methods=['POST'])
def doTasks():
    # 第一阶段处理逻辑
    # ...... 你的业务代码
    task_msg_queue.put("update") # 主动推送更新信号

    # 第二阶段处理逻辑
    # ...... 你的业务代码
    task_msg_queue.put("update")

    # 任务执行结束
    task_msg_queue.put("done")
    return "任务执行完成"

@views.route('/task_listen')
def task_listen():
    def event_stream():
        while True:
            msg = task_msg_queue.get()
            yield f"data: {msg}\n\n"
            if msg == "done":
                break
    return Response(event_stream(), mimetype="text/event-stream")

home.html 新增SSE监听逻辑：

// 页面加载完成后建立SSE连接
const taskEvtSource = new EventSource('/task_listen');
taskEvtSource.onmessage = function(e) {
    if (e.data === 'update' || e.data === 'done') {
        loadValue(); // 收到后端信号就拉取最新进度
    }
}

这个方案不需要轮询，后端能精确控制更新时机，缺点是不支持IE浏览器，需要额外处理连接断开重连的逻辑。

方案3：WebSocket双向通信（适合复杂交互场景） #

如果后续还有更多实时双向交互需求，直接集成Flask-SocketIO做WebSocket连接，后端可以在任意时机向前端发送自定义事件，前端收到事件后触发进度更新，灵活度最高，缺点是需要引入额外依赖，部署时需要配置支持WebSocket的服务。

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