Flask定时刷新抓取数据并更新页面方法

本文介绍如何使用 Flask-APScheduler 在 Flask 服务内部实现每小时自动重爬目标 URL、更新数据并实时反映到网页，避免手动重启或外部调度依赖。

本文介绍如何使用 Flask-APScheduler 在 Flask 服务内部实现每小时自动重爬目标 URL、更新数据并实时反映到网页，避免手动重启或外部调度依赖。

在构建动态内容展示型 Flask 应用（如新闻轮播页、实时资讯看板）时，若后端数据源（如 CMS 页面）频繁更新，但前端又需保持服务长期运行，关键挑战在于：如何让内存中的数据随源头变化而自动刷新，同时确保每次 HTTP 请求获取的是最新结果。直接在全局作用域执行一次爬虫（如原代码中 soup = BeautifulSoup(...)）会导致数据“冻结”——后续所有请求都返回初始快照，时间戳也不再更新。

解决思路是将数据获取逻辑封装为可复用的函数，并通过调度器周期性调用该函数，将结果存入线程安全的共享对象（如模块级变量或单例类属性）。以下是完整、可运行的优化方案：

✅ 推荐方案：Flask 内置调度（Flask-APScheduler）

相比 Windows 任务计划程序（需额外管理进程启停、状态同步）或纯 APScheduler（易与 Flask 生命周期冲突），Flask-APScheduler 是专为 Flask 设计的集成方案，自动处理应用上下文、信号绑定与异常恢复，更适合生产级轻量调度。

? 安装依赖

pip install flask flask-apscheduler beautifulsoup4 requests

? 完整可运行示例（适配你的新闻爬虫场景）

# app.py
from datetime import datetime
from flask import Flask, render_template
from flask_apscheduler import APScheduler
from bs4 import BeautifulSoup
import requests

# === 配置与初始化 ===
class Config:
    SCHEDULER_API_ENABLED = False  # 关闭内置 API（生产环境建议关闭）

app = Flask(__name__)
app.config.from_object(Config())
scheduler = APScheduler()
scheduler.init_app(app)
scheduler.start()

# === 共享数据容器（线程安全）===
class NewsData:
    titles = []
    pictures = []
    last_updated = datetime.now()

# === 核心爬虫函数（被调度器调用）===
def fetch_news():
    try:
        url = "https://news.clemson.edu/tag/extension/"
        headers = {"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()

        soup = BeautifulSoup(response.text, "html.parser")

        # 提取图片（data-src）
        pictures = []
        for img in soup.select('article img.lazyload'):
            src = img.get('data-src')
            if src and src.startswith('http'):
                pictures.append(src)

        # 提取标题
        titles = []
        for header in soup.select('article header'):
            h3 = header.find("h3", class_="entry-title bold")
            if h3 and h3.get_text(strip=True):
                titles.append(h3.get_text(strip=True))

        # 原子性更新共享数据
        NewsData.titles = titles[:5]  # 限制数量，防页面过长
        NewsData.pictures = pictures[:5]
        NewsData.last_updated = datetime.now()

        print(f"[✓] 成功刷新 {len(titles)} 条新闻，时间：{NewsData.last_updated}")

    except Exception as e:
        print(f"[✗] 爬取失败：{e}")

# === 注册定时任务（每小时执行一次）===
@scheduler.task("interval", id="fetch_news_job", hours=1)
def scheduled_fetch():
    fetch_news()

# === Flask 路由（始终返回最新数据）===
@app.route('/')
def home():
    return render_template(
        'home.html',
        titles=NewsData.titles,
        pictures=NewsData.pictures,
        current_time=NewsData.last_updated.strftime("%d/%m/%Y %H:%M:%S")
    )

# === 首次启动时预加载数据（避免首次访问空白）===
if __name__ == '__main__':
    fetch_news()  # 启动时立即抓一次
    app.run(host='0.0.0.0', port=5000, debug=False)  # 生产请改用 Gunicorn

? 模板示例（templates/home.html）

  
Latest News (Updated: {{ current_time }})
  {% for title, pic in zip(titles, pictures) %}
    

      
      
{{ title }}
    
  {% else %}
    
No news available.
  {% endfor %}

⚠️ 关键注意事项

• 线程安全：NewsData 类作为全局状态容器，fetch_news() 中的赋值是原子操作（列表/字符串替换），无需额外加锁；若涉及复杂状态（如字典更新），建议使用 threading.Lock。
• 错误隔离：爬虫异常不会中断调度器，try/except 确保单次失败不影响后续任务。
• 调试技巧：开发时可将 hours=1 临时改为 seconds=30 快速验证刷新效果。
• 生产加固：
  • 添加 requests 超时（已含）、User-Agent（已含）、反爬应对（如随机延迟、Session 复用）；
  • 使用 gunicorn 替代 app.run() 启动多进程服务（注意 APScheduler 默认不跨进程，需改用 Redis 存储任务状态）；
  • 敏感 URL 或 Headers 建议移至配置文件或环境变量。

此方案实现了“一个进程、零外部依赖、数据实时生效”的闭环，既满足你每小时刷新的核心需求，又保持了代码简洁性与可维护性。