from common import *
import json,hashlib,ast
g=pd.read_parquet(R/'games.parquet');m=g[g.mature];checks=[]
def check(name,value):
    checks.append(dict(name=name,passed=bool(value)))
    if not value:raise AssertionError(name)
check('unique source games',len(g)==g.appid.nunique()==93013)
check('no future release in mature set',m.date.lt(pd.Timestamp('2025-09-01',tz='UTC')).all())
p=pd.read_csv(R/'catalogs.csv');check('catalog totals recover mature games',p.n.sum()==len(m));check('core size',int(p.core_shape.sum())==2651);check('core hit count',int((p.core_shape&p.hits.ge(1)).sum())==683)
for dev in p.developer_id.sample(100,random_state=779):
    a=m[m.developer_id.eq(dev)].sort_values(['date','appid']);q=p[p.developer_id.eq(dev)].iloc[0]
    check('catalog '+dev,len(a)==q.n and a.reviews.max()==q.peak and int(a.reviews.ge(1000).sum())==q.hits)
    check('sequence '+dev,' '.join(a.review_band)==q.sequence)
x=pd.read_parquet(R/'latest_transitions.parquet');check('one latest transition per name',len(x)==x.developer_id.nunique())
for row in x.sample(100,random_state=779).itertuples():
    earlier=m[m.developer_id.eq(row.developer_id)&m.date.lt(row.date)]
    check('prior max '+str(row.appid),earlier.reviews.max()==row.prior_best)
    check('prior hits '+str(row.appid),int(earlier.reviews.ge(1000).sum())==row.prior_hits_1000)
e=pd.read_csv(R/'first_strong_events.csv');f=pd.read_csv(R/'first_strong_followups.csv')
for row in e[e.threshold.eq(1000)].sample(80,random_state=779).itertuples():
    a=f[f.threshold.eq(1000)&f.developer_id.eq(row.developer_id)];date=pd.Timestamp(row.anchor_date)
    check('five-year window '+str(row.anchor_appid),pd.to_datetime(a.date,utc=True).le(date+pd.Timedelta(days=365.25*5)).all())
    check('followup count '+str(row.anchor_appid),len(a)==row.followup_n)
    if len(a)>=3:check('three-game floor '+str(row.anchor_appid),a.sort_values('position').head(3).reviews.min()==row.first3_min)
policy=json.loads((R/'collection_policy.json').read_text());req=[json.loads(x) for x in (R/'requests.jsonl').read_text().splitlines()]
check('history sample hash unchanged',hashlib.sha256((R/'history_targets.csv').read_bytes()).hexdigest()==policy['sample_sha256'])
check('collection closed',not policy['active']);check('under request cap',len(req)<=450);check('requests successful',all(x.get('status')==200 and 'error' not in x for x in req));check('request spacing',all(b['epoch']-a['epoch']>=1.99 for a,b in zip(req,req[1:])))
total=0
for a in pd.read_parquet(R/'sample.parquet').itertuples():
    data=pd.read_parquet(R/'completed'/f'{a.appid}.parquet');total+=len(data);check('no duplicate review '+str(a.appid),not data.recommendationid.duplicated().any());check('no account/text '+str(a.appid),set(data.columns)=={'appid','recommendationid','timestamp_created','timestamp_updated','steam_purchase','written_during_early_access'})
    cursor='*';seen=set();files=sorted((R/'pages').glob(f'{a.appid}_*.json'))
    for file in files:
        page=json.loads(file.read_text());check('cursor '+file.stem,page['requested_cursor']==cursor);cursor=page['next_cursor'];seen.update(z['recommendationid'] for z in page['reviews'])
    check('terminal empty '+str(a.appid),len(page['reviews'])==0);check('page records equal exported '+str(a.appid),seen==set(data.recommendationid))
check('review cap and count',total==32974 and total<=40000)
r=pd.read_parquet(R/'historical_reviews.parquet');check('historical union unique',not r.duplicated(['appid','recommendationid']).any());h=pd.read_csv(R/'historical_game_windows.csv');check('historical union games',len(h)==50)
hp=pd.read_csv(R/'historical_prior_states.csv')
for row in hp.itertuples():
    cur=g[g.appid.eq(row.next_appid)].iloc[0];prior=g[g.developer_id.eq(cur.developer_id)&g.date.lt(cur.date)].appid
    counts=r[r.appid.isin(prior)&r.created.lt(cur.date)].groupby('appid').size().reindex(prior,fill_value=0)
    check('historical prior '+str(row.next_appid),counts.max()==row.prior_then_best and int(counts.ge(1000).sum())==row.prior_then_hits)
b=pd.read_csv(R/'backcatalog_events.csv')
for row in b.itertuples():
    ids=[int(x) for x in row.older_appids.split(';')];date=pd.Timestamp(row.date);old=r[r.appid.isin(ids)]
    before=int((old.created.ge(date-pd.Timedelta(days=90))&old.created.lt(date)).sum());after=int((old.created.ge(date)&old.created.lt(date+pd.Timedelta(days=90))).sum())
    check('backcatalog recount '+str(row.new_appid),before==row.pre90 and after==row.post90)
peers=pd.read_csv(R/'backcatalog_control_matches.csv');pairs=pd.read_csv(R/'backcatalog_control_pairs.csv')
for row in pairs[pairs.matched].itertuples():
    ps=peers[peers.new_appid.eq(row.new_appid)&peers.older_appid.eq(row.older_appid)]
    check('matched peer count '+str(row.new_appid)+' '+str(row.older_appid),3<=len(ps)<=5)
    expected=row.pre90*np.median((ps.control_post+.5)/(ps.control_pre+.5));check('expected recount '+str(row.new_appid)+' '+str(row.older_appid),np.isclose(expected,row.expected_post90))
links=pd.read_csv(R/'developer_link_checks.csv');check('developer links checked',len(links)==18 and links.status.eq(200).all())
check('bulk policy unchanged paused',json.loads((R.parent/'network_policy.json').read_text())['network_paused'])
for path in R.glob('*.py'):ast.parse(path.read_text());check('syntax '+path.name,True)
(R/'validation.json').write_text(json.dumps({'passed':len(checks),'new_reviews':total,'new_review_requests':len(req),'developer_link_requests':len(links),'checks':checks},indent=2));print('Passed',len(checks),'checks; new reviews',total)