python脚本

5分钟用python定时自动执行任务

发表于 2025年12月03日阅读 19 评论 0

大家好，我是何三，80后老猿，独立开发者。

继续我们的"5分钟Python自动化"系列。今天我们来解决一个让效率翻倍的刚需——定时自动执行任务。

有这样的场景：你每天都要手动备份重要文件，每周要定时发送工作报告，每月需要清理临时文件，或者你想在半夜网络空闲时下载大文件。这些重复性的定时任务占据了宝贵的时间，还容易忘记执行...

传统做法：设置闹钟提醒，手动执行，或者依赖操作系统的任务计划（配置复杂，跨平台不兼容）。

Python做法：10行代码，创建跨平台的定时任务系统，一次编写，永久自动运行！

今天，我就带你用5分钟，掌握这个让你的程序"活起来"的神奇技能！

环境准备：Python原生支持

在开始之前，你只需要：

安装Python（这个已经装好了）
一个文本编辑器

特别说明：今天的内容主要使用Python标准库，不需要安装任何第三方库！

核心代码：10行定时任务框架

让我们先看看这个最简单的定时任务系统：

import time
import schedule

def daily_task():
    """每天执行的任务"""
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 执行每日任务...")
    # 这里写你的实际任务代码
    # 比如：备份文件、发送邮件等

# 安装schedule库（如果还没安装）
# pip install schedule

# 设置定时规则：每天10:30执行
schedule.every().day.at("10:30").do(daily_task)

print("定时任务已启动，等待执行...")

# 保持程序运行，检查定时任务
while True:
    schedule.run_pending()
    time.sleep(1)

逐行详解：理解定时任务的原理

第1-2行：导入核心工具箱

import time
import schedule

time：Python内置的时间模块
schedule：轻量级的定时任务库（需安装：pip install schedule）

第4-8行：定义要执行的任务函数

def daily_task():
    """每天执行的任务"""
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 执行每日任务...")
    # 这里写你的实际任务代码

这是一个普通的Python函数，你可以在这里写任何想要定时执行的代码。

第11行：设置定时规则

schedule.every().day.at("10:30").do(daily_task)

这是schedule库最优雅的部分，语法接近自然语言： - schedule.every()：每个... - .day：天 - .at("10:30")：在10:30 - .do(daily_task)：执行daily_task函数

第13-17行：启动任务调度器

print("定时任务已启动，等待执行...")

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

schedule.run_pending()：检查是否有到期的任务需要执行
time.sleep(1)：每秒检查一次，避免CPU占用过高

实战演示：创建你的第一个定时任务

安装schedule库： bash pip install schedule
保存脚本：将上面的代码保存为scheduler.py
运行脚本：打开命令行，输入： bash python scheduler.py
查看效果：程序会持续运行，在每天的10:30执行你定义的任务。

运行效果：

定时任务已启动，等待执行...
[2024-01-15 10:30:00] 执行每日任务...
[2024-01-16 10:30:00] 执行每日任务...
[2024-01-17 10:30:00] 执行每日任务...
...（每天自动执行）

举一反三：丰富的定时规则

schedule库提供了极其灵活的定时规则，语法直观易懂：

案例1：多种时间间隔

import schedule
import time

def job():
    print("执行任务...")

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

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

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

# 每周三的13:15执行
schedule.every().wednesday.at("13:15").do(job)

# 每5到10秒之间随机执行
schedule.every(5).to(10).seconds.do(job)

案例2：带参数的定时任务

def personalized_greeting(name, time_of_day):
    print(f"你好，{name}！现在是{time_of_day}")

# 传递参数给任务函数
schedule.every().day.at("09:00").do(personalized_greeting, "张三", "早上")
schedule.every().day.at("18:00").do(personalized_greeting, "李四", "晚上")

案例3：定时备份文件

import shutil
import datetime
import os

def backup_files():
    """每天凌晨备份重要文件"""
    source_folder = "/path/to/important/files"
    backup_folder = f"/backups/{datetime.date.today()}"

    # 创建备份文件夹
    os.makedirs(backup_folder, exist_ok=True)

    # 复制文件
    for filename in os.listdir(source_folder):
        source = os.path.join(source_folder, filename)
        if os.path.isfile(source):
            shutil.copy2(source, backup_folder)

    print(f"[{datetime.datetime.now()}] 文件备份完成：{backup_folder}")

# 每天凌晨2点执行备份
schedule.every().day.at("02:00").do(backup_files)

案例4：定时发送邮件报告

import smtplib
from email.mime.text import MIMEText
import datetime

def send_daily_report():
    """发送每日工作报告"""
    report_content = f"""
    每日工作报告
    日期：{datetime.date.today()}

    1. 今日完成工作：
       - 完成项目A的开发
       - 修复了3个bug
       - 召开了团队会议

    2. 明日计划：
       - 继续项目B的开发
       - 代码review

    报告生成时间：{datetime.datetime.now()}
    """

    # 创建邮件
    msg = MIMEText(report_content, 'plain', 'utf-8')
    msg['Subject'] = f'每日工作报告 - {datetime.date.today()}'
    msg['From'] = 'your_email@example.com'
    msg['To'] = 'boss@example.com'

    # 发送邮件（需要配置SMTP服务器）
    # with smtplib.SMTP('smtp.example.com', 587) as server:
    #     server.login('username', 'password')
    #     server.send_message(msg)

    print(f"[{datetime.datetime.now()}] 日报已生成（模拟发送）")

# 每个工作日下午5:30发送
schedule.every().monday.at("17:30").do(send_daily_report)
schedule.every().tuesday.at("17:30").do(send_daily_report)
schedule.every().wednesday.at("17:30").do(send_daily_report)
schedule.every().thursday.at("17:30").do(send_daily_report)
schedule.every().friday.at("17:30").do(send_daily_report)

进阶技巧：专业级定时任务

使用Python内置的sched模块（无需安装）

import sched
import time
from datetime import datetime

def scheduled_task(task_name):
    print(f"[{datetime.now()}] 执行任务: {task_name}")

# 创建调度器
scheduler = sched.scheduler(time.time, time.sleep)

def schedule_task(delay, task_name):
    """安排一个延迟执行的任务"""
    scheduler.enter(delay, 1, scheduled_task, argument=(task_name,))
    print(f"任务 '{task_name}' 将在 {delay} 秒后执行")

# 安排多个任务
schedule_task(10, "10秒后执行的任务")
schedule_task(30, "30秒后执行的任务")
schedule_task(60, "1分钟后执行的任务")

print("开始执行调度器...")
scheduler.run()

使用APScheduler（企业级调度）

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

def job_function():
    print(f"[{datetime.now()}] 执行定时任务")

# 创建调度器
scheduler = BlockingScheduler()

# 添加定时任务
scheduler.add_job(job_function, 'cron', hour=9, minute=30)  # 每天9:30
scheduler.add_job(job_function, 'interval', minutes=30)     # 每30分钟
scheduler.add_job(job_function, 'cron', day_of_week='mon-fri', hour=17)  # 工作日17:00

print("定时任务已启动，按 Ctrl+C 退出")
try:
    scheduler.start()
except KeyboardInterrupt:
    print("定时任务已停止")

使用操作系统的任务计划（跨平台）

import platform
import subprocess
import sys

def create_system_schedule():
    """创建操作系统级别的定时任务"""
    system = platform.system()

    if system == "Windows":
        # Windows任务计划
        cmd = '''
        schtasks /create /tn "PythonDailyTask" /tr "python D:\scripts\daily_task.py" /sc daily /st 09:00
        '''
        subprocess.run(cmd, shell=True)
        print("Windows定时任务已创建")

    elif system == "Linux" or system == "Darwin":
        # Linux/Mac的crontab
        cron_entry = "0 9 * * * python /home/user/scripts/daily_task.py\n"

        # 获取现有crontab
        result = subprocess.run(['crontab', '-l'], capture_output=True, text=True)
        current_cron = result.stdout

        # 添加新任务
        with open('/tmp/new_cron', 'w') as f:
            f.write(current_cron)
            f.write(cron_entry)

        # 安装新的crontab
        subprocess.run(['crontab', '/tmp/new_cron'])
        print("Linux/Mac定时任务已创建")

    else:
        print(f"不支持的操作系统: {system}")

# 注意：这需要管理员/root权限
# create_system_schedule()

实用场景：定时任务的多种应用

场景1：定时网站监控

import requests
import schedule
import time

def check_website():
    url = "https://your-website.com"
    try:
        response = requests.get(url, timeout=10)
        if response.status_code == 200:
            print(f"[{time.strftime('%H:%M:%S')}] 网站正常访问")
        else:
            print(f"[{time.strftime('%H:%M:%S')}] 网站异常: {response.status_code}")
    except Exception as e:
        print(f"[{time.strftime('%H:%M:%S')}] 网站监控错误: {e}")

# 每5分钟检查一次网站状态
schedule.every(5).minutes.do(check_website)

场景2：定时数据清理

import os
import schedule
import datetime

def clean_temp_files():
    """清理7天前的临时文件"""
    temp_folder = "/tmp"
    cutoff_time = datetime.datetime.now() - datetime.timedelta(days=7)

    deleted_count = 0
    for filename in os.listdir(temp_folder):
        filepath = os.path.join(temp_folder, filename)
        if os.path.isfile(filepath):
            file_mtime = datetime.datetime.fromtimestamp(os.path.getmtime(filepath))
            if file_mtime < cutoff_time:
                os.remove(filepath)
                deleted_count += 1

    print(f"[{datetime.datetime.now()}] 清理了 {deleted_count} 个临时文件")

# 每天凌晨3点清理
schedule.every().day.at("03:00").do(clean_temp_files)

场景3：定时数据同步

import schedule
import sqlite3
import pandas as pd
from datetime import datetime

def sync_data():
    """定时从数据库同步数据到Excel"""
    # 连接数据库
    conn = sqlite3.connect('data.db')

    # 查询数据
    df = pd.read_sql_query("SELECT * FROM daily_sales", conn)

    # 保存到Excel
    filename = f"销售数据_{datetime.now().strftime('%Y%m%d_%H%M')}.xlsx"
    df.to_excel(filename, index=False)

    conn.close()
    print(f"[{datetime.now()}] 数据已同步到: {filename}")

# 每小时的第15分钟执行
schedule.every().hour.at(":15").do(sync_data)

注意事项与最佳实践

1. 处理长时间运行的任务

import threading

def long_running_task():
    """长时间运行的任务"""
    time.sleep(300)  # 模拟5分钟的任务
    print("长时间任务完成")

# 在新线程中执行，避免阻塞调度器
def run_in_thread(job_func):
    job_thread = threading.Thread(target=job_func)
    job_thread.start()

schedule.every().hour.do(run_in_thread, long_running_task)

2. 添加异常处理

def safe_task():
    try:
        # 你的任务代码
        risky_operation()
        print("任务执行成功")
    except Exception as e:
        print(f"任务执行失败: {e}")
        # 可以添加重试逻辑或发送警报

schedule.every().hour.do(safe_task)

3. 记录任务执行日志

import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('scheduler.log'),
        logging.StreamHandler()
    ]
)

def logged_task():
    logging.info("开始执行定时任务")
    # 任务代码
    logging.info("定时任务执行完成")

schedule.every().day.at("08:00").do(logged_task)

结语：让程序为你"值班"

恭喜！在又一个5分钟内，你掌握了一个强大的自动化技能：

✅ 使用schedule库创建直观的定时任务
✅ 掌握各种时间间隔的设置方法
✅ 学会传递参数和创建复杂的任务逻辑
✅ 具备了处理异常和长时间任务的能力

今天你学会的不仅仅是定时执行，而是掌握了让程序24小时为你工作的能力。

想象一下，你还可以： - 设置定时数据备份，再也不怕数据丢失 - 创建定时监控脚本，第一时间发现问题 - 实现定时报表生成，解放重复性工作 - 安排定时清理任务，保持系统整洁

自动化的最高境界，是让机器在你休息时继续创造价值。

现在，就找一个你每天、每周或每月都要手动执行的任务，用今天学到的知识把它自动化吧！在评论区分享你实现的第一个定时任务是什么！

记得关注我，下一篇我们挑战更有趣的Python自动化技巧！

版权声明：如无特殊说明，文章均为何三笔记原创，转载请注明出处

本文链接：https://www.h3blog.com/article/660/

环境准备：Python原生支持
核心代码：10行定时任务框架
逐行详解：理解定时任务的原理
实战演示：创建你的第一个定时任务
举一反三：丰富的定时规则
进阶技巧：专业级定时任务
实用场景：定时任务的多种应用
注意事项与最佳实践
结语：让程序为你"值班"