| |
| """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)<CAPS[b] |
| def took(b): count[b]=count.get(b,0)+1 |
|
|
| def record(text, lang, *, mis="unverified", sent="neutral", ents=None, claims=None, |
| stance="neutral", host="none", events=None, summary=""): |
| return {"language":lang,"script":C.script_for(lang,text), |
| "code_mixed":bool(C.script_for(lang,text)=="Mixed"), |
| "entities":ents or [],"claims":claims or [],"stance":stance, |
| "sentiment":sent,"misinformation":mis,"hostility":{"label":host}, |
| "events":events or [],"summary_en":clip(summary or text,300)} |
|
|
| def emit_rec(E, src, text, lang, **kw): |
| if not room("schema_lock") or len(text.strip())<30: return |
| u=C.schema_prompt(C.content_hash(text)[:12], clip(text,1500)) |
| if E.emit("schema_lock",src,u,record(text,lang,**kw),C.OSINT_RECORD_SCHEMA,heldout_text=text): |
| took("schema_lock") |
|
|
| def do_schema_lock(E): |
| |
| 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) |
| |
| 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"}]) |
| |
| 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) |
| |
| 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")) |
|
|
