File size: 7,121 Bytes
81cffe7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 | """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()
|