AI 自动下载遥感数据：告别手动找数据

一、传统手动下载的痛点

作为遥感/GIS 从业者，你一定经历过这些痛苦时刻：

1. 登录繁琐: 每次都要打开 Copernicus 网站，输入账号密码登录
2. 搜索低效: 在网站上手动框选区域、选择时间范围、筛选数据
3. 下载缓慢: 网页下载经常中断，需要重新登录、重新搜索、重新下载
4. 批量困难: 需要多景数据时，只能一景一景手动下载
5. 管理混乱: 下载的数据散落在各个文件夹，没有统一的命名和管理

痛点对比：

操作	手动方式	AI 自动化
登录	每次手动输入	自动 Token 认证
搜索	框选区域 + 筛选	代码一键搜索
下载	单景下载，易中断	批量下载，断点续传
管理	手动整理	自动分类命名

效率提升: 手动下载 10 景数据约需 2-3 小时，自动化脚本仅需 5-10 分钟！

---

二、环境准备

2.1 注册 Copernicus Data Space

访问 Copernicus Data Space 注册账号（免费）。

2.2 安装 Python 环境

推荐使用 Anaconda/Miniconda：

# 创建环境
conda create -n remote_sensing python=3.11
conda activate remote_sensing

# 安装依赖
pip install requests sentinelsat python-dotenv

2.3 配置账号信息

创建 .env 文件存储账号信息（不要提交到 GitHub）：

# .env 文件
CDSE_USERNAME=your_email@gmail.com
CDSE_PASSWORD=your_password

---

三、核心代码实现

3.1 自动登录获取 Token

#!/usr/bin/env python3
"""
Copernicus Data Space 自动登录
"""

import os
import requests
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

def get_cdse_token():
    """获取 Copernicus Data Space 访问令牌"""
    
    base_url = "https://identity.dataspace.copernicus.eu/auth/realms/CDSE"
    username = os.getenv("CDSE_USERNAME")
    password = os.getenv("CDSE_PASSWORD")
    
    if not username or not password:
        raise ValueError("请在 .env 文件中配置 CDSE_USERNAME 和 CDSE_PASSWORD")
    
    # 登录获取 token
    token_url = f"{base_url}/protocol/openid-connect/token"
    response = requests.post(token_url, data={
        'grant_type': 'password',
        'username': username,
        'password': password,
        'client_id': 'cdse-public'
    })
    
    if response.status_code == 200:
        token = response.json()['access_token']
        print(f"✅ 登录成功！Token 长度：{len(token)}")
        return token
    else:
        print(f"❌ 登录失败：{response.text}")
        return None

if __name__ == "__main__":
    token = get_cdse_token()
    if token:
        print(f"Token 前 20 位：{token[:20]}...")

3.2 自动搜索遥感数据

#!/usr/bin/env python3
"""
自动搜索 Sentinel-1/2 遥感数据
"""

import os
import requests
import json
from datetime import datetime

def search_sentinel_data(token, collection="sentinel-1", 
                         start_date="2025-06-01", end_date="2025-07-01",
                         limit=10):
    """搜索 Sentinel 数据"""
    
    # STAC API 搜索端点
    stac_url = "https://browser.stac.dataspace.copernicus.eu/api/v1/search"
    headers = {'Authorization': f'Bearer {token}'}
    
    # 搜索参数
    payload = {
        'collections': [collection],
        'datetime': f'{start_date}T00:00:00Z/{end_date}T00:00:00Z',
        'limit': limit
    }
    
    response = requests.post(stac_url, headers=headers, json=payload)
    
    if response.status_code == 200:
        try:
            data = response.json()
            features = data.get('features', [])
            print(f"✅ 找到 {len(features)} 条记录")
            
            # 显示搜索结果
            for i, item in enumerate(features[:5], 1):
                props = item.get('properties', {})
                print(f"\n{i}. {item.get('id', 'N/A')}")
                print(f"   日期：{props.get('datetime', 'N/A')}")
                print(f"   大小：{props.get('assets', {}).get('product', {}).get('size', 'N/A')} MB")
            
            return features
        except:
            print(f"❌ 解析响应失败")
            return []
    else:
        print(f"❌ 搜索失败：{response.status_code}")
        return []

if __name__ == "__main__":
    # 加载 token
    from auto_login import get_cdse_token
    token = get_cdse_token()
    
    if token:
        # 搜索 Sentinel-1 SLC 数据
        products = search_sentinel_data(
            token=token,
            collection="sentinel-1",
            start_date="2025-06-01",
            end_date="2025-07-01",
            limit=5
        )

3.3 自动下载 + 断点续传

#!/usr/bin/env python3
"""
自动下载遥感数据 + 断点续传
"""

import os
import requests
import time
from pathlib import Path

def download_product(token, product_id, output_dir="./data", max_retries=3):
    """下载单景产品，支持断点续传"""
    
    # 下载 URL
    download_url = f"https://dataspace.copernicus.eu/browser/api/v1/products/{product_id}/zip"
    headers = {'Authorization': f'Bearer {token}'}
    
    # 创建输出目录
    os.makedirs(output_dir, exist_ok=True)
    output_path = os.path.join(output_dir, f"{product_id}.zip")
    
    # 断点续传
    if os.path.exists(output_path):
        file_size = os.path.getsize(output_path)
        headers['Range'] = f'bytes={file_size}-'
        mode = 'ab'  # 追加模式
        print(f"  📥 断点续传：{file_size / 1024 / 1024:.1f} MB")
    else:
        mode = 'wb'  # 写入模式
        print(f"  📥 开始下载：{product_id}")
    
    # 下载
    for attempt in range(max_retries):
        try:
            response = requests.get(download_url, headers=headers, stream=True, timeout=300)
            
            if response.status_code in [200, 206]:  # 206 表示部分下载
                total_size = int(response.headers.get('Content-Length', 0))
                downloaded = 0
                
                with open(output_path, mode) as f:
                    for chunk in response.iter_content(chunk_size=8192):
                        if chunk:
                            f.write(chunk)
                            downloaded += len(chunk)
                            
                            # 显示进度
                            if total_size > 0:
                                percent = downloaded / total_size * 100
                                print(f"\r  📥 进度：{percent:.1f}%", end="")
                
                print(f"\n  ✅ 下载完成：{output_path}")
                return True
            else:
                print(f"  ⚠️ 下载失败（尝试 {attempt + 1}/{max_retries}）：{response.status_code}")
                time.sleep(5)  # 等待 5 秒后重试
                
        except Exception as e:
            print(f"  ⚠️ 下载异常（尝试 {attempt + 1}/{max_retries}）：{e}")
            time.sleep(5)
    
    print(f"  ❌ 下载失败：{product_id}")
    return False

def batch_download(token, product_ids, output_dir="./data"):
    """批量下载产品"""
    
    print(f"🚀 开始批量下载 {len(product_ids)} 景数据")
    print(f"📁 输出目录：{output_dir}")
    print()
    
    success_count = 0
    fail_count = 0
    
    for i, product_id in enumerate(product_ids, 1):
        print(f"[{i}/{len(product_ids)}] 下载：{product_id}")
        
        if download_product(token, product_id, output_dir):
            success_count += 1
        else:
            fail_count += 1
        
        # 避免请求过快
        if i < len(product_ids):
            time.sleep(2)
    
    print()
    print("="*60)
    print(f"📊 下载完成统计：")
    print(f"  ✅ 成功：{success_count}")
    print(f"  ❌ 失败：{fail_count}")
    print(f"  📁 总计：{len(product_ids)}")
    print("="*60)

if __name__ == "__main__":
    # 加载 token
    from auto_login import get_cdse_token
    token = get_cdse_token()
    
    if token:
        # 示例：下载 3 景 Sentinel-1 数据
        product_ids = [
            "S1A_IW_SLC__1SDV_20250607T101359_20250607T101426_059539_076449_4477",
            "S1A_IW_SLC__1SDV_20250619T101359_20250619T101425_059714_076A41_9B59",
            "S1A_IW_SLC__1SDV_20250701T101358_20250701T101425_059889_07705E_AAEB"
        ]
        
        batch_download(token, product_ids, output_dir="./sentinel1_data")

---

四、效果对比

4.1 时间对比

操作	手动方式	自动化脚本
登录	2-3 分钟	5 秒
搜索	5-10 分钟	10 秒
下载 1 景	10-15 分钟	5-8 分钟
下载 10 景	2-3 小时	30-50 分钟
总计	2-3 小时	30-50 分钟

4.2 功能对比

功能	手动方式	自动化脚本
断点续传	❌ 不支持	✅ 支持
批量下载	❌ 不支持	✅ 支持
自动命名	❌ 手动	✅ 自动
进度显示	❌ 无	✅ 实时进度
错误重试	❌ 手动	✅ 自动重试

---

五、常见问题

Q1: 下载速度慢怎么办？

A: 可以尝试以下方法：

1. 使用国内镜像源（如阿里云 OSS）
2. 调整下载线程数（多线程下载）
3. 选择非高峰时段下载

Q2: 断点续传失败怎么办？

A: 检查以下几点：

1. 确认文件大小与服务器一致
2. 删除部分下载的文件，重新下载
3. 检查网络连接是否稳定

Q3: 如何搜索特定区域的数据？

A: 可以使用 GeoJSON 格式指定搜索区域：

# 添加空间过滤
payload = {
    'collections': ['sentinel-1'],
    'datetime': '2025-06-01T00:00:00Z/2025-07-01T00:00:00Z',
    'intersects': {
        'type': 'Polygon',
        'coordinates': [[[lon1, lat1], [lon2, lat1], [lon2, lat2], [lon1, lat2], [lon1, lat1]]]
    },
    'limit': 10
}

Q4: 如何自动重命名下载的文件？

A: 可以使用产品元数据自动命名：

# 从产品 ID 提取信息
product_id = "S1A_IW_SLC__1SDV_20250607T101359_20250607T101426_059539_076449_4477"
date_str = product_id.split("_")[5][:8]  # 20250607
new_name = f"sentinel1_{date_str}.zip"
os.rename(output_path, os.path.join(output_dir, new_name))

---

六、总结

本教程实现了遥感数据的全自动下载，包括：

1. ✅ 自动登录: 使用 Token 认证，无需手动输入账号密码
2. ✅ 自动搜索: 支持时间范围、空间区域、数据类型筛选
3. ✅ 批量下载: 一次下载多景数据，无需手动操作
4. ✅ 断点续传: 网络中断后可继续下载，无需重新开始
5. ✅ 进度显示: 实时显示下载进度，方便监控

效率提升: 从手动 2-3 小时缩短到自动化 30-50 分钟，效率提升 3-5 倍！

---

七、完整代码下载

完整代码已上传至 GitHub：AI-Remote-Sensing-Automation

# 克隆代码
git clone https://github.com/your-username/AI-Remote-Sensing-Automation.git
cd AI-Remote-Sensing-Automation

# 安装依赖
pip install -r requirements.txt

# 配置账号
cp .env.example .env
# 编辑 .env 文件，填入你的 Copernicus 账号密码

# 运行下载
python auto_download.py

---

作者: 赵志远
单位: 河北地矿集团 / 河北水文工程地质勘察院有限责任公司
发布日期: 2026-04-22
系列: AI + 遥感/GIS 自动化系列（共 12 篇）

---

下一篇预告: 《AI 自动 InSAR 处理：从 Sentinel-1 到形变图》