从SciHub到OData API：Python自动化下载哥白尼数据的完整指南

当欧空局宣布关闭SciHub平台时，许多依赖该平台获取卫星数据的研究人员和开发者陷入了困境。新的Copernicus Data Space Ecosystem平台虽然提供了更规范的API接口，但迁移过程却让不少用户感到无所适从。本文将带你深入了解如何通过Python脚本高效利用OData API，实现批量下载哥白尼数据的全自动化流程。

1. 新旧平台对比与迁移挑战

Copernicus Data Space Ecosystem平台与旧版SciHub在数据获取方式上存在显著差异。旧平台以直接下载链接为主，而新平台则采用了标准的OData协议作为数据访问接口。这种变化虽然提高了系统的规范性和可扩展性，但也带来了新的学习曲线。

主要差异点对比 ：

特性	SciHub平台	Copernicus Data Space Ecosystem
访问方式	直接HTTP下载	OData API标准接口
认证机制	基础HTTP认证	OAuth 2.0令牌认证
查询能力	有限过滤条件	强大的OData过滤语法
并发限制	较严格	更灵活的配额管理
数据组织	按产品类型分类	统一目录服务

迁移过程中最常见的痛点包括：

• 认证流程复杂化，需要获取和管理访问令牌
• 查询语法学习成本高，特别是对于不熟悉OData协议的开发者
• 缺乏现成的批量下载工具，需要自行开发脚本
• 错误处理和重试机制需要额外实现

2. OData API核心概念与认证机制

OData(Open Data Protocol)是一种基于REST的数据访问协议，它标准化了数据的查询和操作方式。在Copernicus平台中，所有数据访问都通过OData接口完成，这为开发者提供了统一的编程模型。

认证流程详解 ：

1. 获取访问令牌是使用API的第一步。Copernicus平台采用OAuth 2.0的密码授权模式：

def get_access_token(username: str, password: str) -> str:
    data = {
        "client_id": "cdse-public",
        "username": username,
        "password": password,
        "grant_type": "password",
    }
    try:
        response = requests.post(
            "https://identity.dataspace.copernicus.eu/auth/realms/CDSE/protocol/openid-connect/token",
            data=data,
        )
        response.raise_for_status()
        return response.json()["access_token"]
    except Exception as e:
        raise Exception(f"认证失败: {str(e)}")

提示：妥善保管访问令牌，建议将其存储在环境变量中而非硬编码在脚本里

2. 使用令牌进行API调用时，需要在请求头中添加Authorization字段：

access_token = get_access_token("your_username", "your_password")
headers = {"Authorization": f"Bearer {access_token}"}
session = requests.Session()
session.headers.update(headers)

3. 高级查询构建与数据筛选

OData的强大之处在于其灵活的查询能力。通过精心构造的filter参数，可以精确筛选出所需的数据产品。以下是一个典型的产品查询URL构建函数：

def build_product_query_url(params):
    filters = []
    
    # 时间范围过滤
    if 'time_range' in params:
        start, end = params['time_range']
        filters.append(f"ContentDate/Start gt {start}T00:00:00.000Z")
        filters.append(f"ContentDate/Start lt {end}T00:00:00.000Z")
    
    # 轨道号过滤
    if 'orbit_number' in params:
        filters.append(
            "Attributes/OData.CSC.IntegerAttribute/any(att:att/Name eq 'relativeOrbitNumber' "
            f"and att/OData.CSC.IntegerAttribute/Value eq {params['orbit_number']})"
        )
    
    # 产品类型过滤
    if 'product_type' in params:
        filters.append(
            "Attributes/OData.CSC.StringAttribute/any(att:att/Name eq 'productType' "
            f"and att/OData.CSC.StringAttribute/Value eq '{params['product_type']}')"
        )
    
    # 组合所有过滤条件
    base_url = "https://catalogue.dataspace.copernicus.eu/odata/v1/Products"
    query = f"{base_url}?$filter={' and '.join(filters)}&$top=1000"
    
    return query

常见查询场景示例 ：

• 获取2023年1月所有S3A卫星的SR_2_LAN___产品：

  params = {
      'time_range': ['2023-01-01', '2023-02-01'],
      'product_type': 'SR_2_LAN___',
      'collection': 'S3A'
  }
  

• 查询特定轨道号(310)的降轨(DESCENDING)数据：

  params = {
      'orbit_number': 310,
      'orbit_direction': 'DESCENDING'
  }
  

4. 构建生产级下载脚本

一个健壮的下载脚本需要考虑多种因素：网络稳定性、大文件传输、进度显示、错误恢复等。以下是关键组件的实现：

带重试机制的下载函数 ：

def download_with_retry(url, filename, max_retries=3, chunk_size=8192):
    retries = 0
    while retries < max_retries:
        try:
            with requests.get(url, stream=True) as r:
                r.raise_for_status()
                total_size = int(r.headers.get('content-length', 0))
                
                with open(filename, 'wb') as f, tqdm(
                    desc=filename,
                    total=total_size,
                    unit='B',
                    unit_scale=True,
                    unit_divisor=1024,
                ) as progress:
                    for chunk in r.iter_content(chunk_size=chunk_size):
                        if chunk:  # 过滤掉保持连接的空chunk
                            f.write(chunk)
                            progress.update(len(chunk))
            
            # 验证文件完整性
            if os.path.getsize(filename) == total_size:
                return True
            else:
                os.remove(filename)
                raise Exception("文件大小不匹配")
                
        except Exception as e:
            retries += 1
            if retries < max_retries:
                time.sleep(5 * retries)  # 指数退避
                continue
            raise

并发下载管理器 ：

from concurrent.futures import ThreadPoolExecutor, as_completed

def batch_download(product_list, max_workers=4):
    successful = []
    failed = []
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_product = {
            executor.submit(
                download_with_retry, 
                product['download_url'],
                product['filename']
            ): product for product in product_list
        }
        
        for future in as_completed(future_to_product):
            product = future_to_product[future]
            try:
                result = future.result()
                if result:
                    successful.append(product['filename'])
            except Exception as e:
                failed.append((product['filename'], str(e)))
    
    return successful, failed

文件完整性检查 ：

def verify_downloads(download_dir, expected_products):
    missing = []
    corrupted = []
    
    for product in expected_products:
        filepath = os.path.join(download_dir, product['filename'])
        if not os.path.exists(filepath):
            missing.append(product['filename'])
            continue
            
        actual_size = os.path.getsize(filepath)
        if actual_size != product['expected_size']:
            corrupted.append(product['filename'])
    
    return missing, corrupted

5. 脚本优化与高级功能

配置管理 ：

建议使用配置文件或环境变量管理脚本参数：

import configparser

def load_config(config_path):
    config = configparser.ConfigParser()
    config.read(config_path)
    
    return {
        'credentials': {
            'username': config.get('AUTH', 'USERNAME'),
            'password': config.get('AUTH', 'PASSWORD')
        },
        'download': {
            'directory': config.get('DOWNLOAD', 'DIRECTORY'),
            'max_workers': config.getint('DOWNLOAD', 'MAX_WORKERS'),
            'retries': config.getint('DOWNLOAD', 'MAX_RETRIES')
        }
    }

日志记录 ：

import logging

def setup_logging(log_file='download.log'):
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(levelname)s - %(message)s',
        handlers=[
            logging.FileHandler(log_file),
            logging.StreamHandler()
        ]
    )
    return logging.getLogger(__name__)

断点续传 ：

def resume_download(url, filename, headers=None):
    if os.path.exists(filename):
        downloaded = os.path.getsize(filename)
        headers = headers or {}
        headers['Range'] = f'bytes={downloaded}-'
    else:
        downloaded = 0
    
    with requests.get(url, headers=headers, stream=True) as r:
        r.raise_for_status()
        total_size = int(r.headers.get('content-length', 0)) + downloaded
        
        mode = 'ab' if downloaded else 'wb'
        with open(filename, mode) as f, tqdm(
            desc=filename,
            total=total_size,
            initial=downloaded,
            unit='B',
            unit_scale=True,
        ) as progress:
            for chunk in r.iter_content(chunk_size=8192):
                if chunk:
                    f.write(chunk)
                    progress.update(len(chunk))
    
    return total_size == os.path.getsize(filename)

6. 实战：完整数据获取流程

结合上述组件，我们可以构建一个端到端的数据获取流程：

1. 初始化配置和日志 ：

   config = load_config('config.ini')
   logger = setup_logging()
   

2. 获取访问令牌 ：

   try:
       token = get_access_token(
           config['credentials']['username'],
           config['credentials']['password']
       )
       headers = {'Authorization': f'Bearer {token}'}
   except Exception as e:
       logger.error(f'认证失败: {str(e)}')
       sys.exit(1)
   

3. 查询目标产品 ：

   query_params = {
       'time_range': ['2023-01-01', '2023-02-01'],
       'product_type': 'SR_2_LAN___',
       'collection': 'S3A'
   }
   query_url = build_product_query_url(query_params)
   
   try:
       response = requests.get(query_url, headers=headers)
       products = response.json()['value']
       logger.info(f'找到 {len(products)} 个匹配产品')
   except Exception as e:
       logger.error(f'查询失败: {str(e)}')
       sys.exit(1)
   

4. 准备下载任务 ：

   download_tasks = []
   for product in products:
       task = {
           'filename': product['Name'],
           'download_url': f"https://zipper.dataspace.copernicus.eu/odata/v1/Products({product['Id']})/$value",
           'expected_size': product['ContentLength']
       }
       download_tasks.append(task)
   

5. 执行批量下载 ：

   successful, failed = batch_download(
       download_tasks,
       max_workers=config['download']['max_workers']
   )
   
   logger.info(f'下载完成: 成功 {len(successful)} 个, 失败 {len(failed)} 个')
   if failed:
       for filename, error in failed:
           logger.warning(f'{filename}: {error}')
   

6. 验证下载结果 ：

   missing, corrupted = verify_downloads(
       config['download']['directory'],
       download_tasks
   )
   
   if missing:
       logger.warning(f'缺失文件: {len(missing)} 个')
   if corrupted:
       logger.warning(f'损坏文件: {len(corrupted)} 个')
   

在实际项目中，这套脚本已经帮助团队将数据获取时间从原来的数小时缩短到几分钟，同时显著提高了下载的可靠性。特别是在处理大批量数据时，合理的并发控制和错误恢复机制能够节省大量时间。