"""Pilot the Chinese content filter on a row-group sample of the cmn_Hani shards. Samples PER_FILE evenly spaced row groups from every shard (about 1,000 documents each), matches configs/zh-filter-terms.toml against the text (after t2s, so traditional pages hit the same simplified terms), and writes to the output folder: pilot-report.json overall and per-category hit rates, per-term document counts, per-TLD and per-domain hit shares (domains with >= 20 sampled docs) pilot-review.md per category, random hits with +-40 characters of context, and the domains whose sampled pages hit most often Nothing is filtered here; the report is for reviewing the lists. """ import argparse import collections import json from pathlib import Path import random import tomllib from concurrent.futures import ProcessPoolExecutor from urllib.parse import urlsplit ROOT=Path(__file__).resolve().parents[2] def load_terms(path): spec=tomllib.loads(Path(path).read_text(encoding='utf-8')) terms={c:[t.lower() for t in ts] for c,ts in spec['categories'].items()} weak={t.lower() for t in spec.get('rules',{}).get('weak',[])} return terms,spec.get('domains',{}),weak def counts_as_hit(hits,weak): """A document hits when it contains a non-weak term, or two different terms.""" found={t for ts in hits.values() for t in ts} return bool(found-weak) or len(found)>=2 def scan(args): path,per_file,terms_path,seed=args import opencc import pyarrow.parquet as pq terms,_,_=load_terms(terms_path) t2s=opencc.OpenCC('t2s'); rng=random.Random(f'{seed}:{path.name}') f=pq.ParquetFile(path); n=f.metadata.num_row_groups groups=sorted(set(round(i*(n-1)/(per_file-1)) for i in range(per_file))) out=[] for g in groups: for r in f.read_row_group(g,columns=['text','url','id']).to_pylist(): text=t2s.convert(r['text']).lower() hits={} for cat,ts in terms.items(): found=[t for t in ts if t in text] if found: hits[cat]=found snippet={} for cat,found in hits.items(): t=rng.choice(found); i=text.find(t) snippet[cat]=(t,text[max(0,i-40):i+len(t)+40].replace('\n',' ')) out.append((urlsplit(r['url']).hostname or '',hits,snippet,r['id'])) return out def main(): p=argparse.ArgumentParser(description=__doc__,formatter_class=argparse.RawDescriptionHelpFormatter) p.add_argument('--data',default=str(ROOT/'datasets/fineweb-2-cmn-hani')) p.add_argument('--terms',default=str(ROOT/'configs/zh-filter-terms.toml')) p.add_argument('--out',default=str(ROOT/'outputs/zh-corpus/term-pilot')) p.add_argument('--per-file',type=int,default=20) p.add_argument('--workers',type=int,default=8) p.add_argument('--seed',type=int,default=0) a=p.parse_args() terms,domains,weak=load_terms(a.terms) files=sorted(Path(a.data).glob('*.parquet')) with ProcessPoolExecutor(a.workers) as ex: docs=[d for part in ex.map(scan,[(f,a.per_file,a.terms,a.seed) for f in files]) for d in part] N=len(docs); cats=list(terms) tlds=set(domains.get('block_tlds',[])) def tld(h): return h.rsplit('.',1)[-1] term_docs=collections.Counter(); cat_docs=collections.Counter() host_n=collections.Counter(); host_hit=collections.Counter() tld_n=collections.Counter(); tld_hit=collections.Counter() any_hit=0; tld_blocked=0; tld_or_hit=0; weak_only=0 tiers={'block':domains.get('block',[]),'block_tier2':domains.get('block_tier2',[])} def tier(h): for name,ds in tiers.items(): if any(h==d or h.endswith('.'+d) for d in ds): return name return None tier_n=collections.Counter() docs=[(h,hits if counts_as_hit(hits,weak) else {},s,i,bool(hits) and not counts_as_hit(hits,weak)) for h,hits,s,i in docs] for host,hits,_,_,spared in docs: if spared: weak_only+=1 for host,hits,_,_,_ in docs: host_n[host]+=1; tld_n[tld(host)]+=1 t=tier(host) if t: tier_n[t]+=1 tld_blocked+=tld(host) in tlds blocked=tld(host) in tlds or t is not None if hits: any_hit+=1; host_hit[host]+=1; tld_hit[tld(host)]+=1 for c,found in hits.items(): cat_docs[c]+=1; term_docs.update(found) tld_or_hit+=bool(hits) or blocked pct=lambda x:round(100*x/N,3) report=dict(documents=N,files=len(files),per_file=a.per_file, any_hit_pct=pct(any_hit),weak_only_spared_pct=pct(weak_only),blocked_tld_pct=pct(tld_blocked), blocked_domain_pct={k:pct(v) for k,v in sorted(tier_n.items())},dropped_total_pct=pct(tld_or_hit), categories={c:dict(docs=cat_docs[c],pct=pct(cat_docs[c])) for c in cats}, terms={c:sorted(((t,term_docs[t]) for t in terms[c]),key=lambda x:-x[1]) for c in cats}, tlds=[(t,n,round(100*tld_hit[t]/n,1)) for t,n in tld_n.most_common(25)], domains=sorted(((h,n,host_hit[h],round(100*host_hit[h]/n,1)) for h,n in host_n.items() if n>=20 and host_hit[h]), key=lambda x:(-x[2],-x[3]))[:200]) out=Path(a.out); out.mkdir(parents=True,exist_ok=True) (out/'pilot-report.json').write_text(json.dumps(report,ensure_ascii=False,indent=1),encoding='utf-8') rng=random.Random(a.seed) with open(out/'pilot-review.md','w',encoding='utf-8') as fh: fh.write(f'# 合规词表 pilot\n\n抽样 {N} 篇，任一类命中 {report["any_hit_pct"]}%，' f'只含一个弱词而保留 {report["weak_only_spared_pct"]}%，屏蔽顶级域名 {report["blocked_tld_pct"]}%，' f'屏蔽域名 {report["blocked_domain_pct"]}，合计丢弃 {report["dropped_total_pct"]}%。\n\n') for c in cats: fh.write(f'## {c}：{cat_docs[c]} 篇（{pct(cat_docs[c])}%）\n\n') fh.write('命中最多的词：'+'，'.join(f'{t} {n}' for t,n in report['terms'][c][:15] if n)+'\n\n') pool=[(h,s[c]) for h,hits,s,_,_ in docs if c in hits] for h,(t,ctx) in rng.sample(pool,min(40,len(pool))): fh.write(f'- `{t}` · {h} · …{ctx}…\n') fh.write('\n') spared=[(h,t,ctx) for h,_,s,_,keep in docs if keep for t,ctx in s.values()] fh.write(f'## 只含一个弱词、予以保留：{weak_only} 篇\n\n') for h,t,ctx in rng.sample(spared,min(40,len(spared))): fh.write(f'- `{t}` · {h} · …{ctx}…\n') fh.write('\n') fh.write('## 命中最多的域名（抽样中至少 20 篇）\n\n| 域名 | 抽样篇数 | 命中篇数 | 命中率 |\n| --- | ---: | ---: | ---: |\n') for h,n,k,r in report['domains'][:60]: fh.write(f'| {h} | {n} | {k} | {r}% |\n') print(json.dumps({k:report[k] for k in ('documents','any_hit_pct','weak_only_spared_pct','blocked_tld_pct','blocked_domain_pct','dropped_total_pct','categories')},ensure_ascii=False,indent=1)) if __name__=='__main__': main()