社区首页 › 编程脚本 › Python批量下载与自动压缩实现：多线程、加密打包与定时任务

脚本专家 发表于 7 天前

Python批量下载与自动压缩实现：多线程、加密打包与定时任务

在日常工作中，频繁处理文件批量下载和自动压缩打包是常见的重复劳动。本文将分享一套基于 Python 的完整自动化方案，涵盖从 URL 列表、网页图片到 GitHub 目录的下载，以及压缩整个文件夹、筛选特定类型文件、甚至带密码加密打包的功能，最后提供一条龙流水线和定时任务调度，帮助开发者和运维人员提升效率。所有代码均可直接运行，注释清晰，适合根据实际需求定制。

一、批量下载模块设计

首先封装一个通用的 BatchDownloader 类，支持多线程并发下载、断点续传和失败重试。核心参数包括保存目录和最大线程数。


import requests
import os
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

class BatchDownloader:
    """批量下载器"""
    def __init__(self, save_dir="downloads", max_workers=5):
      self.save_dir = save_dir
      self.max_workers = max_workers
      os.makedirs(save_dir, exist_ok=True)

    def download_one(self, url, filename=None):
      """下载单个文件，支持断点续传"""
      try:
            if not filename:
                filename = url.split("/")[-1] or "index.html"
            filepath = os.path.join(self.save_dir, filename)
            headers = {}
            if os.path.exists(filepath):
                headers["Range"] = f"bytes={os.path.getsize(filepath)}-"
            resp = requests.get(url, headers=headers, timeout=30, stream=True)
            resp.raise_for_status()
            mode = "ab" if os.path.exists(filepath) else "wb"
            with open(filepath, mode) as f:
                for chunk in resp.iter_content(chunk_size=8192):
                  if chunk:
                        f.write(chunk)
            size = os.path.getsize(filepath)
            print(f"✅ {filename} ({size/1024:.1f}KB)")
            return filepath
      except Exception as e:
            print(f"❌ {filename or url}: {e}")
            return None

    def download_many(self, urls, filenames=None):
      """多线程批量下载"""
      if filenames is None:
            filenames = * len(urls)
      start = time.time()
      success = 0
      with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = {
                executor.submit(self.download_one, url, name): url
                for url, name in zip(urls, filenames)
            }
            for future in as_completed(futures):
                if future.result():
                  success += 1
      elapsed = time.time() - start
      print(f"\n下载完成：{success}/{len(urls)} 成功，耗时 {elapsed:.1f}s")
      return


使用方法：

downloader = BatchDownloader("教程附件", max_workers=10)
with open("urls.txt", "r") as f:
    urls =
downloader.download_many(urls)


1.2 批量下载网页中的图片
借助 BeautifulSoup 解析 HTML，提取所有 img 标签的 src 属性，并过滤常见图片格式。

import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin

def download_images(page_url, save_dir="images"):
    os.makedirs(save_dir, exist_ok=True)
    resp = requests.get(page_url)
    soup = BeautifulSoup(resp.text, "html.parser")
    img_tags = soup.find_all("img")
    img_urls = []
    for img in img_tags:
      src = img.get("src")
      if src:
            full_url = urljoin(page_url, src)
            if any(ext in full_url.lower() for ext in [".jpg", ".jpeg", ".png", ".gif", ".webp"]):
                img_urls.append(full_url)
    print(f"找到 {len(img_urls)} 张图片")
    downloader = BatchDownloader(save_dir)
    downloader.download_many(img_urls)

# 使用示例
download_images("https://example.com/article/1")


1.3 下载 GitHub 仓库中的指定目录（无需克隆整个仓库）
通过 GitHub API 获取目录内容，只下载文件，不克隆仓库。

def download_github_dir(repo, dir_path, save_dir="github_files"):
    api_url = f"https://api.github.com/repos/{repo}/contents/{dir_path}"
    resp = requests.get(api_url)
    files = resp.json()
    urls, names = [], []
    for f in files:
      if f["type"] == "file":
            urls.append(f["download_url"])
            names.append(f["name"])
    downloader = BatchDownloader(save_dir)
    downloader.download_many(urls, names)

# 下载 Spring Boot 的 starters 目录
download_github_dir("spring-projects/spring-boot", "starters")


二、自动压缩打包
下载完成后，常需压缩归档。这里提供三种场景：

2.1 压缩整个文件夹
使用 shutil.make_archive 直接打包，命名自动带上日期。

import shutil
from datetime import datetime

def zip_folder(folder_path, output_name=None):
    if not output_name:
      folder_name = os.path.basename(folder_path)
      today = datetime.now().strftime("%Y%m%d")
      output_name = f"{folder_name}_{today}"
    shutil.make_archive(output_name, "zip", folder_path)
    size = os.path.getsize(output_name + ".zip") / 1024 / 1024
    print(f"已压缩: {output_name}.zip ({size:.1f}MB)")
    return f"{output_name}.zip"

zip_folder("downloads", "课件包_20260630")


2.2 只压缩特定类型文件
利用 zipfile 模块，遍历目录筛选扩展名。

import zipfile

def zip_selected_files(directory, extensions, output_name="selected"):
    with zipfile.ZipFile(f"{output_name}.zip", "w", zipfile.ZIP_DEFLATED) as zf:
      for root, _, files in os.walk(directory):
            for f in files:
                if any(f.endswith(ext) for ext in extensions):
                  filepath = os.path.join(root, f)
                  arcname = os.path.relpath(filepath, directory)
                  zf.write(filepath, arcname)
                  print(f"已添加: {arcname}")
    print(f"压缩完成: {output_name}.zip")

# 压缩 PDF 和 Word 文件
zip_selected_files("课件", [".pdf", ".docx"], "课件PDF版")


2.3 带密码加密压缩
需安装 pyminizip 库（pip install pyminizip），可设置密码并指定压缩等级。

import pyminizip

def zip_with_password(folder_path, password, output_name="protected"):
    pyminizip.compress_dir(folder_path, f"{output_name}.zip", password, 5)
    print(f"已加密压缩: {output_name}.zip")

zip_with_password("机密文件", "mypassword123", "机密文件_加密")


三、自动删除源文件（可选）
压缩后如需清理原始文件，可以调用以下函数：

import shutil

def clean_source(source_path):
    if os.path.isfile(source_path):
      os.remove(source_path)
      print(f"已删除: {source_path}")
    elif os.path.isdir(source_path):
      shutil.rmtree(source_path)
      print(f"已删除目录: {source_path}")


四、完整流水线：下载 → 压缩 → 清理
整合所有步骤，传入 URL 列表文件即可自动完成全流程。

def download_and_pack(urls_file, output_name="打包文件"):
    base_dir = "temp_download"
    with open(urls_file, "r") as f:
      urls =
    print(f"共 {len(urls)} 个文件待下载")
    downloader = BatchDownloader(base_dir, max_workers=10)
    results = downloader.download_many(urls)
    success_count = sum(1 for r in results if r)
    print(f"成功下载: {success_count}/{len(results)}")
    if success_count > 0:
      zip_file = zip_folder(base_dir, output_name)
      shutil.rmtree(base_dir)
      print("临时文件已清理")
      return zip_file

# 执行
download_and_pack("urls.txt", "课程资料包_202607")


五、定时任务自动执行
利用 schedule 库让脚本在每天指定时间运行，适合持续集成场景。

import schedule
import time
from datetime import datetime

def daily_download_task():
    print(f"开始定时下载任务: {datetime.now()}")
    resp = requests.get("https://api.example.com/today/files")
    files = resp.json()
    urls = for f in files]
    names = for f in files]
    downloader = BatchDownloader("每日更新")
    downloader.download_many(urls, names)
    zip_folder("每日更新", f"更新包_{datetime.now().strftime('%Y%m%d')}")
    print("任务完成")

schedule.every().day.at("08:00").do(daily_download_task)
while True:
    schedule.run_pending()
    time.sleep(60)


六、下载工具对比与场景建议
| 方式 | 速度 | 适用场景 |
|------|------|----------|
| requests 单线程 | 慢 | 少量文件 |
| 多线程下载（本文方案） | 快 | 批量文件（推荐） |
| wget 命令 | 中 | Linux 服务器 |
| aria2c | 极快 | 大文件、多文件 |

本文提供的脚本可直接用于课件下载、爬虫数据打包、加密文件分发、每日备份等场景，只需调整参数即可适应不同需求。开发者可以在此基础上扩展文件类型过滤、重试机制、日志记录等高级功能。

热心网友4 发表于 7 天前

Re: Python批量下载与自动压缩实现：多线程、加密打包与定时任务

这个方案很实用，尤其是多线程下载和断点续传的部分，正好解决了我最近批量下载大文件时频繁中断的问题。想问一下，如果URL列表里包含了一些需要登录认证的链接，BatchDownloader类有没有预留传入cookie或header的接口？另外加密打包的部分是用zipfile配合密码实现的吗？期待后续的内容。

热心网友4 发表于 7 天前

Re: Python批量下载与自动压缩实现：多线程、加密打包与定时任务

感谢楼主分享这么实用的工具！批量下载加上断点续传和失败重试，在日常工作中确实能省下不少重复劳动。代码结构也清晰，直接就能跑起来，挺方便的。 另外，我看标题里还提到了“加密打包”和“定时任务”，这部分内容好像没有在正文里展示出来？如果方便的话，希望能补上压缩加密和自动调度的实现细节，这样整个流水线就完整了。再次感谢分享！

热心网友4 发表于 7 天前

Re: Python批量下载与自动压缩实现：多线程、加密打包与定时任务

感谢分享！批量下载器类设计得很不错，多线程和断点续传功能非常实用，代码注释也很清晰，方便直接拿来用。请问能不能再增加一个进度条显示，方便实时查看下载进度？另外，后面提到的加密打包和定时任务部分还没贴完，期待楼主继续更新～

查看完整版本: Python批量下载与自动压缩实现：多线程、加密打包与定时任务