#!/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())