Flask后端如何检测后台运行的网页爬取任务状态？

1. 给任务分配唯一标识与状态定义 #

• 给每个爬虫任务生成唯一task_id（比如用uuid.uuid4().hex），和表格中的每一行绑定（比如给表格行加data-task-id属性对应task_id）。
• 提前统一状态枚举：未运行、运行中、完成、停止，避免状态值混乱。

2. 选择状态存储方案 #

根据项目规模和需求选合适的存储方式：

• 本地调试/轻量项目：用Flask全局字典存储状态，比如在app初始化时定义app.config['TASK_STATUSES'] = {}，键为task_id，值为状态字符串。注意：重启服务后状态会丢失，仅适合测试场景。
• 生产环境：用Redis做状态存储，它的键值对读写速度快，还支持过期清理。比如用redis.set(f"task:{task_id}", "运行中")更新状态，redis.get(f"task:{task_id}")获取状态。如果需要持久化历史状态，可搭配SQLite/MySQL，建task_status表存储task_id、status、start_time、end_time等字段。

3. 后台任务的状态更新逻辑 #

不管用哪种方式运行后台爬虫，都要在关键节点主动更新状态：

• 任务启动前：将状态设为运行中并写入存储。
• 任务正常完成：更新状态为完成，可同步记录结束时间。
• 任务异常终止（爬取出错、手动停止）：捕获异常或监听停止信号，将状态设为停止。
• 未启动的任务：默认状态为未运行。

举个线程实现的简单示例：

import threading
import uuid
from flask import Flask, jsonify

app = Flask(__name__)
app.config['TASK_STATUSES'] = {}

def crawler_task(task_id):
    # 启动时更新状态
    app.config['TASK_STATUSES'][task_id] = "运行中"
    try:
        # 这里写你的爬虫逻辑
        # ...
        # 完成后更新状态
        app.config['TASK_STATUSES'][task_id] = "完成"
    except Exception as e:
        # 异常时设为停止
        app.config['TASK_STATUSES'][task_id] = "停止"

@app.route('/start-task')
def start_task():
    task_id = uuid.uuid4().hex
    # 开线程跑爬虫，不阻塞Flask请求
    threading.Thread(target=crawler_task, args=(task_id,)).start()
    return jsonify({"task_id": task_id})

@app.route('/get-status/<task_id>')
def get_status(task_id):
    status = app.config['TASK_STATUSES'].get(task_id, "未运行")
    return jsonify({"status": status})

4. 前端状态展示与更新 #

在Jinja2渲染的表格中添加状态列，用JS实现状态动态更新：

• 定时轮询方案：用setInterval每隔几秒请求Flask的状态接口，更新对应行的状态文本。示例JS代码：

  // 假设表格每行有data-task-id属性，状态单元格有status-cell类
  setInterval(() => {
      document.querySelectorAll('[data-task-id]').forEach(row => {
          const taskId = row.dataset.taskId;
          fetch(`/get-status/${taskId}`)
              .then(res => res.json())
              .then(data => {
                  row.querySelector('.status-cell').textContent = data.status;
              });
      });
  }, 3000); // 每3秒查询一次
  

• 实时推送方案：用Flask-SocketIO，当任务状态变化时主动推送给前端，无需轮询。比如爬虫任务更新状态时调用socketio.emit('status_update', {'task_id': task_id, 'status': status})，前端监听该事件直接更新表格。

5. 手动停止任务的处理 #

如果需要支持手动停止任务，需给每个任务保存可终止的标识：

• 用线程时：在全局字典中存储task_id对应的线程对象，停止接口触发时修改状态为停止，并终止线程（或让爬虫逻辑循环检查状态，主动退出）。
• 用Celery等任务队列时：通过celery.control.revoke(task_id)终止任务，同步更新状态为停止。

示例线程停止逻辑：

app.config['RUNNING_TASKS'] = {} # 存task_id: thread对象

def crawler_task(task_id):
    app.config['TASK_STATUSES'][task_id] = "运行中"
    while app.config['TASK_STATUSES'][task_id] == "运行中":
        # 爬虫循环逻辑，每次循环检查状态
        # ...
    app.config['TASK_STATUSES'][task_id] = "停止"

@app.route('/stop-task/<task_id>')
def stop_task(task_id):
    if task_id in app.config['TASK_STATUSES']:
        app.config['TASK_STATUSES'][task_id] = "停止"
    return jsonify({"msg": "任务已停止"})

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