脚本自动化全攻略
目录导读
为什么需要批量拉取网络资源？
在日常开发、数据采集、内容迁移或SEO优化工作中，我们经常需要从网络上下载大量文件——图片、CSS/JS文件、PDF文档、视频片段等，手动逐一下载不仅耗时，而且极易出错。
典型场景包括：
- 网站静态资源备份（如全站图片、样式表）
- 爬虫采集到的资源本地化
- CDN资源迁移或镜像站点搭建
- 大规模数据集下载（如开源模型权重、公开数据集）
脚本批量拉取成为最优雅的解决方案，它能自动处理重复请求、错误重试、并发下载，甚至支持断点续传。
核心挑战：如何高效、稳定、不违规地实现批量下载？本文从零到一给出完整方案。
核心工具与脚本语言选择
1 推荐语言
| 语言 | 优势 | 适用场景 |
|---|---|---|
| Python | 生态丰富，urllib/requests/aiohttp | 复杂逻辑、多格式处理 |
| Shell | 轻量快速，配合wget/curl | Linux环境、简单下载任务 |
| Node.js | 异步I/O性能优异 | 高并发、实时流下载 |
2 必备命令行工具
- wget：经典下载器，支持递归、断点续传
- curl：灵活发送请求，适合API资源
- axel / aria2：多线程加速下载
3 通用流程
URL列表准备 → 循环/并发请求 → 错误处理 → 本地存储 → 日志记录实战：Python脚本批量下载文件
1 基础版：单线程顺序下载
import requests
import os
url_list = [
"https://example.com/images/1.jpg",
"https://example.com/css/style.css",
# 可以来自文件或爬虫结果
]
def download_file(url, save_path="downloads"):
os.makedirs(save_path, exist_ok=True)
filename = url.split("/")[-1]
full_path = os.path.join(save_path, filename)
try:
r = requests.get(url, timeout=10)
r.raise_for_status()
with open(full_path, 'wb') as f:
f.write(r.content)
print(f"✅ 下载成功: {filename}")
except Exception as e:
print(f"❌ 下载失败: {url} - {e}")
for url in url_list:
download_file(url)
2 进阶版：多线程并发 + 重试机制
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def download_with_retry(url, max_retries=3, **kwargs):
session = requests.Session()
retries = Retry(total=max_retries, backoff_factor=0.5,
status_forcelist=[500, 502, 503, 504])
session.mount('https://', HTTPAdapter(max_retries=retries))
try:
resp = session.get(url, timeout=15)
return resp.content
except Exception as e:
raise e
def parallel_download(url_list, max_workers=10):
results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_url = {executor.submit(download_with_retry, url): url
for url in url_list}
for future in as_completed(future_to_url):
url = future_to_url[future]
try:
data = future.result()
# 保存文件逻辑
results.append((url, True))
except Exception as e:
results.append((url, False, str(e)))
return results
关键优化点：
- 使用
Session复用TCP连接 - 设置重试策略（应对网络抖动）
- 控制并发数（避免被封IP）
3 高效从URL文件批量读取
将URL列表放在urls.txt文件中，每行一个：
with open('urls.txt', 'r') as f:
urls = [line.strip() for line in f if line.strip()]
实战：Shell脚本批量拉取资源
1 使用wget批量下载
#!/bin/bash
# 从文件读取URL
while IFS= read -r url; do
wget -c -t 3 -P ./downloads "$url"
done < urls.txt
-c：断点续传-t 3：重试3次-P：指定保存目录
2 使用curl + 并发
#!/bin/bash
download_url() {
local url="$1"
local filename=$(basename "$url")
curl -L -o "./downloads/$filename" --retry 3 --retry-delay 2 "$url"
}
export -f download_url
# 从URL列表并行下载（最多5个并发）
xargs -P 5 -I {} bash -c 'download_url "$@"' _ {} < urls.txt
Shell方案优势：无需安装依赖，原生支持管道和并发控制。
常见问题与高频问答
Q1：批量下载时如何避免被封IP？
A：使用以下策略：
- 控制请求间隔：加入
time.sleep(random.uniform(0.5, 1.5)) - 更换User-Agent：模拟不同浏览器
- 使用代理池：轮换IP（如通过代理API）
- 限制并发数：不要超过20个同时连接
Q2：下载链接是动态生成（如带Token）怎么办？
A：先用爬虫或手动生成完整的直链列表，再批量下载，若必须实时构造，可在脚本中集成登录获取Token的逻辑。
Q3：遇到SSL证书错误如何处理？
A：Python中设置verify=False（注意安全风险）：
requests.get(url, verify=False)
建议使用自定义CA证书或忽略特定错误。
Q4：如何验证下载文件完整性？
A：配合哈希校验：
import hashlib
expected_hash = "a1b2c3..." # 预先计算
actual_hash = hashlib.md5(open(file, 'rb').read()).hexdigest()
if actual_hash != expected_hash:
# 重新下载
Q5：下载10000+文件时内存不足怎么办？
A：使用流式下载，避免一次性加载到内存：
with requests.get(url, stream=True) as r:
with open(local_path, 'wb') as f:
for chunk in r.iter_content(chunk_size=8192):
f.write(chunk)
性能优化与避坑指南
1 加速技巧
- 使用异步I/O：
aiohttp库在Python中可实现千级并发 - 多线程分段下载：
aria2支持单文件多连接分块下载 - 文件名冲突处理：添加UUID或时间戳避免覆盖
2 容易踩的坑
| 问题 | 解决方案 |
|---|---|
| 编码错误导致的乱码文件名 | 对URL进行urldecode，并替换特殊字符 |
| 链接重定向丢失 | 设置allow_redirects=True |
| 大文件下载中断 | 使用断点续传（wget -c / Range头） |
| 服务器自动关闭长连接 | 设置keep-alive头，降低请求频率 |
3 合法性提醒
- 遵守目标网站的
robots.txt协议 - 不下载受版权保护的资源
- 控制请求频率,避免对服务器造成压力
批量拉取网络资源是数据工作者、运维人员和开发者必备技能，通过Python或Shell脚本，我们可以将重复劳动自动化，实现高效、可靠的文件下载，核心思路是：准备URL列表 → 选择合适的工具 → 加入错误处理和并发控制 → 记录日志便于排查。
对于10万级以上文件,建议采用异步框架+消息队列架构；对于常规任务，上述Python多线程方案已足够，始终记住：稳定的网络策略比一味追求速度更重要。
