from pathlib import Path
import pandas as pd,numpy as np,json,math
R=Path(__file__).resolve().parent
sample=pd.read_parquet(R/'sample.parquet');rows=[];outliers=[]
for p in sorted((R/'store_summaries').glob('*.json')):
    d=json.loads(p.read_text());rows.append({k:d[k] for k in ['appid','source','retrieved_at','total','positive','english','english_positive']})
    for x in d.get('props',{}).get('filter_options',{}).get('rgLanguageOutliers',[]):
        outliers.append(dict(appid=d['appid'],language=x.get('strLanguage'),count=x.get('nReviews'),positive=x.get('nReviewsPositive'),filtered=x.get('bFilteredReviews'),source='store_scored_language_subset'))
if not rows:raise SystemExit('No observations yet')
d=sample.merge(pd.DataFrame(rows),on='appid',validate='one_to_one')
d['english_share']=d.english_y/d.total.replace(0,np.nan)
d=d.rename(columns={'english_x':'english_support','english_y':'english_count'})
d['price_band']=d.price_band.fillna('missing').astype(str)
d['nonenglish_count']=d.total-d.english_count;d['nonenglish_positive']=d.positive-d.english_positive
d['english_pct']=100*d.english_positive/d.english_count.replace(0,np.nan);d['nonenglish_pct']=100*d.nonenglish_positive/d.nonenglish_count.replace(0,np.nan)
d['english_minus_nonenglish_pp']=d.english_pct-d.nonenglish_pct
d['low_english_10']=d.english_share.le(.1);d['low_english_25']=d.english_share.le(.25);d['majority_nonenglish']=d.english_share.lt(.5)
d['low_english_absolute']=d.english_count.lt(100)
d['substantial_elsewhere_low_english']=d.nonenglish_count.ge(500)&d.english_count.lt(100)&d.english_share.le(.1)
d['share_bin']=pd.cut(d.english_share,[-.001,.1,.5,1],labels=['<=10%','10-50%','>50%']).astype(str)
d['snapshot_count_delta']=d.total-d.reviews
t=pd.read_parquet(R.parent/'fable-verification-2026-09-06/tags.parquet');taglists=t[t.appid.isin(d.appid)].sort_values('rank').groupby('appid').tag.agg(list)
d['tags']=d.appid.map(taglists).map(lambda x:'; '.join(x) if isinstance(x,list) else '')
d.to_parquet(R/'observations_all.parquet',index=False)
expected_n=len(sample)
if (R/'collection_amendment.json').exists():
    amend=json.loads((R/'collection_amendment.json').read_text());expected_n=amend['primary_n']
    d=d[d.sample_order.lt(expected_n)].copy();d['planned_weight']=d.weight
    d['sample_n']=amend['effective_per_stratum'];d['weight']=d.population_n/d.sample_n
d.to_parquet(R/'observations.parquet',index=False);d.drop(columns=['desc']).to_csv(R/'observations.csv',index=False)
pd.DataFrame(outliers).to_csv(R/'store_language_subsets.csv',index=False)
complete=len(d)==expected_n
if not complete:
    print(json.dumps({'collected':len(d),'sample':expected_n,'complete':False,'unweighted_low_english10':int(d.low_english_10.sum())}));raise SystemExit
def estimate(y,domain=None):
    if domain is None:domain=pd.Series(True,index=d.index)
    yy=y.fillna(False).astype(float)*domain.astype(float);zz=domain.astype(float)
    den=float((d.weight*zz).sum());num=float((d.weight*yy).sum());p=num/den if den else np.nan
    residual=yy-p*zz;v=0
    for _,a in d.groupby('stratum'):
        N=float(a.population_n.iloc[0]);n=len(a);v+=N*N*(1-n/N)*residual.loc[a.index].var(ddof=1)/n
    se=math.sqrt(max(0,v))/den if den else np.nan
    n_domain=int(domain.sum());k_domain=int((y.fillna(False)&domain).sum())
    boundary=k_domain in [0,n_domain]
    return dict(sample_n=n_domain,sample_k=k_domain,weighted_domain_n=den,estimate=p,se=se,lo95=np.nan if boundary else max(0,p-1.96*se),hi95=np.nan if boundary else min(1,p+1.96*se),interval_note='No interval: boundary estimate lacks sample variation' if boundary else 'Design-based linearized 95% interval')
rows=[]
domains=[('all',pd.Series(True,index=d.index)),('english_supported',d.english_support),('english_not_listed',~d.english_support)]
domains += [('period:'+p,d.period.eq(p)) for p in sorted(d.period.unique())]
domains += [('review_band:'+p,d.review_band.eq(p)) for p in sorted(d.review_band.unique())]
domains += [('price:'+p,d.price_band.eq(p)) for p in sorted(d.price_band.unique())]
vn_ids=set(t[t.tag.isin(['Visual Novel','FMV'])].appid)
sexual_ids=set(t[t.tag.isin(['Sexual Content','Nudity','Hentai'])].appid)
rpg_ids=set(t[t.tag.eq('RPG')].appid)
domains += [('excluding_VN_FMV',~d.appid.isin(vn_ids)),('excluding_sexual_tags',~d.appid.isin(sexual_ids)),('RPG_excluding_VN_FMV',d.appid.isin(rpg_ids)&~d.appid.isin(vn_ids))]
for name,dom in domains:
    for metric in ['low_english_10','low_english_25','majority_nonenglish','low_english_absolute','substantial_elsewhere_low_english']:
        rows.append(dict(domain=name,metric=metric,**estimate(d[metric],dom)))
pd.DataFrame(rows).to_csv(R/'weighted_estimates.csv',index=False)
rows=[]
for tag in ['Action','Adventure','RPG','Strategy','Simulation','Puzzle','Horror','Action Roguelike','Roguelike Deckbuilder','Auto Battler','Turn-Based Tactics','JRPG','Visual Novel','FMV','Idler','Incremental']:
    ids=set(t[t.tag.eq(tag)].appid);dom=d.appid.isin(ids)
    for metric in ['low_english_10','majority_nonenglish']:rows.append(dict(tag=tag,metric=metric,**estimate(d[metric],dom)))
pd.DataFrame(rows).to_csv(R/'tag_estimates.csv',index=False)
# Exact counts on the observed games; review-weighted estimate is secondary to game shares.
review_den=float((d.weight*d.total).sum());review_ratio=float((d.weight*d.english_count).sum()/review_den)
review_residual=d.english_count-review_ratio*d.total;review_variance=0
for _,a in d.groupby('stratum'):
    N=float(a.population_n.iloc[0]);n=len(a);review_variance+=N*N*(1-n/N)*review_residual.loc[a.index].var(ddof=1)/n
review_se=math.sqrt(max(0,review_variance))/review_den
summary={'sample_n':len(d),'population_n':int(round(d.weight.sum())),'total_observed_reviews':int(d.total.sum()),'observed_english_reviews':int(d.english_count.sum()),'weighted_review_english_share':review_ratio,'weighted_review_english_share_lo95':max(0,review_ratio-1.96*review_se),'weighted_review_english_share_hi95':min(1,review_ratio+1.96*review_se),'english_support_sample_n':int(d.english_support.sum()),'low_english10_sample_n':int(d.low_english_10.sum()),'low_english10_with_support':int((d.low_english_10&d.english_support).sum()),'count_drift_max_abs':int(d.snapshot_count_delta.abs().max()),'store_source':int(d.source.eq('store_props').sum()),'api_source':int(d.source.eq('api_fallback').sum())}
(R/'summary.json').write_text(json.dumps(summary,indent=2));print(json.dumps(summary,indent=2))
d[d.low_english_10].sort_values('nonenglish_count',ascending=False).drop(columns=['desc']).to_csv(R/'low_english_discoveries.csv',index=False)
d[d.substantial_elsewhere_low_english].sort_values('nonenglish_count',ascending=False).drop(columns=['desc']).to_csv(R/'discovery_shortlist.csv',index=False)
d[d.english_count.ge(50)&d.nonenglish_count.ge(50)].sort_values('english_minus_nonenglish_pp').drop(columns=['desc']).to_csv(R/'within_game_reception.csv',index=False)
paired=d[d.english_count.ge(50)&d.nonenglish_count.ge(50)].copy()
paired['abs_gap']=paired.english_minus_nonenglish_pp.abs()
gap_summary={'eligible_pairs':len(paired),'english_higher':int(paired.english_minus_nonenglish_pp.gt(0).sum()),'nonenglish_higher':int(paired.english_minus_nonenglish_pp.lt(0).sum()),'median_absolute_gap_pp':float(paired.abs_gap.median()),'absolute_gap_10pp_or_more':int(paired.abs_gap.ge(10).sum()),'weighted_abs_gap10_share_among_paired':float(np.average(paired.abs_gap.ge(10),weights=paired.weight)),'note':'Descriptive current recommendation differences, not nationality, culture, causal localization effects, or prospective tests.'}
(R/'reception_summary.json').write_text(json.dumps(gap_summary,indent=2))