🐍 Python Day83：定时任务 — 让脚本按时自动运行

🕐 预计用时：3-4 小时 ｜ 🎯 目标：掌握 schedule、APScheduler 和 crontab 三种定时任务方案

📖 今日目录

1. 定时任务是什么？
2. schedule 库（简单场景）
3. APScheduler（生产环境）
4. 系统 crontab（Linux/Mac）
5. Windows 任务计划程序
6. 实战：自动化办公流水线
7. 今日练习
8. 今日小结

1. 定时任务是什么？

# 定时任务 = 让程序在指定时间自动执行
# 就像给程序设闹钟

# 常见场景：
# ✅ 每天早上 8:00 发送日报邮件
# ✅ 每小时抓取一次数据
# ✅ 每周一生成周报
# ✅ 每月 1 号生成工资表
# ✅ 每 5 分钟检查一次服务器状态

# 三种方案：
# 1. schedule    → 纯 Python，简单轻量，适合脚本
# 2. APScheduler → 功能强大，支持持久化，适合生产环境
# 3. crontab     → 系统级调度，最稳定，适合服务器

2. schedule 库

# 安装
# pip install schedule

import schedule
import time
from datetime import datetime

# === 基础用法 ===
def job():
    print(f'[{datetime.now().strftime("%H:%M:%S")}] 任务执行了！')

# 每 10 秒执行一次
schedule.every(10).seconds.do(job)

# 每 5 分钟执行一次
schedule.every(5).minutes.do(job)

# 每小时执行一次
schedule.every(1).hours.do(job)

# 每天 10:30 执行
schedule.every().day.at("10:30").do(job)

# 每天下午 6 点执行
schedule.every().day.at("18:00").do(job)

# 每周一执行
schedule.every().monday.do(job)

# 每周一 09:00 执行
schedule.every().monday.at("09:00").do(job)

# 每月 1 号执行（需要自定义）
schedule.every().day.at("00:00").do(lambda: job() if datetime.now().day == 1 else None)

# === 主循环 ===
print('⏰ 定时任务已启动，按 Ctrl+C 停止')
while True:
    schedule.run_pending()   # 检查并执行到期的任务
    time.sleep(1)            # 每秒检查一次

# === 传递参数 ===
def send_report(recipient, report_type):
    print(f'发送 {report_type} 报告给 {recipient}')

schedule.every().day.at("18:00").do(send_report, recipient='boss@company.com', report_type='日报')

# === 取消任务 ===
job = schedule.every().day.at("18:00").do(send_report, recipient='boss@company.com', report_type='日报')
schedule.cancel_job(job)

# === 取消所有任务 ===
schedule.clear()

# === 获取所有任务 ===
for job in schedule.get_jobs():
    print(f'任务: {job.job_func}, 下次执行: {job.next_run}')

# === 标签（方便管理）===
schedule.every().day.at("08:00").do(job).tag('morning', 'report')
schedule.every().day.at("18:00").do(job).tag('evening', 'report')

# 按标签取消
schedule.clear('morning')  # 取消所有 morning 标签的任务

💡 schedule 的局限：

• 只能在前台运行（程序关闭就停止）
• 不支持持久化（重启后任务丢失）
• 不支持并发（一个任务阻塞会影响其他任务）
• 适合简单的个人脚本，不适合生产环境

3. APScheduler

# 安装
# pip install apscheduler

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.date import DateTrigger
from datetime import datetime, timedelta

# === 三种调度器 ===
# 1. BlockingScheduler — 阻塞式（前台运行，适合独立脚本）
# 2. BackgroundScheduler — 后台式（不阻塞，适合 Web 应用）
# 3. AsyncIOScheduler — 异步式（配合 asyncio）

# === 三种触发器 ===
# 1. IntervalTrigger — 间隔触发（每 N 秒/分/小时）
# 2. CronTrigger — Cron 表达式触发（每天几点、每周几等）
# 3. DateTrigger — 指定时间触发（一次性）

# === 示例1：间隔执行 ===
scheduler = BlockingScheduler()

def my_job():
    print(f'[{datetime.now()}] 间隔任务执行')

scheduler.add_job(my_job, IntervalTrigger(seconds=30), id='job1')

# === 示例2：Cron 表达式 ===
def daily_report():
    print(f'[{datetime.now()}] 生成日报')

# 每天 18:00 执行
scheduler.add_job(daily_report, CronTrigger(hour=18, minute=0), id='daily_report')

# 每周一 09:00 执行
scheduler.add_job(daily_report, CronTrigger(day_of_week='mon', hour=9, minute=0), id='weekly_report')

# 每月 1 号 00:00 执行
scheduler.add_job(daily_report, CronTrigger(day=1, hour=0, minute=0), id='monthly_report')

# 每 5 分钟执行
scheduler.add_job(daily_report, CronTrigger(minute='*/5'), id='every_5min')

# 工作日每天 09:00-18:00 每小时执行
scheduler.add_job(daily_report, CronTrigger(day_of_week='mon-fri', hour='9-18'), id='workday_hourly')

# === 示例3：指定时间（一次性）===
scheduler.add_job(daily_report, DateTrigger(run_date='2026-05-13 10:00:00'), id='one_time')

# === 启动 ===
# scheduler.start()  # 取消注释后运行

# === 带参数的 Job ===
def send_email(to, subject, body):
    print(f'发送邮件: {subject} → {to}')

scheduler.add_job(
    send_email,
    CronTrigger(hour=18),
    args=['boss@company.com'],           # 位置参数
    kwargs={'subject': '日报', 'body': '...'},  # 关键字参数
    id='email_job',
)

# === Job 管理 ===
# 暂停
scheduler.pause_job('daily_report')

# 恢复
scheduler.resume_job('daily_report')

# 修改
scheduler.reschedule_job('daily_report', trigger=CronTrigger(hour=19))

# 删除
scheduler.remove_job('daily_report')

# 获取所有 Job
for job in scheduler.get_jobs():
    print(f'{job.id}: next_run={job.next_run_time}')

# === 事件监听 ===
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR

def job_listener(event):
    if event.exception:
        print(f'❌ Job {event.job_id} 执行失败: {event.exception}')
    else:
        print(f'✅ Job {event.job_id} 执行成功')

scheduler.add_listener(job_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)

📋 CronTrigger 参数速查

4. 系统 crontab

# crontab 是 Linux/Mac 的系统级定时任务调度器
# 最稳定、最可靠，服务器首选

# === crontab 表达式格式 ===
# ┌───────────── 分 (0-59)
# │ ┌───────────── 时 (0-23)
# │ │ ┌───────────── 日 (1-31)
# │ │ │ ┌───────────── 月 (1-12)
# │ │ │ │ ┌───────────── 星期 (0-6, 0=周日)
# │ │ │ │ │
# * * * * * command

# === 常用示例 ===
# 每分钟执行
# * * * * * /usr/bin/python3 /home/user/script.py

# 每 5 分钟执行
# */5 * * * * /usr/bin/python3 /home/user/script.py

# 每小时执行
# 0 * * * * /usr/bin/python3 /home/user/script.py

# 每天 18:00 执行
# 0 18 * * * /usr/bin/python3 /home/user/script.py

# 每天 9:00 和 18:00 执行
# 0 9,18 * * * /usr/bin/python3 /home/user/script.py

# 工作日 9:00 执行（周一到周五）
# 0 9 * * 1-5 /usr/bin/python3 /home/user/script.py

# 每月 1 号 00:00 执行
# 0 0 1 * * /usr/bin/python3 /home/user/script.py

# 每周一 09:00 执行
# 0 9 * * 1 /usr/bin/python3 /home/user/script.py

# === 管理 crontab ===

# 查看当前 crontab
# crontab -l

# 编辑 crontab
# crontab -e

# 删除所有 crontab（危险！）
# crontab -r

# === 用 Python 管理 crontab ===
# pip install python-crontab
from crontab import CronTab

# 创建 cron 对象（当前用户）
cron = CronTab(user=True)

# 添加任务
job = cron.new(command='/usr/bin/python3 /home/user/daily_report.py')
job.setall('0 18 * * *')  # 每天 18:00
job.set_comment('每日报告')

# 写入
cron.write()

# 查看所有任务
for job in cron:
    print(f'{job} | {job.comment}')

# 删除任务
cron.remove(job)
cron.write()

# 清空所有
cron.remove_all()
cron.write()

💡 crontab 注意事项：

• 使用绝对路径：Python 路径、脚本路径、文件路径都要用绝对路径
• 环境变量：cron 环境变量和终端不同，可能找不到包
• 日志：重定向输出到日志文件 >> /var/log/cron.log 2>&1
• 权限：确保脚本有执行权限 chmod +x script.py

5. Windows 任务计划程序

# Windows 没有 crontab，但有"任务计划程序"

# === 方式1：GUI 创建 ===
# 1. 搜索"任务计划程序"
# 2. 创建基本任务
# 3. 设置触发器（每天/每周/每月）
# 4. 设置操作（启动程序 → python.exe）
# 5. 设置参数（脚本路径）

# === 方式2：命令行创建 ===
# schtasks /create /tn "DailyReport" /tr "python C:\scripts\daily.py" /sc daily /st 18:00
# schtasks /query /tn "DailyReport"
# schtasks /delete /tn "DailyReport"

# === 方式3：Python subprocess ===
import subprocess

# 创建每日任务
subprocess.run([
    'schtasks', '/create',
    '/tn', 'PythonDailyReport',
    '/tr', r'python C:\scripts\daily_report.py',
    '/sc', 'daily',
    '/st', '18:00',
    '/f'  # 强制覆盖
], check=True)

# 查询任务
result = subprocess.run(['schtasks', '/query', '/tn', 'PythonDailyReport'], capture_output=True, text=True)
print(result.stdout)

# 删除任务
subprocess.run(['schtasks', '/delete', '/tn', 'PythonDailyReport', '/f'], check=True)

6. 实战：自动化办公流水线

"""
实战：完整的自动化办公流水线
每天 17:30 自动执行：
1. 从数据库/API 获取今日数据
2. 用 openpyxl 生成 Excel 报表
3. 用 email 发送报表给团队
4. 记录执行日志
"""
import schedule
import time
from datetime import datetime
import logging
import traceback

# 配置日志
logging.basicConfig(
    filename='automation.log',
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

def daily_pipeline():
    """每日自动化流水线"""
    logging.info('========== 流水线开始 ==========')

    try:
        # 步骤1：获取数据
        logging.info('步骤1：获取今日数据...')
        data = fetch_today_data()
        logging.info(f'  获取到 {len(data)} 条数据')

        # 步骤2：生成报表
        logging.info('步骤2：生成 Excel 报表...')
        report_file = generate_report(data)
        logging.info(f'  报表已生成: {report_file}')

        # 步骤3：发送邮件
        logging.info('步骤3：发送邮件...')
        send_report_email(report_file)
        logging.info('  邮件发送成功')

        # 步骤4：备份
        logging.info('步骤4：备份数据...')
        backup_data(data)
        logging.info('  备份完成')

        logging.info('========== 流水线完成 ✅ ==========')

    except Exception as e:
        logging.error(f'========== 流水线失败 ❌ ==========')
        logging.error(f'错误: {e}')
        logging.error(traceback.format_exc())

        # 发送告警邮件
        try:
            send_alert_email(f'自动化流水线失败: {e}')
        except Exception:
            pass

def fetch_today_data():
    """模拟获取数据"""
    # 实际项目中从数据库或 API 获取
    import pandas as pd
    return pd.DataFrame({'订单号': [1, 2], '金额': [100, 200]})

def generate_report(data):
    """模拟生成报表"""
    filename = f'report_{datetime.now().strftime("%Y%m%d")}.xlsx'
    # data.to_excel(filename, index=False)
    return filename

def send_report_email(filepath):
    """模拟发送邮件"""
    # 实际调用 smtplib
    pass

def send_alert_email(message):
    """模拟发送告警"""
    pass

def backup_data(data):
    """模拟备份"""
    pass

# === 注册定时任务 ===
schedule.every().day.at("17:30").do(daily_pipeline)

# 也可以测试时立即执行
# daily_pipeline()

# === 主循环 ===
print('🤖 自动化流水线已启动')
print('   每天 17:30 自动执行')
print('   按 Ctrl+C 停止')

while True:
    schedule.run_pending()
    time.sleep(30)

7. 今日练习

🏋️ 练习 1：简单定时器

# 用 schedule 实现：
# 每 10 秒打印一次当前时间
# 每分钟打印一次"一分钟过去了"
# 运行 3 分钟后自动停止

🏋️ 练习 2：APScheduler 实战

# 用 APScheduler 实现：
# 每天 9:00 打印"早安"
# 每天 18:00 打印"下班了"
# 工作日每小时执行一次检查任务
# 添加事件监听，记录执行日志

🏋️ 练习 3：自动化脚本

# 写一个完整的自动化脚本：
# 每天 18:00 自动读取数据文件
# 生成 Excel 报表
# 发送邮件给指定收件人
# 记录执行日志到文件

8. 今日小结

🚀 明日预告：Day 84 — 自动化办公工具箱综合项目

把 openpyxl + 邮件 + 定时任务组合起来，构建一个完整的自动化办公工具箱：Excel 报表自动生成 + 邮件自动发送 + 定时执行 + 日志记录。这是自动化阶段的终极实战！