原创 河山 牛瞎哞 | 2026年5月7日 19:05 上海
ZENODO 是一个由 CERN（欧洲核子研究组织）数据中心所支持的科研数据存储库，使研究人员能够保存和分享他们的研究成果。但是网页直接点击下载数据的速度较慢，并且容易中断。Zenodo 官方 REST API 支持 Records 和 Files 访问，所以网页下载慢时，用 API/命令行更稳定。在此分享更加快速稳定的数据下载方法：Python 获取文件列表 + aria2c 断点续传下载 + MD5 校验。
1. 网页下载速度#
直接点击下载大概只有 100-200 kB/s 的下载速度，并且不稳定，经常中断或变为 10-20 kB/s，下载 1GB 的数据需要数小时。
2. API/命令行下载速度#
用学校 wifi，实测的下载速度大约为 1 MB/s。
3. Python 代码#
Python 获取文件列表 + aria2c 断点续传下载 + MD5 校验。
需要的 Python 包和下载工具#
PowerShell：
python -m pip install -U pip
python -m pip install requests tqdm PySocks
# requests：访问 Zenodo API，获取文件列表和 MD5
# tqdm：进度条，可选，建议装
# PySocks：避免报错
# 安装后检查
python -c "import requests, tqdm; print('Python packages OK')"
# aria2c.exe 不是 Python 包，不用装在虚拟环境里。
# 它是一个独立下载工具，安装在 Windows 系统层面。
winget install -e --id aria2.aria2
# 安装完成后，关闭当前 PowerShell，重新打开，再检查
aria2c --version
修改目标下载目录#
# 修改为你的目标下载目录
OUT_DIR = Path(r"D:\Zenodo_Download\CropLayer_2020")
指定想要部分下载的数据#
TARGET_FILES = [] # 是下载所有文件
TARGET_FILES = [
"cf_county_guangdong.tar.gz",
"cf_county_hainan.tar.gz",
"cf_county_yunnan.tar.gz",
]
# 是下载指定文件，例如广东、海南、云南的数据
并发参数不要太大，避免 Zenodo 限流#
# aria2 并发参数：不要开太大，避免触发 Zenodo 限流
CONNECTIONS_PER_FILE = 4 # 单个文件连接数
PARALLEL_FILES = 2 # 同时下载几个文件
完整代码#
下载 https://doi.org/10.5281/zenodo.14726428
from pathlib import Path
from urllib.parse import quote
import hashlib
import subprocess
import requests
import sys
# =========================
# 基本参数
# =========================
RECORD_ID = "14726428"
# 修改为你的目标下载目录
OUT_DIR = Path(r"D:\Zenodo_Download\CropLayer_2020")
# 如果为空列表，则下载全部文件
# 如果只想下载部分省份，就写入文件名，例如：
# TARGET_FILES = ["cf_county_hainan.tar.gz", "cf_county_guangdong.tar.gz"]
TARGET_FILES = [
"cf_county_guangdong.tar.gz",
"cf_county_hainan.tar.gz",
"cf_county_yunnan.tar.gz",
"cf_county_jilin.tar.gz",
"cf_county_anhui.tar.gz",
]
# aria2 并发参数：不要开太大，避免触发 Zenodo 限流
CONNECTIONS_PER_FILE = 4 # 单个文件连接数
PARALLEL_FILES = 2 # 同时下载几个文件
# 给请求加 User-Agent，避免默认 Python requests 被部分站点识别为低优先级请求
HEADERS = {
"User-Agent": "Zenodo-CropLayer-Downloader/1.0"
}
# =========================
# 工具函数
# =========================
def fetch_record_metadata(record_id: str) -> dict:
"""从 Zenodo API 获取 record 元数据。"""
api_url = f"https://zenodo.org/api/records/{record_id}"
response = requests.get(api_url, headers=HEADERS, timeout=60)
response.raise_for_status()
return response.json()
def normalize_checksum(checksum: str) -> str | None:
"""提取 md5 值。Zenodo 常见形式为 md5:xxxxxxxx。"""
if not checksum:
return None
if checksum.startswith("md5:"):
return checksum.split("md5:", 1)[1]
return checksum
def build_download_url(record_id: str, filename: str) -> str:
"""构造 Zenodo 文件下载链接。"""
encoded_name = quote(filename)
return f"https://zenodo.org/records/{record_id}/files/{encoded_name}?download=1"
def write_aria2_input(files: list[dict], url_file: Path) -> None:
"""生成 aria2c 输入文件。"""
with url_file.open("w", encoding="utf-8") as f:
for item in files:
name = item["key"]
url = build_download_url(RECORD_ID, name)
f.write(url + "\n")
f.write(f" out={name}\n")
def run_aria2(url_file: Path, out_dir: Path) -> None:
"""调用 aria2c 进行断点续传下载。"""
cmd = [
"aria2c",
"-i", str(url_file),
"-d", str(out_dir),
"-c", # 断点续传
"-x", str(CONNECTIONS_PER_FILE),
"-s", str(CONNECTIONS_PER_FILE),
"-j", str(PARALLEL_FILES),
"--max-tries=10",
"--retry-wait=20",
"--timeout=60",
"--connect-timeout=30",
"--continue=true",
"--auto-file-renaming=false",
"--allow-overwrite=true",
]
print("\n开始调用 aria2c 下载：")
print(" ".join(cmd))
subprocess.run(cmd, check=True)
def md5sum(path: Path, block_size: int = 1024 * 1024 * 8) -> str:
"""计算本地文件 md5。"""
h = hashlib.md5()
with path.open("rb") as f:
while True:
block = f.read(block_size)
if not block:
break
h.update(block)
return h.hexdigest()
def check_files(files: list[dict], out_dir: Path) -> None:
"""校验下载文件的 MD5。"""
print("\n开始 MD5 校验：")
failed = []
for item in files:
name = item["key"]
expected_md5 = normalize_checksum(item.get("checksum", ""))
local_file = out_dir / name
if not local_file.exists():
print(f"[缺失] {name}")
failed.append(name)
continue
if not expected_md5:
print(f"[跳过] {name}：Zenodo 未提供 checksum")
continue
real_md5 = md5sum(local_file)
if real_md5.lower() == expected_md5.lower():
print(f"[通过] {name}")
else:
print(f"[失败] {name}")
print(f" 期望: {expected_md5}")
print(f" 实际: {real_md5}")
failed.append(name)
if failed:
print("\n以下文件校验失败或缺失，建议删除后重新下载：")
for name in failed:
print(" -", name)
else:
print("\n全部文件 MD5 校验通过。")
# =========================
# 主流程
# =========================
def main():
OUT_DIR.mkdir(parents=True, exist_ok=True)
print("正在读取 Zenodo record 元数据...")
record = fetch_record_metadata(RECORD_ID)
all_files = record.get("files", [])
if not all_files:
print("没有读取到 files 字段，请检查 record ID 或网络连接。")
sys.exit(1)
print(f"Zenodo record: {record.get('metadata', {}).get('title', '')}")
print(f"文件总数: {len(all_files)}")
if TARGET_FILES:
target_set = set(TARGET_FILES)
files = [f for f in all_files if f["key"] in target_set]
missing = target_set - {f["key"] for f in files}
if missing:
print("\n以下 TARGET_FILES 没有在 Zenodo 记录中找到：")
for name in missing:
print(" -", name)
sys.exit(1)
else:
files = all_files
print("\n本次将下载以下文件：")
for f in files:
size_gb = f.get("size", 0) / 1024**3
print(f" - {f['key']} ({size_gb:.2f} GB)")
url_file = OUT_DIR / "zenodo_aria2_urls.txt"
write_aria2_input(files, url_file)
print(f"\n已生成 aria2 下载清单：{url_file}")
run_aria2(url_file, OUT_DIR)
check_files(files, OUT_DIR)
if __name__ == "__main__":
main()