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

脚本专家

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

一、批量下载模块设计

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

2. import requests
3. import os
4. from concurrent.futures import ThreadPoolExecutor, as_completed
5. import time
7. class BatchDownloader:
8.     """批量下载器"""
9.     def __init__(self, save_dir="downloads", max_workers=5):
10.         self.save_dir = save_dir
11.         self.max_workers = max_workers
12.         os.makedirs(save_dir, exist_ok=True)
14.     def download_one(self, url, filename=None):
15.         """下载单个文件，支持断点续传"""
16.         try:
17.             if not filename:
18.                 filename = url.split("/")[-1] or "index.html"
19.             filepath = os.path.join(self.save_dir, filename)
20.             headers = {}
21.             if os.path.exists(filepath):
22.                 headers["Range"] = f"bytes={os.path.getsize(filepath)}-"
23.             resp = requests.get(url, headers=headers, timeout=30, stream=True)
24.             resp.raise_for_status()
25.             mode = "ab" if os.path.exists(filepath) else "wb"
26.             with open(filepath, mode) as f:
27.                 for chunk in resp.iter_content(chunk_size=8192):
28.                     if chunk:
29.                         f.write(chunk)
30.             size = os.path.getsize(filepath)
31.             print(f"✅ {filename} ({size/1024:.1f}KB)")
32.             return filepath
33.         except Exception as e:
34.             print(f"❌ {filename or url}: {e}")
35.             return None
37.     def download_many(self, urls, filenames=None):
38.         """多线程批量下载"""
39.         if filenames is None:
40.             filenames = [None] * len(urls)
41.         start = time.time()
42.         success = 0
43.         with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
44.             futures = {
45.                 executor.submit(self.download_one, url, name): url
46.                 for url, name in zip(urls, filenames)
47.             }
48.             for future in as_completed(futures):
49.                 if future.result():
50.                     success += 1
51.         elapsed = time.time() - start
52.         print(f"\n下载完成：{success}/{len(urls)} 成功，耗时 {elapsed:.1f}s")
53.         return [f.result() for f in futures]

复制代码

使用方法：

2. downloader = BatchDownloader("教程附件", max_workers=10)
3. with open("urls.txt", "r") as f:
4.     urls = [line.strip() for line in f if line.strip()]
5. downloader.download_many(urls)

复制代码

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

2. import requests
3. from bs4 import BeautifulSoup
4. from urllib.parse import urljoin
6. def download_images(page_url, save_dir="images"):
7.     os.makedirs(save_dir, exist_ok=True)
8.     resp = requests.get(page_url)
9.     soup = BeautifulSoup(resp.text, "html.parser")
10.     img_tags = soup.find_all("img")
11.     img_urls = []
12.     for img in img_tags:
13.         src = img.get("src")
14.         if src:
15.             full_url = urljoin(page_url, src)
16.             if any(ext in full_url.lower() for ext in [".jpg", ".jpeg", ".png", ".gif", ".webp"]):
17.                 img_urls.append(full_url)
18.     print(f"找到 {len(img_urls)} 张图片")
19.     downloader = BatchDownloader(save_dir)
20.     downloader.download_many(img_urls)
22. # 使用示例
23. download_images("https://example.com/article/1")

复制代码

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

2. def download_github_dir(repo, dir_path, save_dir="github_files"):
3.     api_url = f"https://api.github.com/repos/{repo}/contents/{dir_path}"
4.     resp = requests.get(api_url)
5.     files = resp.json()
6.     urls, names = [], []
7.     for f in files:
8.         if f["type"] == "file":
9.             urls.append(f["download_url"])
10.             names.append(f["name"])
11.     downloader = BatchDownloader(save_dir)
12.     downloader.download_many(urls, names)
14. # 下载 Spring Boot 的 starters 目录
15. download_github_dir("spring-projects/spring-boot", "starters")

复制代码

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

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

2. import shutil
3. from datetime import datetime
5. def zip_folder(folder_path, output_name=None):
6.     if not output_name:
7.         folder_name = os.path.basename(folder_path)
8.         today = datetime.now().strftime("%Y%m%d")
9.         output_name = f"{folder_name}_{today}"
10.     shutil.make_archive(output_name, "zip", folder_path)
11.     size = os.path.getsize(output_name + ".zip") / 1024 / 1024
12.     print(f"已压缩: {output_name}.zip ({size:.1f}MB)")
13.     return f"{output_name}.zip"
15. zip_folder("downloads", "课件包_20260630")

复制代码

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

2. import zipfile
4. def zip_selected_files(directory, extensions, output_name="selected"):
5.     with zipfile.ZipFile(f"{output_name}.zip", "w", zipfile.ZIP_DEFLATED) as zf:
6.         for root, _, files in os.walk(directory):
7.             for f in files:
8.                 if any(f.endswith(ext) for ext in extensions):
9.                     filepath = os.path.join(root, f)
10.                     arcname = os.path.relpath(filepath, directory)
11.                     zf.write(filepath, arcname)
12.                     print(f"已添加: {arcname}")
13.     print(f"压缩完成: {output_name}.zip")
15. # 压缩 PDF 和 Word 文件
16. zip_selected_files("课件", [".pdf", ".docx"], "课件PDF版")

复制代码

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

2. import pyminizip
4. def zip_with_password(folder_path, password, output_name="protected"):
5.     pyminizip.compress_dir(folder_path, f"{output_name}.zip", password, 5)
6.     print(f"已加密压缩: {output_name}.zip")
8. zip_with_password("机密文件", "mypassword123", "机密文件_加密")

复制代码

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

2. import shutil
4. def clean_source(source_path):
5.     if os.path.isfile(source_path):
6.         os.remove(source_path)
7.         print(f"已删除: {source_path}")
8.     elif os.path.isdir(source_path):
9.         shutil.rmtree(source_path)
10.         print(f"已删除目录: {source_path}")

复制代码

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

2. def download_and_pack(urls_file, output_name="打包文件"):
3.     base_dir = "temp_download"
4.     with open(urls_file, "r") as f:
5.         urls = [line.strip() for line in f if line.strip()]
6.     print(f"共 {len(urls)} 个文件待下载")
7.     downloader = BatchDownloader(base_dir, max_workers=10)
8.     results = downloader.download_many(urls)
9.     success_count = sum(1 for r in results if r)
10.     print(f"成功下载: {success_count}/{len(results)}")
11.     if success_count > 0:
12.         zip_file = zip_folder(base_dir, output_name)
13.         shutil.rmtree(base_dir)
14.         print("临时文件已清理")
15.         return zip_file
17. # 执行
18. download_and_pack("urls.txt", "课程资料包_202607")

复制代码

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

2. import schedule
3. import time
4. from datetime import datetime
6. def daily_download_task():
7.     print(f"开始定时下载任务: {datetime.now()}")
8.     resp = requests.get("https://api.example.com/today/files")
9.     files = resp.json()
10.     urls = [f["url"] for f in files]
11.     names = [f["name"] for f in files]
12.     downloader = BatchDownloader("每日更新")
13.     downloader.download_many(urls, names)
14.     zip_folder("每日更新", f"更新包_{datetime.now().strftime('%Y%m%d')}")
15.     print("任务完成")
17. schedule.every().day.at("08:00").do(daily_download_task)
18. while True:
19.     schedule.run_pending()
20.     time.sleep(60)

复制代码

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

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

热心网友4

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

热心网友4

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

热心网友4

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