自动刷新 Flask 应用中的 CSV 数据：定时任务实现教程

pip install APScheduler

from flask import Flask
from apscheduler.schedulers.background import BackgroundScheduler
import time
import pandas as pd

def update_csv():
    """
    这个函数负责抓取数据并更新 CSV 文件。
    你需要在这里实现你的数据抓取逻辑。
    """
    print("Updating CSV file...")
    # 模拟数据抓取和 CSV 文件更新
    data = {'col1': [1, 2], 'col2': [3, 4]}
    df = pd.DataFrame(data)
    df.to_csv('data.csv', index=False) # 将数据保存到 CSV 文件
    print("CSV file updated.")

def create_app():
    app = Flask(__name__)
    app.config['SECRET_KEY'] = "abcjzllkk"

    # 创建调度器
    scheduler = BackgroundScheduler()
    # 添加定时任务，每 10 分钟执行一次 update_csv 函数
    scheduler.add_job(update_csv, 'interval', minutes=10)
    # 启动调度器
    scheduler.start()

    from .views import views
    from .auth import auth

    app.register_blueprint(views, url_prefix="/")
    app.register_blueprint(auth, url_prefix="/")

    return app

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

import fcntl

def update_csv():
    """
    更新 CSV 文件，并使用文件锁定机制。
    """
    print("Updating CSV file...")
    try:
        with open('data.csv', 'w') as f:
            # 获取文件锁
            fcntl.flock(f.fileno(), fcntl.LOCK_EX)
            # 模拟数据抓取和 CSV 文件更新
            data = {'col1': [1, 2], 'col2': [3, 4]}
            df = pd.DataFrame(data)
            df.to_csv(f, index=False)
            # 释放文件锁
            fcntl.flock(f.fileno(), fcntl.LOCK_UN)
            print("CSV file updated.")
    except Exception as e:
        print(f"Error updating CSV: {e}")
    finally:
        if 'f' in locals() and not f.closed:
            # 确保文件锁被释放
            fcntl.flock(f.fileno(), fcntl.LOCK_UN)
            f.close()