mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-10-27 12:40:35 +00:00
feat(platform): add membership groups to Country type (#7352)
* feat(platform): add membership groups to Country type Add membership groups (G7, G20, EU, NATO, OECD, OPEC, BRICS) to the Country type, as suggested by maintainer in PR #7333. Changes: - Updated country_data.json with 'groups' field for each country - Added Country.groups property returning list of group memberships - Added Country.is_member_of(group) method for membership checks - Added comprehensive tests for group functionality This enables filtering and analysis by economic/political groupings: c = Country('Germany') c.groups # ['G7', 'G20', 'EU', 'NATO', 'OECD'] c.is_member_of('EU') # True Follow-up to #7333 * fix(platform): add Indonesia to BRICS (joined Jan 2025) * refactor(platform): use TypedDict for country data typing - Replace dict[str, Any] with CountryData TypedDict for proper type inference - Add official data sources to module docstring - Add _sources metadata section to country_data.json with references Sources: - G7: g7italy.it - G20: g20.org - EU: european-union.europa.eu - NATO: nato.int - OECD: oecd.org - OPEC: opec.org - BRICS: brics-russia2024.ru * fix(platform): update group membership reference URLs Replace broken/inaccessible URLs with stable Wikipedia references: - G7, G20, OECD, OPEC, BRICS → Wikipedia (stable, well-maintained) - EU, NATO → Updated to current official URLs * feat: add developer script to regenerate country_data.json from sources * fix: resolve codespell lint errors in update_country_data.py * fix: resolve black/ruff lint issues in update_country_data.py * feat(core): normalize unicode country names for ASCII input matching * fix(core): remove Angola from OPEC and Bangladesh from NATO Angola withdrew from OPEC effective Jan 2024. The OPEC scraper falls back to a hardcoded list which still included AO — removed it. Bangladesh was incorrectly scraped as a NATO member due to the scraper iterating all wikitables on the page. Now filters tables by their nearest h2 section header to only process member state tables. * fix(core): normalize country names to ASCII in update script Strip diacritical marks at generation time so country_data.json uses ASCII-only names (e.g., Turkiye, Curacao, Cote d'Ivoire). * Update openbb_platform/core/openbb_core/provider/utils/update_country_data.py Co-authored-by: Danglewood <85772166+deeleeramone@users.noreply.github.com> * fix imports * fix: resolve mypy type narrowing for expected counts * restore reference.json --------- Co-authored-by: Danglewood <85772166+deeleeramone@users.noreply.github.com>
This commit is contained in:
John Seong
@@ -1,2 +1,3 @@
|
||||
""" Autogenerated OpenBB module."""
|
||||
"""Autogenerated OpenBB module."""
|
||||
|
||||
### THIS FILE IS AUTO-GENERATED. DO NOT EDIT. ###
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -3,20 +3,58 @@
|
||||
This module provides a Country type that inherits from str for type checker compatibility
|
||||
while providing full access to ISO 3166 country data via a static dataset.
|
||||
|
||||
Providers can access alpha_2, alpha_3, name, and numeric properties as needed.
|
||||
Providers can access alpha_2, alpha_3, name, numeric, and groups properties as needed.
|
||||
|
||||
Supported membership groups:
|
||||
- G7: Group of Seven major advanced economies
|
||||
- G20: Group of Twenty major economies
|
||||
- EU: European Union member states
|
||||
- NATO: North Atlantic Treaty Organization members
|
||||
- OECD: Organisation for Economic Co-operation and Development members
|
||||
- OPEC: Organization of the Petroleum Exporting Countries
|
||||
- BRICS: Brazil, Russia, India, China, South Africa (+ 2024 expansion)
|
||||
|
||||
References:
|
||||
- ISO 3166-1: https://en.wikipedia.org/wiki/ISO_3166-1
|
||||
- G7: https://en.wikipedia.org/wiki/G7
|
||||
- G20: https://en.wikipedia.org/wiki/G20
|
||||
- EU: https://european-union.europa.eu/principles-countries-history/eu-countries_en
|
||||
- NATO: https://www.nato.int/en/about-us/organization/nato-member-countries
|
||||
- OECD: https://en.wikipedia.org/wiki/OECD
|
||||
- OPEC: https://en.wikipedia.org/wiki/OPEC
|
||||
- BRICS: https://en.wikipedia.org/wiki/BRICS
|
||||
"""
|
||||
|
||||
import json
|
||||
import unicodedata
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from typing import Any, TypedDict
|
||||
|
||||
from pydantic_core import core_schema
|
||||
|
||||
|
||||
def _load_country_data() -> dict[str, dict[str, str]]:
|
||||
class CountryData(TypedDict, total=False):
|
||||
"""Type definition for country data dictionary."""
|
||||
|
||||
alpha_2: str
|
||||
alpha_3: str
|
||||
name: str
|
||||
numeric: str
|
||||
groups: list[str]
|
||||
|
||||
|
||||
_COMMON_ALIASES: dict[str, list[str]] = {
|
||||
"TR": ["turkey"],
|
||||
}
|
||||
|
||||
|
||||
def _strip_accents(text: str) -> str:
|
||||
"""Strip diacritical marks from text (e.g., Curaçao -> Curacao)."""
|
||||
nfd = unicodedata.normalize("NFD", text)
|
||||
return "".join(c for c in nfd if not unicodedata.combining(c))
|
||||
|
||||
|
||||
def _load_country_data() -> dict[str, CountryData]:
|
||||
"""Load country data from JSON and build lookup indices.
|
||||
|
||||
Returns a dict with lookup keys (alpha_2, alpha_3, name variants) mapping to country data.
|
||||
@@ -25,24 +63,30 @@ def _load_country_data() -> dict[str, dict[str, str]]:
|
||||
with open(data_path, encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
|
||||
lookup: dict[str, dict[str, str]] = {}
|
||||
lookup: dict[str, CountryData] = {}
|
||||
for country in data["countries"]:
|
||||
# Index by alpha_2 (case-insensitive)
|
||||
lookup[country["alpha_2"].upper()] = country
|
||||
lookup[country["alpha_2"].lower()] = country
|
||||
|
||||
# Index by alpha_3 (case-insensitive)
|
||||
lookup[country["alpha_3"].upper()] = country
|
||||
lookup[country["alpha_3"].lower()] = country
|
||||
|
||||
# Index by name (case-insensitive)
|
||||
name_lower = country["name"].lower()
|
||||
lookup[name_lower] = country
|
||||
lookup[country["name"]] = country
|
||||
|
||||
# Index by snake_case name
|
||||
ascii_lower = _strip_accents(name_lower)
|
||||
if ascii_lower != name_lower:
|
||||
lookup[ascii_lower] = country
|
||||
|
||||
for alias in _COMMON_ALIASES.get(country["alpha_2"], []):
|
||||
lookup[alias] = country
|
||||
|
||||
snake_name = name_lower.replace(" ", "_").replace(",", "").replace("'", "")
|
||||
lookup[snake_name] = country
|
||||
ascii_snake = _strip_accents(snake_name)
|
||||
if ascii_snake != snake_name:
|
||||
lookup[ascii_snake] = country
|
||||
|
||||
return lookup
|
||||
|
||||
@@ -74,9 +118,13 @@ class Country(str):
|
||||
'USA'
|
||||
>>> c.name
|
||||
'United States'
|
||||
>>> c.groups
|
||||
['G7', 'G20', 'NATO', 'OECD']
|
||||
>>> c.is_member_of("G7")
|
||||
True
|
||||
"""
|
||||
|
||||
_country_data: dict[str, str]
|
||||
_country_data: CountryData
|
||||
|
||||
def __new__(cls, value: Any) -> "Country":
|
||||
"""Create a new Country instance.
|
||||
@@ -109,7 +157,7 @@ class Country(str):
|
||||
return instance
|
||||
|
||||
@staticmethod
|
||||
def _lookup_country(value: Any) -> dict[str, str]:
|
||||
def _lookup_country(value: Any) -> CountryData:
|
||||
"""Look up a country from various input formats.
|
||||
|
||||
Parameters
|
||||
@@ -119,8 +167,8 @@ class Country(str):
|
||||
|
||||
Returns
|
||||
-------
|
||||
dict[str, str]
|
||||
The country data dictionary with alpha_2, alpha_3, name, numeric.
|
||||
CountryData
|
||||
The country data dictionary with alpha_2, alpha_3, name, numeric, groups.
|
||||
|
||||
Raises
|
||||
------
|
||||
@@ -129,19 +177,20 @@ class Country(str):
|
||||
"""
|
||||
val = str(value).strip()
|
||||
|
||||
# Convert lower_snake_case to lookup key
|
||||
if "_" in val:
|
||||
val = val.replace("_", " ")
|
||||
|
||||
# Try direct lookup
|
||||
lookup_key = val.lower()
|
||||
if lookup_key in _COUNTRY_LOOKUP:
|
||||
return _COUNTRY_LOOKUP[lookup_key]
|
||||
|
||||
# Try original case
|
||||
if val in _COUNTRY_LOOKUP:
|
||||
return _COUNTRY_LOOKUP[val]
|
||||
|
||||
ascii_key = _strip_accents(lookup_key)
|
||||
if ascii_key in _COUNTRY_LOOKUP:
|
||||
return _COUNTRY_LOOKUP[ascii_key]
|
||||
|
||||
raise ValueError(
|
||||
f"Invalid country: '{value}'. "
|
||||
"Accepts ISO 3166-1 alpha-2 codes (e.g., 'US'), "
|
||||
@@ -169,6 +218,44 @@ class Country(str):
|
||||
"""ISO 3166-1 numeric code (e.g., '840'), if available."""
|
||||
return self._country_data.get("numeric")
|
||||
|
||||
@property
|
||||
def groups(self) -> list[str]:
|
||||
"""List of membership groups this country belongs to.
|
||||
|
||||
Available groups: G7, G20, EU, NATO, OECD, OPEC, BRICS.
|
||||
|
||||
Examples
|
||||
--------
|
||||
>>> c = Country("US")
|
||||
>>> c.groups
|
||||
['G7', 'G20', 'NATO', 'OECD']
|
||||
"""
|
||||
return self._country_data.get("groups", [])
|
||||
|
||||
def is_member_of(self, group: str) -> bool:
|
||||
"""Check if country is a member of a specific group.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
group : str
|
||||
The group to check membership for (e.g., 'G7', 'EU', 'NATO').
|
||||
Case-insensitive.
|
||||
|
||||
Returns
|
||||
-------
|
||||
bool
|
||||
True if the country is a member of the group.
|
||||
|
||||
Examples
|
||||
--------
|
||||
>>> c = Country("Germany")
|
||||
>>> c.is_member_of("G7")
|
||||
True
|
||||
>>> c.is_member_of("OPEC")
|
||||
False
|
||||
"""
|
||||
return group.upper() in [g.upper() for g in self.groups]
|
||||
|
||||
@classmethod
|
||||
def __get_pydantic_core_schema__(cls, _source_type: Any, _handler: Any) -> Any:
|
||||
"""Return the Pydantic core schema for validation."""
|
||||
|
||||
@@ -0,0 +1,607 @@
|
||||
# ruff: noqa: T201
|
||||
"""Developer script to regenerate country_data.json from authoritative sources.
|
||||
|
||||
Sources:
|
||||
- ISO 3166-1: pycountry library
|
||||
- G7: https://en.wikipedia.org/wiki/G7
|
||||
- G20: https://en.wikipedia.org/wiki/G20
|
||||
- EU: https://european-union.europa.eu/principles-countries-history/eu-countries_en
|
||||
- NATO: https://en.wikipedia.org/wiki/Member_states_of_NATO
|
||||
- OECD: https://en.wikipedia.org/wiki/OECD
|
||||
- OPEC: https://en.wikipedia.org/wiki/OPEC
|
||||
- BRICS: https://en.wikipedia.org/wiki/BRICS
|
||||
|
||||
Usage:
|
||||
python update_country_data.py [--dry-run] [--output PATH]
|
||||
|
||||
Requirements:
|
||||
pip install pycountry beautifulsoup4 requests
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import unicodedata
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
|
||||
import requests
|
||||
|
||||
try:
|
||||
import pycountry
|
||||
from bs4 import BeautifulSoup
|
||||
except ImportError as e:
|
||||
missing_package = str(e).split("'")[1] if "'" in str(e) else "unknown"
|
||||
raise ImportError(
|
||||
f"Missing required package: {missing_package}. "
|
||||
"Please install dependencies: pip install pycountry beautifulsoup4 requests"
|
||||
) from e
|
||||
|
||||
SCRIPT_DIR = Path(__file__).parent
|
||||
DEFAULT_OUTPUT = SCRIPT_DIR / "country_data.json"
|
||||
|
||||
HEADERS = {
|
||||
"User-Agent": (
|
||||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
|
||||
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
||||
)
|
||||
}
|
||||
|
||||
# Expected member counts for sanity checks (approximate — update when membership changes)
|
||||
EXPECTED_COUNTS: dict[str, int | tuple[int, int]] = {
|
||||
"G7": 7,
|
||||
"G20": 20, # 19 countries + EU (we only count countries)
|
||||
"EU": 27,
|
||||
"NATO": 32,
|
||||
"OECD": 38,
|
||||
"OPEC": (12, 14), # fluctuates
|
||||
"BRICS": (9, 11), # 2024 expansion + Indonesia 2025
|
||||
}
|
||||
|
||||
# ── Name normalization ──────────────────────────────────────────────────────
|
||||
# Map common display names → pycountry names or alpha_2 codes
|
||||
NAME_OVERRIDES: dict[str, str | None] = {
|
||||
"brunei": "BN",
|
||||
"bolivia": "BO",
|
||||
"congo": "CG",
|
||||
"democratic republic of the congo": "CD",
|
||||
"côte d'ivoire": "CI",
|
||||
"cote d'ivoire": "CI",
|
||||
"ivory coast": "CI",
|
||||
"czech republic": "CZ",
|
||||
"czechia": "CZ",
|
||||
"eswatini": "SZ",
|
||||
"swaziland": "SZ",
|
||||
"iran": "IR",
|
||||
"south korea": "KR",
|
||||
"korea": "KR",
|
||||
"korea, republic of": "KR",
|
||||
"republic of korea": "KR",
|
||||
"north korea": "KP",
|
||||
"north macedonia": "MK",
|
||||
"laos": "LA",
|
||||
"moldova": "MD",
|
||||
"russia": "RU",
|
||||
"russian federation": "RU",
|
||||
"syria": "SY",
|
||||
"taiwan": "TW",
|
||||
"tanzania": "TZ",
|
||||
"türkiye": "TR",
|
||||
"turkey": "TR",
|
||||
"united states of america": "US",
|
||||
"united states": "US",
|
||||
"usa": "US",
|
||||
"united kingdom": "GB",
|
||||
"uk": "GB",
|
||||
"venezuela": "VE",
|
||||
"vietnam": "VN",
|
||||
"palestine": "PS",
|
||||
"the netherlands": "NL",
|
||||
"netherlands": "NL",
|
||||
"uae": "AE",
|
||||
"united arab emirates": "AE",
|
||||
"saudi arabia": "SA",
|
||||
"south africa": "ZA",
|
||||
"european union": None, # skip — not a country
|
||||
"african union": None,
|
||||
}
|
||||
|
||||
|
||||
def _strip_accents(text: str) -> str:
|
||||
"""Strip diacritical marks (e.g., Côte d'Ivoire -> Cote d'Ivoire)."""
|
||||
nfd = unicodedata.normalize("NFD", text)
|
||||
return "".join(c for c in nfd if not unicodedata.combining(c))
|
||||
|
||||
|
||||
def _build_name_index() -> dict[str, str]:
|
||||
"""Build lowercase name → alpha_2 lookup from pycountry + overrides."""
|
||||
idx: dict[str, str] = {}
|
||||
for c in pycountry.countries:
|
||||
idx[c.name.lower()] = c.alpha_2 # type: ignore
|
||||
if hasattr(c, "common_name"):
|
||||
idx[c.common_name.lower()] = c.alpha_2 # type: ignore
|
||||
if hasattr(c, "official_name"):
|
||||
idx[c.official_name.lower()] = c.alpha_2 # type: ignore
|
||||
idx.update(NAME_OVERRIDES) # type: ignore
|
||||
return idx
|
||||
|
||||
|
||||
NAME_INDEX = _build_name_index()
|
||||
|
||||
|
||||
def resolve_country(name: str) -> str | None:
|
||||
"""Resolve a display name to ISO alpha_2, or None if not a country."""
|
||||
clean = re.sub(r"\[.*?\]", "", name).strip() # remove wiki refs like [1]
|
||||
clean = re.sub(r"\(.*?\)", "", clean).strip() # remove parentheticals
|
||||
clean = clean.strip("* \t\n")
|
||||
key = clean.lower()
|
||||
|
||||
if key in NAME_INDEX:
|
||||
return NAME_INDEX[key]
|
||||
|
||||
# Try pycountry fuzzy search as fallback
|
||||
try:
|
||||
results = pycountry.countries.search_fuzzy(clean)
|
||||
if results:
|
||||
return results[0].alpha_2 # type: ignore
|
||||
except LookupError:
|
||||
pass
|
||||
|
||||
return None
|
||||
|
||||
|
||||
# ── Scrapers ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def fetch_soup(url: str) -> BeautifulSoup:
|
||||
"""Fetch a URL and return parsed BeautifulSoup."""
|
||||
resp = requests.get(url, headers=HEADERS, timeout=30)
|
||||
resp.raise_for_status()
|
||||
return BeautifulSoup(resp.text, "html.parser")
|
||||
|
||||
|
||||
def scrape_g7() -> set[str]:
|
||||
"""Scrape G7 members from Wikipedia.
|
||||
|
||||
G7 membership is extremely stable (no changes since Russia left in 2014),
|
||||
so we use the infobox and validate against hardcoded list.
|
||||
"""
|
||||
soup = fetch_soup("https://en.wikipedia.org/wiki/G7")
|
||||
hardcoded = {"CA", "FR", "DE", "IT", "JP", "GB", "US"}
|
||||
|
||||
members = set()
|
||||
# Look for the infobox — it has a row labeled "Members" with country links
|
||||
infobox = soup.find("table", class_="infobox")
|
||||
if infobox:
|
||||
for tr in infobox.find_all("tr"):
|
||||
th = tr.find("th")
|
||||
if th and "member" in th.get_text().lower():
|
||||
td = tr.find("td")
|
||||
if td:
|
||||
for li in td.find_all("li"):
|
||||
a = li.find("a")
|
||||
if a:
|
||||
code = resolve_country(a.get_text())
|
||||
if code:
|
||||
members.add(code)
|
||||
|
||||
return members if 6 <= len(members) <= 9 else hardcoded
|
||||
|
||||
|
||||
def scrape_g20() -> set[str]:
|
||||
"""Scrape G20 members from Wikipedia."""
|
||||
soup = fetch_soup("https://en.wikipedia.org/wiki/G20")
|
||||
hardcoded = {
|
||||
"AR",
|
||||
"AU",
|
||||
"BR",
|
||||
"CA",
|
||||
"CN",
|
||||
"FR",
|
||||
"DE",
|
||||
"IN",
|
||||
"ID",
|
||||
"IT",
|
||||
"JP",
|
||||
"KR",
|
||||
"MX",
|
||||
"RU",
|
||||
"SA",
|
||||
"ZA",
|
||||
"TR",
|
||||
"GB",
|
||||
"US",
|
||||
}
|
||||
|
||||
members = set()
|
||||
# Find the member states section
|
||||
for header in soup.find_all(["h2", "h3"]):
|
||||
if "member" in header.get_text().lower():
|
||||
# Scan the next sibling elements for a table or list
|
||||
for sib in header.find_next_siblings():
|
||||
if sib.name in ("h2", "h3"):
|
||||
break
|
||||
for a in sib.find_all("a"):
|
||||
code = resolve_country(a.get_text())
|
||||
if code:
|
||||
members.add(code)
|
||||
|
||||
# Filter to reasonable count (G20 has 19 country members + EU)
|
||||
return members if len(members) >= 19 else hardcoded
|
||||
|
||||
|
||||
def scrape_eu() -> set[str]:
|
||||
"""Scrape EU member states from Wikipedia's member states table."""
|
||||
hardcoded = {
|
||||
"AT",
|
||||
"BE",
|
||||
"BG",
|
||||
"HR",
|
||||
"CY",
|
||||
"CZ",
|
||||
"DK",
|
||||
"EE",
|
||||
"FI",
|
||||
"FR",
|
||||
"DE",
|
||||
"GR",
|
||||
"HU",
|
||||
"IE",
|
||||
"IT",
|
||||
"LV",
|
||||
"LT",
|
||||
"LU",
|
||||
"MT",
|
||||
"NL",
|
||||
"PL",
|
||||
"PT",
|
||||
"RO",
|
||||
"SK",
|
||||
"SI",
|
||||
"ES",
|
||||
"SE",
|
||||
}
|
||||
|
||||
soup = fetch_soup(
|
||||
"https://en.wikipedia.org/wiki/Member_state_of_the_European_Union"
|
||||
)
|
||||
members = set()
|
||||
|
||||
# Find the first wikitable sortable — this is the member states table
|
||||
table = soup.find("table", class_="wikitable sortable")
|
||||
if table:
|
||||
for row in table.find_all("tr")[1:]: # skip header
|
||||
cells = row.find_all("td")
|
||||
if not cells:
|
||||
continue
|
||||
# Country name is typically in the first cell with a link
|
||||
a = cells[0].find("a")
|
||||
if a:
|
||||
code = resolve_country(a.get_text())
|
||||
if code:
|
||||
members.add(code)
|
||||
|
||||
return members if 25 <= len(members) <= 30 else hardcoded
|
||||
|
||||
|
||||
def scrape_nato() -> set[str]:
|
||||
"""Scrape NATO member states from Wikipedia."""
|
||||
soup = fetch_soup("https://en.wikipedia.org/wiki/Member_states_of_NATO")
|
||||
members = set()
|
||||
|
||||
for table in soup.find_all("table", class_="wikitable"):
|
||||
prev_header = table.find_previous("h2")
|
||||
if not prev_header or "member" not in prev_header.get_text().lower():
|
||||
continue
|
||||
for row in table.find_all("tr"):
|
||||
cells = row.find_all("td")
|
||||
if not cells:
|
||||
continue
|
||||
# First cell with a link is usually the country
|
||||
for a in cells[0].find_all("a"):
|
||||
code = resolve_country(a.get_text())
|
||||
if code:
|
||||
members.add(code)
|
||||
break
|
||||
|
||||
hardcoded = {
|
||||
"AL",
|
||||
"BE",
|
||||
"BG",
|
||||
"CA",
|
||||
"HR",
|
||||
"CZ",
|
||||
"DK",
|
||||
"EE",
|
||||
"FI",
|
||||
"FR",
|
||||
"DE",
|
||||
"GR",
|
||||
"HU",
|
||||
"IS",
|
||||
"IT",
|
||||
"LV",
|
||||
"LT",
|
||||
"LU",
|
||||
"ME",
|
||||
"NL",
|
||||
"MK",
|
||||
"NO",
|
||||
"PL",
|
||||
"PT",
|
||||
"RO",
|
||||
"SK",
|
||||
"SI",
|
||||
"ES",
|
||||
"SE",
|
||||
"TR",
|
||||
"GB",
|
||||
"US",
|
||||
}
|
||||
return members if len(members) >= 30 else hardcoded
|
||||
|
||||
|
||||
def scrape_oecd() -> set[str]:
|
||||
"""Scrape OECD member states from Wikipedia."""
|
||||
soup = fetch_soup("https://en.wikipedia.org/wiki/OECD")
|
||||
members = set()
|
||||
|
||||
# Find "Member countries" section
|
||||
for header in soup.find_all(["h2", "h3"]):
|
||||
htext = header.get_text().lower()
|
||||
if "member" in htext and "country" in htext or "countries" in htext:
|
||||
for sib in header.find_next_siblings():
|
||||
if sib.name in ("h2", "h3"):
|
||||
break
|
||||
for a in sib.find_all("a"):
|
||||
code = resolve_country(a.get_text())
|
||||
if code:
|
||||
members.add(code)
|
||||
break
|
||||
|
||||
hardcoded = {
|
||||
"AU",
|
||||
"AT",
|
||||
"BE",
|
||||
"CA",
|
||||
"CL",
|
||||
"CO",
|
||||
"CR",
|
||||
"CZ",
|
||||
"DK",
|
||||
"EE",
|
||||
"FI",
|
||||
"FR",
|
||||
"DE",
|
||||
"GR",
|
||||
"HU",
|
||||
"IS",
|
||||
"IE",
|
||||
"IL",
|
||||
"IT",
|
||||
"JP",
|
||||
"KR",
|
||||
"LV",
|
||||
"LT",
|
||||
"LU",
|
||||
"MX",
|
||||
"NL",
|
||||
"NZ",
|
||||
"NO",
|
||||
"PL",
|
||||
"PT",
|
||||
"SK",
|
||||
"SI",
|
||||
"ES",
|
||||
"SE",
|
||||
"CH",
|
||||
"TR",
|
||||
"GB",
|
||||
"US",
|
||||
}
|
||||
return members if len(members) >= 35 else hardcoded
|
||||
|
||||
|
||||
def scrape_opec() -> set[str]:
|
||||
"""Scrape OPEC member states from Wikipedia."""
|
||||
soup = fetch_soup("https://en.wikipedia.org/wiki/OPEC")
|
||||
members = set()
|
||||
|
||||
# Find "Current members" or "Member countries" section
|
||||
for header in soup.find_all(["h2", "h3"]):
|
||||
htext = header.get_text().lower()
|
||||
if ("current" in htext and "member" in htext) or (
|
||||
"member" in htext and ("country" in htext or "countries" in htext)
|
||||
):
|
||||
for sib in header.find_next_siblings():
|
||||
if sib.name in ("h2", "h3"):
|
||||
break
|
||||
for a in sib.find_all("a"):
|
||||
code = resolve_country(a.get_text())
|
||||
if code:
|
||||
members.add(code)
|
||||
if members:
|
||||
break
|
||||
|
||||
hardcoded = {
|
||||
"DZ",
|
||||
"CG",
|
||||
"GQ",
|
||||
"GA",
|
||||
"IR",
|
||||
"IQ",
|
||||
"KW",
|
||||
"LY",
|
||||
"NG",
|
||||
"SA",
|
||||
"AE",
|
||||
"VE",
|
||||
}
|
||||
return members if len(members) >= 10 else hardcoded
|
||||
|
||||
|
||||
def scrape_brics() -> set[str]:
|
||||
"""Scrape BRICS member states from Wikipedia."""
|
||||
soup = fetch_soup("https://en.wikipedia.org/wiki/BRICS")
|
||||
members = set()
|
||||
|
||||
# Find "Member states" section
|
||||
for header in soup.find_all(["h2", "h3"]):
|
||||
if "member" in header.get_text().lower():
|
||||
for sib in header.find_next_siblings():
|
||||
if sib.name in ("h2", "h3"):
|
||||
break
|
||||
# Look in tables for member states
|
||||
if sib.name == "table":
|
||||
for row in sib.find_all("tr"):
|
||||
for a in row.find_all("a"):
|
||||
code = resolve_country(a.get_text())
|
||||
if code:
|
||||
members.add(code)
|
||||
for a in sib.find_all("a"):
|
||||
code = resolve_country(a.get_text())
|
||||
if code:
|
||||
members.add(code)
|
||||
if members:
|
||||
break
|
||||
|
||||
# 2024 expansion: Egypt, Ethiopia, Iran, UAE + Indonesia (Jan 2025)
|
||||
hardcoded = {"BR", "RU", "IN", "CN", "ZA", "EG", "ET", "IR", "AE", "ID"}
|
||||
return members if len(members) >= 9 else hardcoded
|
||||
|
||||
|
||||
# ── Main logic ──────────────────────────────────────────────────────────────
|
||||
|
||||
GROUP_SCRAPERS = {
|
||||
"G7": scrape_g7,
|
||||
"G20": scrape_g20,
|
||||
"EU": scrape_eu,
|
||||
"NATO": scrape_nato,
|
||||
"OECD": scrape_oecd,
|
||||
"OPEC": scrape_opec,
|
||||
"BRICS": scrape_brics,
|
||||
}
|
||||
|
||||
|
||||
def build_country_data() -> dict:
|
||||
"""Build the complete country_data.json structure."""
|
||||
# 1. Scrape all group memberships
|
||||
group_members: dict[str, set[str]] = {}
|
||||
for group_name, scraper in GROUP_SCRAPERS.items():
|
||||
print(f"Scraping {group_name}...", end=" ", flush=True)
|
||||
try:
|
||||
members = scraper()
|
||||
group_members[group_name] = members
|
||||
print(f"✅ {len(members)} members")
|
||||
except Exception as e:
|
||||
print(f"❌ Error: {e}")
|
||||
group_members[group_name] = set()
|
||||
|
||||
# 2. Sanity checks
|
||||
print("\n── Sanity Checks ──")
|
||||
for group_name, members in group_members.items():
|
||||
expected = EXPECTED_COUNTS.get(group_name)
|
||||
count = len(members)
|
||||
if expected is None:
|
||||
continue
|
||||
if isinstance(expected, tuple):
|
||||
lo, hi = expected
|
||||
ok = lo <= count <= hi
|
||||
expected_str = f"{lo}-{hi}"
|
||||
elif isinstance(expected, int):
|
||||
ok = abs(count - expected) <= 2 # allow ±2 tolerance
|
||||
expected_str = str(expected)
|
||||
status = "✅" if ok else "⚠️"
|
||||
print(f" {status} {group_name}: got {count}, expected ~{expected_str}")
|
||||
if not ok:
|
||||
print(f" Members: {sorted(members)}")
|
||||
|
||||
# 3. Build alpha_2 → groups mapping
|
||||
country_groups: dict[str, list[str]] = {}
|
||||
for group_name, members in group_members.items():
|
||||
for alpha_2 in members:
|
||||
country_groups.setdefault(alpha_2, []).append(group_name)
|
||||
|
||||
# Sort group lists for consistency
|
||||
for groups in country_groups.values():
|
||||
groups.sort()
|
||||
|
||||
# 4. Build country list from pycountry
|
||||
countries = []
|
||||
for c in sorted(pycountry.countries, key=lambda x: x.alpha_2): # type: ignore
|
||||
entry: dict = {
|
||||
"alpha_2": c.alpha_2, # type: ignore
|
||||
"alpha_3": c.alpha_3, # type: ignore
|
||||
"name": _strip_accents(c.name), # type: ignore
|
||||
"numeric": c.numeric, # type: ignore
|
||||
}
|
||||
groups = country_groups.get(c.alpha_2) # type: ignore
|
||||
if groups:
|
||||
entry["groups"] = groups
|
||||
countries.append(entry)
|
||||
|
||||
return {
|
||||
"_last_updated": date.today().isoformat(),
|
||||
"_sources": {
|
||||
"iso_3166": "pycountry library",
|
||||
"G7": "https://en.wikipedia.org/wiki/G7",
|
||||
"G20": "https://en.wikipedia.org/wiki/G20",
|
||||
"EU": "https://european-union.europa.eu/principles-countries-history/eu-countries_en",
|
||||
"NATO": "https://en.wikipedia.org/wiki/Member_states_of_NATO",
|
||||
"OECD": "https://en.wikipedia.org/wiki/OECD",
|
||||
"OPEC": "https://en.wikipedia.org/wiki/OPEC",
|
||||
"BRICS": "https://en.wikipedia.org/wiki/BRICS",
|
||||
},
|
||||
"countries": countries,
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
"""CLI entrypoint to regenerate country_data.json."""
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Regenerate country_data.json from authoritative sources."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--output",
|
||||
"-o",
|
||||
type=Path,
|
||||
default=DEFAULT_OUTPUT,
|
||||
help=f"Output path (default: {DEFAULT_OUTPUT})",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--dry-run",
|
||||
action="store_true",
|
||||
help="Print summary without writing file.",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
print("🌍 Updating country_data.json\n")
|
||||
data = build_country_data()
|
||||
|
||||
countries_with_groups = sum(1 for c in data["countries"] if "groups" in c)
|
||||
total = len(data["countries"])
|
||||
print("\n── Summary ──")
|
||||
print(f" Total countries: {total}")
|
||||
print(f" Countries with group memberships: {countries_with_groups}")
|
||||
print(f" Last updated: {data['_last_updated']}")
|
||||
|
||||
if args.dry_run:
|
||||
print("\n[DRY RUN] Would write to:", args.output)
|
||||
# Print first 5 entries as sample
|
||||
print("\nSample (first 5 entries):")
|
||||
sample = {
|
||||
"_last_updated": data["_last_updated"],
|
||||
"countries": data["countries"][:5],
|
||||
}
|
||||
print(json.dumps(sample, indent=2, ensure_ascii=False))
|
||||
else:
|
||||
with open(args.output, "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, indent=2, ensure_ascii=False)
|
||||
f.write("\n")
|
||||
print(f"\n✅ Written to: {args.output}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -97,3 +97,95 @@ class TestCountry:
|
||||
def test_country_param_alias(self):
|
||||
"""CountryParam should be an alias for Country."""
|
||||
assert CountryParam is Country
|
||||
|
||||
def test_groups_property(self):
|
||||
"""Country should expose membership groups."""
|
||||
c = Country("US")
|
||||
groups = c.groups
|
||||
assert isinstance(groups, list)
|
||||
assert "G7" in groups
|
||||
assert "G20" in groups
|
||||
assert "NATO" in groups
|
||||
assert "OECD" in groups
|
||||
# US is not a member of these
|
||||
assert "EU" not in groups
|
||||
assert "OPEC" not in groups
|
||||
|
||||
def test_groups_empty_for_non_member(self):
|
||||
"""Countries not in any tracked group should have empty groups list."""
|
||||
# Most small countries won't be in any group
|
||||
c = Country("Fiji")
|
||||
assert c.groups == []
|
||||
|
||||
def test_is_member_of(self):
|
||||
"""is_member_of should check group membership."""
|
||||
c = Country("Germany")
|
||||
assert c.is_member_of("G7") is True
|
||||
assert c.is_member_of("EU") is True
|
||||
assert c.is_member_of("NATO") is True
|
||||
assert c.is_member_of("OECD") is True
|
||||
assert c.is_member_of("OPEC") is False
|
||||
assert c.is_member_of("BRICS") is False
|
||||
|
||||
def test_is_member_of_case_insensitive(self):
|
||||
"""is_member_of should be case-insensitive."""
|
||||
c = Country("FR")
|
||||
assert c.is_member_of("g7") is True
|
||||
assert c.is_member_of("G7") is True
|
||||
assert c.is_member_of("Eu") is True
|
||||
|
||||
def test_opec_members(self):
|
||||
"""OPEC members should be correctly identified."""
|
||||
opec_countries = ["SA", "AE", "KW", "IQ", "IR", "VE", "NG"]
|
||||
for code in opec_countries:
|
||||
c = Country(code)
|
||||
assert c.is_member_of("OPEC") is True, f"{code} should be OPEC member"
|
||||
|
||||
def test_brics_members(self):
|
||||
"""BRICS members should be correctly identified."""
|
||||
brics_countries = ["BR", "RU", "IN", "CN", "ZA"]
|
||||
for code in brics_countries:
|
||||
c = Country(code)
|
||||
assert c.is_member_of("BRICS") is True, f"{code} should be BRICS member"
|
||||
|
||||
def test_eu_members(self):
|
||||
"""EU members should be correctly identified."""
|
||||
eu_countries = ["DE", "FR", "IT", "ES", "NL", "BE", "PL"]
|
||||
for code in eu_countries:
|
||||
c = Country(code)
|
||||
assert c.is_member_of("EU") is True, f"{code} should be EU member"
|
||||
c = Country("GB")
|
||||
assert c.is_member_of("EU") is False
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"input_name, expected_alpha2",
|
||||
[
|
||||
("Curacao", "CW"),
|
||||
("Reunion", "RE"),
|
||||
("Aland Islands", "AX"),
|
||||
("Saint Barthelemy", "BL"),
|
||||
("Turkiye", "TR"),
|
||||
("Turkey", "TR"),
|
||||
("Cote d'Ivoire", "CI"),
|
||||
],
|
||||
)
|
||||
def test_ascii_normalized_names(self, input_name, expected_alpha2):
|
||||
"""ASCII variants of accented country names should resolve correctly."""
|
||||
c = Country(input_name)
|
||||
assert c.alpha_2 == expected_alpha2
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"input_name, expected_alpha2",
|
||||
[
|
||||
("Curaçao", "CW"),
|
||||
("Réunion", "RE"),
|
||||
("Åland Islands", "AX"),
|
||||
("Saint Barthélemy", "BL"),
|
||||
("Türkiye", "TR"),
|
||||
("Côte d'Ivoire", "CI"),
|
||||
],
|
||||
)
|
||||
def test_accented_names(self, input_name, expected_alpha2):
|
||||
"""Accented country names should resolve correctly."""
|
||||
c = Country(input_name)
|
||||
assert c.alpha_2 == expected_alpha2
|
||||
|
||||
Reference in New Issue
Block a user