Automating USGS EarthExplorer Bulk Downloads with requests

import os
import json
import time
import logging
from pathlib import Path
from urllib.parse import urljoin
from requests import Session
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

USGS_M2M_BASE = "https://m2m.cr.usgs.gov/api/api/json/stable/"
DATASET_ID = "LANDSAT_8_C2_L1"
OUTPUT_DIR = Path("./usgs_downloads")
OUTPUT_DIR.mkdir(exist_ok=True)

def get_session() -> Session:
    """Configure session with connection pooling and exponential backoff."""
    session = Session()
    retry_strategy = Retry(
        total=5,
        backoff_factor=1.5,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET", "POST"]
    )
    adapter = HTTPAdapter(
        max_retries=retry_strategy,
        pool_connections=10,
        pool_maxsize=10
    )
    session.mount("https://", adapter)
    session.headers.update({"Content-Type": "application/json"})
    return session

def authenticate(session: Session, username: str, password: str) -> str:
    """Exchange credentials for a temporary M2M token."""
    payload = {"username": username, "password": password}
    resp = session.post(urljoin(USGS_M2M_BASE, "login"), json=payload)
    resp.raise_for_status()
    token = resp.json()["data"]
    logger.info("Authentication successful. Token valid for ~2 hours.")
    return token

def search_scenes(session: Session, token: str, bbox: list, start_date: str, end_date: str) -> list:
    """Retrieve scene metadata for a bounding box and date range."""
    payload = {
        "datasetName": DATASET_ID,
        "maxResults": 50,
        "sceneFilter": {
            "acquisitionFilter": {"start": start_date, "end": end_date},
            "spatialFilter": {"filterType": "mbr", "lowerLeft": bbox[:2], "upperRight": bbox[2:]}
        }
    }
    headers = {"X-Auth-Token": token}
    resp = session.post(urljoin(USGS_M2M_BASE, "scene-search"), json=payload, headers=headers)
    resp.raise_for_status()
    return resp.json()["data"]["results"]

def get_download_urls(session: Session, token: str, scene_ids: list) -> dict:
    """Resolve temporary download URLs for a list of scene IDs."""
    payload = {"datasetName": DATASET_ID, "sceneIds": scene_ids}
    headers = {"X-Auth-Token": token}
    resp = session.post(urljoin(USGS_M2M_BASE, "download-options"), json=payload, headers=headers)
    resp.raise_for_status()
    
    url_map = {}
    for scene in resp.json()["data"]:
        # Filter for primary product (usually Level-1 or Level-2 GeoTIFF/ZIP)
        primary = next((opt for opt in scene.get("downloadOptions", []) if opt.get("primary")), None)
        if primary:
            url_map[scene["entityId"]] = primary["url"]
    return url_map

def download_file(session: Session, url: str, dest_path: Path, chunk_size: int = 8192) -> None:
    """Stream large files to disk with progress logging."""
    with session.get(url, stream=True) as resp:
        resp.raise_for_status()
        total_size = int(resp.headers.get("content-length", 0))
        downloaded = 0
        
        with open(dest_path, "wb") as f:
            for chunk in resp.iter_content(chunk_size=chunk_size):
                f.write(chunk)
                downloaded += len(chunk)
                if total_size:
                    logger.info(f"{dest_path.name}: {downloaded/1024/1024:.1f}MB / {total_size/1024/1024:.1f}MB")
    logger.info(f"Download complete: {dest_path}")

def run_pipeline(username: str, password: str, bbox: list, start_date: str, end_date: str) -> None:
    session = get_session()
    token = authenticate(session, username, password)
    
    scenes = search_scenes(session, token, bbox, start_date, end_date)
    if not scenes:
        logger.warning("No scenes found for the specified criteria.")
        return
        
    scene_ids = [s["entityId"] for s in scenes]
    url_map = get_download_urls(session, token, scene_ids)
    
    for scene_id, dl_url in url_map.items():
        dest = OUTPUT_DIR / f"{scene_id}.tar.gz"
        if dest.exists():
            logger.info(f"Skipping existing: {dest.name}")
            continue
        try:
            download_file(session, dl_url, dest)
        except Exception as e:
            logger.error(f"Failed to download {scene_id}: {e}")

if __name__ == "__main__":
    # Replace with actual credentials or environment variables
    run_pipeline(
        username="YOUR_USGS_USERNAME",
        password="YOUR_USGS_PASSWORD",
        bbox=[-122.5, 37.7, -122.3, 37.9],  # San Francisco Bay Area
        start_date="2023-06-01",
        end_date="2023-06-30"
    )