"""Independent normalization of the session-5 raw export. No Fable data imported."""
from pathlib import Path
import json,re,unicodedata,hashlib
import numpy as np
import pandas as pd
import pyarrow.parquet as pq
R=Path(__file__).resolve().parent
S=R.parent/'2026-09-05/exports'
def norm(s): return re.sub(r'\s+',' ',unicodedata.normalize('NFKC',s).casefold()).strip()
def clean(s):
    s=re.sub(r'\[img[^\]]*\].*?\[/img\]',' ',s,flags=re.I|re.S)
    return re.sub(r'\s+',' ',re.sub(r'https?://\S+|\[[^\]]*\]|<[^>]+>',' ',s)).strip()
names=pd.read_parquet(S/'tag_names.parquet').set_index('tagid').name.to_dict()
rows=[]
for batch in pq.ParquetFile(S/'games.parquet').iter_batches(batch_size=4096):
    for row in batch.to_pylist():
        d=json.loads(row['data_json']);b=d.get('basic_info',{});rel=d.get('release',{})
        devs=[norm(x['name']) for x in b.get('developers',[]) if x.get('name','').strip()]
        pubs=[norm(x['name']) for x in b.get('publishers',[]) if x.get('name','').strip()]
        o=d.get('best_purchase_option',{});price=np.nan
        if row['source_country']=='US' and o.get('packageid') and not o.get('bundleid') and o.get('included_game_count',1)<=1 and not o.get('free_with_master_sub_appid'):
            p=o.get('original_price_in_cents',o.get('final_price_in_cents'))
            if p is not None and int(p)>0:price=int(p)/100
        timestamp=row['original_steam_release_date'] or row['steam_release_date']
        if timestamp is not None and timestamp<=0:timestamp=None
        desc=clean(d.get('full_description_bbcode',''));short=clean(row['short_description'] or '')
        cats={x for v in d.get('categories',{}).values() if isinstance(v,list) for x in v if isinstance(x,int)}
        langs=[x.get('elanguage') for x in d.get('supported_languages',[]) if x.get('supported')]
        rows.append(dict(appid=row['appid'],name=row['name'],date=timestamp,steam_date=row['steam_release_date'],
          original_release=row['original_release_date'],released=not row['is_coming_soon'],free=bool(row['is_free']),
          explicit=bool(set(d.get('content_descriptorids',[]))&{3,4}),ea=bool(row['is_early_access']),
          known_ea=bool(row['is_early_access'] or row['original_steam_release_date'] or row['release_from_early_access_date']),
          reviews=row['reviews_filtered_count'] or 0,pct=row['reviews_filtered_positive_percent'],price=price,
          dev=devs[0] if len(devs)==1 else '',dev_full=' | '.join(devs),ndev=len(devs),pub=' | '.join(pubs),
          selfpub=bool(len(devs)==1 and pubs==devs),desc=desc,short=short,words=len(desc.split()),
          lang_count=len(langs),english=0 in langs,chinese=6 in langs,japanese=10 in langs,russian=8 in langs,
          demo=bool(d.get('related_items',{}).get('demos') or d.get('related_items',{}).get('standalone_demos')),
          controller=28 in cats,achievements=22 in cats,workshop=30 in cats))
g=pd.DataFrame(rows)
for c in ['date','steam_date','original_release']:g[c]=pd.to_datetime(g[c],unit='s',utc=True,errors='coerce')
# Previously independently verified date correction; preserve raw export.
override=json.loads((R.parent/'portfolios-2026-09-06/date_overrides.json').read_text())
for a,v in override.items():g.loc[g.appid.eq(int(a)),'date']=pd.Timestamp(v['date'],tz='UTC')
g['valid']=g.released & g.date.notna() & g.date.lt(pd.Timestamp('2026-09-06',tz='UTC'))
g['year']=g.date.dt.year;g['month']=g.date.dt.month;g['quarter']=g.date.dt.strftime('%Y')+'Q'+g.date.dt.quarter.astype(str)
g['price_band']=pd.cut(g.price,[0,5,7,12,20,40,np.inf],labels=['<=5','5-7','7-12','12-20','20-40','>40']).astype(str)
g['length_band']=pd.cut(g.words,[-1,100,200,400,800,np.inf],labels=['<=100','101-200','201-400','401-800','>800']).astype(str)
g['h50']=g.reviews*g.price*35.9>=50000
for n in [50,100,556,1000]:g[f'r{n}']=g.reviews>=n
g['joint']=g.r100 & g.pct.ge(80)
t=pd.read_parquet(S/'game_tags.parquet');t=t[t.appid.isin(g.appid)].copy();t['tag']=t.tagid.map(names)
g['primary']=g.appid.map(t[t['rank']==1].set_index('appid').tag).fillna('unknown')
g.to_parquet(R/'games.parquet',index=False);t.to_parquet(R/'tags.parquet',index=False)
audit={'games':len(g),'valid':int(g.valid.sum()),'single_dev':int(g.ndev.eq(1).sum()),'no_dev':int(g.ndev.eq(0).sum()),'multi_dev':int(g.ndev.gt(1).sum()),'date_override':override,'source':'../2026-09-05/exports/games.parquet','source_sha256':hashlib.sha256((S/'games.parquet').read_bytes()).hexdigest()}
(R/'source_audit.json').write_text(json.dumps(audit,indent=2));print(audit)