from flask import Flask
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from datetime import datetime

app = Flask(__name__)
app.config['SCHEDULER_JOBSTORES'] = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.db')
}

# 創(chuàng)建調(diào)度器實例
scheduler = BackgroundScheduler(
    jobstores=app.config['SCHEDULER_JOBSTORES'],
    daemon=True
)

# 定義定時任務(wù)函數(shù)
def daily_report():
    """每天上午9點執(zhí)行的日報任務(wù)"""
    current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(f"[日報] 執(zhí)行時間: {current_time}")
    # 這里添加實際的日報生成邏輯

def every_5_minutes():
    """每5分鐘執(zhí)行一次的任務(wù)"""
    current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(f"[5分鐘任務(wù)] 執(zhí)行時間: {current_time}")
    # 這里添加實際的任務(wù)邏輯

def on_startup():
    """應(yīng)用啟動時添加定時任務(wù)"""
    # 添加每天9:00執(zhí)行的任務(wù)
    scheduler.add_job(
        id='daily_report_job',
        func=daily_report,
        trigger='cron',
        hour=9,
        minute=0,
        replace_existing=True
    )
    
    # 添加每5分鐘執(zhí)行的任務(wù)
    scheduler.add_job(
        id='every_5_minutes_job',
        func=every_5_minutes,
        trigger='interval',
        minutes=5,
        replace_existing=True
    )
    
    # 添加一次性任務(wù)（應(yīng)用啟動后10秒執(zhí)行）
    scheduler.add_job(
        id='startup_check',
        func=lambda: print("應(yīng)用啟動完成!"),
        trigger='date',
        run_date=datetime.now() + timedelta(seconds=10)
    )

# 啟動調(diào)度器
if not scheduler.running:
    scheduler.start()
    on_startup()

@app.route('/')
def home():
    return """
    <h1>Flask 定時任務(wù)示例</h1>
    <p>已啟動的定時任務(wù)：</p>
    <ul>
        <li>日報任務(wù)：每天9:00執(zhí)行</li>
        <li>5分鐘任務(wù)：每5分鐘執(zhí)行一次</li>
    </ul>
    <p>查看控制臺輸出獲取執(zhí)行詳情</p>
    """

@app.route('/jobs')
def list_jobs():
    """查看當(dāng)前所有定時任務(wù)"""
    jobs = scheduler.get_jobs()
    job_list = [f"ID: {job.id}, 下次執(zhí)行: {job.next_run_time}" for job in jobs]
    return "
".join(job_list)

@app.route('/add_job/<minutes>')
def add_job(minutes):
    """動態(tài)添加定時任務(wù)"""
    try:
        minutes = int(minutes)
        job_id = f"dynamic_job_{minutes}min"
        
        scheduler.add_job(
            id=job_id,
            func=lambda: print(f"動態(tài)添加的任務(wù) - 每 {minutes} 分鐘執(zhí)行一次"),
            trigger='interval',
            minutes=minutes,
            replace_existing=True
        )
        
        return f"成功添加每 {minutes} 分鐘執(zhí)行的任務(wù)!"
    except ValueError:
        return "參數(shù)錯誤，請輸入有效的分鐘數(shù)"

@app.route('/remove_job/<job_id>')
def remove_job(job_id):
    """移除定時任務(wù)"""
    if scheduler.get_job(job_id):
        scheduler.remove_job(job_id)
        return f"已移除任務(wù): {job_id}"
    return "任務(wù)不存在"

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

pip install flask apscheduler

trigger='cron', hour=9, minute=0  # 每天9:00執(zhí)行

trigger='interval', minutes=5  # 每5分鐘執(zhí)行

trigger='date', run_date=datetime.now() + timedelta(seconds=10)

jobstores={
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.db')
}

# 在應(yīng)用工廠函數(shù)中初始化調(diào)度器
def create_app():
    app = Flask(__name__)
    
    if os.environ.get('WERKZEUG_RUN_MAIN') == 'true' or not app.debug:
        # 確保只在主進(jìn)程中初始化調(diào)度器
        scheduler = BackgroundScheduler(daemon=True)
        scheduler.start()
        # 添加任務(wù)...
    
    return app

# Celery配置示例
from celery import Celery
from celery.schedules import crontab

app = Celery('tasks', broker='redis://localhost:6379/0')

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    # 每10分鐘執(zhí)行
    sender.add_periodic_task(600, task1.s(), name='每10分鐘任務(wù)')
    
    # 每天午夜執(zhí)行
    sender.add_periodic_task(
        crontab(hour=0, minute=0),
        task2.s(),
        name='每日午夜任務(wù)'
    )

@app.task
def task1():
    print("執(zhí)行每10分鐘任務(wù)")

@app.task
def task2():
    print("執(zhí)行每日任務(wù)")

def job_with_error_handling():
    try:
        # 業(yè)務(wù)代碼
        pass
    except Exception as e:
        app.logger.error(f"任務(wù)執(zhí)行失敗: {str(e)}")
        # 發(fā)送警報郵件
        send_alert_email(f"任務(wù)失敗: {str(e)}")
        
def send_alert_email(message):
    # 實現(xiàn)郵件發(fā)送邏輯
    pass