#!/usr/bin/env python3
"""A08: bounded Europe PMC corpus audit and artificial-data design analysis.
Raw third-party documents remain in a caller-selected cache outside public assets.
No human participants or agent performance are inferred from these tools.
"""
import argparse, concurrent.futures, datetime, hashlib, json, math, pathlib, re, urllib.parse, urllib.request, xml.etree.ElementTree as ET
from collections import Counter
import numpy as np
from scipy import stats

QUERY = '(TITLE_ABS:"school leadership" OR TITLE_ABS:"principal leadership") AND FIRST_PDATE:[2017-01-01 TO 2026-10-15] AND OPEN_ACCESS:Y AND IN_EPMC:Y NOT (TITLE_ABS:nurs* OR TITLE_ABS:"medical school" OR TITLE_ABS:"medical education")'
BASE='https://www.ebi.ac.uk/europepmc/webservices/rest/'
XLINK='{http://www.w3.org/1999/xlink}href'
# Relevance is an explicitly human-authored screening specification, not an AI effect estimate.
# The three exclusions are adjacent professional training/caregiver contexts, outside the benchmark target.
EXCLUDED_RELEVANCE={
 'PMC13472988':'University physical-education trainee monitoring; outside district decision-support target.',
 'PMC13261048':'Tele-delivered caregiver autism coaching; outside school/district governance target.',
 'PMC10081929':'Interview-based policy review retained only for contextual reading, excluded from primary research benchmark construction.'
}

def write_json(path, obj):
 pathlib.Path(path).write_text(json.dumps(obj,ensure_ascii=False,indent=2)+'\n')

def sha(b): return hashlib.sha256(b).hexdigest()
def text(e): return '' if e is None else ' '.join(''.join(e.itertext()).split())
def retrieve(url, path):
 p=pathlib.Path(path)
 if p.exists(): return p.read_bytes()
 req=urllib.request.Request(url,headers={'User-Agent':'JournalOS-A08-research-audit/1.0'})
 with urllib.request.urlopen(req,timeout=55) as r: b=r.read()
 p.write_bytes(b); return b

def rights(root):
 perms=root.find('./front/article-meta/permissions')
 urls=[]; article_urls=[]
 if perms is None: return {'status':'unresolved','licenseName':None,'articleLicenseUrls':[],'allLicenseUrls':[]}
 for lic in perms.findall('license'):
  t=text(lic).lower(); links=[e.attrib.get(XLINK,e.attrib.get('href','')) for e in lic.iter()]
  links += [text(e) for e in lic.iter() if e.tag.endswith('license_ref')]
  links=[u for u in links if 'creativecommons.org/' in u]
  urls+=links
  # A license that only waives data rights cannot license the article.
  data_only=('data made available' in t or 'data included' in t or 'associated data' in t) and all('/publicdomain/zero/' in u for u in links)
  if not data_only:
   article_urls += [u for u in links if not ('/publicdomain/zero/' in u and 'data' in t and ('waiver' in t or any('/licenses/' in v for v in links)))]
 urls=sorted(set(u.rstrip('/')+'/' for u in urls))
 article_urls=sorted(set(u.rstrip('/')+'/' for u in article_urls))
 bad=any(re.search(r'/licenses/[^/]*(?:nc|nd)',u) for u in article_urls) or bool(re.search(r'no commercial re[ -]?use|non[ -]?commercial|no[ -]?derivatives',text(perms).lower()))
 eligible=[u for u in article_urls if re.search(r'/licenses/by(?:-sa)?/(?:[1-4]\.0)/',u) or '/publicdomain/zero/1.0/' in u]
 if bad: status='restricted-or-conflicting'
 elif len(eligible)==1: status='eligible-article-license'
 elif len(eligible)>1: status='multiple-article-licenses-needs-review'
 else: status='unresolved'
 name=None
 if len(eligible)==1:
  m=re.search(r'/licenses/(by(?:-sa)?)/([1-4]\.0)/',eligible[0])
  name=('CC '+m.group(1).upper()+' '+m.group(2)) if m else 'CC0 1.0'
 return {'status':status,'licenseName':name,'articleLicenseUrls':article_urls,'allLicenseUrls':urls,'copyrightNotice':text(perms.find('copyright-statement'))}

def authors(root):
 out=[]
 for c in root.findall('./front/article-meta/contrib-group/contrib'):
  if c.get('contrib-type','author')!='author': continue
  # Some publisher JATS omits contrib-type for both authors and editors.
  # An explicit editorial role still excludes that contributor from authorship;
  # CRediT 'Writing - review & editing' is an author role, not an editor role.
  roles=[text(r).strip().lower() for r in c.findall('role')]
  if roles and all(re.fullmatch(r'(?:(?:academic|handling|associate|guest) )?editor',r) for r in roles): continue
  n=c.find('name')
  if n is not None: out.append(' '.join(filter(None,[text(n.find('given-names')),text(n.find('surname'))])))
  elif c.find('collab') is not None: out.append(text(c.find('collab')))
 return out

def audit_record(rec, raw):
 root=ET.fromstring(raw); meta=root.find('./front/article-meta'); pmcid=rec['pmcid']
 dois=[text(x).lower().removeprefix('https://doi.org/') for x in meta.findall('article-id') if x.get('pub-id-type')=='doi']
 core_doi=rec.get('doi','').lower(); type_=root.get('article-type','')
 related=[{'type':x.get('related-article-type'),'href':x.get(XLINK),'idType':x.get('ext-link-type')} for x in meta.findall('related-article')]
 corrections=rec.get('commentCorrectionList',{})
 flags=[]
 if rec.get('isRetracted')=='Y': flags.append('core:isRetracted')
 pubtypes=rec.get('pubTypeList',{}).get('pubType',[])
 for t in pubtypes:
  if any(s in t.lower() for s in ['retract','expression of concern','withdraw']):flags.append('pubType:'+t)
 if re.match(r'^(retract|withdraw|expression of concern)',rec.get('title',''),re.I): flags.append('notice-title')
 if type_ in ('retraction','expression-of-concern'):flags.append('jats:'+type_)
 # Related records can be corrections rather than retractions; preserve relation semantics.
 relations_blob=json.dumps({'related':related,'corrections':corrections}).lower()
 if any(s in relations_blob for s in ['retracted-article','retractionof','retractionin','expressionofconcern','withdrawn']): flags.append('status-relation')
 rr=rights(root); body=root.find('body'); paragraphs=0 if body is None else len(body.findall('.//p'))
 relevant=pmcid not in EXCLUDED_RELEVANCE
 research=type_=='research-article' and not any('preprint' in t.lower() for t in pubtypes)
 reasons=[]
 if not relevant:reasons.append('relevance-or-design-exclusion')
 if not research:reasons.append('not-primary-research-article')
 if rr['status']!='eligible-article-license':reasons.append('article-rights-not-cleared')
 if not paragraphs:reasons.append('full-text-body-unavailable')
 if flags:reasons.append('status-signal-needs-review')
 if core_doi and core_doi not in dois:reasons.append('doi-conflict')
 return {'pmcid':pmcid,'pmid':rec.get('pmid'),'doi':rec.get('doi'),'sourceVersionDois':dois,'title':text(meta.find('./title-group/article-title')),'authors':authors(root),'journal':text(root.find('./front/journal-meta/journal-title-group/journal-title')),'publisher':text(root.find('./front/journal-meta/publisher/publisher-name')),'firstPublicationDate':rec.get('firstPublicationDate'),'publicationDatePrecisionNote':'Europe PMC firstPublicationDate may normalize incomplete dates; consult JATS dates for an exact day.','jatsPublicationDates':[{'type':d.get('pub-type') or d.get('publication-format'),'date':{'year':text(d.find('year')),'month':text(d.find('month')),'day':text(d.find('day'))}} for d in meta.findall('pub-date')],'articleVersion':text(meta.find('article-version')) or None,'versionIdentifiers':[{"type":v.get('pub-id-type'),"value":text(v)} for v in meta.findall('article-id')],'versionHistory':[{"type":v.get('event-type'),"identifiers":[{"type":z.get('pub-id-type'),"value":text(z)} for z in v.findall('article-id')],"version":text(v.find('article-version')),"date":{"year":text(v.find('./pub-date/year')),"month":text(v.find('./pub-date/month')),"day":text(v.find('./pub-date/day'))}} for v in meta.findall('./pub-history/event') if v.findall('article-id')],'articleType':type_,'publicationTypes':pubtypes,'fullTextUrl':BASE+pmcid+'/fullTextXML','sourceUrl':'https://europepmc.org/articles/'+pmcid,'sourceTextSha256':sha(raw),'sourceBytes':len(raw),'bodyParagraphs':paragraphs,'license':rr,'coreLicenseSummary':rec.get('license'),'noticeScreen':{'flags':flags,'relatedArticles':related,'commentCorrections':corrections,'interpretation':'No screened signal is not a certification of absence of correction, retraction or integrity concerns.'},'relevanceScreen':{'include':relevant,'reason':EXCLUDED_RELEVANCE.get(pmcid,'School/district leadership or adjacent implementation/wellbeing decision context; final case selection requires independent human review.')},'candidateStatus':'source-eligible-for-benchmark-construction' if not reasons else 'excluded-or-unresolved','exclusionReasons':reasons,'versionIdentity':pmcid+'; SHA-256 '+sha(raw),'verificationDate':'2026-10-15'}

def corpus_audit(cache_dir, output, download=False):
 cache=pathlib.Path(cache_dir);cache.mkdir(parents=True,exist_ok=True)
 url=BASE+'search?'+urllib.parse.urlencode({'query':QUERY,'format':'json','resultType':'core','pageSize':36})
 p=cache/'search-core.json'
 if not p.exists() and not download: raise ValueError('Missing frozen search response; use --download for a fresh explicitly dated audit.')
 search_bytes=retrieve(url,p); search=json.loads(search_bytes); recs=search['resultList']['result']
 if len(recs)!=36: raise ValueError('Expected exactly the first 36 results; review and amend the audit for a different response.')
 def one(rec):
  try:
   path=cache/(rec['pmcid']+'.xml')
   if not path.exists() and not download: raise ValueError('Missing cached JATS')
   raw=retrieve(BASE+rec['pmcid']+'/fullTextXML',path)
   return audit_record(rec,raw)
  except Exception as e:return {'pmcid':rec.get('pmcid'),'title':rec.get('title'),'candidateStatus':'unresolved-fetch-or-parse','error':type(e).__name__+': '+str(e)}
 with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool: rows=list(pool.map(one,recs))
 counts=Counter(r['candidateStatus'] for r in rows)
 licenses=Counter(r.get('license',{}).get('status','unresolved') for r in rows)
 types=Counter(r.get('articleType','unresolved') for r in rows)
 ids=[r['pmcid'] for r in rows]; doi=[r.get('doi','').lower() for r in rows if r.get('doi')]
 out={'analysis':'Europe PMC exact-query bounded corpus feasibility audit','verificationDate':'2026-10-15','actualHumanParticipants':0,'query':QUERY,'searchUrl':url,'databaseHitCount':search['hitCount'],'auditedRecords':len(rows),'selection':'First 36 records of one default relevance-ranked response, not all hits and not a systematic review. Ranking may change; frozen response hash preserves this snapshot.','searchResponseSha256':sha(search_bytes),'deduplication':{'uniquePmcids':len(set(ids)),'uniqueNonemptyDois':len(set(doi)),'duplicatePmcidCount':len(ids)-len(set(ids)),'duplicateDoiCount':len(doi)-len(set(doi)),'scope':'This frozen corpus only, not the journal catalog.'},'counts':dict(counts),'licenseCounts':dict(licenses),'articleTypeCounts':dict(types),'records':rows,'limits':['No human benchmark was run.','No source-retrieval or agent answer accuracy is inferred from a full-text download.','Article-level permissions do not establish rights to every third-party figure or dataset; these are not redistributed.','No agent authorship, human peer review or clinical/educational effectiveness is established.']}
 write_json(output,out);return out

def blocked_estimate(rows):
 # One complete mean score per person, assigned within exactly four prespecified strata.
 ids=[x['participant_id'] for x in rows]
 if len(ids)!=len(set(ids)): raise ValueError('Duplicate participant')
 strata=sorted({x['stratum'] for x in rows})
 if len(strata)!=4:raise ValueError('Expected four strata')
 effects=[]; variances=[]
 for s in strata:
  group=[x for x in rows if x['stratum']==s]
  vals=[]
  for arm in [0,1]:
   a=np.array([float(x['mean_score']) for x in group if x['arm']==arm])
   if len(a)<2 or not np.isfinite(a).all() or ((a<0)|(a>24)).any():raise ValueError('Need at least two finite 0–24 means per arm/stratum')
   vals.append(a)
  if sum(x['arm'] in [0,1] for x in group)!=len(group):raise ValueError('Arm must be 0 or 1')
  effects.append(vals[1].mean()-vals[0].mean());variances.append(vals[1].var(ddof=1)/len(vals[1])+vals[0].var(ddof=1)/len(vals[0]))
 effect=float(np.mean(effects));se=float(np.sqrt(sum(variances)/16))
 # Welch–Satterthwaite degrees of freedom for eight independent cell variance terms.
 terms=[]
 for s in strata:
  for arm in [0,1]:
   a=np.array([float(x['mean_score']) for x in rows if x['stratum']==s and x['arm']==arm])
   terms.append((float(a.var(ddof=1)/(16*len(a))),len(a)-1))
 df=float(sum(v for v,d in terms)**2/sum(v*v/d for v,d in terms)) if any(v for v,d in terms) else 0
 if se<=0:raise ValueError('Zero variance; no conventional t inference')
 q=float(stats.t.ppf(.975,df));return {'effect_points':effect,'standard_error':se,'df':df,'ci95':[effect-q*se,effect+q*se],'p_two_sided':float(2*stats.t.sf(abs(effect/se),df)),'estimand':'Equal-stratum difference in mean participant 12-case evidence-fidelity score','inference':'Complete-case Neyman variance and Welch–Satterthwaite t inference; not the prespecified imputed confirmatory analysis if outcomes are missing.'}

def simulate(output, reps=10000, seed=20261015):
 if reps<100:raise ValueError('At least 100 replicates required')
 rng=np.random.default_rng(seed);rows=[]
 # Marginal case variance 1. Case reliability/correlation creates person-mean variance.
 for n in [80,160,240]:
  if n%8:raise ValueError('Balanced four-stratum design requires a multiple of eight')
  for rho in [.10,.30,.50]:
   each=n//8;mean_var=rho+(1-rho)/12;df=n-8
   # 4 strata x 2 arms, independent subject means with no ceiling or dropout.
   control=rng.normal(0,math.sqrt(mean_var),(reps,4,each))
   treatment=rng.normal(0,math.sqrt(mean_var),(reps,4,each))
   delta=(treatment.mean(2)-control.mean(2)).mean(1)
   v1=treatment.var(2,ddof=1)/(16*each);v0=control.var(2,ddof=1)/(16*each)
   var=(v1+v0).sum(1);df=var**2/((v1*v1+v0*v0).sum(1)/(each-1))
   se=np.sqrt(var);critical=stats.t.ppf(.975,df)
   null=float(np.mean(abs(delta/se)>critical));power=float(np.mean(abs((delta+.25)/se)>critical))
   rows.append({'plannedParticipants':n,'casesPerParticipant':12,'withinPersonCaseCorrelation':rho,'effect_caseSD':.25,'personMeanVariance_caseSD2':mean_var,'power':power,'powerMonteCarloSE':math.sqrt(power*(1-power)/reps),'nullRejectionRate':null,'nullMonteCarloSE':math.sqrt(null*(1-null)/reps)})
 out={'analysis':'Artificial-data design sensitivity, not human or agent outcomes','seed':seed,'replicatesPerScenario':reps,'scenarios':rows,'assumptions':['Four equal strata; equal treatment/control counts within each.','12 equally weighted cases share an equicorrelated Gaussian person effect; marginal case variance is one.','No source/case-treatment random effects, missing responses, ceiling effects, cross-participant contamination or severity weighting.','Target effect 0.25 marginal case standard deviations is assumed, not estimated from education evidence.','Primary estimand is restricted to the frozen cases.'],'software':{'numpy':np.__version__,'scipy':__import__('scipy').__version__},'actualHumanParticipants':0}
 write_json(output,out);return out

def main():
 p=argparse.ArgumentParser(description=__doc__); sub=p.add_subparsers(dest='command',required=True)
 a=sub.add_parser('audit');a.add_argument('--cache',required=True);a.add_argument('--output',required=True);a.add_argument('--download',action='store_true')
 a=sub.add_parser('simulate');a.add_argument('--output',required=True);a.add_argument('--replicates',type=int,default=10000);a.add_argument('--seed',type=int,default=20261015)
 a=sub.add_parser('estimate-complete');a.add_argument('--input',required=True);a.add_argument('--output',required=True)
 args=p.parse_args()
 if args.command=='audit':out=corpus_audit(args.cache,args.output,args.download)
 elif args.command=='simulate':out=simulate(args.output,args.replicates,args.seed)
 else:
  data=json.loads(pathlib.Path(args.input).read_text())
  if data.get('dataKind') not in ('synthetic-fixture','study-data'):raise ValueError('Explicit provenance label required; this label is not ethics certification.')
  rows=data['rows']
  if data['dataKind']=='study-data' and len(rows)!=160:raise ValueError('Actual confirmatory complete-cohort design expects 160; amend before analyzing another design.')
  if data['dataKind']=='study-data':
   cells=Counter((r['stratum'],r['arm']) for r in rows)
   if len(cells)!=8 or set(cells.values())!={20}:raise ValueError('Actual study requires exactly 20 per arm in each of four strata.')
   if any(r.get('caseCount')!=12 for r in rows):raise ValueError('Actual complete-cohort input requires 12 scored cases per person; missing cases require the specified missing-data analysis.')
  out=blocked_estimate(rows);out['inputSha256']=sha(pathlib.Path(args.input).read_bytes());out['dataKind']=data['dataKind'];write_json(args.output,out)
 print(json.dumps({k:v for k,v in out.items() if k not in ('records','scenarios')},ensure_ascii=False))
if __name__=='__main__':main()