#!/usr/bin/env python3
"""A10 reproducible public-source audit and artificial design sensitivity.
No participant data are included, inferred or transmitted. Python >=3.11.
"""
import argparse,collections,csv,hashlib,io,json,math,pathlib,zipfile
import numpy as np
from scipy import stats


def sha256(p):
 h=hashlib.sha256()
 with open(p,'rb') as f:
  for b in iter(lambda:f.read(1<<20),b''):h.update(b)
 return h.hexdigest()


def zip_rows(p):
 z=zipfile.ZipFile(p)
 names=[n for n in z.namelist() if n.lower().endswith('.csv')]
 if len(names)!=1:raise ValueError('Expected exactly one CSV in source archive')
 return z, csv.DictReader(io.TextIOWrapper(z.open(names[0]),encoding='utf-8-sig'))


def audit(directory,membership=None):
 z,rs=zip_rows(directory)
 required={'NCESSCH','LEAID','UPDATED_STATUS','SCH_TYPE','G_7_OFFERED','G_8_OFFERED','SCHOOL_YEAR'}
 if not required.issubset(rs.fieldnames):raise ValueError('Unexpected directory schema')
 rows=0;ids=set();dupes=0;bad=[];years=collections.Counter();status=collections.Counter();frame={};schools_per_lea=collections.Counter();blank=collections.Counter()
 for r in rs:
  rows+=1;k=r['NCESSCH'];dupes+=k in ids;ids.add(k);years[r['SCHOOL_YEAR']]+=1;status[r['UPDATED_STATUS']]+=1
  if not (len(k)==12 and k.isdigit() and len(r['LEAID'])==7 and r['LEAID'].isdigit()):bad.append(k)
  for c in required:blank[c]+=not bool(r[c].strip())
  if r['UPDATED_STATUS'] in {'1','3','4','5','8'} and r['SCH_TYPE']=='1' and r['G_7_OFFERED']=='Yes' and r['G_8_OFFERED']=='Yes':
   frame[k]=r;schools_per_lea[r['LEAID']]+=1
 z.close()
 out={'analysis':'Executed NCES CCD directory sampling-frame feasibility audit; no participant study','verifiedAt':'2026-10-16','directoryArchiveSha256':sha256(directory),'directoryRows':rows,'uniqueNCESSCH':len(ids),'duplicateNCESSCH':dupes,'malformedIdentifierCount':len(bad),'schoolYears':dict(years),'updatedStatusCounts':dict(sorted(status.items())),'requiredFieldBlankCounts':dict(blank),'candidateFrameDefinition':{'UPDATED_STATUS':['1','3','4','5','8'],'SCH_TYPE':'1','G_7_OFFERED':'Yes','G_8_OFFERED':'Yes'},'candidateSchools':len(frame),'candidateLEAs':len(schools_per_lea),'candidateLEAsWithAtLeastTwoSchools':sum(n>=2 for n in schools_per_lea.values()),'candidateSchoolsInLEAsWithAtLeastTwoSchools':sum(n for n in schools_per_lea.values() if n>=2),'limitations':['Historical administrative frame, not current recruitment eligibility or school consent.','Directory contains no individual student records, belonging scores, cultural identity, cultural authorization or permissions.','NCESSCH and LEAID are strings; names and truncated IDs are not linkage keys.','Statuses are administrative categories, not independent observations of classroom operation.','Selection requires confirmation against updated records, accessible curricula and governance agreements.']}
 if membership:
  z,rs=zip_rows(membership);fields=rs.fieldnames;n=0;total_ids=set();d=0;units=collections.Counter();dms=collections.Counter();blank_dms=collections.Counter();negative=collections.Counter();counts=collections.Counter();selected=0;membership_schools=set();unmatched=set()
  # Long-file schema is discovered and recorded; no guessed race/cell-zero conversion.
  value='STUDENT_COUNT' if 'STUDENT_COUNT' in fields else None
  keycols=[c for c in ['NCESSCH','GRADE','RACE_ETHNICITY','SEX','TOTAL_INDICATOR','DMS_FLAG'] if c in fields]
  for r in rs:
   n+=1;membership_schools.add(r.get('NCESSCH',''));units[r.get('TOTAL_INDICATOR','')]+=1;dms[r.get('DMS_FLAG','')]+=1
   if not r.get('STUDENT_COUNT','').strip():blank_dms[r.get('DMS_FLAG','')]+=1
   if r.get('TOTAL_INDICATOR')=='Education Unit Total':
    k=r['NCESSCH'];d+=k in total_ids;total_ids.add(k)
   if value:
    v=r[value].strip();counts[v]+=1 if v in {'','-1','-2','-9'} else 0
    try:
     if float(v)<0:negative[v]+=1
    except ValueError:counts['non_numeric']+=1
   if r.get('NCESSCH') not in ids:unmatched.add(r.get('NCESSCH'))
  out['membershipAudit']={'archiveSha256':sha256(membership),'rows':n,'columns':fields,'descriptiveLongCellColumns':keycols,'schoolTotalJoinRestriction':'TOTAL_INDICATOR == Education Unit Total','schoolTotalRows':units['Education Unit Total'],'uniqueSchoolTotalNCESSCH':len(total_ids),'duplicateSchoolTotalNCESSCH':d,'totalIndicatorCounts':dict(units),'fullLongCellDuplicateCheck':'Not performed; selected school-total uniqueness audited separately','uniqueSchools':len(membership_schools),'schoolsAbsentFromDirectoryCount':len(unmatched),'dataManagementFlagCounts':dict(dms),'blankStudentCountByDataManagementFlag':dict(blank_dms),'negativeValueCounts':dict(negative),'specialValueCounts':{k:v for k,v in counts.items() if v},'interpretation':'Long records are grade/race/sex/category cells; never join on NCESSCH alone without choosing a cell. Negative or nonnumeric values remain unavailable, not zero. Race/ethnicity reporting categories are not cultural identities. Exact meanings of special codes require this release documentation.'};z.close()
 return out


def design(seed=20261016,reps=10000):
 if reps<1000:raise ValueError('At least 1,000 repetitions required')
 rng=np.random.default_rng(seed);out=[]
 for pairs in [24,32,40]:
  for rho in [.05,.10]:
   for spill in [0.,.25]:
    retention_probability=rng.beta(6.8,1.2,(reps,pairs,2))
    m=np.maximum(rng.binomial(40,retention_probability),12)
    # Baseline/endline correlation .60 for both school and individual residuals.
    variance=.8*(rho+(1-rho)/m)
    noise=rng.normal(size=(reps,pairs))*np.sqrt(variance.sum(axis=2))
    effect=.30*(1-spill)
    differences=noise+effect
    se=differences.std(axis=1,ddof=1)/math.sqrt(pairs)
    statistic=differences.mean(axis=1)/se;critical=stats.t.ppf(.975,pairs-1)
    power=float((abs(statistic)>critical).mean());null=(abs(noise.mean(axis=1)/(noise.std(axis=1,ddof=1)/math.sqrt(pairs)))>critical).mean()
    out.append({'pairs':pairs,'schools':pairs*2,'outcomeICC':rho,'comparisonExposureFraction':spill,'assignedEffectSD':effect,'power':power,'monteCarloSE':math.sqrt(power*(1-power)/reps),'nullRejectionRate':float(null),'meanArtificialRetainedSize':float(m.mean()),'retainedSizeCV':float(m.std()/m.mean()),'retainedMinimum':int(m.min()),'retainedMaximum':int(m.max())})
 return {'analysis':'Artificial paired-school change-score design sensitivity; no observed effects','seed':seed,'replicatesPerScenario':reps,'assumptions':{'initialStudentsPerSchool':40,'meanRetainedTargetBeforeTruncation':34,'retentionModel':'School-specific p ~ Beta(6.8,1.2); retained m ~ Binomial(40,p), lower-truncated by maximum(m,12)','retainedSizeBounds':[12,40],'baselineEndlineCorrelation':.6,'individualOutcomeSD':1,'unattenuatedAssignedEffectSD':.3,'test':'Two-sided paired t test, alpha .05, pairs-1 df','matching':'No extra residual correlation benefit assumed for matched schools','exclusions':'No nonnormal bounds, informative attrition, whole-school dropout, measurement noninvariance, variable compliance or network interference modeled; comparison exposure is a scalar attenuation scenario, not a fitted spillover model.'},'scenarios':out}


def paired_estimate(differences):
 x=np.asarray(differences,dtype=float)
 if x.ndim!=1 or len(x)<3 or not np.isfinite(x).all():raise ValueError('At least 3 finite, prespecified independent pair differences required')
 se=float(x.std(ddof=1)/math.sqrt(len(x)))
 if se<=0:raise ValueError('Zero sampling variation cannot support this t interval')
 est=float(x.mean());q=float(stats.t.ppf(.975,len(x)-1))
 return {'estimate':est,'standardError':se,'confidenceInterval':[est-q*se,est+q*se],'df':len(x)-1,'interpretation':'Mathematical pair-summary utility only; not a complete participant analysis or authorization check.'}


def main():
 p=argparse.ArgumentParser();s=p.add_subparsers(dest='command',required=True)
 a=s.add_parser('audit');a.add_argument('--directory',required=True);a.add_argument('--membership');a.add_argument('--output',required=True)
 a=s.add_parser('design');a.add_argument('--seed',type=int,default=20261016);a.add_argument('--replicates',type=int,default=10000);a.add_argument('--output',required=True)
 a=p.parse_args();r=audit(a.directory,a.membership) if a.command=='audit' else design(a.seed,a.replicates)
 pathlib.Path(a.output).write_text(json.dumps(r,indent=2)+'\n')
if __name__=='__main__':main()