from pathlib import Path
import pandas as pd,numpy as np,json,hashlib,ast
R=Path(__file__).resolve().parent;checks=[]
def check(name,value):
    checks.append({'name':name,'passed':bool(value)})
    if not value:raise AssertionError(name)
spec=json.loads((R/'study_spec.json').read_text());s=pd.read_parquet(R/'sample.parquet');d=pd.read_parquet(R/'observations.parquet')
check('sample unchanged',hashlib.sha256((R/'sample.csv').read_bytes()).hexdigest()==spec['sample_sha256'])
check('480 unique targets',len(s)==s.appid.nunique()==480)
check('all balanced prefix games collected',set(s[s.sample_order.lt(456)].appid)==set(d.appid) and len(d)==456)
check('weights recover frame',np.isclose(d.weight.sum(),13536))
check('forty per stratum',s.groupby('stratum').size().eq(40).all())
check('thirty eight completed per primary stratum',d.groupby('stratum').size().eq(38).all())
check('review counts nonnegative',(d[['total','english_count','positive','english_positive','nonenglish_count','nonenglish_positive']]>=0).all().all())
check('English is a subset',(d.english_count<=d.total).all())
check('positive counts consistent',(d.english_positive<=d.positive).all() and (d.nonenglish_positive<=d.nonenglish_count).all())
check('language share formula',np.allclose(d.english_share,d.english_count/d.total))
for row in d.itertuples():
    raw=json.loads((R/'store_summaries'/f'{row.appid}.json').read_text())
    check('source outcome '+str(row.appid),raw['total']==row.total and raw['english']==row.english_count)
    if raw['source']=='store_props':
        o=raw['props']['summary_options'];check('filtered source '+str(row.appid),o['summaryGlobalNoOutliers']['bFilteredReviews'] and o['summaryYourLanguage']['bFilteredReviews'])
        check('store identity '+str(row.appid),raw['props']['appid']==row.appid)
weights=pd.read_csv(R/'weighted_estimates.csv')
for threshold,label in [(.1,'low_english_10'),(.25,'low_english_25')]:
    x=weights[weights.domain.eq('all')&weights.metric.eq(label)].iloc[0]
    check('independent weighted '+label,np.isclose(x.estimate,np.average(d.english_share.le(threshold),weights=d.weight)))
has_interval=weights.lo95.notna()&weights.hi95.notna()
check('estimate intervals ordered',((weights.loc[has_interval,'lo95']<=weights.loc[has_interval,'estimate'])&(weights.loc[has_interval,'estimate']<=weights.loc[has_interval,'hi95'])).all())
req=[json.loads(x) for x in (R/'requests.jsonl').read_text().splitlines()]
check('request cap including probes',len(req)+4<=1400)
check('request pacing',all(b['started_epoch']-a['started_epoch']>=1.99 for a,b in zip(req,req[1:])))
errors=[x for x in req if 'error' in x]
check('one recorded timeout ends collection',len(errors)==1 and errors[0]==req[-1] and 'Read timed out' in errors[0]['error'])
check('all earlier requests succeeded',all('error' not in x and x['status']==200 for x in req[:-1]))
for p in (R/'api_summaries').glob('*.json'):
    raw=json.loads(p.read_text());q=raw['query_summary']
    check('API count '+p.stem,q['total_positive']+q['total_negative']==q['total_reviews'])
    check('no reviewer records '+p.stem,'reviews' not in raw and 'author' not in raw)
if (R/'language_profiles.csv').exists():
    profiles=pd.read_csv(R/'language_profiles.csv');chosen=pd.read_parquet(R/'profile_sample.parquet')
    check('all profile targets',set(profiles.appid)==set(chosen.appid))
    recon=pd.read_csv(R/'profile_reconciliation.csv')
    # This deliberately reports changing/cached count mismatch rather than
    # converting a negative residual into an invented positive remainder.
    check('profile counts reconcile',recon.residual_count.ge(0).all() and recon.residual_positive.ge(0).all() and (recon.residual_positive<=recon.residual_count).all())
    for a,x in profiles.groupby('appid'):
        check('profile sum '+str(a),np.isclose(x.loc[x.language.ne('all'),'share'].sum(),1))
if (R/'available_language_profiles.csv').exists():
    profiles=pd.read_csv(R/'available_language_profiles.csv');recon=pd.read_csv(R/'available_profile_reconciliation.csv')
    check('available subset IDs in primary',set(profiles.appid)<=set(d.appid))
    check('available subset residuals valid',(recon.residual_count>=0).all() and (recon.residual_positive>=0).all() and (recon.residual_positive<=recon.residual_count).all())
    for a,x in profiles.groupby('appid'):check('available profile sum '+str(a),np.isclose(x.share.sum(),1))
policy=json.loads((R/'collection_policy.json').read_text());check('scoped collection closed',not policy['active'])
check('bulk collection remains closed',json.loads((R.parent/'network_policy.json').read_text())['network_paused'])
for p in R.glob('*.py'):ast.parse(p.read_text());check('syntax '+p.name,True)
(R/'validation.json').write_text(json.dumps({'passed':len(checks),'checks':checks,'requests_including_probes':len(req)+4,'minimum_interval':min(b['started_epoch']-a['started_epoch'] for a,b in zip(req,req[1:]))},indent=2))
print('Passed',len(checks),'checks; requests',len(req)+4)