🐍 第19课：定时任务

—— 让程序自动干活

📌 本课目标

1️⃣ schedule 库

import schedule  # pip install schedule
import time

# 定义任务
def job_hello():
    print("Hello!")

def job_backup():
    print("执行备份...")

def job_report():
    print("生成日报...")

# 调度规则
schedule.every(10).seconds.do(job_hello)          # 每10秒
schedule.every(5).minutes.do(job_hello)            # 每5分钟
schedule.every().hour.do(job_hello)                   # 每小时
schedule.every().day.at("10:30").do(job_backup)     # 每天10:30
schedule.every().monday.at("09:00").do(job_report)  # 每周一9:00
schedule.every().wednesday.at("14:00").do(job_report)
schedule.every().hour.at(":17").do(job_hello)       # 每小时第17分钟

# 运行调度器
while True:
    schedule.run_pending()
    time.sleep(1)

# 带参数的任务
def greet(name):
    print(f"Hello, {name}!")

schedule.every(10).seconds.do(greet, name="张三")

# 取消任务
job = schedule.every(5).seconds.do(job_hello)
schedule.cancel_job(job)

# 清除所有任务
schedule.clear()

2️⃣ APScheduler 高级调度

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.date import DateTrigger
from datetime import datetime

scheduler = BackgroundScheduler()

# 1. 间隔触发
scheduler.add_job(
    job_backup,
    trigger='interval',
    minutes=30,
    id='backup_job',
    name='自动备份',
)

# 2. Cron 触发（最灵活）
scheduler.add_job(
    job_report,
    trigger='cron',
    hour=9,
    minute=0,
    day_of_week='mon-fri',
    id='daily_report',
    name='工作日日报',
)

# 3. 单次触发
scheduler.add_job(
    job_hello,
    trigger='date',
    run_date=datetime(2026, 6, 1, 9, 0, 0),
)

# 4. 带参数
scheduler.add_job(
    greet, 'interval', seconds=30,
    args=['张三'],
    kwargs={'title': 'Mr.'},
)

# 启动
scheduler.start()

# 查看所有任务
for job in scheduler.get_jobs():
    print(f"{job.id}: {job.name}, 下次执行: {job.next_run_time}")

# 暂停/恢复/删除
scheduler.pause_job('backup_job')
scheduler.resume_job('backup_job')
scheduler.remove_job('backup_job')

# 关闭
scheduler.shutdown(wait=True)

3️⃣ Cron 表达式

# Cron 表达式示例
# 每天上午9点       → hour=9, minute=0
# 工作日9点         → hour=9, day_of_week='mon-fri'
# 每月1号8点        → day=1, hour=8
# 每季度首月1号     → month='1,4,7,10', day=1
# 每5分钟           → minute='*/5'
# 每小时第0和30分钟  → minute='0,30'

4️⃣ 任务装饰器

from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

# 用装饰器注册任务
@scheduler.scheduled_job('cron', hour=9, minute=0, id='morning_report')
def morning_report():
    print("生成早报...")

@scheduler.scheduled_job('interval', minutes=30, id='health_check')
def health_check():
    print("健康检查...")

@scheduler.scheduled_job('cron', day_of_week='fri', hour=17, id='weekly_summary')
def weekly_summary():
    print("生成周报...")

# BlockingScheduler 会阻塞主线程
# scheduler.start()

5️⃣ 错误处理与日志

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_MISSED
import logging

# 配置日志
logging.basicConfig()
logger = logging.getLogger('apscheduler')
logger.setLevel(logging.DEBUG)

scheduler = BackgroundScheduler()

# 事件监听
def job_error_listener(event):
    if event.exception:
        logger.error(f"任务 {event.job_id} 失败: {event.exception}")
    elif event.code == EVENT_JOB_MISSED:
        logger.warning(f"任务 {event.job_id} 错过执行")

scheduler.add_listener(job_error_listener, EVENT_JOB_ERROR | EVENT_JOB_MISSED)

# 任务错误处理
def robust_job():
    try:
        # 可能失败的操作
        risky_operation()
    except Exception as e:
        logger.error(f"任务执行失败: {e}")
        # 可以发通知、记录数据库等
        notify_admin(f"定时任务失败: {e}")

6️⃣ 验证脚本

#!/usr/bin/env python3
"""第19课 定时任务验证"""
import schedule
import time

def test_schedule_basic():
    """schedule基本测试"""
    counter = {"value": 0}

    def increment():
        counter["value"] += 1

    schedule.clear()
    schedule.every(1).seconds.do(increment)

    # 手动运行3次
    for _ in range(3):
        schedule.run_pending()
        time.sleep(1.1)

    assert counter["value"] >= 2, f"Expected >=2, got {counter['value']}"
    schedule.clear()
    print("✅ schedule基本测试通过")

def test_schedule_at_time():
    """schedule定时测试"""
    counter = {"value": 0}

    def increment():
        counter["value"] += 1

    schedule.clear()
    schedule.every().day.at("23:59").do(increment)

    jobs = schedule.get_jobs()
    assert len(jobs) == 1
    schedule.clear()
    print("✅ schedule定时测试通过")

def test_apscheduler():
    """APScheduler测试"""
    from apscheduler.schedulers.background import BackgroundScheduler
    from datetime import datetime, timedelta

    counter = {"value": 0}

    def increment():
        counter["value"] += 1

    scheduler = BackgroundScheduler()
    scheduler.add_job(increment, 'interval', seconds=1, id='test')
    scheduler.start()

    time.sleep(3)
    scheduler.shutdown(wait=False)
    assert counter["value"] >= 2
    print("✅ APScheduler测试通过")

def test_cron_trigger():
    """Cron触发器测试"""
    from apscheduler.triggers.cron import CronTrigger

    trigger = CronTrigger(hour=9, minute=0, day_of_week='mon-fri')
    # 获取下次触发时间
    from datetime import datetime
    next_time = trigger.get_next_fire_time(None, datetime.now())
    assert next_time is not None
    print("✅ Cron触发器测试通过")

if __name__ == "__main__":
    test_schedule_basic()
    test_schedule_at_time()
    test_apscheduler()
    test_cron_trigger()
    print("\n🎉 第19课全部验证通过！")

🔑 本课要点