杰瑞科技汇

Python schedule如何实现定时下载?

schedule 库本身是一个非常轻量级的任务调度库,它不负责真正的下载,而是负责在特定的时间点或时间间隔触发你的下载函数,真正的下载工作需要由其他库(如 requests)来完成。

Python schedule如何实现定时下载?-图1
(图片来源网络,侵删)

下面我将分步为你讲解,从安装到编写完整的下载脚本。


第一步:安装必要的库

你需要安装 schedule 库来处理任务调度,以及 requests 库来执行 HTTP 下载,如果你的下载文件很大,还需要 tqdm 库来显示一个美观的进度条。

打开你的终端或命令行工具,运行以下命令:

pip install schedule requests tqdm

第二步:理解 schedule 的基本用法

schedule 的核心思想非常简单:

Python schedule如何实现定时下载?-图2
(图片来源网络,侵删)
  1. 定义一个任务函数:比如一个名为 download_file() 的函数。
  2. 使用 schedule 装饰器或函数来安排任务:告诉 schedule 在什么时间执行这个函数。
  3. 启动调度器:运行一个无限循环,不断地检查任务是否到期,如果到期就执行它。

常用的时间安排方法:

  • schedule.every(10).minutes.do(job) # 每10分钟执行一次
  • schedule.every().hour.do(job) # 每小时执行一次
  • schedule.every().day.at("10:30").do(job) # 每天上午10:30执行一次
  • schedule.every().monday.do(job) # 每周一执行一次
  • schedule.every().wednesday.at("13:15").do(job) # 每周三下午13:15执行一次
  • schedule.every().minute.at(":17").do(job) # 每分钟的17秒执行一次

第三步:编写一个完整的定时下载脚本

下面是一个完整的示例,它会每天上午 10:30 自动下载一个示例文件(这里以 GitHub 的一个测试文件为例),并显示下载进度。

我们将所有逻辑封装在一个 main() 函数中,这样代码结构更清晰。

import schedule
import requests
import time
import os
from tqdm import tqdm
# --- 1. 定义下载任务函数 ---
def download_file():
    """
    一个具体的下载函数,用于下载指定URL的文件。
    """
    # 要下载的文件URL
    file_url = "https://raw.githubusercontent.com/psf/requests/main/README.rst"
    # 保存到本地的文件名
    local_filename = "requests_readme.rst"
    print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] 开始下载文件: {file_url}")
    try:
        # 发起HTTP GET请求,stream=True表示流式下载
        with requests.get(file_url, stream=True) as r:
            r.raise_for_status()  # 如果请求失败 (状态码不是 2xx), 则抛出异常
            # 获取文件总大小 (单位: bytes)
            total_size = int(r.headers.get('content-length', 0))
            # 使用 tqdm 创建进度条
            # initial=0, total=total_size, unit='B', unit_scale=True, unit_divisor=1024
            progress_bar = tqdm(
                total=total_size, 
                unit='B', 
                unit_scale=True, 
                unit_divisor=1024,
                desc=local_filename
            )
            # 分块写入文件
            with open(local_filename, 'wb') as f:
                for chunk in r.iter_content(chunk_size=8192): # 每次读取 8KB
                    f.write(chunk)
                    progress_bar.update(len(chunk))
            progress_bar.close()
            # 检查文件是否真的下载了
            if os.path.exists(local_filename):
                print(f"✅ 下载成功!文件已保存为: {os.path.abspath(local_filename)}")
            else:
                print("❌ 下载失败:文件未创建。")
    except requests.exceptions.RequestException as e:
        print(f"❌ 下载过程中发生错误: {e}")
    except Exception as e:
        print(f"❌ 发生未知错误: {e}")
# --- 2. 定义主函数,用于启动调度器 ---
def run_scheduler():
    """
    设置任务并启动调度器的无限循环。
    """
    # --- 3. 安排任务 ---
    # 设置每天上午10:30执行下载任务
    # 注意:时间格式是 "HH:MM"
    schedule.every().day.at("10:30").do(download_file)
    # 你也可以添加其他任务,比如每小时检查一次
    # schedule.every().hour.do(another_task)
    print("🕒 任务调度器已启动...")
    print(f"⏰ 已设置任务:每天 10:30 执行下载。")
    print("脚本正在运行,按 Ctrl+C 停止...")
    # --- 4. 启动调度器 ---
    while True:
        # run_pending() 会检查所有已安排的任务,如果到了执行时间,就运行它。
        schedule.run_pending()
        # time.sleep(1) 让程序暂停1秒,避免CPU空转占用过高资源
        time.sleep(1)
# --- 5. 程序入口 ---
if __name__ == "__main__":
    run_scheduler()

如何运行和测试这个脚本?

  1. 将代码保存:将上面的代码保存为一个 Python 文件,scheduled_download.py
  2. 直接运行:在终端中运行 python scheduled_download.py
  3. 观察输出:你会看到脚本启动的提示,它会一直等待,直到你设定的 10:30
  4. 立即测试:为了快速验证脚本是否工作,你可以临时修改时间,比如改成 schedule.every(10).seconds.do(download_file),这样它就会每10秒执行一次下载,你可以看到进度条和成功信息,测试成功后,再改回你想要的时间。

第四步:高级用法与最佳实践

同时运行多个任务

schedule 默认是单线程的,任务会按顺序执行,如果一个任务耗时很长,它会阻塞其他任务的执行,如果你需要同时运行多个任务,可以结合 threading 模块。

Python schedule如何实现定时下载?-图3
(图片来源网络,侵删)
import threading
import schedule
import time
def job1():
    print("Job 1 开始...")
    time.sleep(5) # 模拟耗时任务
    print("Job 1 完成。")
def job2():
    print("Job 2 开始...")
    time.sleep(2) # 模拟耗时任务
    print("Job 2 完成。")
def run_threaded(job_func):
    """在独立线程中运行任务"""
    job_thread = threading.Thread(target=job_func)
    job_thread.start()
# 安排任务,并使用 run_threaded 包装器
schedule.every(10).seconds.do(run_threaded, job1)
schedule.every(5).seconds.do(run_threaded, job2)
while True:
    schedule.run_pending()
    time.sleep(1)

如何让脚本在后台持续运行(作为服务)

上面的 while True 循环会让你的脚本一直占用一个终端窗口,如果你想让它在后台运行,你可以:

  • 使用 nohup (Linux/macOS):

    nohup python your_script.py > output.log 2>&1 &
    • nohup: 让命令忽略挂断信号。
    • > output.log: 将标准输出重定向到 output.log 文件。
    • 2>&1: 将标准错误也重定向到同一个文件。
    • &: 让命令在后台运行。
  • 使用 systemd (推荐用于生产环境): 创建一个 .service 文件,让 systemd 来管理你的 Python 脚本,它可以实现开机自启、自动重启、日志管理等功能,非常专业。

停止调度器

schedule 提供了一个 schedule.clear() 方法来清除所有已安排的任务。

# ... 在你的循环中添加一个停止条件 ...
stop_flag = False
while True:
    if stop_flag:
        print("接收到停止信号,正在清除所有任务...")
        schedule.clear()
        print("调度器已停止。")
        break
    schedule.run_pending()
    time.sleep(1)
    # ... 可以通过某种方式(如信号、网络请求)来设置 stop_flag = True ...

希望这个详细的指南能帮助你成功使用 Python schedule 库实现定时下载功能!

分享:
扫描分享到社交APP
上一篇
下一篇