Python 任务调度 _ 用 schedule 包轻松实现定时任务

pip install schedule

pip install schedule -i https://mirrors.aliyun.com/pypi/simple/

import schedule  # 导入schedule包

import time      # 导入时间模块，用来控制循环

# 1. 定义要执行的任务（就是一个普通函数）

def print_hello():

   print(f"当前时间：{time.strftime('%H:%M:%S')}，执行任务啦！")

# 2. 设置定时规则：每1秒执行一次print_hello

schedule.every(1).seconds.do(print_hello)

# 3. 循环检查任务（关键！没有这步任务不会执行）

while True:

   schedule.run_pending()  # 检查有没有到点要执行的任务

   time.sleep(0.1)         # 暂停0.1秒，避免占用太多CPU（别省这行！）

import schedule

import time

def daily_task():

   print(f"每日任务：{time.strftime('%Y-%m-%d %H:%M:%S')} - 抓取今日数据")

def weekly_task():

   print(f"每周任务：{time.strftime('%Y-%m-%d %H:%M:%S')} - 生成周报表")

# 每天15:15执行每日任务

schedule.every().day.at("15:15").do(daily_task)

# 每周一18:00执行每周任务

schedule.every().monday.at("18:00").do(weekly_task)

# 循环检查

while True:

   schedule.run_pending()

   time.sleep(60)  # 这里可以设60秒，因为任务是按分钟/小时的，不用查那么勤

import schedule

import time

# 带参数的任务函数

def send_msg(username, msg):

   print(f"{time.strftime('%H:%M:%S')} - 给{username}发送消息：{msg}")

# 传参：位置参数直接跟在函数后面，关键字参数用key=value

schedule.every(2).seconds.do(

   send_msg,  # 任务函数

   username="张三",  # 第一个参数

   msg="该打卡了！"   # 第二个参数

)

# 循环检查

while True:

   schedule.run_pending()

   time.sleep(0.1)

import schedule

import time

from functools import partial  # 导入partial

def send_email(to, subject, content):

   print(f"{time.strftime('%H:%M:%S')} - 给{to}发邮件：主题[{subject}]，内容[{content}]")

# 打包函数和参数

email_task = partial(

   send_email,

   to="test@example.com",

   subject="每日报表",

   content="今日数据已更新"

)

# 直接用打包好的函数

schedule.every(3).seconds.do(email_task)

while True:

   schedule.run_pending()

   time.sleep(0.1)

import schedule

import time

from schedule import repeat, every  # 导入repeat和every装饰器

# 用装饰器指定定时规则：每2秒执行一次

@repeat(every(2).seconds)

def print_time():

   print(f"当前时间：{time.strftime('%H:%M:%S')}")

# 循环检查（这步还是不能少）

while True:

   schedule.run_pending()

   time.sleep(0.1)

from schedule import repeat, every

import schedule

import time

@repeat(every(3).seconds, username="李四")  # 参数直接跟在定时规则后面

def greet(username):

   print(f"{time.strftime('%H:%M:%S')} - 你好，{username}！")

while True:

   schedule.run_pending()

   time.sleep(0.1)

import schedule

import time

def task1():

   print("执行任务1")

def task2():

   print("执行任务2")

# 1. 保存任务对象（关键！不保存就没法取消）

job1 = schedule.every(1).seconds.do(task1)

job2 = schedule.every(2).seconds.do(task2)

# 2. 模拟5秒后取消task1

time.sleep(5)

print("取消任务1")

schedule.cancel_job(job1)  # 取消task1

# 3. 继续循环，此时只有task2在执行

while True:

   schedule.run_pending()

   time.sleep(0.1)

import schedule

import time

def task1():

   print("任务1")

def task2():

   print("任务2")

# 添加两个任务

schedule.every(1).seconds.do(task1)

schedule.every(2).seconds.do(task2)

# 3秒后清空所有任务

time.sleep(3)

print("清空所有任务")

schedule.clear()

# 循环继续，但没有任务执行了

while True:

   schedule.run_pending()

   time.sleep(0.1)

import schedule

import time

def crawl_data():

   print("抓取数据")

def send_report():

   print("发送报表")

def check_api():

   print("检查接口")

# 给任务加标签：tag参数指定标签

schedule.every(1).seconds.do(crawl_data, tag="数据抓取")

schedule.every(2).seconds.do(send_report, tag="报表")

schedule.every(3).seconds.do(check_api, tag="数据抓取")  # 这个也归为“数据抓取”

# 2秒后，取消所有“数据抓取”标签的任务

time.sleep(2)

print("取消所有数据抓取任务")

# 1. 先获取所有“数据抓取”标签的任务

crawl_jobs = schedule.get_jobs(tag="数据抓取")

# 2. 逐个取消

for job in crawl_jobs:

   schedule.cancel_job(job)

# 后续只有“发送报表”任务在执行

while True:

   schedule.run_pending()

   time.sleep(0.1)

import schedule

import time

def fast_task():  # 快任务：1秒执行一次

   print(f"快任务：{time.strftime('%H:%M:%S')}")

def slow_task():  # 慢任务：2秒执行一次，但要耗时3秒

   print(f"慢任务开始：{time.strftime('%H:%M:%S')}")

   time.sleep(3)  # 模拟任务耗时3秒

   print(f"慢任务结束：{time.strftime('%H:%M:%S')}")

# 添加任务

schedule.every(1).seconds.do(fast_task)

schedule.every(2).seconds.do(slow_task)

# 单线程循环

while True:

   schedule.run_pending()

   time.sleep(0.1)

import schedule

import time

import threading  # 导入多线程模块

def fast_task():

   print(f"快任务：{time.strftime('%H:%M:%S')}")

def slow_task():

   print(f"慢任务开始：{time.strftime('%H:%M:%S')}")

   time.sleep(3)

   print(f"慢任务结束：{time.strftime('%H:%M:%S')}")

# 关键：写一个包装函数，把任务放到线程里执行

def run_in_thread(func):

   def wrapper():

       threading.Thread(target=func).start()  # 启动新线程执行任务

   return wrapper

# 给慢任务用包装函数，快任务可以不用（也能用，不影响）

schedule.every(1).seconds.do(fast_task)

schedule.every(2).seconds.do(run_in_thread(slow_task))  # 慢任务走多线程

# 循环不变

while True:

   schedule.run_pending()

   time.sleep(0.1)

lock = threading.Lock()  # 创建锁

def write_file(content):

   with lock:  # 加锁，同一时间只有一个线程能执行下面的代码

       with open("data.txt", "a") as f:

           f.write(content + "n")

import schedule

import time

import requests

import threading

# 1. 定义抓取函数

def crawl_weather():

   city = "北京"

   api_url = f"https://wttr.in/{city}?format=3"  # 公开天气API，返回简洁结果

   try:

       response = requests.get(api_url, timeout=10)

       response.raise_for_status()  # 抛出自定义异常

       weather = response.text.strip()

       # 保存到文件

       with open("weather_log.txt", "a", encoding="utf-8") as f:

           f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} - {city}天气：{weather}n")

       print(f"抓取成功：{weather}")

   except Exception as e:

       print(f"抓取失败：{str(e)}")

# 2. 多线程包装（避免API请求阻塞）

def run_in_thread(func):

   def wrapper():

       threading.Thread(target=func).start()

   return wrapper

# 3. 设置每天9点执行

schedule.every().day.at("09:00").do(run_in_thread(crawl_weather))

# 4. 循环运行

while True:

   schedule.run_pending()

   time.sleep(60)  # 每60秒检查一次

import schedule

import time

import smtplib

import threading

from email.mime.text import MIMEText

from email.header import Header

# 1. 邮件配置（替换成你的信息）

SMTP_SERVER = "smtp.qq.com"  # QQ邮箱SMTP服务器

SMTP_PORT = 587              # 端口

FROM_EMAIL = "你的QQ邮箱@qq.com"

AUTH_CODE = "你的邮箱授权码"  # 不是密码，是SMTP授权码

TO_EMAIL = "接收邮箱@example.com"

# 2. 发送邮件函数

def send_weekly_report():

   subject = "每周工作报表"

   content = """

   本周工作总结：

   1. 完成数据抓取脚本开发

   2. 修复API超时问题

   3. 优化定时任务性能

   """

   # 构造邮件

   msg = MIMEText(content, "plain", "utf-8")

   msg["From"] = Header(FROM_EMAIL)

   msg["To"] = Header(TO_EMAIL)

   msg["Subject"] = Header(subject, "utf-8")

   try:

       # 连接SMTP服务器发送

       with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:

           server.starttls()  # 开启TLS加密

           server.login(FROM_EMAIL, AUTH_CODE)

           server.sendmail(FROM_EMAIL, TO_EMAIL, msg.as_string())

       print("邮件发送成功")

   except Exception as e:

       print(f"邮件发送失败：{str(e)}")

# 3. 多线程包装

def run_in_thread(func):

   def wrapper():

       threading.Thread(target=func).start()

   return wrapper

# 4. 设置每周一18点执行

schedule.every().monday.at("18:00").do(run_in_thread(send_weekly_report))

# 5. 循环运行

while True:

   schedule.run_pending()

   time.sleep(60)

import schedule

import time

import threading

# 1. 模拟社交媒体发帖（实际替换成平台API调用）

def post_to_weibo(content):

   # 这里是模拟，实际需要调用微博API的post接口

   print(f"{time.strftime('%Y-%m-%d %H:%M:%S')} - 微博发帖成功：{content}")

# 2. 定时发帖任务（每小时发不同内容）

def scheduled_post():

   hour = time.localtime().tm_hour  # 获取当前小时

   if 9 <= hour <= 18:  # 只在工作时间发帖

       content = f"工作时间打卡-{hour}点，今天也要加油！"

   else:

       content = f"休息时间分享-{hour}点，晚安～"

   post_to_weibo(content)

# 3. 多线程包装

def run_in_thread(func):

   def wrapper():

       threading.Thread(target=func).start()

   return wrapper

# 4. 设置每小时执行一次

schedule.every(1).hours.do(run_in_thread(scheduled_post))

# 5. 循环运行

while True:

   schedule.run_pending()

   time.sleep(60)

print("当前所有任务：", schedule.get_jobs())  # 打印任务列表，看有没有重复

import schedule

import time

# 统一异常捕获的包装函数

def catch_exception(func):

   def wrapper(*args, **kwargs):

       try:

           return func(*args, **kwargs)

       except Exception as e:

           print(f"任务执行出错：{str(e)}")  # 打印错误，但不中断脚本

   return wrapper

# 用包装函数装饰任务

@catch_exception

def risky_task():

   # 这个任务可能抛出异常（比如除以0）

   1 / 0

schedule.every(1).seconds.do(risky_task)

while True:

   schedule.run_pending()

   time.sleep(0.1)

import schedule

import time

from datetime import datetime, timezone

# 要按UTC时间12:00执行，本地时间（UTC+8）就是20:00

def utc_task():

   utc_time = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

   print(f"UTC时间任务执行：{utc_time}")

# 本地时间20:00 = UTC时间12:00

schedule.every().day.at("20:00").do(utc_task)

while True:

   schedule.run_pending()

   time.sleep(60)