| |
| """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 |
|
|
| 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: |
| 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"])) |
|
|
| |
| 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: |
| 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()) |
|
|
