基于python下载欧空局哨兵数据，采用token方式下载

import requests

def download_sentinel_data(username, password,downing_path,params):
    """
    下载哨兵数据的封装函数
    
    参数:
    username: 用户名
    password: 密码  
    params: 参数字典，包含:
        - collection: 数据集 ('S2A', 'S2B', 'SENTINEL-2')
        - start_date: 开始日期 ('YYYY-MM-DD')
        - end_date: 结束日期 ('YYYY-MM-DD')
        - cloud_cover: 云量阈值
        - bbox: 四至范围 [min_lon, min_lat, max_lon, max_lat]
        - mgrs_tile: MGRS网格 (如 'T50SMJ')
        - top: 返回数量
    """
    def get_token():
        return requests.post(
            "https://identity.dataspace.copernicus.eu/auth/realms/CDSE/protocol/openid-connect/token",
            data={'client_id': 'cdse-public', 'username': username, 'password': password, 'grant_type': 'password'}
        ).json()['access_token']

    # 构建查询
    filters = [f"Collection/Name eq '{params['collection']}'"]
    filters.append("contains(Name,'MSIL2A')") #只筛选L2A级数据
    filters.append(f"ContentDate/Start ge {params['start_date']}T00:00:00.000Z")
    filters.append(f"ContentDate/Start le {params['end_date']}T23:59:59.999Z")
    filters.append(f"Attributes/OData.CSC.DoubleAttribute/any(att:att/Name eq 'cloudCover' and att/Value lt {params['cloud_cover']})")

    if params.get('bbox'):
        bbox = params['bbox']
        filters.append(f"OData.CSC.Intersects(area=geography'SRID=4326;POLYGON(({bbox[0]} {bbox[1]}, {bbox[2]} {bbox[1]}, {bbox[2]} {bbox[3]}, {bbox[0]} {bbox[3]}, {bbox[0]} {bbox[1]}))')")

    if params.get('mgrs_tile'):
        filters.append(f"Attributes/OData.CSC.StringAttribute/any(att:att/Name eq 'tileId' and att/Value eq '{params['mgrs_tile']}')")

    search_url = f"https://catalogue.dataspace.copernicus.eu/odata/v1/Products?$filter={' and '.join(filters)}&$top={params['top']}"

    # 搜索下载
    products = requests.get(search_url, headers={'Authorization': f'Bearer {get_token()}'}).json()['value']
    print(f"找到 {len(products)} 个产品")

    for product in products:
        download_url = f"https://zipper.dataspace.copernicus.eu/odata/v1/Products({product['Id']})/$value"
        response = requests.get(download_url, headers={'Authorization': f'Bearer {get_token()}'}, stream=True)
        
        if response.status_code == 200:
            with open(f"{downing_path}/{product['Name']}.zip", 'wb') as f:
                for chunk in response.iter_content(8192):
                    f.write(chunk)
            print(f"下载完成: {product['Name']}")
        else:
            print(f"下载失败 {product['Name']}: {response.status_code}")

# 使用示例
if __name__ == "__main__":
    params = {
        'collection': 'SENTINEL-2',
        'start_date': '2024-01-01',
        'end_date': '2024-01-10', 
        'cloud_cover': 10,
        'bbox': [116.0, 39.0, 117.0, 40.0],#北京为例
        'mgrs_tile': None,
        'top': 2
    }
    downing_path =r".\Desktop\py_code\temp"
    download_sentinel_data("账户", "密码",downing_path,params)