#!/usr/bin/env python """Stage 2: schema_lock (G3) + ner (NER gate) -- the two buckets missing from v8a.""" import os,sys,json,glob,zipfile,random,argparse sys.path.insert(0,"/workspace/aditi") import contract as C import pandas as pd GH="/workspace/aditi/ghdata"; random.seed(1947) def clip(t,n=1500): return (str(t) or "").strip()[:n] def pq(p): try: return pd.read_parquet(p) except Exception: return None BUDGET={"schema_lock":420_000,"ner":300_000} CAPS={"schema_lock":6310,"ner":9000} count={} def room(b): return count.get(b,0) misinformation slot f=f"{GH}/HinFakeNews1/HinFakeNews1.csv" if os.path.exists(f): d=pd.read_csv(f,encoding="utf-8-sig",on_bad_lines="skip") d.columns=[c.strip().lower() for c in d.columns] tc=next((c for c in ("text","news","headline","content","title") if c in d.columns),None) lc=next((c for c in ("label","class","target") if c in d.columns),None) for _,r in d.sample(frac=1,random_state=1947).iterrows(): if not room("schema_lock"): break mis="false" if str(r.get(lc,"")).strip().lower() in ("1","fake","false") else "verified" emit_rec(E,"HinFakeNews1",clip(r.get(tc,""),1500),"hi",mis=mis) d=pq(f"{GH}/urdu_fake_news/data/train-00000-of-00001.parquet") if d is not None: for _,r in d.iterrows(): if not room("schema_lock"): break lab=str(r.get("label",r.get("news_label",""))).lower() mis="false" if lab in ("1","fake","false") else "verified" emit_rec(E,"urdu_fake_news",clip(r.get("news",r.get("text","")),1500),"ur",mis=mis) f=f"{GH}/BanFakeNews-2.0/All60Kdataset.csv" if os.path.exists(f): d=pd.read_csv(f,encoding="utf-8-sig",on_bad_lines="skip",nrows=20000) d.columns=[c.strip().lower() for c in d.columns] tc=next((c for c in ("content","text","headline","title") if c in d.columns),None) lc=next((c for c in ("label","class") if c in d.columns),None) for _,r in d.sample(frac=1,random_state=1947).iterrows(): if not room("schema_lock"): break mis="false" if str(r.get(lc,"")).strip() in ("0","fake","1") and str(r.get(lc,""))=="0" else "verified" emit_rec(E,"BanFakeNews-2.0",clip(r.get(tc,""),1500),"bn",mis=mis) # --- claim-detection sources -> claims slot for nm,pat,tc in (("claimbuster",f"{GH}/claimbuster/data/train-*.parquet","text"), ("checkthat-binary",f"{GH}/checkthat-binary/data/train-*.parquet","text")): g=glob.glob(pat) if not g: continue d=pq(g[0]) if d is None: continue col=tc if tc in d.columns else d.columns[0] for _,r in d.iterrows(): if not room("schema_lock"): break t=clip(r.get(col,""),1500) emit_rec(E,nm,t,"en",claims=[{"claim":clip(t,300),"verdict":"not_enough_evidence"}]) # --- IndicSentiment (validation split is trainable; TEST is firewalled) for f in sorted(glob.glob(f"{GH}/IndicSentiment/data/validation/*.json")): lg=os.path.basename(f).split(".")[0] try: rows=[json.loads(l) for l in open(f,encoding="utf-8") if l.strip()] except Exception: continue for r in rows: if not room("schema_lock"): break t=clip(r.get("INDIC REVIEW",r.get("ENGLISH REVIEW","")),1200) s=C.canon("sentiment",str(r.get("LABEL","neutral")).lower(),strict=False) or "neutral" emit_rec(E,"indicsentiment",t,lg,sent=s) # --- naamapadam -> entities slot, 11 languages for z in sorted(glob.glob(f"{GH}/naamapadam/data/*_IndicNER_v1.0.zip")): lg=os.path.basename(z).split("_")[0] try: zf=zipfile.ZipFile(z) except Exception: continue tr=[n for n in zf.namelist() if "train" in n.lower()] if not tr: continue per=0 with zf.open(tr[0]) as fh: for line in fh: if per>=200 or not room("schema_lock"): break try: r=json.loads(line) except Exception: continue w,t=r.get("words"),(r.get("ner") or r.get("ner_tags")) if not w or not t: continue ents=C.bio_entities(w,[str(x) for x in t]) if not ents: continue txt=" ".join(w) if len(txt)<30: continue u=C.schema_prompt(C.content_hash(txt)[:12],clip(txt,1500)) if E.emit("schema_lock",f"naamapadam-record-{lg}",u, record(txt,lg,ents=ents),C.OSINT_RECORD_SCHEMA,heldout_text=txt): took("schema_lock"); per+=1 def do_ner(E): NT={0:"O",1:"B-PER",2:"I-PER",3:"B-ORG",4:"I-ORG",5:"B-LOC",6:"I-LOC"} for z in sorted(glob.glob(f"{GH}/naamapadam/data/*_IndicNER_v1.0.zip")): lg=os.path.basename(z).split("_")[0] try: zf=zipfile.ZipFile(z) except Exception: continue tr=[n for n in zf.namelist() if "train" in n.lower()] if not tr: continue per=0 with zf.open(tr[0]) as fh: for line in fh: if per>=800 or not room("ner"): break try: r=json.loads(line) except Exception: continue w,t=r.get("words"),(r.get("ner") or r.get("ner_tags")) if not w or not t: continue tags=[NT.get(int(x),str(x)) if str(x).isdigit() else str(x) for x in t] ents=C.bio_entities(w,tags); txt=" ".join(w) if len(txt)<25: continue u=C.P_NER.format(iid=C.content_hash(txt)[:12],text=clip(txt,1200)) if E.emit("ner",f"naamapadam-{lg}",u,{"entities":ents},C.NER_SCHEMA,heldout_text=txt): took("ner"); per+=1 f=f"{GH}/COMI-LINGUA/NER_train.csv" if os.path.exists(f): try: d=pd.read_csv(f,encoding="utf-8-sig",on_bad_lines="skip") d.columns=[c.strip().lower() for c in d.columns] tc=next((c for c in ("text","sentence","tokens") if c in d.columns),None) for _,r in d.iterrows(): if not room("ner"): break txt=clip(r.get(tc,""),1200) if len(txt)<25: continue u=C.P_NER.format(iid=C.content_hash(txt)[:12],text=txt) if E.emit("ner","comilingua-ner",u,{"entities":[]},C.NER_SCHEMA,heldout_text=txt): took("ner") except Exception: pass STAGES={"schema_lock":do_schema_lock,"ner":do_ner} if __name__=="__main__": ap=argparse.ArgumentParser(); ap.add_argument("--out",required=True); a=ap.parse_args() E=C.Emitter(a.out,BUDGET,tokenizer_dir="/workspace/models/gemma4-qat-unquantized") for s in STAGES: print(f"--- stage {s}",flush=True) try: STAGES[s](E) except Exception: import traceback; traceback.print_exc() print(f" {s}: {count.get(s,0)} recs, {E.spent.get(s,0)} sup-tok",flush=True) E.close(); print(E.report(a.out+".report.txt"))