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(dict(name=name,passed=bool(value)))
    if not value:raise AssertionError(name)
s=pd.read_parquet(R/'sample.parquet');d=pd.read_parquet(R/'game_histories.parquet');spec=json.loads((R/'study_spec.json').read_text())
check('sample hash unchanged',hashlib.sha256((R/'sample.csv').read_bytes()).hexdigest()==spec['sample_sha256'])
check('sixty complete unique games',len(d)==d.appid.nunique()==60 and set(d.appid)==set(s.appid))
check('sample weights recover frame',np.isclose(d.weight.sum(),65839))
check('five games per stratum',d.groupby('stratum').size().eq(5).all())
m=pd.read_csv(R/'monthly_arrivals.csv');end=pd.Timestamp('2026-09-01',tz='UTC');total=0
for a in d.itertuples():
    raw=pd.read_parquet(R/'completed'/f'{a.appid}.parquet');total+=len(raw)
    check('unique review IDs '+str(a.appid),not raw.recommendationid.duplicated().any())
    check('Steam purchase '+str(a.appid),raw.steam_purchase.all())
    check('no text or accounts '+str(a.appid),set(raw.columns)=={'appid','recommendationid','timestamp_created','timestamp_updated','steam_purchase','written_during_early_access'})
    dates=pd.to_datetime(raw.timestamp_created,unit='s',utc=True)
    check('monthly recount '+str(a.appid),m[m.appid.eq(a.appid)]['count'].sum()==dates.lt(end).sum()==a.through_august)
    check('summary mismatch within documented tolerance '+str(a.appid),abs(a.reconciliation_difference)<=max(3,.01*a.summary_total))
    if a.release_clock_valid:
        origin=pd.Timestamp(a.date)
        for days in [30,90,365]:check('day recount '+str(a.appid)+' '+str(days),int(dates.lt(origin+pd.Timedelta(days=days)).sum())==getattr(a,'release_d'+str(days)))
        check('horizons monotone '+str(a.appid),a.release_d30<=a.release_d90<=a.release_d365 and (pd.isna(a.release_d730) or a.release_d365<=a.release_d730))
    files=sorted((R/'pages').glob(f'{a.appid}_*.json'));cursor='*';ids=set()
    for p in files:
        page=json.loads(p.read_text());check('cursor chain '+p.stem,page['requested_cursor']==cursor);cursor=page['next_cursor'];ids.update(x['recommendationid'] for x in page['reviews'])
    check('empty terminal page '+str(a.appid),not json.loads(files[-1].read_text())['reviews'])
    check('pages match final records '+str(a.appid),ids==set(raw.recommendationid))
policy=json.loads((R/'collection_policy.json').read_text());req=[json.loads(x) for x in (R/'requests.jsonl').read_text().splitlines()]
check('collection closed',not policy['active']);check('under request cap',len(req)<=800);check('under review cap',total<=65000)
check('request pacing',all(b['epoch']-a['epoch']>=1.99 for a,b in zip(req,req[1:])))
check('all requests successful',all('error' not in x and x['status']==200 for x in req))
check('no automatic bulk resume',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),'requests':len(req),'reviews':total,'checks':checks},indent=2));print('Passed',len(checks),'checks;',len(req),'requests;',total,'reviews')