做遥感这行，最耗时间的往往不是算法，而是数据准备。尤其是 Sentinel-2，一景 L2A 产品动辄几百 MB 到 1GB 出头，真正用 Python 批量下载过 Sentinel-2 数据的人，大概率都经历过脚本跑到一半被限流、token 失效、断点续传失效、下载下来的压缩包损坏这类破事。2024 年以后这套流程又踩了一个大变化：旧的下载通道退出了历史舞台，网上大量“老教程”里的代码现在直接跑不通。这篇文章把我在生产环境里一直在用的 Python 批量下载 Sentinel-2 数据方案整体梳理一遍，代码可直接落地，最后附一份完整避坑清单，适合刚接触遥感的同学，也适合已经被限流折磨过的老手参考。
1. 数据源之选：2024 年还有哪几条路能批量拿 Sentinel-2
1.1 为什么你的老脚本突然全军覆没
2023 年之前，绝大多数 Python 批量下载 Sentinel-2 的教程，核心依赖都是sentinelsat这个库，访问的是 SciHub 这个老平台。后来欧洲空间局把数据分发主力切换到了 Copernicus Data Space Ecosystem（后面统一叫 CDSE），老 SciHub 通道陆续关闭，sentinelsat已经无法正常连上默认端点。我 2024 年复跑以前的脚本时遇到的第一行报错就是连接超时，换sentinelsat的镜像 URL 也只是治标不治本，因为认证方式也从“用户名+密码直连”变成了 OAuth 2.0 客户端凭证模式。
这意味着你找新教程时，一定要认准接口地址包含dataspace.copernicus.eu的方案。CDSE 官方提供了两套常见接口：一个是 STAC API，适合做条件筛选，能按时间、经纬度范围、云量等条件查出哪些产品符合需求；另一个是 OData API，适合拿具体产品 ID 去换取下载地址。两套接口配合使用，基本能覆盖“筛选到下载”的完整链路。
1.2 官方平台、AWS 公开数据、云端处理库怎么选
除了 CDSE 官方下载站，还有三条路经常被提到，但适应场景差别很大。
第一是 AWS 公开数据集。Sentinel-2 L2A 的全球归档在 AWS S3 上有一个s3://sentinel-s2-l2a桶，不需要账号，直接用aws s3 sync就能拉数据，速度经常比官方站还稳。但这套数据按瓦片目录存储，路径规则是tiles/UTM带号/纬度带/方格号/年/月/日/0，你得先搞清楚自己要的区域落在哪个瓦片。它的定位适合“知道瓦片号、要大批量补历史数据”的人，不适合随手按任意 Polygon 搜产品。
第二是 Google Earth Engine，也是我经常被问到的。GEE 最大的优势是数据已经在云端，筛完影像直接ee.ImageCollection就能算指数、做时序，完全不需要把数据下载回本地。但如果你就是需要原始的.SAFE产品包，或者要在本地用自己的分割模型跑推理，那 GEE 的getDownloadURL导出路径也很绕，并不比 OData 批量下载省事。我的原则是：只要能云端出结果就不下载，必须要原始包才走 CDSE。
第三是sentinelhub这个 Python 库。它封装了 CDSE 的认证和请求流程，代码看起比裸写requests优雅，但说实话，它为了兼容更多业务场景，抽象层次较高，新手上手反而容易懵。我更习惯直接调用 REST API，至少出问题的时候你能看清是认证挂了还是下载链接失效。下面的代码都以裸requests为主。
2. 动手前必须建好的地基：Python 依赖与 CDSE 客户端凭证
2.1 环境准备与依赖安装
我自己长期用的是 Python 3.11，3.8 以上都没问题。建议新建一个虚拟环境，避免把系统 Python 装成“全家桶”：
conda create -n s2batch python=3.11 -y conda activate s2batch pip install requests tqdm这套流程其实只依赖requests和tqdm，没有额外重量级库。tqdm纯粹是为了看进度条，批量下载几十个场景的时候，你能直观判断当前是卡住了还是在正常传输。如果需要把SAFE产品包解压并读进数组做预处理，到时候再补rasterio、geopandas也不迟。
2.2 CDSE 平台注册与客户端凭证申请
在 CDSE 官网注册普通用户之后，还需要创建一个 OAuth Client，才能拿到脚本里用的client_id和client_secret。操作路径大致是：登录 CDSE 账户后，进入 Dashboard 或者 User Settings，找到 OAuth 相关菜单，创建一个新的 Client Application。创建时会让你填应用名称和重定向 URL，本地脚本场景下重定向 URL 填http://localhost就能通过。创建完页面会展示两个关键字符串：
Client ID，类似一长串 UUID，相当于脚本的“用户名”Client Secret，一长串随机字符，相当于“密码”
这两项要保存好，Secret 只显示一次，丢了只能重新生成。
2.3 凭证管理：别把密钥写死在代码里
我见过不少同学图省事，直接把 client_secret 写进.py文件，结果不小心把代码传到公司 Git 仓库，第二天收到安全告警。稳妥做法是放到环境变量里：
export CDSE_CLIENT_ID="你的-client-id" export CDSE_CLIENT_SECRET="你的-client-secret"Python 脚本里统一从环境变量读取：
import os CLIENT_ID = os.getenv("CDSE_CLIENT_ID") CLIENT_SECRET = os.getenv("CDSE_CLIENT_SECRET") if not CLIENT_ID or not CLIENT_SECRET: raise RuntimeError("请先设置 CDSE_CLIENT_ID 和 CDSE_CLIENT_SECRET 环境变量")这样做的另一个好处是，你可以在多台机器上跑同一个脚本，换机器只改环境变量，不用动代码。
3. 单个产品下载链路先跑通：从 token 到 OData 的完整调用
3.1 获取访问 token
CDSE 的 OAuth 2.0 客户端凭证模式，核心是先拿一个短期 token，再用它去请求下载接口。token 默认有效期不长，大概几十分钟到一小时，所以正确的姿势是：每次批量任务开始时获取一次，运行过程中捕获401 Unauthorized再自动刷新，而不是为每个请求重复申请。
import requests TOKEN_URL = "https://identity.dataspace.copernicus.eu/auth/realms/CDSE/protocol/openid-connect/token" def get_access_token(client_id: str, client_secret: str) -> str: resp = requests.post( TOKEN_URL, data={ "grant_type": "client_credentials", "client_id": client_id, "client_secret": client_secret, }, timeout=30, ) resp.raise_for_status() return resp.json()["access_token"]这条请求发出去后，正常会返回一个 JSON，里面带access_token字段。注意这里用的是表单格式的data，不是 JSON 格式的json，用错的话会一直报 401。
3.2 用 STAC 按时间和经纬度筛选产品
拿到 token 后，第一步是筛选出要下载的产品。CDSE 的 STAC 入口地址是：
https://catalogue.dataspace.copernicus.eu/stac下面这段代码会搜索 2024 年 6 月覆盖北京城区的 L2A 影像：
STAC_URL = "https://catalogue.dataspace.copernicus.eu/stac" def search_sentinel2(bbox, date_from, date_to, token, max_items=20): collection = "SENTINEL-2" url = f"{STAC_URL}/collections/{collection}/items" params = { "bbox": ",".join(str(x) for x in bbox), "datetime": f"{date_from}/{date_to}", "limit": max_items, "sortby": "properties.datetime", } headers = {"Authorization": f"Bearer {token}"} resp = requests.get(url, params=params, headers=headers, timeout=60) resp.raise_for_status() return resp.json().get("features", [])这里两个容易搞混的点：一是datetime参数前后都要有Z，表示 UTC 时间；二是bbox必须按西、南、东、北的顺序填写，不是习惯里的“左上右下”。
筛选结果里，每个 feature 的id就是产品号，properties.datetime是成像时间，properties.eo:cloud_cover是整体云量。L2A 产品还会带s2:product_type之类的额外属性，按需筛选。
3.3 拿到产品 ID 后如何拼接下载地址
STAC 能帮你找到产品元数据，但它返回的资产链接下载起来并不是最省事的。实际操作里我更推荐用 OData 接口，按产品 ID 直接触发下载：
https://catalogue.dataspace.copernicus.eu/odata/v1/Products('<product_id>')/$value拼接Authorization: Bearer <token>请求头，用stream=True把下载流写入文件：
def download_product(product_id: str, token: str, save_path: str): url = f"https://catalogue.dataspace.copernicus.eu/odata/v1/Products({product_id})/$value" headers = {"Authorization": f"Bearer {token}"} with requests.get(url, headers=headers, stream=True, timeout=(30, 600)) as r: r.raise_for_status() with open(save_path, "wb") as f: for chunk in r.iter_content(chunk_size=1024 * 1024): if chunk: f.write(chunk)需要注意，OData 下载返回的默认可能是 ZIP 压缩包，也可能是.SAFE文件夹被打包成一个压缩流，命名通常像S2B_MSIL2A_20240601T022559_N0510_R132_T50TMK_20240601T071234.SAFE.zip。你在循环里保存文件时，最好也按这个后缀落盘，方便后续解压。
4. 批量化的三种做法：for 循环、并发下载与断点续传
4.1 最稳的 for 循环：一次一个跑不死
如果机器配置一般，或者只是偶尔补十来个场景，没必要上并发。直接遍历 STAC 查到的 feature 列表，逐个下载就行。这里的关键是要加上“跳过已存在文件”的逻辑，避免任务中断后从头再来。
import os def download_batch(features, out_dir, token): os.makedirs(out_dir, exist_ok=True) for feat in features: pid = feat["id"] st = feat["properties"]["datetime"] date_str = st[:10] title = f"{date_str}_{pid}.SAFE.zip" save_path = os.path.join(out_dir, title) if os.path.exists(save_path) and os.path.getsize(save_path) > 1024: print("已存在，跳过：", title) continue print("开始下载：", title) try: download_product(pid, token, save_path) except requests.HTTPError as e: if e.response.status_code == 401: print("token 过期，刷新后重试") token = get_access_token(CLIENT_ID, CLIENT_SECRET) download_product(pid, token, save_path) else: print("下载失败：", pid, e, save_path)这段代码已经覆盖了最核心的容错：文件存在就跳过、token 失效就刷新。看起来简单，但它解决了九成生产环境报障。
4.2 线程池并发：注意限流阈值
单线程下载一景 L2A 可能要 5 到 10 分钟，如果一次要下 60 景，耗时确实感人。用concurrent.futures.ThreadPoolExecutor可以提高效率，但不要盲目把线程数开到 16，CDSE 对并发请求有隐式限流，短时间密集请求很容易触发429 Too Many Requests。
我实测下来，4 到 6 个并发 worker 是一个比较稳妥的区间。下面是一个简化版并发框架：
from concurrent.futures import ThreadPoolExecutor, as_completed def download_batch_concurrent(features, out_dir, token, workers=4): os.makedirs(out_dir, exist_ok=True) def one_task(feat): pid = feat["id"] date_str = feat["properties"]["datetime"][:10] title = f"{date_str}_{pid}.SAFE.zip" save_path = os.path.join(out_dir, title) if os.path.exists(save_path) and os.path.getsize(save_path) > 1024: return None try: download_product(pid, token, save_path) return title except Exception as e: return (title, repr(e)) with ThreadPoolExecutor(max_workers=workers) as pool: futures = [pool.submit(one_task, feat) for feat in features] for fut in as_completed(futures): result = fut.result() if result is None: continue if isinstance(result, tuple): print("失败任务：", result[0], result[1]) else: print("完成：", result)并发模式下 token 是多个线程共享的，所以如果其中一个线程遇到 401 后在代码里直接刷新全局 token，其他线程还在用旧 token，可能连续报错。更稳的方案是在one_task里不管 401 还是 403，统一由外层重新获取 token 后再次调用one_task，这里为了篇幅不再展开。
4.3 断点续传：别让 800MB 的努力白费
网络波动是绕不开的难题。CDSE 的下载速度不是一直稳定，尤其是跨洋传输时，可能下到 600MB 突然连接断开，导致整个文件作废。解决办法是下载时先写到.part临时文件，同时用 HTTP 的Range头实现真正的断点续传。
不过需要说明，OData$value接口对 Range 请求的支持并不是每一次都稳定，部分 CDN 节点可能忽略 Range 头直接返回全量文件，这种情况下断点续传会退化成“从头再来”。所以我的实际经验是：用临时文件 + 进度判断兜底，即使没有 Range，至少能通过重试机制保住已完成的文件。如果你对速度有极端要求，优先用 4.2 的并发 + 重试，而不是指望 OData 服务的续传能力。
5. 避坑清单：我把下载失败和踩雷的过程都记录在这里
5.1 认证类问题：401 与 token 过期
最典型的故障是脚本跑了几分钟突然报401 Unauthorized。原因通常是 token 过期，或者一次性下载任务跨过了多个小时。不要每次请求都重新获取 token，这会增加平台压力；合理的做法是捕获 401 后只刷新一次，并在刷新后立即重试当前文件。我把这个逻辑写进任务函数，就能避免“前一张图下了 800MB 后白跑”的尴尬。
5.2 限流类问题：429 需要配合退避重试
CDSE 对同一个账号的并发下载有速率限制，高频请求会触发429 Too Many Requests。很多人的第一反应是提高线程数绕过，实际只会更快触发限流。正确的做法是控制并发数，并在遇到 429 时等待一段时间再重试。我习惯写一个简单的指数退避：
import time max_retries = 5 for attempt in range(max_retries): try: r.raise_for_status() break except requests.HTTPError as e: if e.response.status_code == 429 and attempt < max_retries - 1: time.sleep(5 * (attempt + 1)) continue raise5.3 云量过滤不准：页面显示和接口不一致
STAC 查询结果里的eo:cloud_cover代表整景影像的云量估算，跟哨兵官网浏览界面上的“云覆盖”不一定完全一致，因为不同产品的云量来源于不同算法，L1C 和 L2A 的云量也不同。如果你对云量有硬性要求，比如只要云量低于 20% 的影像，建议在搜索结果里统一过滤eo:cloud_cover <= 20.0，而且下单前再人工抽查几景，别盲信单字段。
5.4 时间边界容易错：UTC 还是本地时间
Sentinel-2 的所有元数据时间都是 UTC。如果你要下载“北京时间 6 月每天”的数据，而搜索时直接用了2024-06-01T00:00:00+08:00，部分解析逻辑可能达不到预期。稳妥做法是先把北京时间转换成 UTC：例如北京 6 月 1 日 00:00 对应 UTC 5 月 31 日 16:00。查询结果返回的时间字段同样带Z，要展示成当地时区时自行转换。
5.5 大文件损坏：下载完后要检查大小和压缩包完整度
Sentinel-2 的压缩包通常几百 MB 到 1GB 以上，偶尔会遇到“下载完成但解压失败”。我在脚本里加了一道校验：比对本地文件大小和响应头里的Content-Length，如果两者相差超过阈值就删掉重下。更稳妥的做法是解压前用zipfile测试压缩包完整度：
import zipfile def verify_zip(path): try: with zipfile.ZipFile(path) as zf: bad = zf.testzip() return bad is None except zipfile.BadZipFile: return False如果返回False，直接删除文件，下次运行脚本时，断点续传逻辑会重新下载它。
5.6 数据产品类别别选错：L1C 还是 L2A
Sentinel-2 在 STAC 集合里统一归在SENTINEL-2下面，但产品 ID 能明显区分级别，MSIL1C是 L1C 顶层反射率，MSIL2A是大气校正后的地表反射率。做定量遥感一般直接用 L2A，做辐射定标算法研究可能需要 L1C。批量下载前最好确认你自己的处理流程依赖哪个级别，不然 60 个 L1C 下载完才发现要的是 L2A，存储和带宽都得重来一遍。
6. 下载完成后的清单校验与数据整理
6.1 快速检查产物是否完整
每次批量下载跑完之后，我会用一个几行的脚本扫一遍目录，列出所有体积小于某个阈值的文件：
for root, _, files in os.walk(out_dir): for fp in files: fpath = os.path.join(root, fp) size_mb = os.path.getsize(fpath) / 1024 / 1024 if size_mb < 50: print("可疑小文件：", fpath, round(size_mb, 2), "MB")这个阈值取决于产品，Sentinel-2 L2A 如果把数据区裁得很小，压出来的 ZIP 也可能很小，所以要结合你自己业务里的正常大小来定。
6.2 统一的目录组织方式
下载完的文件如果全部堆在一个文件夹里，后面找数据会崩溃。我习惯按“级别/年份/瓦片号/日期”的结构整理：
sentinel2_aerial/ ├── L2A/ │ ├── 2024/ │ │ ├── T50TMK/ │ │ │ ├── 20240601/ │ │ │ │ └── S2B_MSIL2A_20240601T022559_N0510_R132_T50TMK_20240601T071234.SAFE.zip │ │ │ └── 20240626/移动文件用脚本批量完成，比在资源管理器里手动拖快得多。目录名里的瓦片号T50TMK一般可以从产品 ID 里直接解析出来，按字符串切片就行。
6.3 后续怎么衔接预处理
下载完 ZIP 后，解压得到.SAFE文件夹，里面包含GRANULE、MTD_MSIL2A.xml等文件。后续接入rasterio读取，最省事的方式是用sentinelhub或者rasterio直接读取MTD_MSIL2A.xml引导的数据路径。但要注意，.SAFE文件夹里的 JP2 文件各波段是独立存放的，做 RGB 合成时要自己指定蓝、绿、红对应的 10m 波段文件，通常在GRANULE/.../IMG_DATA/R10m/下面，文件名类似T50TMK_20240601T022559_B02_10m.jp2对应蓝波段，B03 绿、B04 红。这是下载流程之后最常被问到的问题，先在这里预热一下。
关于这套下载方案，我个人在实际操作中最深的体会是：宁可把脚本写得“保守”一点，也不要追求一次性并发拉满。下载 Sentinel-2 数据的瓶颈往往不在本地网速，而在远端的稳定性和账号限流策略。452 和 401 处理好了，断点续传逻辑写好了，剩下的只是等待时间。此外建议你把这份脚本好好保存，随着处理任务增多，你一定会频繁往回翻“当时到底是怎么下载的”。如果后面还需要把下载下来的 L2A 做波段合成、云掩膜或者时序分析，可以在这个基础上继续往下写。
