"""Count-only comparisons. No individual review text or positivity is used."""
import json
import numpy as np
from design import *
g=pd.read_parquet(R/'classified.parquet')

def rates(z):
 n=len(z)
 return dict(n=n,median=float(z.reviews.median()) if n else None,
             under10=int(z.reviews.lt(10).sum()),**{f'ge{k}':int(z.reviews.ge(k).sum()) for k in [50,100,200,1000]},
             developers=int(z.loc[z.developer_id.str.startswith('name:'),'developer_id'].nunique()),
             publishers=int(z.loc[z.publisher_id.str.startswith('name:'),'publisher_id'].nunique()))

def expected(b,c,threshold=100,text=False):
 if not len(b) or not len(c):return (np.nan,0.0,0.0)
 c=c[['reviews','year','date_bin','price_bucket','text_bin']].copy();c['y']=c.reviews.ge(threshold).astype(float)
 prediction=pd.Series(float(c.y.mean()),index=b.index)
 full=pd.Series(False,index=b.index);dated=full.copy()
 levels=[['year'],['date_bin'],['date_bin','price_bucket']]
 if text:levels.append(['date_bin','price_bucket','text_bin'])
 for cols in levels:
  ck=c[cols].fillna('missing').astype(str).agg('|'.join,axis=1);bk=b[cols].fillna('missing').astype(str).agg('|'.join,axis=1)
  stats=c.groupby(ck).y.agg(['mean','size']);valid=stats[stats['size'].ge(20)]
  vals=bk.map(valid['mean']);prediction=vals.fillna(prediction)
  full=vals.notna()
  if cols==['date_bin']:dated=vals.notna()
 return float(prediction.sum()),float(full.mean()),float(dated.mean())

def comparison(pool,definition='strict',extras=False):
 b=pool[pool[definition]];c=pool[~pool[definition]]
 r={f'b_{k}':v for k,v in rates(b).items()}|{f'c_{k}':v for k,v in rates(c).items()}
 r['n']=len(pool);r['share']=len(b)/len(pool) if len(pool) else None
 for k in [50,100,200,1000]:
  e,full,date=expected(b,c,k)
  r[f'expected{k}']=e;r[f'ratio{k}']=r[f'b_ge{k}']/e if e and np.isfinite(e) else None
  if k==100:r['price_date_support']=full;r['date_support']=date
 e,full,date=expected(b,c,100,True);r['text_ratio100']=r['b_ge100']/e if e and np.isfinite(e) else None;r['text_support']=full
 r['b_developer_weighted_rate100']=float(b.assign(y=b.reviews.ge(100)).groupby('developer_id').y.mean().mean()) if len(b) else None
 r['c_developer_weighted_rate100']=float(c.assign(y=c.reviews.ge(100)).groupby('developer_id').y.mean().mean()) if len(c) else None
 if extras and len(b):
  for field,short in [('developer_id','dev'),('publisher_id','pub')]:
   key=b.groupby(field).reviews.sum().idxmax();bb=b[b[field].ne(key)];cc=c[c[field].ne(key)];e,_,_=expected(bb,cc)
   r[f'removed_{short}']=key;r[f'without_{short}_n']=len(bb);r[f'without_{short}_ge100']=int(bb.reviews.ge(100).sum())
   r[f'without_{short}_ratio100']=float(bb.reviews.ge(100).sum()/e) if e and np.isfinite(e) else None
  bb=b.drop(b.nlargest(min(3,len(b)),'reviews').index);e,_,_=expected(bb,c)
  r['without_top3_n']=len(bb);r['without_top3_ge100']=int(bb.reviews.ge(100).sum());r['without_top3_ratio100']=float(bb.reviews.ge(100).sum()/e) if e and np.isfinite(e) else None
 return r

rows=[];sensitivity=[];members=[]
for period in ['2019_2022','2023_2025','2026_JanAug']:
 base=g[g.period.eq(period)]
 for host in HOSTS:
  tagged=base[base[f'host10_{host}']];pool=tagged[tagged['text_'+host]]
  if not len(pool):continue
  r=dict(period=period,host=host,tagged_n=len(tagged),host_text_coverage=len(pool)/len(tagged),**comparison(pool,extras=True));rows.append(r)
  for label,sub,definition in [('extended',pool,'extended'),('direct',pool,'direct'),('tag_only',tagged,'strict'),
       ('rank5',base[base[f'host5_{host}']&base['text_'+host]],'strict'),('rank20',base[base[f'host20_{host}']&base['text_'+host]],'strict'),
       ('no_core',pool[~pool.core],'strict'),('core',pool[pool.core],'strict'),('no_rogue',pool[~pool.rogue],'strict')]:
   sensitivity.append(dict(period=period,host=host,variant=label,**comparison(sub,definition)))
  b=pool[pool.strict].copy()
  b['host']=host
  members.append(b[['appid','name','developer','publisher','developer_id','publisher_id','host','period','first_date','reviews','usd_list_price','core','rogue','direct','short_description','description','evidence','store_url']])
 print('Finished',period,flush=True)
result=pd.DataFrame(rows);result.to_csv(R/'host_results.csv',index=False)
pd.DataFrame(sensitivity).to_csv(R/'sensitivity.csv',index=False)
pd.concat(members).to_parquet(R/'build_members.parquet',index=False)

# All passing and failing screen conditions are visible; no unseen holdout claim.
screen=result[result.period.eq('2023_2025')].copy()
screen['enough_host']=screen.n.ge(100)
screen['sparse']=screen.share.le(.15)
screen['enough_build']=screen.b_n.ge(10)
screen['repeated_response']=screen.b_ge100.ge(5)
creator_success=pd.concat(members).query("period == '2023_2025' and reviews >= 100").groupby('host').developer_id.nunique()
screen['responding_developers']=screen.host.map(creator_success).fillna(0).astype(int)
screen['enough_creators']=screen.responding_developers.ge(5)
screen['favorable']=screen.ratio100.ge(1.5)
screen['screen_pass']=screen[['enough_host','sparse','enough_build','repeated_response','enough_creators','favorable']].all(axis=1)
screen.to_csv(R/'sparse_screen.csv',index=False)

# Comparable calendar supply within each year. These are surviving catalog counts.
supply=[]
for year in [2023,2024,2025,2026]:
 base=g[g.year.eq(year)&g.month.le(8)]
 for host in HOSTS:
  pool=base[base[f'host10_{host}']&base['text_'+host]]
  supply.append(dict(year=year,host=host,n=len(pool),strict=int(pool.strict.sum()),extended=int(pool.extended.sum()),share=float(pool.strict.mean()) if len(pool) else None))
pd.DataFrame(supply).to_csv(R/'jan_aug_supply.csv',index=False)
print('MAIN\n',result[result.period.eq('2023_2025')][['host','n','b_n','share','b_ge100','c_ge100','b_median','c_median','ratio100','text_ratio100','b_ge1000','price_date_support','host_text_coverage']].round(3).to_string(index=False))
print('SPARSE PASSES\n',screen[screen.screen_pass][['host','b_n','b_ge100','responding_developers','ratio100','without_pub_ratio100','without_top3_ratio100']].round(3).to_string(index=False))