Blame for refresh_data.py

#!/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()