115 lines
3.9 KiB
Python
115 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Extract GDP per capita (current US$) for 12 economies from the World Bank API.
|
|
|
|
Lane: Extract (data extraction / tables).
|
|
Source: World Bank API v2, indicator NY.GDP.PCAP.CD (GDP per capita, current US$),
|
|
country ISO3 codes, per_page=100, newest-first; take the most recent
|
|
record for each country. Retrieved (UTC): see fetched_at below.
|
|
|
|
Outputs (same directory as this script):
|
|
gdp-per-capita.md - human-readable table (display rounded to 2 dp)
|
|
gdp-per-capita.csv - machine-readable table (raw values)
|
|
gdp-per-capita.json- parsed table + provenance
|
|
|
|
Verdict (see sibling verify_gdp_per_capita.py): re-fetches the API and diffs
|
|
the artifact; exit 0 == artifact matches the live API response.
|
|
"""
|
|
import csv
|
|
import io
|
|
import json
|
|
import urllib.request
|
|
from collections import OrderedDict
|
|
from datetime import datetime, timezone
|
|
|
|
API_BASE = "https://api.worldbank.org/v2/country/{country}/indicator/NY.GDP.PCAP.CD"
|
|
INDICATOR = "NY.GDP.PCAP.CD"
|
|
COUNTRIES = OrderedDict([
|
|
("Australia", "AUS"),
|
|
("Brazil", "BRA"),
|
|
("Canada", "CAN"),
|
|
("China", "CHN"),
|
|
("France", "FRA"),
|
|
("Germany", "DEU"),
|
|
("India", "IND"),
|
|
("Italy", "ITA"),
|
|
("Japan", "JPN"),
|
|
("South Korea", "KOR"),
|
|
("United Kingdom", "GBR"),
|
|
("United States", "USA"),
|
|
])
|
|
FUNDED = "ledger (Hermes Agent extract lane, Nous free tier)"
|
|
OUT_DIR = "/root/.hermes/wizard-outputs/ledger"
|
|
|
|
|
|
def fetch(country_iso3):
|
|
url = "{}?format=json&per_page=100".format(API_BASE.format(country=country_iso3))
|
|
req = urllib.request.Request(url, headers={"User-Agent": "ledger-extract/1.0"})
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
raw = r.read()
|
|
try:
|
|
data = json.loads(raw.decode("utf-8"))
|
|
except Exception as e:
|
|
raise RuntimeError("bad JSON for %s: %r" % (country_iso3, raw[:300])) from e
|
|
results = data[1]
|
|
if not results:
|
|
raise RuntimeError("empty results for %s" % country_iso3)
|
|
r = results[0] # newest-first
|
|
return r["date"], float(r["value"]), r.get("decimal")
|
|
|
|
|
|
def main():
|
|
rows = []
|
|
provenance = {
|
|
"source": "https://data.worldbank.org/indicator/NY.GDP.PCAP.CD",
|
|
"indicator": "NY.GDP.PCAP.CD (GDP per capita, current US$)",
|
|
"fetched_at": datetime.now(timezone.utc).isoformat(),
|
|
"countries": {name: iso for name, iso in COUNTRIES.items()},
|
|
"funded_by": FUNDED,
|
|
}
|
|
for name, iso in COUNTRIES.items():
|
|
yr, val, dec = fetch(iso)
|
|
provenance.setdefault("decoder", {})[iso] = dec
|
|
rows.append({"country": name, "year": int(yr), "value": val})
|
|
|
|
out = io.StringIO()
|
|
w = csv.DictWriter(out, fieldnames=["country", "year", "value"])
|
|
w.writeheader()
|
|
for r in rows:
|
|
w.writerow(r)
|
|
csv_text = out.getvalue()
|
|
|
|
enc = json.dumps(provenance, indent=2)
|
|
with open(OUT_DIR + "/gdp-per-capita.json", "w") as f:
|
|
f.write(enc + "\n")
|
|
with open(OUT_DIR + "/gdp-per-capita.csv", "w") as f:
|
|
f.write(csv_text)
|
|
|
|
md = build_md(rows, provenance)
|
|
with open(OUT_DIR + "/gdp-per-capita.md", "w") as f:
|
|
f.write(md)
|
|
|
|
print("fetched_at: %s" % provenance["fetched_at"])
|
|
print("rows: %d" % len(rows))
|
|
print("wrote: "
|
|
"gdp-per-capita.md gdp-per-capita.csv gdp-per-capita.json")
|
|
|
|
|
|
def build_md(rows, prov):
|
|
lines = []
|
|
lines.append("# GDP per capita, current US$ (World Bank, NY.GDP.PCAP.CD)")
|
|
lines.append("")
|
|
lines.append("Source: World Bank API `NY.GDP.PCAP.CD`, per-ISO3 newest record,"
|
|
" fetched `%s`." % prov["fetched_at"])
|
|
lines.append("Funded by: %s" % prov["funded_by"])
|
|
lines.append("")
|
|
lines.append("| Country | GDP per capita (current US$) | Year |")
|
|
lines.append("|---|---|---|")
|
|
for r in rows:
|
|
lines.append("| %s | %.2f | %d |" % (r["country"], r["value"], r["year"]))
|
|
lines.append("")
|
|
return "\n".join(lines)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|