#!/usr/bin/env python3 """Reproducible public-data snapshot for ASCEND (all World Bank economies); no synthetic fallback. Usage: python3 refresh_data.py # rebuild data.json from live public APIs python3 refresh_data.py --validate # re-validate the committed data.json Optional: if geo/access_summary.json exists (produced by access_pipeline.py), its modelled geographic-access indicators are merged as a separate, clearly labelled group. """ from pathlib import Path import argparse, concurrent.futures, datetime, gzip, html, json, math, re, time import urllib.request, urllib.parse BASE = Path(__file__).resolve().parent DEST = BASE / 'data.json' RAW = BASE / 'data_raw' ACCESS = BASE / 'geo' / 'access_summary.json' YEARS = (2000, 2024) VERSION = '2.0.0' WB_NOTE = 'World Bank open data (CC BY 4.0); indicator attribution retained. Check dataset terms before redistribution.' WHO_NOTE = 'WHO Global Health Observatory open data (CC BY-NC-SA 3.0 IGO); attribution retained.' EPMC_NOTE = 'Europe PMC bibliographic service. Article and abstract rights remain with their respective holders; linked records retain identifiers and attribution.' CT_NOTE = 'ClinicalTrials.gov public registry data; sponsor submissions are not independent verification of efficacy. See ClinicalTrials.gov terms and conditions.' GOV_NOTE = 'Official Brazilian Ministry of Health public API. Connector proof only; schema field names retained, no individual patient records distributed.' # key -> definition. `source` is 'worldbank' or 'who'. `scale` guides chart axes. `target` is the # Lancet Commission on Global Surgery 2030 target where one exists (Meara et al., Lancet 2015). INDICATORS = { 'spending': {'id':'SH.XPD.CHEX.PC.CD','source':'worldbank','label':'Current health expenditure per capita','short':'Health spending / person','unit':'current US$','decimals':0,'scale':'log','better':'higher'}, 'spending_ppp': {'id':'SH.XPD.CHEX.PP.CD','source':'worldbank','label':'Current health expenditure per capita, PPP','short':'Health spending / person (PPP)','unit':'international $','decimals':0,'scale':'log','better':'higher'}, 'physicians': {'id':'SH.MED.PHYS.ZS','source':'worldbank','label':'Physicians','short':'Physicians','unit':'per 1,000 people','decimals':2,'scale':'linear','better':'higher'}, 'life': {'id':'SP.DYN.LE00.IN','source':'worldbank','label':'Life expectancy at birth','short':'Life expectancy','unit':'years','decimals':1,'scale':'linear','better':'higher'}, 'oop': {'id':'SH.XPD.OOPC.CH.ZS','source':'worldbank','label':'Out-of-pocket expenditure','short':'Out-of-pocket share','unit':'% of current health expenditure','decimals':1,'scale':'linear','better':'lower'}, 'population': {'id':'SP.POP.TOTL','source':'worldbank','label':'Population, total','short':'Population','unit':'people','decimals':0,'scale':'log','better':None}, 'sao': {'id':'SH.MED.SAOP.P5','source':'worldbank','label':'Specialist surgical workforce','short':'Surgical workforce (SAO)','unit':'per 100,000 population','decimals':1,'scale':'linear','better':'higher','target':20,'targetLabel':'LCoGS 2030 target: at least 20 per 100,000'}, 'volume': {'id':'SH.SGR.PROC.P5','source':'worldbank','label':'Number of surgical procedures','short':'Surgical volume','unit':'per 100,000 population','decimals':0,'scale':'linear','better':'higher','target':5000,'targetLabel':'LCoGS 2030 target: at least 5,000 per 100,000'}, 'catastrophic': {'id':'SH.SGR.CRSK.ZS','source':'worldbank','label':'Risk of catastrophic expenditure for surgical care','short':'Catastrophic cost risk (surgery)','unit':'% of people at risk','decimals':1,'scale':'linear','better':'lower','target':0,'targetLabel':'LCoGS 2030 target: 100% protection (0% at risk)'}, 'impoverishing':{'id':'SH.SGR.IRSK.ZS','source':'worldbank','label':'Risk of impoverishing expenditure for surgical care','short':'Impoverishing cost risk (surgery)','unit':'% of people at risk','decimals':1,'scale':'linear','better':'lower','target':0,'targetLabel':'LCoGS 2030 target: 100% protection (0% at risk)'}, 'uhc': {'id':'UHC_INDEX_REPORTED','source':'who','label':'UHC service coverage index (SDG 3.8.1)','short':'UHC service coverage index','unit':'index 0–100','decimals':0,'scale':'linear','better':'higher'}, 'cata10': {'id':'FINPROTECTION_CATA_TOT_10_POP','source':'who','dim1':'RESIDENCEAREATYPE_TOTL','label':'Population with household health spending above 10% of total consumption or income (SDG 3.8.2)','short':'Catastrophic health spending (>10%)','unit':'% of population','decimals':1,'scale':'linear','better':'lower'}, } GROUPS = [ {'id':'context','label':'System context','indicators':['spending','spending_ppp','physicians','life','oop','population'],'note':'National averages; contextual observations, not measures of surgical access.'}, {'id':'surgical','label':'Surgical system (LCoGS indicators)','indicators':['sao','volume','catastrophic','impoverishing'],'note':'Four of the six Lancet Commission on Global Surgery indicators as compiled in World Development Indicators. Sparse and irregular reporting years; the latest value can be a decade old.'}, {'id':'coverage','label':'Coverage & financial protection','indicators':['uhc','cata10'],'note':'SDG 3.8 monitoring indicators from WHO and World Bank.'}, ] ACCESS_GROUP = {'id':'access','label':'Geographic access (modelled)','indicators':['access2h','access1h','traveltime'],'note':'Population-weighted travel time to the nearest healthcare facility, computed by ASCEND from the Malaria Atlas Project 2019 travel-time surface and WorldPop 2020 counts. Any facility, not Bellwether surgical capacity; a modelled upper bound on surgical access.'} def utcnow(): return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec='seconds') def fetch(job): name,url,note = job record = {'id':name,'name':name.replace('_',' ').title(),'url':url,'accessedAt':utcnow(),'licenseNote':note} attempts=6 if 'ghoapi' in url else 4 for attempt in range(attempts): try: req = urllib.request.Request(url,headers={'User-Agent':'ASCEND-research-prototype/2.0','Accept':'application/json','Accept-Encoding':'gzip'}) with urllib.request.urlopen(req,timeout=90) as response: body=response.read() if body[:2] == b'\x1f\x8b': body=gzip.decompress(body) payload=json.loads(body.decode('utf-8')) record.update(status='success',httpStatus=response.status,bytes=len(body),attempts=attempt+1) (RAW/f'{name}.json').write_text(json.dumps(payload,ensure_ascii=False)) return name,payload,record except Exception as exc: record.update(status='failed',error=str(exc)[:300],attempts=attempt+1) time.sleep(2+3*attempt) cached=RAW/f'{name}.json' if cached.exists(): try: payload=json.loads(cached.read_text()) record.update(status='cached',note='Live retrieval failed; the previously retrieved raw response in data_raw/ was reused unchanged. Timestamp is the file modification time.', accessedAt=datetime.datetime.fromtimestamp(cached.stat().st_mtime,datetime.timezone.utc).isoformat(timespec='seconds')) return name,payload,record except Exception: pass return name,None,record def clean_text(text): return html.unescape(re.sub('<[^>]+>', ' ', text or '')).strip() def wb_rows(payload): return payload[1] if isinstance(payload,list) and len(payload)>1 and isinstance(payload[1],list) else [] def to_series(values): """{year:value} -> sorted [[year,value],...] with nulls dropped and finite check.""" out=[] for year in range(YEARS[0],YEARS[1]+1): v=values.get(year) if v is None: continue v=float(v) if not math.isfinite(v): continue out.append([year,round(v,4)]) return out def validate(data): assert data['kind'] == 'observed/retrieved' assert data['version'] == VERSION ids=[c['id'] for c in data['countries']] assert len(ids)==len(set(ids)) and len(ids) >= 190, len(ids) keys=set(data['indicators']) for g in data['indicatorGroups']: assert set(g['indicators']) <= keys observations=0; nonempty=0 for c in data['countries']: assert re.fullmatch(r'[A-Z]{3}',c['id']) and c['name'] and c['region'] and c['income'] assert c['lat'] is None or -90 <= c['lat'] <= 90 assert c['lon'] is None or -180 <= c['lon'] <= 180 assert set(c['series']) == keys for key,series in c['series'].items(): years=[y for y,_ in series] assert years==sorted(years) and len(set(years))==len(years) assert all(YEARS[0] <= y <= YEARS[1] for y in years) assert all(isinstance(v,(int,float)) and math.isfinite(v) for _,v in series) observations+=len(series); nonempty+=bool(series) for source in data['sources']: assert source['url'].startswith('https://') and source['accessedAt'] and source['licenseNote'] assert data['evidence']['classification'] == 'related literature; not model effect estimates' for p in data['evidence']['publications']: assert p['title'] and p['url'].startswith('https://') assert data['trials']['classification'] == 'registered studies; not evidence of efficacy or surgery relevance' for p in data['trials']['studies']: assert p['nctId'].startswith('NCT') and p['url'].startswith('https://clinicaltrials.gov/study/') for c in data['connectorProofs']: assert 'schemaFields' in c and 'patientData' not in c coverage={k:sum(1 for c in data['countries'] if c['series'][k]) for k in keys} return {'countries':len(ids),'indicators':len(keys),'series':len(ids)*len(keys),'nonEmptySeries':nonempty,'observations':observations, 'countriesWithAnyValue':coverage,'publications':len(data['evidence']['publications']),'trials':len(data['trials']['studies'])} def build(): RAW.mkdir(parents=True,exist_ok=True) jobs=[('worldbank_countries','https://api.worldbank.org/v2/country?format=json&per_page=400',WB_NOTE)] for key,meta in INDICATORS.items(): if meta['source']=='worldbank': jobs.append((f'worldbank_{key}',f"https://api.worldbank.org/v2/country/all/indicator/{meta['id']}?format=json&date={YEARS[0]}:{YEARS[1]}&per_page=20000",WB_NOTE)) else: jobs.append((f'who_{key}',f"https://ghoapi.azureedge.net/api/{meta['id']}",WHO_NOTE)) queries={ 'europe_pmc_access':'(TITLE_ABS:"surgical access" OR TITLE_ABS:"access to surgery" OR TITLE_ABS:"access to surgical care") AND (TITLE_ABS:"global surgery" OR TITLE_ABS:"health system" OR TITLE_ABS:"low- and middle-income")', 'europe_pmc_capacity':'(TITLE_ABS:"surgical capacity" OR TITLE_ABS:"surgical workforce") AND (TITLE_ABS:financing OR TITLE_ABS:access OR TITLE_ABS:"health system")', 'europe_pmc_indicators':'TITLE_ABS:"Lancet Commission on Global Surgery" AND TITLE_ABS:indicators' } for key,q in queries.items(): jobs.append((key,'https://www.ebi.ac.uk/europepmc/webservices/rest/search?'+urllib.parse.urlencode({'query':q,'format':'json','pageSize':6,'resultType':'core','sort':'CITED desc'}),EPMC_NOTE)) jobs.extend([ ('demas_cnes','https://apidadosabertos.saude.gov.br/cnes/estabelecimentos?codigo_uf=35&limit=1&offset=0',GOV_NOTE), ('demas_sih','https://apidadosabertos.saude.gov.br/assistencia-a-saude/sih-procedimentos-hospitalares?nu_ano=2023&limit=1&offset=0',GOV_NOTE), ('clinical_trials','https://clinicaltrials.gov/api/v2/studies?'+urllib.parse.urlencode({'query.term':'AREA[BriefTitle]("surgical access" OR "access to surgery" OR "access to surgical care" OR "global surgery" OR "surgical capacity" OR "surgical workforce" OR "surgical care delivery" OR "surgical outreach") AND AREA[StudyType]INTERVENTIONAL','pageSize':6,'format':'json'}),CT_NOTE) ]) responses={}; sources=[] with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool: for name,payload,record in pool.map(fetch,jobs): responses[name]=payload;sources.append(record) print(name,record['status'],record.get('bytes',''),flush=True) (RAW/'request_manifest.json').write_text(json.dumps(sources,indent=2)) metadata=wb_rows(responses['worldbank_countries']) if not metadata: raise RuntimeError('Country metadata unavailable; no substitute snapshot was fabricated. Inspect raw request manifest.') economies=[m for m in metadata if m.get('region',{}).get('value') and m['region']['value']!='Aggregates'] # indicator tables keyed by iso3 tables={} for key,meta in INDICATORS.items(): if meta['source']=='worldbank': rows=wb_rows(responses.get(f'worldbank_{key}')) table={} for row in rows: iso=row.get('countryiso3code') if not iso or row.get('value') is None: continue table.setdefault(iso,{})[int(row['date'])]=row['value'] tables[key]=(table,'success' if rows else 'unavailable',f'worldbank_{key}') else: payload=responses.get(f'who_{key}') or {} table={} for row in payload.get('value',[]): if row.get('SpatialDimType')!='COUNTRY' or row.get('NumericValue') is None: continue if (row.get('Dim1') or None)!=meta.get('dim1'): continue table.setdefault(row['SpatialDim'],{})[int(row['TimeDim'])]=row['NumericValue'] tables[key]=(table,'success' if table else 'unavailable',f'who_{key}') access=None if ACCESS.exists(): access=json.loads(ACCESS.read_text()) for k,meta in access['indicators'].items(): INDICATORS[k]={**meta,'source':'ascend-geo'} GROUPS.append(ACCESS_GROUP) sources.append({'id':'ascend_geo_access','name':'ASCEND geographic-access pipeline','url':access['provenance']['methodUrl'],'accessedAt':access['generatedAt'],'licenseNote':access['provenance']['licenseNote'],'status':'computed','note':access['provenance']['summary']}) for s in access['provenance']['inputs']: sources.append({'id':s['id'],'name':s['name'],'url':s['url'],'accessedAt':access['generatedAt'],'licenseNote':s['licenseNote'],'status':'success'}) countries=[] for meta in sorted(economies,key=lambda m:m['name']): iso=meta['id'] c={'id':iso,'iso2':meta.get('iso2Code'),'name':meta['name'],'region':meta['region']['value'].strip(),'income':meta['incomeLevel']['value'].strip(), 'capital':meta.get('capitalCity') or None, 'lat':float(meta['latitude']) if meta.get('latitude') else None,'lon':float(meta['longitude']) if meta.get('longitude') else None, 'series':{}} for key in INDICATORS: if key in tables: c['series'][key]=to_series(tables[key][0].get(iso,{})) elif access: v=access['countries'].get(iso,{}).get(key) c['series'][key]=[[access['referenceYear'],round(float(v),4)]] if v is not None and math.isfinite(float(v)) else [] countries.append(c) indicator_defs={} for key,meta in INDICATORS.items(): status=tables[key][1] if key in tables else ('computed' if access else 'unavailable') indicator_defs[key]={**meta,'sourceId':tables[key][2] if key in tables else 'ascend_geo_access','retrievalStatus':status, 'countriesWithData':sum(1 for c in countries if c['series'][key])} publications=[]; seen=set() for key,q in queries.items(): response=responses[key] or {} for p in response.get('resultList',{}).get('result',[]): ident=p.get('doi') or p.get('pmid') or p['id'] if ident in seen: continue seen.add(ident) source=p.get('source','MED');id_=p['id'] publications.append({'id':id_,'title':clean_text(p.get('title')),'authors':p.get('authorString',''),'year':int(p['pubYear']) if p.get('pubYear') else None, 'doi':p.get('doi'),'pmid':p.get('pmid'),'url':f'https://europepmc.org/article/{source}/{id_}', 'abstract':clean_text(p.get('abstractText')),'journal':p.get('journalInfo',{}).get('journal',{}).get('title',''), 'citedBy':p.get('citedByCount'), 'classification':'related literature; not model effect estimates','query':q,'sourceId':key}) studies=[] for row in (responses['clinical_trials'] or {}).get('studies',[]): p=row['protocolSection']; ident=p['identificationModule']; design=p.get('designModule',{}); status=p.get('statusModule',{}) locations=p.get('contactsLocationsModule',{}).get('locations',[]) countries_in=sorted({loc.get('country') for loc in locations if loc.get('country')}) studies.append({'nctId':ident['nctId'],'title':ident.get('briefTitle',''),'status':status.get('overallStatus',''), 'phase':design.get('phases',[]),'studyType':design.get('studyType',''),'briefSummary':p.get('descriptionModule',{}).get('briefSummary',''), 'url':'https://clinicaltrials.gov/study/'+ident['nctId'],'countries':countries_in,'locations':[{k:loc[k] for k in ['facility','city','country'] if k in loc} for loc in locations[:6]], 'sourceId':'clinical_trials','lastUpdate':status.get('lastUpdatePostDateStruct',{}).get('date')}) proofs=[] for name,key in [('demas_cnes','estabelecimentos'),('demas_sih','sih_procedimentos_hospitalares')]: response=responses[name] or {}; rows=response.get(key,[]) proofs.append({'id':name,'label':'CNES establishment schema' if name=='demas_cnes' else 'SIH hospital-procedure schema', 'status':'success' if rows else 'unavailable','schemaFields':sorted(rows[0].keys()) if rows else [],'sampleRowsRetrieved':len(rows), 'sourceId':name,'note':'Connectivity and schema proof only (example national registry). One response row does not establish data completeness, historical coverage, surgical relevance, or analytic validity.'}) data={'version':VERSION,'kind':'observed/retrieved','accessedAt':utcnow(),'yearRange':list(YEARS), 'description':'All World Bank economies: contextual health-system indicators, Lancet Commission on Global Surgery indicators, SDG 3.8 coverage and protection indicators'+(', and ASCEND-modelled geographic access' if access else '')+'. Observational comparisons do not establish intervention effects.', 'indicatorGroups':GROUPS,'indicators':indicator_defs,'countries':countries, 'sources':sources,'apiRegistry':[{k:v for k,v in r.items() if k not in ['bytes']} for r in sources], 'evidence':{'classification':'related literature; not model effect estimates','selection':'Three explicit Europe PMC queries on surgical access, capacity and LCoGS indicators, six most-cited results per query, deduplicated. Convenience discovery set, not a systematic review. No relevance or risk-of-bias adjudication.','status':'success' if publications else 'unavailable','publications':publications}, 'trials':{'classification':'registered studies; not evidence of efficacy or surgery relevance','selection':'First six ClinicalTrials.gov API results for interventional studies with surgical access, capacity, workforce or delivery terms in the title. No country filter; registry coverage of health-system interventions is thin.','status':'success' if studies else 'unavailable','studies':studies}, 'connectorProofs':proofs, 'limitations':['All World Bank economies are included; small states and territories often lack observations and remain visibly missing.', 'Latest values can refer to different years; the observation year is always shown next to the value.', 'Expenditure is shown in current US dollars and, separately, in PPP international dollars; neither is inflation-adjusted across years.', 'Missing values remain null; no interpolation, imputation or carry-forward in fixed-year mode.', 'Lancet Commission on Global Surgery indicators are compiled from heterogeneous national reports with irregular years and definitions.', 'National ecological indicators cannot estimate effects of a surgical intervention.', 'Bibliographic and registry retrieval is not a systematic evidence synthesis.', 'No patient-level records are included in this distributable snapshot.']+ (['Geographic access is modelled from a travel-time surface to any healthcare facility and 2020 population counts; it is not the LCoGS Bellwether 2-hour indicator and overstates surgical access.'] if access else [])} data['validation']=validate(data) DEST.parent.mkdir(parents=True,exist_ok=True) DEST.write_text(json.dumps(data,ensure_ascii=False,separators=(',',':'))) print(json.dumps(data['validation'],indent=2)) print('bytes',DEST.stat().st_size) if __name__ == '__main__': parser=argparse.ArgumentParser();parser.add_argument('--validate',action='store_true');args=parser.parse_args() if args.validate: print(json.dumps(validate(json.loads(DEST.read_text())),indent=2)) else: build()