CMM Harvester

A production-grade data harvesting pipeline for Critical Minerals & Materials (CMM) research. The system collects, normalizes, and enriches data from 37 nontraditional sources spanning trade flows, legislative/regulatory actions, geopolitical events, satellite imagery, mining operations, sanctions, patents, and more. It implements a Bronze/Silver/Gold lakehouse architecture with optional Dagster orchestration.

Author: Nancy Washton Version: 0.1.0 License: MIT Python: >= 3.12

Overview

The CMM Harvester addresses a core challenge in critical minerals research: relevant data is scattered across dozens of government APIs, international databases, satellite platforms, and open-data portals. This pipeline consolidates those sources into a unified, queryable data lake.

What It Does

Harvests raw data from 37 APIs (trade, policy, events, mining, sanctions, energy, patents, satellite)
Stores immutable raw responses in Parquet format (Bronze layer)
Cleans and normalizes records with ISO standards (Silver layer)
Enriches with CMM-specific metadata: HS code mapping, supply chain stage classification, relevance scoring (Gold layer)
Orchestrates scheduled harvests via Dagster (148 software-defined assets)

Why It Matters

Critical minerals like rare earth elements, cobalt, lithium, and graphite underpin modern technology (EV batteries, wind turbines, semiconductors, defense systems). Understanding supply chain risks requires cross-referencing trade data with mining locations, regulatory actions, geopolitical events, and environmental compliance — exactly what this pipeline enables.

Data Sources

37 Implemented Harvesters Across 11 Domains

Trade & Economics (6 sources)

Source	API	Schedule	Description
`census_trade`	US Census Bureau International Trade	Daily	US imports/exports by 10-digit HS code — cobalt, lithium, REE, nickel, etc.
`eurostat_comext`	EU Eurostat COMEXT	Weekly	EU trade flows for critical mineral commodities
`wits_trade`	World Bank WITS	Weekly	Global bilateral trade for mineral ores (niobium, tantalum, vanadium)
`imf_dots`	IMF Direction of Trade Statistics	Weekly	Aggregate bilateral trade values between countries
`ilostat`	ILO Statistics	Weekly	Employment in mining sectors by country
`world_bank`	World Bank Development Indicators	Weekly	GDP, governance, mining revenue indicators

Policy & Legislation (3 sources)

Source	API	Schedule	Description
`congress_gov`	Congress.gov	Daily	Bills and hearings mentioning critical minerals, rare earths, mining
`govinfo_gpo`	GovInfo GPO	Daily	Federal Register documents, committee reports, hearing transcripts
`regulations_gov`	Regulations.gov	Daily	Federal rulemaking: proposed rules, final rules, public notices, comments

Sanctions & Screening (3 sources)

Source	API	Schedule	Description
`bis_screening`	BIS Consolidated Screening List	Daily	Export-controlled entities relevant to mineral trade
`ofac_sdn`	OFAC SDN List	Daily	Specially Designated Nationals — sanctions on mineral actors
`sam_gov`	SAM.gov Entity API	Daily	Federal contractor registrations for mining/mineral entities

Government & Spending (2 sources)

Source	API	Schedule	Description
`usaspending`	USASpending.gov	Daily	Federal contracts and grants for critical minerals programs
`eiti`	EITI Summary Data	Weekly	Extractive Industries Transparency Initiative disclosures

Mining & Minerals (5 sources)

Source	API	Schedule	Description
`usgs_mrds`	USGS Mineral Resources Data System	Daily	149K+ global mine/deposit locations with commodity data
`ipis_mining`	IPIS Mining Sites	Weekly	11K+ artisanal mining sites in Congo, Great Lakes region
`osm_mining`	OpenStreetMap Overpass	Weekly	Mining-tagged features from OpenStreetMap globally
`blm_mining`	BLM Mining Claims	Daily	US Bureau of Land Management mining claim records
`rmi_smelters`	Responsible Minerals Initiative	Weekly	Audited smelter/refiner list for conflict minerals (3TG)

Energy & Infrastructure (3 sources)

Source	API	Schedule	Description
`eia_electricity`	EIA Open Data	Daily	US electricity generation, fuel mix, capacity by plant/state
`stb_waybill`	Surface Transportation Board	Monthly	Rail freight waybill data for mineral logistics
`usgs_water_quality`	USGS Water Quality Portal	Daily	Water quality near mining operations (933K+ samples)

Satellite & Remote Sensing (2 sources)

Source	API	Schedule	Description
`copernicus_sentinel`	Copernicus Data Space	Monthly	Sentinel-2 imagery for mining site monitoring (NDVI, change detection)
`viirs_nightlights`	NOAA VIIRS	Monthly	Nighttime light intensity for facility activity detection

Geopolitical Events (2 sources)

Source	API	Schedule	Description
`acled`	Armed Conflict Location & Event Data	Weekly	Conflict events in CMM-producing countries (DRC, Zambia, Peru, etc.)
`gdelt`	GDELT Project	Every 30 min	Global events related to mining, critical minerals, supply chains

Environmental Compliance (1 source)

Source	API	Schedule	Description
`epa_echo`	EPA ECHO	Daily	Mining facility inspections, violations, enforcement — RCRA/CWA/CAA

Corporate & Financial (3 sources)

Source	API	Schedule	Description
`sec_edgar`	SEC EDGAR	Daily	10-K, 10-Q, 8-K filings + XBRL facts for mining companies
`sec_formsd`	SEC EFTS	Daily	Form D private placement filings (mining/mineral ventures)
`courtlistener`	CourtListener RECAP	Daily	Federal court opinions involving mining and minerals

Scientific & Technology (3 sources)

Source	API	Schedule	Description
`arxiv`	arXiv	Daily	Scientific preprints on mineral processing, extraction, recycling
`osti`	DOE OSTI	Daily	Department of Energy technical reports on critical minerals
`uspto_patentsview`	USPTO PatentsView	Daily	Patent filings related to mineral extraction and processing (IPC codes)

Additional Data (2 sources)

Source	API	Schedule	Description
`bgs_minerals`	British Geological Survey	Weekly	Global mineral production statistics by country and commodity
`global_trade_alert`	Global Trade Alert	Weekly	Trade policy interventions affecting mineral commodities

Architecture

                   +-----------+     +-------------+     +---------------+     +------------------+
  37 APIs -------> |  Bronze   | --> |   Silver    | --> |     Gold      | --> | Corpus / LanceDB |
  (raw fetch)      | (raw API  |     | (cleaned,   |     | (HS-mapped,   |     | (query-ready,    |
                   |  response)|     |  normalized,|     |  supply chain |     |  vector-indexed) |
                   |           |     |  deduped)   |     |  classified,  |     |                  |
                   |  Parquet  |     |  Parquet    |     |  scored)      |     |  Parquet + JSON   |
                   +-----------+     +-------------+     +---------------+     +------------------+
                        |
                   Content-hashed
                   with provenance

Layers

Bronze (raw): Immutable raw API responses stored as Parquet files. Each record includes full provenance metadata (source URL, fetch timestamp, SHA-256 content hash, API response code). Formats: JSON, XML, CSV, GeoTIFF.
Silver (clean/normalized): Cleaned and deduplicated records. Countries normalized to ISO 3166 alpha-3 codes, dates to ISO 8601, text fields trimmed. Content-hash-based deduplication prevents duplicate records across harvests.
Gold (CMM-enriched): Records enriched with critical minerals metadata. HS codes mapped to mineral names (from configs/minerals.yaml), supply chain stage classified (upstream/midstream/downstream/logistics/policy), and relevance score computed via keyword matching.
Corpus/LanceDB: Query-ready output for downstream MPII_CMM pipelines. (Phase 5 — not yet implemented.)

File Layout

data/
├── bronze/{source_id}/{YYYY-MM-DD}/{source_id}_{timestamp}_{uuid}.parquet
├── silver/{source_id}/{YYYY-MM-DD}/...
├── gold/{source_id}/{YYYY-MM-DD}/...
└── dead_letter/{source_id}/{source_id}_{timestamp}_{uuid}_error.json

Parquet Schema (Bronze)

Every Bronze parquet file contains these columns:

Column	Type	Description
`id`	VARCHAR	UUID for each record
`layer`	VARCHAR	Always `"bronze"`
`source_id`	VARCHAR	e.g., `"census_trade"`, `"usgs_mrds"`
`provenance`	STRUCT	Source URL, fetch timestamp, SHA-256 hash, response code, content type, record count, license
`raw_data`	STRUCT	Full API response (varies per source — nested fields accessible via DuckDB)
`cleaned_data`	NULL	Populated in Silver layer
`cmm_data`	NULL	Populated in Gold layer
`created_at`	TIMESTAMP WITH TIME ZONE	When the record was harvested

Setup

Prerequisites

Python 3.12+
uv (Python package manager)
API keys for authenticated sources (see .env.example)

Installation

# Install uv if needed
curl -LsSf https://astral.sh/uv/install.sh | sh

# Clone and install
git clone <repository-url>
cd cmm-harvester

# Install project and dev dependencies (all extras)
make dev

# Copy and fill in API keys
cp .env.example .env
# Edit .env with your keys

# Verify installation
make check

API Keys

The following API keys are needed for authenticated sources. Most sources work without keys (public APIs). Copy .env.example to .env and fill in values:

Key	Source	Sign-up URL
`CONGRESS_API_KEY`	Congress.gov	https://api.congress.gov/sign-up/
`REGULATIONS_GOV_API_KEY`	Regulations.gov	https://open.gsa.gov/api/regulationsgov/
`ACLED_API_KEY` + `ACLED_EMAIL`	ACLED	https://developer.acleddata.com/
`EIA_API_KEY`	EIA	https://www.eia.gov/opendata/register.php
`COPERNICUS_CLIENT_ID` + `SECRET`	Copernicus	https://dataspace.copernicus.eu/
`SAM_GOV_CLIENT_ID` + `SECRET`	SAM.gov	https://open.gsa.gov/api/entity-api/
`FIRMS_MAP_KEY`	NASA FIRMS	https://firms.modaps.eosdis.nasa.gov/api/area/
`MINDAT_API_KEY`	Mindat.org	https://www.mindat.org/a/api
`COURTLISTENER_API_KEY`	CourtListener	https://www.courtlistener.com/sign-in/
`GCP_PROJECT_ID`	GDELT BigQuery	Google Cloud Console

Usage

CLI Commands

# Harvest a single source
cmm-harvest harvest --source census_trade

# Dry run (fetch but don't persist to parquet)
cmm-harvest harvest --source census_trade --dry-run

# Check harvest status (files per layer per source)
cmm-harvest status

# Promote Gold data to Corpus/ (not yet implemented)
cmm-harvest promote --source census_trade

# Launch Dagster UI for orchestrated harvesting
make dagster

Batch Harvesting

Run all sources in parallel:

# Fast sources (< 1 min each)
for src in census_trade eurostat_comext bis_screening ofac_sdn gdelt stb_waybill \
           viirs_nightlights sec_edgar rmi_smelters osti courtlistener sec_formsd \
           arxiv osm_mining eiti bgs_minerals ilostat; do
    cmm-harvest harvest --source $src &
done
wait

# Medium sources (1-5 min each)
for src in congress_gov govinfo_gpo regulations_gov usaspending \
           uspto_patentsview imf_dots ipis_mining usgs_mrds epa_echo \
           blm_mining world_bank usgs_water_quality; do
    cmm-harvest harvest --source $src &
done
wait

# Slow sources (5+ min each)
for src in acled eia_electricity wits_trade sam_gov; do
    cmm-harvest harvest --source $src &
done
wait

Data Exploration with DuckDB

All Bronze-layer parquet files can be queried directly with DuckDB — no loading or ETL required:

# Install DuckDB
brew install duckdb   # macOS
# or: pip install duckdb

# Start interactive session
duckdb

Example Queries

-- Record counts per source
SELECT source_id, COUNT(*) AS records
FROM 'data/bronze/**/*.parquet'
GROUP BY source_id
ORDER BY records DESC;

-- Explore a source schema (e.g., eia_electricity)
DESCRIBE SELECT raw_data.* FROM 'data/bronze/eia_electricity/**/*.parquet' LIMIT 1;

-- Sample rows from census_trade
SELECT
    raw_data.I_COMMODITY AS hs_code,
    raw_data.I_COMMODITY_LDESC AS commodity,
    raw_data.CTY_NAME AS country,
    raw_data.GEN_VAL_MO AS import_value_usd,
    raw_data._direction AS direction
FROM 'data/bronze/census_trade/**/*.parquet'
LIMIT 10;

-- Cross-reference USGS mines with EPA compliance
SELECT
    m.raw_data.site_name AS mine_name,
    m.raw_data.commod1 AS primary_commodity,
    m.raw_data.latitude AS lat,
    m.raw_data.longitude AS lon,
    e.raw_data.FacName AS epa_facility,
    e.raw_data.FacState AS state
FROM 'data/bronze/usgs_mrds/**/*.parquet' m
JOIN 'data/bronze/epa_echo/**/*.parquet' e
    ON ABS(CAST(m.raw_data.latitude AS DOUBLE) - CAST(e.raw_data.FacLat AS DOUBLE)) < 0.1
   AND ABS(CAST(m.raw_data.longitude AS DOUBLE) - CAST(e.raw_data.FacLong AS DOUBLE)) < 0.1
LIMIT 20;

-- REE trade flows (US imports from Census data)
SELECT
    raw_data.CTY_NAME AS country,
    raw_data.I_COMMODITY_LDESC AS commodity,
    TRY_CAST(raw_data.GEN_VAL_MO AS BIGINT) AS import_value_usd,
    raw_data.time AS period
FROM 'data/bronze/census_trade/**/*.parquet'
WHERE raw_data.I_COMMODITY LIKE '2846%'   -- REE compounds
   OR raw_data.I_COMMODITY LIKE '2805%'   -- REE metals
ORDER BY import_value_usd DESC
LIMIT 20;

-- Critical minerals legislation (GovInfo GPO)
SELECT
    raw_data.title AS title,
    raw_data.dateIssued AS date_issued,
    raw_data.packageId AS package_id
FROM 'data/bronze/govinfo_gpo/**/*.parquet'
WHERE lower(raw_data.title) LIKE '%critical mineral%'
   OR lower(raw_data.title) LIKE '%rare earth%'
ORDER BY raw_data.dateIssued DESC
LIMIT 20;

-- Regulatory actions on critical minerals (Regulations.gov)
SELECT
    raw_data.attributes.title AS title,
    raw_data.attributes.agencyId AS agency,
    raw_data.attributes.documentType AS doc_type,
    raw_data.attributes.docketId AS docket_id,
    raw_data.attributes.postedDate AS posted
FROM 'data/bronze/regulations_gov/**/*.parquet'
WHERE raw_data.attributes.documentType IN ('Rule', 'Proposed Rule', 'Notice')
  AND (lower(raw_data.attributes.title) LIKE '%critical mineral%'
    OR lower(raw_data.attributes.title) LIKE '%rare earth%')
ORDER BY raw_data.attributes.postedDate DESC;

Harvested Data Inventory

As of 2026-03-04, the Bronze layer contains 2,247,880 total records across 30 sources in 48 parquet files (150 MB).

Source	Records	Description
`usgs_water_quality`	933,123	Water quality samples near mining operations
`eurostat_comext`	651,223	EU trade flows for mineral commodities
`eia_electricity`	183,079	US electricity generation by plant/state
`usgs_mrds`	149,336	Global mine and deposit locations
`bgs_minerals`	110,024	Global mineral production statistics
`epa_echo`	81,370	EPA facility inspections and violations
`world_bank`	33,916	Development indicators
`blm_mining`	20,000	BLM mining claims
`ipis_mining`	11,828	Artisanal mining sites (Congo/Great Lakes)
`osm_mining`	11,392	OpenStreetMap mining features
`osti`	8,160	DOE technical reports
`wits_trade`	7,054	Global bilateral mineral trade
`govinfo_gpo`	6,914	Federal documents (bills, hearings, reports)
`sec_formsd`	6,600	SEC Form D private placements
`congress_gov`	6,250	Congressional bills and hearings
`census_trade`	5,227	US import/export by HS code
`usaspending`	4,164	Federal contracts and grants
`ilostat`	4,092	Mining employment by country
`ofac_sdn`	4,047	OFAC sanctioned entities
`regulations_gov`	3,001	Federal rulemaking documents
`acled`	2,294	Conflict events in mining regions
`gdelt`	1,422	Global mining/mineral news events
`eiti`	1,324	Extractive industry transparency data
`courtlistener`	960	Federal court opinions (mining cases)
`sec_edgar`	809	SEC filings (10-K, 10-Q, 8-K, 20-F)
`arxiv`	118	Scientific preprints on mineral processing
`imf_dots`	80	IMF bilateral trade statistics
`stb_waybill`	47	Rail freight waybill data
`rmi_smelters`	24	Audited smelter/refiner facilities
`viirs_nightlights`	2	Nighttime light composites

Sources with 0 records (API key or availability issues): sam_gov, bis_screening, uspto_patentsview, copernicus_sentinel, global_trade_alert, firms_thermal, mindat.

Key Findings from Initial Data Exploration

The following findings emerged from cross-referencing the harvested datasets using DuckDB.

REE Mining and Processing Sites

United States:

USGS MRDS identifies 2,408 US sites associated with rare earth elements (16 processing plants, 216 active producers)
EPA ECHO shows Molycorp/Mountain Pass (CA) as the primary REE processing facility with extensive compliance history
Key operations: Freeport-McMoRan (AZ copper/REE), Greens Creek Mine (AK), Red Dog Mine (AK), Newmont Carlin (NV)

International:

Australia: 126 REE-associated sites (Mount Weld, Lynas)
China/Mongolia: 50 known sites in USGS MRDS (likely significant undercounting)
South America: 106 sites (Brazil, Argentina)
Canada: 64 sites
Congo/DRC: 2,660 artisanal mining sites (IPIS data) — coltan, wolframite, cobalt

REE Trade Flows

US Imports (Census Trade Data):

China dominates at ~62% of US REE imports by value
Other significant suppliers: France (La Rochelle refinery), Japan (recycled REE), Estonia, Malaysia
HS 2846 (REE compounds) and HS 2805 (REE metals) tracked monthly

EU Trade (Eurostat COMEXT):

France is the largest EU supplier (domestic refining)
China is the second-largest EU source
RE compound trade concentrated in a few major facilities

Legislative and Regulatory Activity

Key Recent Legislation (GovInfo GPO):

Critical Mineral Consistency Act of 2025 (CRPT-119hrpt519, Feb 2026)
Critical Mineral Dominance Act (Nov 2025)
Securing America's Critical Minerals Supply Act (Sep 2025)
H.R. 1 Energy Bill with critical minerals processing provisions (2023)

Key Congressional Hearings:

"Digging Deeper: Building Our Critical Minerals Workforce" (Jun 2024)
"Opportunities to Counter PRC's Control of Critical Mineral Supply Chains" (Sep 2023)

Key Regulatory Actions (Regulations.gov):

USTR: Plurilateral Agreement on Trade in Critical Minerals (Feb 2026, open for comment)
BIS: Section 232 Investigation — Critical Minerals as national security issue (Apr 2025)
USGS: Final 2025 List of Critical Minerals (Nov 2025) — official designation triggering downstream policy
IRS: 45X Advanced Manufacturing Production Credit for critical mineral processing (Oct 2024)
Forest Service: Locatable Minerals proposed rule on mining on federal lands (Feb 2026, open for comment)
FAR: Federal procurement — domestically nonavailable critical minerals (Oct 2024)
DHS: Uyghur Forced Labor Prevention Act Entity List — includes REE supply chain (Nov 2024)
BLM: Batch rescission of mining claim regulations (Jul–Sep 2025)

Cross-Reference: Mining States

State-level analysis joining USGS mines, EPA facilities, and EIA electricity data for mining-intensive states:

State	USGS Mine Sites	EPA Mining Facilities	EIA Generation (GWh)
Arizona	3,200+	45+	75,000+
Nevada	2,800+	30+	40,000+
Alaska	400+	15+	6,500+
Montana	1,200+	20+	15,000+
Wyoming	800+	25+	45,000+
Utah	900+	20+	38,000+
Colorado	1,500+	35+	55,000+

Data Gaps Identified

Chinese mining data: Only 50 REE sites in USGS MRDS for China — likely significant undercounting given China produces ~60% of global REE
International environmental compliance: No equivalent to EPA ECHO for non-US facilities
International energy data: EIA covers US only; no harvester for IEA or equivalent international energy data
WITS trade data: Does not include HS 2846/2805 (REE compounds/metals) — only mineral ores
SAM.gov, BIS screening, USPTO: Returned 0 records due to API key/availability issues

Configuration

Source Configuration (YAML)

Each source is configured in configs/sources/{source_id}.yaml:

# configs/sources/census_trade.yaml
source_id: census_trade
name: US Census Bureau International Trade
base_url: https://api.census.gov
auth_type: none
rate_limit_rps: 5.0
retry_max_attempts: 3
retry_backoff_factor: 2.0
schedule: "0 2 * * *"
data_format: json
description: US import/export data by 10-digit HS commodity code
enabled: true
cmm_hs_codes:
  - "2605000000"   # Cobalt ores
  - "2604000000"   # Nickel ores
  - "2603000000"   # Copper ores
  - "2846100000"   # Cerium compounds (REE)
  - "2846900000"   # Other rare-earth compounds
  - "2805300000"   # Rare-earth metals
  # ... more codes

Environment Settings

Settings are loaded from environment variables and .env file via Pydantic Settings:

class HarvesterSettings(BaseSettings):
    data_dir: Path = Path("data")
    configs_dir: Path = CONFIGS_DIR
    log_level: str = "INFO"
    log_format: str = "json"       # "json" or "console"
    dry_run: bool = False

CMM Commodity Reference

The pipeline tracks 11 critical mineral groups defined in configs/minerals.yaml, each with HS codes, STCC codes (rail freight), IPC codes (patents), NAICS codes (industry classification), and keywords:

Mineral	Symbol	Key HS Codes	Supply Chain Stages
Cobalt	Co	2605 (ores), 8105 (mattes), 2822 (oxides)	Upstream, Midstream
Lithium	Li	2836.91 (carbonate), 2825.20 (hydroxide)	Upstream, Midstream, Downstream
Nickel	Ni	2604 (ores), 7501 (mattes), 7502 (unwrought)	Upstream, Midstream
Rare Earth Elements	REE	2846 (compounds), 2805.30 (metals)	Upstream, Midstream, Downstream
Manganese	Mn	2602 (ores), 7202.11 (ferromanganese)	Upstream, Midstream
Graphite	C	2504 (natural), 3801 (artificial)	Upstream, Midstream, Downstream
Copper	Cu	2603 (ores), 7403.11 (cathodes)	Upstream, Midstream, Downstream
Tungsten	W	2611 (ores), 8101 (powders)	Upstream, Midstream
Tin	Sn	2609 (ores), 8001 (unwrought)	Upstream, Midstream
Tantalum	Ta	2615.90 (ores), 8103 (unwrought)	Upstream, Midstream, Downstream
Platinum Group Metals	PGM	7110 (Pt, Pd, Rh unwrought)	Upstream, Midstream, Downstream

Each commodity also carries:

STCC codes: For matching rail waybill freight data
IPC codes: For matching USPTO patent classifications
NAICS codes: For matching EPA facility industry codes
Keywords: For text-based matching in legislation, news, filings

Development

Prerequisites

make dev   # Installs all extras + dev dependencies via uv

Commands

make fmt        # Format code (ruff format + ruff check --fix)
make lint       # Check formatting and linting (ruff)
make typecheck  # Run mypy in strict mode
make test       # Run pytest (204 tests)
make test-cov   # Run tests with coverage report (>=80% required)
make check      # All of the above (lint + typecheck + test)
make clean      # Remove caches (.mypy_cache, .pytest_cache, .ruff_cache, __pycache__)

Tooling

Tool	Purpose	Config Location
`uv`	Package management	`pyproject.toml`, `uv.lock`
`ruff`	Linting + formatting	`pyproject.toml [tool.ruff]` — line length 100, Python 3.12
`mypy`	Type checking (strict)	`pyproject.toml [tool.mypy]` — Pydantic plugin enabled
`pytest`	Testing	`pyproject.toml [tool.pytest]` — asyncio_mode="auto"
`respx`	HTTP mocking	Used in tests to mock httpx responses
`pre-commit`	Git hooks	`.pre-commit-config.yaml`

Testing

# Run all tests
uv run pytest -x --tb=short

# Run a specific test file
uv run pytest tests/harvesters/test_census_trade.py -v

# Run with coverage
uv run pytest --cov --cov-report=term-missing --cov-report=html

# Skip integration tests (for CI)
uv run pytest -m "not integration"

Tests use respx to mock HTTP responses, tmp_path for temporary file I/O, and pytest-asyncio for async test support. Each harvester has a corresponding test file that validates:

Correct parsing of API responses
Record structure and field extraction
Pagination behavior (mocks return < page_size to exit loops after 1 iteration)
Error handling for API failures

CI Pipeline

GitHub Actions (.github/workflows/ci.yml) runs on every push to master and PR:

Format check (ruff format --check)
Lint (ruff check)
Type check (mypy src)
Test (pytest -x --tb=short -m "not integration" with coverage)

Project Structure

cmm-harvester/
├── pyproject.toml                  # Project metadata, dependencies, tool config
├── Makefile                        # Development automation
├── Dockerfile                      # Multi-stage build (harvester + dagster targets)
├── .env.example                    # API key template
├── .pre-commit-config.yaml         # Pre-commit hooks
├── .github/workflows/ci.yml       # GitHub Actions CI
│
├── configs/
│   ├── sources/                    # 37 YAML source configurations
│   │   ├── census_trade.yaml
│   │   ├── congress_gov.yaml
│   │   └── ...
│   ├── minerals.yaml               # CMM commodity reference (11 minerals, HS/STCC/IPC codes)
│   ├── mine_coordinates.yaml       # Facility GPS locations for satellite monitoring
│   ├── schedules.yaml              # Cron schedules per source
│   └── sec_edgar_companies.yaml    # Mining companies for SEC filtering
│
├── src/cmm_harvester/
│   ├── __init__.py                 # Version: 0.1.0
│   ├── py.typed                    # PEP 561 marker
│   │
│   ├── core/                       # Core infrastructure
│   │   ├── config.py               # YAML + env config loading
│   │   ├── exceptions.py           # Exception hierarchy (HarvesterError -> ...)
│   │   ├── hashing.py              # SHA-256 content hashing
│   │   ├── http_client.py          # Async HTTP client (httpx + tenacity retry + rate limit)
│   │   ├── logging_setup.py        # Structlog JSON/console configuration
│   │   ├── models.py               # Pydantic models (HarvestRecord, SourceConfig, etc.)
│   │   └── storage.py              # StorageManager (Parquet + JSON + dead letter)
│   │
│   ├── harvesters/                 # 37 harvester implementations
│   │   ├── base.py                 # BaseHarvester abstract class
│   │   ├── acled.py                # Armed Conflict Location & Event Data
│   │   ├── arxiv.py                # arXiv scientific preprints
│   │   ├── census_trade.py         # US Census Trade API
│   │   ├── congress_gov.py         # Congress.gov bills & hearings
│   │   └── ... (33 more)
│   │
│   ├── processing/                 # Bronze -> Silver -> Gold pipeline
│   │   ├── silver.py               # Cleaning, normalization, deduplication
│   │   ├── gold.py                 # HS code mapping, supply chain classification, scoring
│   │   ├── normalizers.py          # Country (ISO 3166) and date (ISO 8601) normalization
│   │   ├── dedup.py                # Content-hash deduplication
│   │   ├── raster.py               # GeoTIFF/satellite imagery processing
│   │   ├── satellite_analysis.py   # NDVI, cloud masking, change detection
│   │   └── promoter.py             # Gold -> Corpus/LanceDB promotion
│   │
│   ├── cli/
│   │   └── main.py                 # Typer CLI: harvest, status, promote
│   │
│   └── dagster/                    # Dagster orchestration
│       ├── assets.py               # 148 software-defined assets (37 sources x 4 layers)
│       ├── jobs.py                 # Job definitions
│       ├── resources.py            # Dagster resources
│       ├── schedules.py            # Cron schedule definitions
│       └── sensors.py              # Event sensors
│
├── tests/                          # 204 tests
│   ├── conftest.py                 # Shared fixtures
│   ├── test_http_client.py
│   ├── test_storage.py
│   ├── harvesters/                 # Per-harvester tests (37 files)
│   ├── processing/                 # Silver, gold, normalizer tests
│   └── fixtures/                   # Mock API response data
│
└── data/                           # Lakehouse output (gitignored)
    ├── bronze/                     # 48 parquet files, 30 sources, 2.2M records
    ├── silver/                     # (empty — not yet processed)
    ├── gold/                       # (empty — not yet processed)
    └── dead_letter/                # (empty — no errors)

Exception Hierarchy

HarvesterError (base)
├── ConfigError
│   └── SourceConfigError
├── HttpClientError
│   ├── RateLimitError
│   └── AuthenticationError
├── StorageError
│   ├── StorageWriteError
│   └── StorageReadError
├── ProcessingError
│   ├── DeduplicationError
│   └── NormalizationError
├── PromotionError
└── RasterError

Adding a New Source

Create source config — configs/sources/{source_id}.yaml:

source_id: my_new_source
name: My New Data Source
base_url: https://api.example.com
auth_type: api_key          # none | api_key | bearer | oauth2
auth_env_var: MY_SOURCE_KEY # env var name for API key
rate_limit_rps: 5.0
schedule: "0 2 * * *"       # Daily at 2 AM UTC
data_format: json
description: Description of the data source
cmm_hs_codes: []            # Optional: HS codes to filter

Create harvester — src/cmm_harvester/harvesters/{source_id}.py:

from __future__ import annotations

from typing import Any

from cmm_harvester.harvesters.base import BaseHarvester

_MAX_PAGES = 10
_PAGE_SIZE = 100

class MyNewSource(BaseHarvester):
    """Harvest data from My New Data Source."""

    source_id = "my_new_source"

    async def fetch(self) -> list[dict[str, Any]]:
        all_records: list[dict[str, Any]] = []
        for page in range(_MAX_PAGES):
            resp = await self.client.get(
                "/api/endpoint",
                params={"page": page, "limit": _PAGE_SIZE},
            )
            data = resp.json()
            records = data.get("results", [])
            all_records.extend(records)
            if len(records) < _PAGE_SIZE:
                break
        self.log.info("fetch_complete", count=len(all_records))
        return all_records

Add tests — tests/harvesters/test_{source_id}.py:

import httpx
import respx

from cmm_harvester.harvesters.my_new_source import MyNewSource

@respx.mock
async def test_my_new_source_fetch(sample_source_config, settings, storage):
    config = sample_source_config.model_copy(
        update={"source_id": "my_new_source", "base_url": "https://api.example.com"}
    )
    respx.get("https://api.example.com/api/endpoint").mock(
        return_value=httpx.Response(200, json={"results": [{"id": 1}, {"id": 2}]})
    )
    async with MyNewSource(config=config, settings=settings, storage=storage) as h:
        records = await h.fetch()
    assert len(records) == 2

Register in src/cmm_harvester/cli/main.py (HARVESTER_REGISTRY) and src/cmm_harvester/dagster/assets.py (ALL_SOURCES).
Verify: make check

Dagster Orchestration

The pipeline includes full Dagster integration for production scheduling:

# Launch Dagster UI
make dagster
# Opens at http://localhost:3000

Assets

148 software-defined assets are generated via a factory pattern (37 sources x 4 layers):

{source_id}_bronze — harvest from API
{source_id}_silver — depends on bronze, runs cleaning/normalization
{source_id}_gold — depends on silver, runs CMM enrichment
{source_id}_corpus — depends on gold, promotes to Corpus/LanceDB

Schedules

Each source has a cron schedule defined in configs/schedules.yaml. Default schedules:

Every 30 min: GDELT
Daily at 2 AM: Trade, policy, sanctions, energy, corporate, patent sources
Weekly: ACLED, Eurostat, WITS, BGS, EITI, IPIS, ILO, World Bank, OSM
Monthly: Satellite (Copernicus, VIIRS), STB Waybill

Docker

# Build CLI image
docker build --target harvester -t cmm-harvester .

# Run a harvest
docker run --env-file .env -v $(pwd)/data:/app/data cmm-harvester harvest --source census_trade

# Build Dagster image
docker build --target dagster -t cmm-dagster .

# Run Dagster UI
docker run --env-file .env -v $(pwd)/data:/app/data -p 3000:3000 cmm-dagster

Build Targets

Target	Base	Extras	Purpose
`harvester`	python:3.12-slim	core only	CLI harvesting
`dagster`	python:3.12-slim	+ dagster	Orchestration UI

Pagination

All harvesters that query paginated APIs implement multi-page fetching with configurable limits:

Harvester	Pagination Type	Page Size	Max Pages
`congress_gov`	Offset (`offset += limit`)	50 bills / 25 hearings	10
`courtlistener`	Cursor (follow `next` URL)	API default	5
`regulations_gov`	Page number (`page[number]`)	API default	10
`govinfo_gpo`	Cursor (`offsetMark`)	100	10
`osti`	Page index (0-based)	100	10
`usaspending`	Page number (1-based POST)	100	10
`arxiv`	Start offset	200	5
`sam_gov`	Offset	100	10
`gdelt`	N/A (API hard cap)	250 max	1
`bis_screening`	Offset	100	10
`sec_formsd`	Start/count	100	10
`uspto_patentsview`	Offset (JSON options)	100	10

All pagination loops break early when the API returns fewer records than the page size, preventing unnecessary requests.

HTTP Client

The HarvesterHttpClient provides resilient HTTP access for all harvesters:

Async: Built on httpx.AsyncClient for concurrent requests
Retry: 3 attempts with exponential backoff (1s, 2s, 4s) via tenacity
Rate limiting: Per-source token bucket (configurable rate_limit_rps)
Authentication: None, API key (X-Api-Key), Bearer token, or custom headers
Error handling: Typed exceptions for 429 (rate limit), 401/403 (auth), 4xx/5xx (client/server)
Methods: get() (relative URL), post(), get_absolute() (for cursor pagination), download()

Name		Name	Last commit message	Last commit date
Latest commit History 17 Commits
.github/workflows		.github/workflows
configs		configs
src/cmm_harvester		src/cmm_harvester
tests		tests
.env.example		.env.example
.gitignore		.gitignore
.pre-commit-config.yaml		.pre-commit-config.yaml
.python-version		.python-version
Dockerfile		Dockerfile
Makefile		Makefile
README.md		README.md
pyproject.toml		pyproject.toml

Folders and files

Latest commit

History

Repository files navigation

CMM Harvester

Table of Contents

Overview

What It Does

Why It Matters

Data Sources

37 Implemented Harvesters Across 11 Domains

Trade & Economics (6 sources)

Policy & Legislation (3 sources)

Sanctions & Screening (3 sources)

Government & Spending (2 sources)

Mining & Minerals (5 sources)

Energy & Infrastructure (3 sources)

Satellite & Remote Sensing (2 sources)

Geopolitical Events (2 sources)

Environmental Compliance (1 source)

Corporate & Financial (3 sources)

Scientific & Technology (3 sources)

Additional Data (2 sources)

Architecture

Layers

File Layout

Parquet Schema (Bronze)

Setup

Prerequisites

Installation

API Keys

Usage

CLI Commands

Batch Harvesting

Data Exploration with DuckDB

Example Queries

Harvested Data Inventory

Key Findings from Initial Data Exploration

REE Mining and Processing Sites

REE Trade Flows

Legislative and Regulatory Activity

Cross-Reference: Mining States

Data Gaps Identified

Configuration

Source Configuration (YAML)

Environment Settings

CMM Commodity Reference

Development

Prerequisites

Commands

Tooling

Testing

CI Pipeline

Project Structure

Exception Hierarchy

Adding a New Source

Dagster Orchestration

Assets

Schedules

Docker

Build Targets

Pagination

HTTP Client

About

Topics

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages