from pathlib import Path
import json,re,unicodedata
import numpy as np
import pandas as pd
from source_data import load_games
R=Path(__file__).resolve().parent;ROOT=R.parent
x=pd.read_parquet(R/'games.parquet');tr=pd.read_parquet(R/'transitions.parquet');p=pd.read_csv(R/'portfolios.csv')
def norm(v):return re.sub(r'\s+',' ',unicodedata.normalize('NFKC',str(v)).casefold()).strip()
def stats(z):
 return dict(n=len(z),ge100=int(z.reviews.ge(100).sum()),ge1000=int(z.reviews.ge(1000).sum()),
 liked100=int((z.reviews.ge(100)&z.positive_pct.ge(80)).sum()),under10=int(z.reviews.lt(10).sum()),
 median_reviews=float(z.reviews.median()) if len(z) else None,
 below_predecessor=int(z.reviews.lt(z.prev_reviews).sum()) if 'prev_reviews' in z else None,
 median_raw_ratio=float((z.reviews/z.prev_reviews).median()) if len(z) and 'prev_reviews' in z else None,
 beats_predecessor_cohort=int(z.cohort_pct.gt(z.prev_cohort_pct).sum()) if 'prev_cohort_pct' in z else None)

# Repeat the key predecessor comparisons in a more personally relevant, small
# observed catalog subset; no claim that these are solo or independent developers.
first=x.groupby('dev').first_date.min();lastord=x.groupby('dev').ordinal.max()
tr['first_observed_year']=tr.dev.map(first).dt.year
tr['catalog_n_now']=tr.dev.map(lastord)
rows=[]
for period,z in tr.groupby('period'):
 for label,mask in [('all',pd.Series(True,index=z.index)),('catalog2to10',z.catalog_n_now.between(2,10)),
 ('first2010plus_catalog2to10',z.first_observed_year.ge(2010)&z.catalog_n_now.between(2,10)),
 ('no_recorded_ea',~z.known_ea_history&~z.prev_known_ea_history.astype(bool)),
 ('no_shared_franchise',~z.shared_franchise),('unshared_pages',z.page_names.eq(1)&z.prev_page_names.eq(1))]:
  for band,a in z[mask].groupby('prior_band'):
   rows.append(dict(period=period,restriction=label,prior_band=band,**stats(a)))
pd.DataFrame(rows).to_csv(R/'robustness.csv',index=False)

# Return probability is not inferred: this counts only the presence of another
# eligible release for old first-release cohorts, regardless current follow-up age.
firstrows=x.sort_values('first_date').groupby('dev').head(1).copy()
firstrows['n_catalog']=firstrows.dev.map(lastord)
firstrows=firstrows[firstrows.year.between(2010,2020)]
firstrows['first_review_band_now']=pd.cut(firstrows.reviews,[-1,9,99,999,9999,np.inf],labels=['<10','10-99','100-999','1000-9999','10000+']).astype(str)
presence=[]
for band,z in firstrows.groupby('first_review_band_now'):
 presence.append(dict(first_review_band_now=band,n=len(z),no_later_paid=int(z.n_catalog.eq(1).sum()),with_later_paid=int(z.n_catalog.ge(2).sum())))
pd.DataFrame(presence).to_csv(R/'later_release_presence.csv',index=False)

# Gap and positioning, specifically for currently substantially reviewed predecessors.
detail=[]
for period,z in tr.groupby('period'):
 z=z[z.prev_reviews.ge(1000)]
 for restriction,mask in [('all',pd.Series(True,index=z.index)),('small_catalog',z.catalog_n_now.between(2,10)),('no_shared_franchise',~z.shared_franchise)]:
  for axis in ['gap_band','similarity_band']:
   for label,a in z[mask].groupby(axis):detail.append(dict(period=period,restriction=restriction,axis=axis,value=label,**stats(a)))
pd.DataFrame(detail).to_csv(R/'strong_predecessor_details.csv',index=False)

# Direction sensitivity: top-20 Jaccard instead of top-10.
T=pd.read_parquet(ROOT/'analysis-2026-09-05/tags.parquet')
admin={'Indie','Singleplayer','Early Access','Free to Play','Great Soundtrack','Controller','Steam Machine','Software','Utilities'}
tags=T[~T.tag.isin(admin)].groupby('appid').tag.agg(set).to_dict()
tr['similarity20']=[len(tags.get(a,set())&tags.get(b,set()))/len(tags.get(a,set())|tags.get(b,set())) if tags.get(a,set())|tags.get(b,set()) else np.nan for a,b in zip(tr.appid,tr.prev_appid)]
tr['similarity20_band']=pd.cut(tr.similarity20,[-.001,.25,.4999999,1],labels=['distant','middle','close']).astype(str)
sim=[]
for period,z in tr.groupby('period'):
 for strength,mask in [('all',pd.Series(True,index=z.index)),('prior1000plus',z.prev_reviews.ge(1000))]:
  for label,a in z[mask].groupby('similarity20_band'):sim.append(dict(period=period,predecessor=strength,value=label,**stats(a)))
pd.DataFrame(sim).to_csv(R/'similarity20_sensitivity.csv',index=False)

# Freeze 18 randomly selected pairs for a semantic inspection of tag distance.
audit=[]
for band,z in tr[tr.period.eq('2023_2025')].groupby('similarity_band'):
 for a in z.sample(min(len(z),6),random_state=690617).itertuples():
  audit.append(dict(appid=a.appid,name=a.name,previous_appid=a.prev_appid,previous_name=a.prev_name,developer=a.credit,similarity_band=band,
    previous_tags=sorted(tags.get(a.prev_appid,set())),current_tags=sorted(tags.get(a.appid,set())),short_description=a.short_description))
(R/'similarity_audit.json').write_text(json.dumps(audit,indent=2,ensure_ascii=False))

# Named examples are illustrations, including smaller and larger follow-ups.
# Full credits are used for the gallery to reveal co-developed titles excluded
# from the aggregate single-credit study.
NAMES=['a327ex','Daniel Mullins Games','Terry Cavanagh','Suspicious Developments','Mossmouth','Hopoo Games','increpare games','Zachtronics','Sokpop Collective','PUNKCAKE Delicieux']
c=pd.read_parquet(ROOT/'2026-09-05/exports/creators.parquet');g=load_games()
c=c[c.role.eq('developers')].copy();c['norm']=c.name.map(norm)
gallery=[]
for name in NAMES:
 ids=c[c.norm.eq(norm(name))].appid
 for a in g[g.appid.isin(ids)&g.valid_released].sort_values('first_date').itertuples():
  credits=c[c.appid.eq(a.appid)].name.tolist()
  gallery.append(dict(credited_developer=name,appid=a.appid,name=a.name,release=str(a.first_date.date()),reviews=a.reviews,positive_pct=a.positive_pct,
     is_free=a.is_free,in_main_study=a.appid in set(x.appid),all_developer_credits=' | '.join(credits),short_description=a.short_description,store_url=a.store_url))
pd.DataFrame(gallery).to_csv(R/'case_catalogs.csv',index=False)
tr.to_parquet(R/'transitions_enriched.parquet',index=False)
# An older substantial game can coexist with a quiet immediate predecessor.
older=[];older_restricted=[]
for period,z in tr.groupby('period'):
 for label,a in [('prev_lt100_no_older1000',z[z.prev_reviews.lt(100)&z.previous_max_now.lt(1000)]),
                 ('prev_lt100_older1000',z[z.prev_reviews.lt(100)&z.previous_max_now.ge(1000)]),
                 ('prev100to999_older1000',z[z.prev_reviews.between(100,999)&z.previous_max_now.ge(1000)]),
                 ('prev1000plus',z[z.prev_reviews.ge(1000)])]:
  older.append(dict(period=period,group=label,**stats(a)))
 for restriction,mask in [('at_least_third_release',z.ordinal.ge(3)),('thirdplus_smallcatalog',z.ordinal.ge(3)&z.catalog_n_now.le(10)),
                          ('thirdplus_price5to20',z.ordinal.ge(3)&z.usd_list_price.between(5,20))]:
  for strength in [False,True]:
   a=z[mask&z.prev_reviews.lt(100)&z.previous_max_now.ge(1000).eq(strength)]
   older_restricted.append(dict(period=period,restriction=restriction,older1000=strength,**stats(a)))
pd.DataFrame(older).to_csv(R/'older_catalog_strength.csv',index=False)
pd.DataFrame(older_restricted).to_csv(R/'older_catalog_sensitivity.csv',index=False)

# A second outcome-hidden audit asks about actual change, not tag-distance labels.
q=tr[tr.period.eq('2023_2025')&tr.prev_reviews.ge(1000)&tr.similarity_band.eq('distant')].sample(n=24,random_state=690631)
desc=x.set_index('appid').short_description.to_dict()
audit2=[]
for a in q.itertuples():
 audit2.append(dict(appid=a.appid,previous_appid=int(a.prev_appid),developer=a.credit,name=a.name,previous_name=a.prev_name,
  short_description=a.short_description,previous_description=desc.get(a.prev_appid,''),previous_tags=sorted(tags.get(a.prev_appid,set())),current_tags=sorted(tags.get(a.appid,set()))))
if not (R/'direction_pilot_blind.json').exists():
 (R/'direction_pilot_blind.json').write_text(json.dumps(audit2,indent=2,ensure_ascii=False))
# Preserve the frozen pilot after a source-date correction. Flag any pair whose
# predecessor changed rather than silently replacing its sampled member.
if (R/'direction_labels.json').exists():
 lab=pd.DataFrame(json.loads((R/'direction_labels.json').read_text()))
 out=lab.merge(x[['appid','first_date','reviews','positive_pct','prev_reviews','prev_first_date','prev_appid','original_release_date','known_ea_history']],on='appid')
 out['pair_still_valid']=out.previous_appid.eq(out.prev_appid)
 out.to_csv(R/'direction_pilot_outcomes.csv',index=False)
print('ROBUSTNESS\n',pd.DataFrame(rows).query("period == '2023_2025' and prior_band in ['1000-9999','10000+']").round(3).to_string(index=False))
print('PRESENCE\n',pd.DataFrame(presence).to_string(index=False))
print('STRONG PRIOR\n',pd.DataFrame(detail).query("period == '2023_2025'").round(3).to_string(index=False))
print('SIM20\n',pd.DataFrame(sim).query("period == '2023_2025'").round(3).to_string(index=False))