File size: 8,013 Bytes
e2492ec | 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 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 | #!/usr/bin/env python3
"""Merge Jev verdicts into the final gmeow ad corpus + stats/report.
Inputs : data/cleaned/gmeow/ads_jev.jsonl (one row per unique text)
Outputs: data/cleaned/gmeow/ads_final.jsonl label=ad, kind != none
data/cleaned/gmeow/rejected.jsonl Jev said clean / kind none
data/cleaned/gmeow/campaigns.jsonl near-dup clusters
data/cleaned/gmeow/stats.json
data/cleaned/gmeow/report.md
"""
from __future__ import annotations
import collections
import json
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
from clean_gmeow_ads import normalize, parse_source_msg, signature # noqa: E402
OUT = Path("data/cleaned/gmeow")
STORE = Path("data/raw/gmeow/bans.jsonl")
def load(p: Path) -> list[dict]:
if not p.exists():
return []
return [json.loads(l) for l in p.open(encoding="utf-8") if l.strip()]
def main() -> int:
rows = load(OUT / "ads_jev.jsonl")
latest: dict[str, dict] = {}
for r in rows: # last judgement wins if a text was re-judged
latest[r["text"]] = r
rows = list(latest.values())
ads = [r for r in rows if r["label"] == "ad" and r["kind"] != "none"]
rejected = [r for r in rows if r not in ads]
ads.sort(key=lambda r: (-r["n_accounts"], -r["n_records"], r["text"]))
# account/chat coverage measured on raw records, not summed per text
accounts: set[int] = set()
chats: set[int] = set()
ad_texts = {r["text"] for r in ads}
if STORE.exists():
for line in STORE.open(encoding="utf-8"):
try:
rec = json.loads(line)
except Exception: # noqa: BLE001
continue
_, msg = parse_source_msg(rec["info"])
if normalize(msg) in ad_texts:
accounts.add(rec["tgid"])
if rec["chat_id"] is not None:
chats.add(rec["chat_id"])
with (OUT / "ads_final.jsonl").open("w", encoding="utf-8") as fh:
for r in ads:
fh.write(json.dumps({
"text": r["text"],
"label": "ad",
"kind": r["kind"],
"scope": r["scope"],
"severity": r["severity"],
"conf": r["jev_conf"],
"quality": r["quality"],
"rule_kind": r["rule_kind"],
"n_records": r["n_records"],
"n_accounts": r["n_accounts"],
"n_chats": r["n_chats"],
"reason": r["reason"],
"channel": r["channel"],
"first_seen": r["first_seen"],
"last_seen": r["last_seen"],
}, ensure_ascii=False) + "\n")
with (OUT / "rejected.jsonl").open("w", encoding="utf-8") as fh:
for r in rejected:
fh.write(json.dumps({
"text": r["text"], "label": r["label"], "kind": r["kind"],
"reason": r["reason"], "n_records": r["n_records"],
"rule_kind": r["rule_kind"], "jev_conf": r["jev_conf"],
}, ensure_ascii=False) + "\n")
campaigns: dict[str, list[dict]] = collections.defaultdict(list)
for r in ads:
campaigns[signature(r["text"])].append(r)
camp_rows = []
for sig, members in campaigns.items():
members.sort(key=lambda r: -r["n_accounts"])
camp_rows.append({
"campaign": sig,
"variants": len(members),
"accounts": sum(m["n_accounts"] for m in members),
"kind": collections.Counter(m["kind"] for m in members).most_common(1)[0][0],
"canonical": members[0]["text"],
})
camp_rows.sort(key=lambda c: (-c["accounts"], -c["variants"]))
with (OUT / "campaigns.jsonl").open("w", encoding="utf-8") as fh:
for c in camp_rows:
fh.write(json.dumps(c, ensure_ascii=False) + "\n")
kind_dist = collections.Counter(r["kind"] for r in ads)
stats = {
"unique_texts_judged": len(rows),
"ads": len(ads),
"rejected": len(rejected),
"rejected_detail": dict(collections.Counter(
f"{r['label']}/{r['kind']}" for r in rejected)),
"kind_dist": dict(kind_dist),
"scope_dist": dict(collections.Counter(r["scope"] for r in ads)),
"severity_avg": round(sum(r["severity"] or 0 for r in ads) / max(1, len(ads)), 2),
"severity_hist": dict(collections.Counter(
round(r["severity"] or 0) for r in ads)),
"accounts_covered": len(accounts),
"chats_covered": len(chats),
"raw_records_covered": sum(r["n_records"] for r in ads),
"campaigns": len(camp_rows),
"rule_agreement": round(
sum(1 for r in ads if r["rule_kind"] == r["kind"]) / max(1, len(ads)), 3),
"quality_flags": dict(collections.Counter(
f for r in ads for f in r["quality"])),
"time_span": [min(r["first_seen"] for r in ads),
max(r["last_seen"] for r in ads)] if ads else [],
}
(OUT / "stats.json").write_text(
json.dumps(stats, ensure_ascii=False, indent=2), encoding="utf-8")
lines = [
"# gmeow 封禁广告语料 · 初次清理（Jev 实时判定）", "",
f"- 去重文本：{stats['unique_texts_judged']}，Jev 判为广告 {stats['ads']}，"
f"剔除 {stats['rejected']}（{stats['rejected_detail']}）",
f"- 覆盖封禁记录 {stats['raw_records_covered']} 条 / 账号 "
f"{stats['accounts_covered']} 个 / 时间跨度 {stats['time_span']}",
f"- 类别分布：{stats['kind_dist']}",
f"- 载体：{stats['scope_dist']}；严重度均值 {stats['severity_avg']}"
f"（分布 {stats['severity_hist']}）",
f"- 近重复战役簇 {stats['campaigns']} 个；规则与 Jev 一致率 "
f"{stats['rule_agreement']}",
"",
"## 每类样例（按覆盖账号数排序）", "",
]
for kind in ("scam", "promo", "porn", "crypto", "gambling", "porn_bait",
"spam_flood"):
top = [r for r in ads if r["kind"] == kind][:5]
if not top:
continue
lines += [f"### {kind}（{kind_dist[kind]} 条）", ""]
for r in top:
text = r["text"].replace("\n", " ⏎ ")
lines.append(f"- `{r['n_accounts']}号` {text[:110]}")
lines.append("")
lines += ["## 最大战役簇", "", "| 账号数 | 变体 | 类别 | 代表文本 |", "|---:|---:|---|---|"]
for c in camp_rows[:10]:
text = c["canonical"].replace("|", "\\|").replace("\n", " ")
lines.append(f"| {c['accounts']} | {c['variants']} | {c['kind']} | {text[:70]} |")
lines += [
"",
"## 文件与复跑",
"",
"| 文件 | 内容 |",
"|---|---|",
"| `data/raw/gmeow/bans.jsonl` | 原始封禁记录（采集器增量去重写入） |",
"| `data/cleaned/gmeow/ads_jev.jsonl` | 每条唯一文本的 Jev 四问判定（实时追加） |",
"| `data/cleaned/gmeow/ads_final.jsonl` | 最终广告语料（label=ad 且类别非 none） |",
"| `data/cleaned/gmeow/rejected.jsonl` | Jev 判非广告/无类别的剔除项（含禁言次数，供复核） |",
"| `data/cleaned/gmeow/campaigns.jsonl` | 联系人/链接/数字归一后的近重复战役簇 |",
"",
"```bash",
"# 持续采集（页面只保留最近 100 条，约 4 条/分钟）",
"python scripts/collect_gmeow_bans.py --interval 45 --duration 7200",
"# 实时 Jev 清洗（新增文本自动判定，已判过的跳过）",
"python scripts/jev_clean_gmeow.py --watch --interval 20",
"# 重新生成最终语料与统计",
"python scripts/report_gmeow.py",
"```",
]
(OUT / "report.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
print(json.dumps(stats, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
|