Automating Government Portal Downloads

pip install requests==2.32.3 beautifulsoup4==4.12.3 tenacity==8.5.0 \
    geopandas==1.0.1 lxml==5.3.0 pyproj==3.7.0

import geopandas as gpd
import pyproj
print(gpd.__version__)       # expect 1.0+
print(pyproj.proj_version)   # expect 9.x

export GOV_PORTAL_BASE_URL="https://data.example.gov/api/3"
export GOV_API_KEY="your_api_key_here"
export DOWNLOAD_DIR="./data/raw/government"
export MAX_RETRIES=4
export CHUNK_SIZE=65536

import os
import requests
from bs4 import BeautifulSoup

PORTAL_BASE = os.environ["GOV_PORTAL_BASE_URL"]

def detect_portal_type(session: requests.Session, base_url: str) -> str:
    """
    Return 'ckan', 'arcgis', 'wfs', or 'html' based on capability probes.
    """
    probes = {
        "ckan":   f"{base_url}/action/status_show",
        "arcgis": f"{base_url}/rest/info?f=json",
        "wfs":    f"{base_url}?service=WFS&request=GetCapabilities",
    }
    for portal_type, url in probes.items():
        try:
            resp = session.get(url, timeout=10)
            if resp.status_code == 200:
                return portal_type
        except requests.exceptions.RequestException:
            continue
    return "html"

def extract_download_links_from_html(session: requests.Session, page_url: str) -> list[str]:
    """
    Fall back to BeautifulSoup when no structured API is found.
    Targets <a href> pointing at known spatial extensions.
    """
    spatial_exts = {".geojson", ".gpkg", ".shp", ".zip", ".kml", ".gml"}
    resp = session.get(page_url, timeout=20)
    resp.raise_for_status()
    soup = BeautifulSoup(resp.text, "lxml")
    links = []
    for tag in soup.find_all("a", href=True):
        href: str = tag["href"]
        if any(href.lower().endswith(ext) for ext in spatial_exts):
            links.append(href if href.startswith("http") else page_url.rstrip("/") + "/" + href.lstrip("/"))
    return links

import time
from dataclasses import dataclass, field

@dataclass
class TokenCache:
    access_token: str = ""
    expires_at: float = 0.0

    def is_valid(self, buffer_seconds: int = 120) -> bool:
        return bool(self.access_token) and time.time() < (self.expires_at - buffer_seconds)

def build_authenticated_session(
    base_url: str,
    api_key: str | None = None,
    oauth_token_url: str | None = None,
    client_id: str | None = None,
    client_secret: str | None = None,
) -> tuple[requests.Session, TokenCache]:
    session = requests.Session()
    session.headers.update({
        "User-Agent": "GeospatialETL/1.0 (contact: pipeline@example.org)",
        "Accept": "application/json, application/geo+json, */*",
    })
    token_cache = TokenCache()

    if api_key:
        # CKAN-style or generic key-in-header auth
        session.headers["X-CKAN-API-Key"] = api_key

    if oauth_token_url and client_id and client_secret:
        resp = session.post(oauth_token_url, data={
            "grant_type": "client_credentials",
            "client_id": client_id,
            "client_secret": client_secret,
        }, timeout=15)
        resp.raise_for_status()
        payload = resp.json()
        token_cache.access_token = payload["access_token"]
        token_cache.expires_at = time.time() + payload.get("expires_in", 3600)
        session.headers["Authorization"] = f"Bearer {token_cache.access_token}"

    return session, token_cache

import logging
from typing import Iterator

logger = logging.getLogger(__name__)

def iter_ckan_resources(
    session: requests.Session,
    base_url: str,
    dataset_id: str,
    rows_per_page: int = 1000,
) -> Iterator[dict]:
    """
    Yield individual CKAN resource metadata dicts, page by page.
    """
    start = 0
    while True:
        params = {
            "id": dataset_id,
            "rows": rows_per_page,
            "start": start,
        }
        resp = session.get(f"{base_url}/action/package_show", params=params, timeout=30)
        resp.raise_for_status()
        body = resp.json()
        resources: list[dict] = body.get("result", {}).get("resources", [])
        if not resources:
            logger.info("CKAN pagination complete at offset %d", start)
            break
        for resource in resources:
            yield resource
        start += len(resources)

def iter_arcgis_features(
    session: requests.Session,
    layer_url: str,
    where_clause: str = "1=1",
    out_fields: str = "*",
    result_offset: int = 0,
    page_size: int = 1000,
) -> Iterator[dict]:
    """
    Yield ArcGIS REST feature dicts, handling exceededTransferLimit.
    """
    while True:
        params = {
            "where": where_clause,
            "outFields": out_fields,
            "f": "geojson",
            "resultOffset": result_offset,
            "resultRecordCount": page_size,
        }
        resp = session.get(f"{layer_url}/query", params=params, timeout=60)
        resp.raise_for_status()
        body = resp.json()
        features: list[dict] = body.get("features", [])
        if not features:
            break
        for feat in features:
            yield feat
        if not body.get("exceededTransferLimit", False):
            break
        result_offset += len(features)
        logger.debug("ArcGIS page complete, next offset=%d", result_offset)

import hashlib
from pathlib import Path
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests.exceptions as req_exc

@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    retry=retry_if_exception_type((req_exc.ConnectionError, req_exc.Timeout, req_exc.ChunkedEncodingError)),
    reraise=True,
)
def download_with_resume(
    url: str,
    dest: Path,
    session: requests.Session,
    chunk_size: int = 65536,
) -> str:
    """
    Stream-download url to dest, resuming from existing partial file.
    Returns the hex SHA-256 digest of the completed file.
    """
    headers: dict[str, str] = {}
    dest.parent.mkdir(parents=True, exist_ok=True)

    existing_size = dest.stat().st_size if dest.exists() else 0
    if existing_size > 0:
        # Probe server support for range requests
        head = session.head(url, timeout=10)
        if head.headers.get("Accept-Ranges") == "bytes":
            headers["Range"] = f"bytes={existing_size}-"
            logger.info("Resuming %s from byte %d", url, existing_size)

    resp = session.get(url, headers=headers, stream=True, timeout=60)
    if resp.status_code == 416:
        # File already complete (server says range not satisfiable)
        logger.info("File already complete: %s", dest)
    else:
        resp.raise_for_status()
        write_mode = "ab" if "Range" in headers else "wb"
        with dest.open(write_mode) as fh:
            for chunk in resp.iter_content(chunk_size=chunk_size):
                if chunk:
                    fh.write(chunk)

    # Compute SHA-256 over the full file for integrity tracking
    sha256 = hashlib.sha256()
    with dest.open("rb") as fh:
        for chunk in iter(lambda: fh.read(chunk_size), b""):
            sha256.update(chunk)
    digest = sha256.hexdigest()
    logger.info("Download complete: %s  sha256=%s", dest, digest)
    return digest

import json
import shapely
import geopandas as gpd
from pathlib import Path

def validate_spatial_file(
    path: Path,
    expected_crs: str | None = "EPSG:4326",
    expected_geom_types: set[str] | None = None,
) -> dict:
    """
    Open the file with geopandas, check CRS and geometry types.
    Returns a validation summary dict; raises ValueError on critical failures.
    """
    gdf = gpd.read_file(path)
    summary: dict = {
        "path": str(path),
        "rows": len(gdf),
        "columns": list(gdf.columns),
        "crs": str(gdf.crs),
        "geom_types": list(gdf.geometry.geom_type.unique()),
        "invalid_geom_count": int((~shapely.is_valid(gdf.geometry.values)).sum()),
    }

    if expected_crs and str(gdf.crs) != expected_crs:
        raise ValueError(f"CRS mismatch: expected {expected_crs}, got {gdf.crs}")

    if expected_geom_types:
        actual = set(summary["geom_types"])
        if not actual.issubset(expected_geom_types):
            raise ValueError(f"Unexpected geometry types: {actual - expected_geom_types}")

    if summary["invalid_geom_count"] > 0:
        import warnings
        warnings.warn(f"{summary['invalid_geom_count']} invalid geometries in {path}")

    return summary

def write_metadata_sidecar(path: Path, metadata: dict) -> None:
    """
    Write a JSON sidecar file adjacent to the spatial asset for lineage tracking.
    """
    sidecar = path.with_suffix(".meta.json")
    with sidecar.open("w") as fh:
        json.dump(metadata, fh, indent=2, default=str)
    logger.info("Metadata sidecar written: %s", sidecar)

import sqlite3
from datetime import datetime, timezone
from pathlib import Path

def init_registry(db_path: Path) -> sqlite3.Connection:
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS downloads (
            url TEXT PRIMARY KEY,
            local_path TEXT,
            sha256 TEXT,
            last_modified TEXT,
            downloaded_at TEXT,
            validation_status TEXT
        )
    """)
    conn.commit()
    return conn

def is_already_current(conn: sqlite3.Connection, url: str, server_last_modified: str | None) -> bool:
    row = conn.execute(
        "SELECT last_modified FROM downloads WHERE url = ?", (url,)
    ).fetchone()
    if row is None:
        return False
    if server_last_modified is None:
        return True  # No ETag/Last-Modified to compare; skip re-download
    return row[0] == server_last_modified

def record_download(
    conn: sqlite3.Connection,
    url: str,
    local_path: Path,
    sha256: str,
    last_modified: str | None,
    validation_status: str,
) -> None:
    conn.execute("""
        INSERT OR REPLACE INTO downloads
            (url, local_path, sha256, last_modified, downloaded_at, validation_status)
        VALUES (?, ?, ?, ?, ?, ?)
    """, (
        url, str(local_path), sha256, last_modified,
        datetime.now(timezone.utc).isoformat(), validation_status,
    ))
    conn.commit()

def make_request_with_token_refresh(
    session: requests.Session,
    token_cache: TokenCache,
    url: str,
    token_url: str,
    client_id: str,
    client_secret: str,
    **kwargs,
) -> requests.Response:
    if not token_cache.is_valid():
        resp = session.post(token_url, data={
            "grant_type": "client_credentials",
            "client_id": client_id,
            "client_secret": client_secret,
        }, timeout=15)
        resp.raise_for_status()
        payload = resp.json()
        token_cache.access_token = payload["access_token"]
        token_cache.expires_at = time.time() + payload.get("expires_in", 3600)
        session.headers["Authorization"] = f"Bearer {token_cache.access_token}"

    response = session.get(url, **kwargs)
    if response.status_code == 401:
        # Force refresh and retry once
        token_cache.expires_at = 0.0
        return make_request_with_token_refresh(
            session, token_cache, url, token_url, client_id, client_secret, **kwargs
        )
    response.raise_for_status()
    return response

import json
from pathlib import Path
import geopandas as gpd

def check_schema_drift(
    gdf: gpd.GeoDataFrame,
    contract_path: Path,
) -> dict:
    """
    Compare gdf columns against a JSON schema contract.
    Returns a dict with 'added', 'removed', and 'type_changed' keys.
    """
    with contract_path.open() as fh:
        contract: dict = json.load(fh)

    baseline_cols: set[str] = set(contract.get("columns", []))
    actual_cols: set[str] = set(gdf.columns) - {"geometry"}

    drift = {
        "added":   sorted(actual_cols - baseline_cols),
        "removed": sorted(baseline_cols - actual_cols),
    }

    if drift["added"] or drift["removed"]:
        logger.warning("Schema drift detected: %s", drift)
        # Quarantine path for manual review
        return drift

    return drift  # Empty drift = schema matches contract

from lxml import etree

def iter_wfs_features(
    session: requests.Session,
    wfs_url: str,
    type_name: str,
    page_size: int = 500,
) -> Iterator[bytes]:
    """
    Yield raw WFS feature XML chunks, handling startIndex pagination.
    """
    start_index = 0
    while True:
        params = {
            "service": "WFS",
            "version": "2.0.0",
            "request": "GetFeature",
            "typeName": type_name,
            "outputFormat": "application/json",
            "count": page_size,
            "startIndex": start_index,
        }
        resp = session.get(wfs_url, params=params, timeout=60)
        resp.raise_for_status()
        body = resp.json()
        features = body.get("features", [])
        if not features:
            break
        yield from features
        start_index += len(features)
        # WFS 2.0 signals end via numberReturned < count
        if body.get("numberReturned", page_size) < page_size:
            break

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests.exceptions as req_exc

@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    retry=retry_if_exception_type((req_exc.ConnectionError, req_exc.Timeout, req_exc.HTTPError)),
    reraise=True,
)
def fetch_with_backoff(session: requests.Session, url: str, params: dict | None = None) -> dict:
    response = session.get(url, params=params, timeout=30)
    if response.status_code == 429:
        retry_after = int(response.headers.get("Retry-After", 30))
        logger.info("Rate-limited. Waiting %d seconds.", retry_after)
        time.sleep(retry_after)
        response = session.get(url, params=params, timeout=30)
    response.raise_for_status()
    return response.json()

from pathlib import Path
import sqlite3

def run_portal_ingestion(
    portal_url: str,
    dataset_id: str,
    output_dir: Path,
    db_path: Path,
    api_key: str | None = None,
) -> list[dict]:
    """
    Full pipeline: discover → authenticate → paginate → download → validate → store.
    Returns a list of validation summary dicts for each downloaded resource.
    """
    conn = init_registry(db_path)
    session, token_cache = build_authenticated_session(portal_url, api_key=api_key)
    results: list[dict] = []

    for resource in iter_ckan_resources(session, portal_url, dataset_id):
        url: str = resource.get("url", "")
        if not url:
            continue

        # Check Last-Modified to skip unchanged files (idempotency)
        head = session.head(url, timeout=10)
        last_mod = head.headers.get("Last-Modified")
        if is_already_current(conn, url, last_mod):
            logger.info("Skipping unchanged resource: %s", url)
            continue

        dest = output_dir / Path(url).name
        try:
            sha256 = download_with_resume(url, dest, session)
            summary = validate_spatial_file(dest)
            write_metadata_sidecar(dest, {**summary, "source_url": url, "sha256": sha256})
            record_download(conn, url, dest, sha256, last_mod, "valid")
            results.append(summary)
        except Exception as exc:
            logger.error("Failed to ingest %s: %s", url, exc)
            record_download(conn, url, dest, "", last_mod, f"error: {exc}")

    return results