laya-ad / code /scripts /clean_gmeow_ads.py

laya-ad v2.2: six precision dirs

e2492ec verified 8 days ago

14.7 kB

#!/usr/bin/env python3
"""Initial cleaning for gmeow ban records -> unique ad texts.
Input : data/raw/gmeow/bans.jsonl   (collector output)
Output: data/cleaned/gmeow/ads_rules.jsonl        one row per unique text
        data/cleaned/gmeow/campaigns.jsonl  near-dup groups (contact/URL stripped)
        data/cleaned/gmeow/stats.json
        data/cleaned/gmeow/report.md
Stages: parse -> normalize -> drop non-text placeholders -> exact dedupe ->
heuristic kind labeling (7 kinds used by the classifier) -> campaign grouping.
"""
from __future__ import annotations

import argparse
import collections
import hashlib
import json
import re
import unicodedata
from pathlib import Path

KINDS = ["crypto", "gambling", "porn", "porn_bait", "promo", "scam", "spam_flood"]

# Regex rules scored like keywords; needed for short lures ("抢了这么多u").
REGEX_RULES: dict[str, list[tuple[str, int]]] = {
    "crypto": [
        (r"\d+\s*[uU](?![A-Za-z])", 4),          # 200U / 10u
        (r"(?:抢|领|送|白嫖|空投|白送|开局)[^。！\n]{0,8}[uU]", 4),
        (r"(?:空投|白送|免费领|零撸|撸毛)", 4),
        (r"(?:回血|带单|跟单|月化|翻倍|策略|走势|合约)", 2),
    ],
    "gambling": [
        (r"(?:包赢|稳赚|必中|红单|上岸|回血)", 3),
        (r"(?:实力平台|顶级平台|安全首选|押金保障)", 3),
        (r"(?:pc28|pg电子|pp电子|jdb|百家乐|时时彩)", 5),
    ],
    "scam": [
        (r"(?:一单|每单|单结|结算)[^。！\n]{0,6}\d+", 4),
        (r"\d+\s*(?:元|块|刀|u|U)\s*(?:一单|每单|/单)", 4),
        (r"(?:日赚|月入|日入|躺赚|日结|轻松赚)\s*\d*", 4),
        (r"(?:收资|资单|收单|代收|跑分|洗u|黑u|白资)", 4),
        (r"(?:解封|找回|追款|查档|户籍|定位|开房记录)", 3),
        (r"(?:兼职|招聘|招人|找人|做事|干活|小白可做|无本)", 2),
        (r"(?:留资|资方|卡商|号商|接码|实名卡)", 4),
        (r"(?:扫码|二维码|捡签|有码)", 2),
    ],
    "porn_bait": [
        (r"(?:看|点|戳)(?:我)?(?:的)?(?:主页|头像|简介|煮页|朋友圈|签名)", 3),
    ],
    "promo": [
        (r"(?:加群|进群|频道|客服|主号|双向|机器人)", 2),
        (r"(?:代开|中转站|全套业务|会员代开|充值折扣|服务器|专线)", 2),
        (r"(?:一家|专业|源头|实力|诚信|靠谱)", 1),
    ],
}

# keyword -> weight. Weights are per kind; first hit wins ties by list order.
RULES: dict[str, list[tuple[str, int]]] = {
    "crypto": [
        ("空投", 4), ("usdt", 4), ("泰达", 4), ("钱包", 3), ("助记词", 4),
        ("合约", 3), ("eth", 3), ("btc", 3), ("比特币", 3), ("炒币", 4),
        ("挖矿", 3), ("交易所", 3), ("量化", 3), ("搬砖", 3), ("u币", 4),
        ("波场", 3), ("trx", 2), ("defi", 3), ("币圈", 3), ("usd", 2),
    ],
    "gambling": [
        ("博彩", 5), ("赌博", 5), ("彩票", 4), ("时时彩", 5), ("快三", 5),
        ("棋牌", 4), ("娱乐城", 4), ("上分", 3), ("下分", 3), ("红单", 4),
        ("包赔", 3), ("走势", 2), ("pc28", 5), ("pg电子", 4), ("电子游艺", 4),
        ("百家乐", 5), ("六合彩", 5), ("足球", 2), ("篮球", 2), ("稳赚", 2),
        ("平台", 1), ("押金", 2), ("担保", 2), ("充", 1), ("提现", 2),
    ],
    "porn": [
        ("楼凤", 5), ("上门", 3), ("全套", 4), ("快餐", 3), ("课表", 4),
        ("约课", 4), ("约炮", 5), ("做爱", 5), ("裸聊", 5), ("福利视频", 4),
        ("幼女", 5), ("萝莉", 4), ("学生妹", 4), ("人妻", 3), ("少妇", 3),
        ("母狗", 5), ("骚逼", 5), ("调教", 3), ("sm圈", 4), ("选妃", 4),
        ("外围", 4), ("探花", 4), ("福利姬", 4), ("视频", 1), ("妹妹", 2),
        ("生活圈", 3), ("实照", 3), ("真人", 2), ("口嗨", 2), ("上门服务", 3),
    ],
    "porn_bait": [
        ("看主页", 4), ("点我头像", 4), ("点击头像", 4), ("看简介", 4),
        ("简介看", 4), ("煮页", 4), ("主页", 2), ("签名", 2), ("看片", 3),
        ("进群看", 3), ("点击进入", 3), ("戳这里", 3), ("扫码", 3),
    ],
    "promo": [
        ("广告代发", 5), ("代发", 3), ("群发", 4), ("引流", 4), ("涨粉", 4),
        ("刷量", 4), ("推广", 3), ("曝光", 2), ("项目招商", 3), ("招商", 2),
        ("招代理", 3), ("代理", 1), ("频道", 1), ("进群", 1), ("客服", 2),
    ],
    "scam": [
        ("兼职", 4), ("日赚", 5), ("日结", 4), ("月入", 4), ("刷单", 5),
        ("垫付", 4), ("解封", 5), ("找回", 3), ("追款", 4), ("贷款", 3),
        ("接码", 5), ("卖号", 4), ("买号", 4), ("号商", 5), ("社工库", 5),
        ("实名", 3), ("接单", 3), ("跑分", 4), ("代收", 3), ("洗钱", 5),
        ("洗u", 4), ("黑u", 4), ("假钞", 5), ("空壳", 3), ("盗号", 4),
        ("劫持", 4), ("协议号", 4), ("蓝标", 3), ("赚钱", 2), ("做单", 3),
        ("一单", 3), ("佣金", 3), ("结算", 2), ("私聊", 2), ("咨询", 2),
    ],
    "spam_flood": [
        ("加我", 1), ("互粉", 2), ("互关", 2), ("点赞", 2), ("评论", 2),
    ],
}

PLACEHOLDER = re.compile(
    r"相片內含有廣告|相片内含有广告|Image contains Ads|TG違規內容|TG违规内容", re.I)
URL_RE = re.compile(r"(?:https?://|www\.)\S+|t\.me/\S+", re.I)
HANDLE_RE = re.compile(r"@[A-Za-z0-9_]{3,}")
PHONE_RE = re.compile(r"(?<!\d)1[3-9]\d{9}(?!\d)")
ZERO_WIDTH = dict.fromkeys(map(ord, "\u200b\u200c\u200d\ufeff\u2060"))


def _fold_ascii_width(text: str) -> str:
    """Fold full-width alphanumerics only; Chinese punctuation stays intact."""
    out = []
    for ch in text:
        o = ord(ch)
        if 0xFF21 <= o <= 0xFF3A or 0xFF41 <= o <= 0xFF5A or 0xFF10 <= o <= 0xFF19:
            out.append(chr(o - 0xFEE0))
        elif ch == "\u3000":
            out.append(" ")
        else:
            out.append(ch)
    return "".join(out)


def normalize(text: str) -> str:
    text = text.translate(ZERO_WIDTH)
    text = _fold_ascii_width(text)
    text = text.replace("\r\n", "\n").replace("\r", "\n")
    lines = []
    for ln in text.split("\n"):
        ln = re.sub(r"[ \t\u00a0]+", " ", ln).strip()
        if ln:
            lines.append(ln)
    return "\n".join(lines)


def strip_contacts(text: str) -> tuple[str, int, int]:
    urls = len(URL_RE.findall(text))
    handles = len(HANDLE_RE.findall(text))
    core = URL_RE.sub(" ", text)
    core = HANDLE_RE.sub(" ", core)
    core = PHONE_RE.sub(" ", core)
    return core, urls, handles


def signature(text: str) -> str:
    """Near-dup key: case/space/punctuation-insensitive, contacts+digits masked."""
    s = text.lower()
    s = URL_RE.sub(" <url> ", s)
    s = HANDLE_RE.sub(" <at> ", s)
    s = PHONE_RE.sub(" <phone> ", s)
    s = re.sub(r"\d+", " <n> ", s)
    s = re.sub(r"[\W_]+", "", s, flags=re.U)
    return hashlib.sha1(s.encode("utf-8")).hexdigest()[:16]


def _score_blob(blob: str) -> tuple[dict[str, int], dict[str, list[str]]]:
    low = blob.lower()
    scores = {k: 0 for k in KINDS}
    hits: dict[str, list[str]] = {k: [] for k in KINDS}
    for kind, rules in RULES.items():
        for kw, w in rules:
            if kw in low:
                scores[kind] += w
                hits[kind].append(kw)
    for kind, rules in REGEX_RULES.items():
        for pat, w in rules:
            if re.search(pat, blob):
                scores[kind] += w
                hits[kind].append(pat)
    return scores, hits


def classify(text: str, channel: str | None = None) -> tuple[str, int, list[str]]:
    scores, hits = _score_blob(text)
    if channel:
        cs, _ = _score_blob(channel)
        for kind, val in cs.items():
            scores[kind] += val // 2
    best = max(scores, key=lambda k: (scores[k], -KINDS.index(k)))
    if scores[best] <= 0:
        return "unknown", 0, []
    if scores[best] < 3 and not (channel and scores[best] >= 2):
        return "unknown", scores[best], hits[best][:8]
    return best, scores[best], hits[best][:8]


def quality_flags(text: str, urls: int, handles: int) -> list[str]:
    flags: list[str] = []
    core, _, _ = strip_contacts(text)
    compact = re.sub(r"\s+", "", core)
    if len(compact) < 4 and not urls and not handles:
        flags.append("low_signal")
    if PLACEHOLDER.search(text):
        flags.append("placeholder")
    if len(compact) >= 4 and not (urls or handles or PHONE_RE.search(text)) \
            and not re.search(r"联系|私|进群|频道|头像|简介|主页|客服|加", text):
        flags.append("no_contact")
    if re.search(r"(.)\1{5,}", compact):
        flags.append("repeated_chars")
    return flags


def parse_source_msg(info: str) -> tuple[str | None, str]:
    """ExReply records are shaped 'Source: <channel>\\nMsg: <message>'."""
    m = re.match(r"Source:\s*(.*?)\nMsg:\s*(.*)$", info, re.S)
    if m:
        return m.group(1).strip(), m.group(2).strip()
    return None, info


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--store", type=Path, default=Path("data/raw/gmeow/bans.jsonl"))
    ap.add_argument("--out", type=Path, default=Path("data/cleaned/gmeow"))
    args = ap.parse_args()
    args.out.mkdir(parents=True, exist_ok=True)

    raw = [json.loads(l) for l in args.store.open(encoding="utf-8") if l.strip()]
    groups: dict[str, dict] = {}
    dropped = collections.Counter()

    for rec in raw:
        channel, msg = parse_source_msg(rec["info"])
        text = normalize(msg)
        if not text:
            dropped["empty"] += 1
            continue
        flags = quality_flags(text, *strip_contacts(text)[1:])
        if "placeholder" in flags:
            dropped["placeholder_image"] += 1
            continue
        key = text
        g = groups.get(key)
        if g is None:
            g = groups[key] = {
                "text": text,
                "channel": channel,
                "reasons": collections.Counter(),
                "tgids": set(),
                "chat_ids": set(),
                "first_seen": rec["recorded_at"],
                "last_seen": rec["recorded_at"],
                "sources": collections.Counter(),
                "n": 0,
                "flags": flags,
            }
        g["n"] += 1
        g["reasons"][rec["reason"]] += 1
        g["tgids"].add(rec["tgid"])
        if rec["chat_id"] is not None:
            g["chat_ids"].add(rec["chat_id"])
        g["sources"][rec["source"]] += 1
        g["first_seen"] = min(g["first_seen"], rec["recorded_at"])
        g["last_seen"] = max(g["last_seen"], rec["recorded_at"])

    rows = []
    campaigns: dict[str, list[dict]] = collections.defaultdict(list)
    for g in groups.values():
        kind, score, hit_words = classify(g["text"], g["channel"])
        core_letters = len(re.sub(r"[\W_]+", "", g["text"], flags=re.U))
        row = {
            "text": g["text"],
            "label": "ad",
            "kind": kind,
            "kind_score": score,
            "kind_hits": hit_words[:8],
            "quality": g["flags"],
            "n_records": g["n"],
            "n_accounts": len(g["tgids"]),
            "n_chats": len(g["chat_ids"]),
            "reasons": dict(g["reasons"]),
            "source_channel": g["channel"],
            "first_seen": g["first_seen"],
            "last_seen": g["last_seen"],
            "sources": dict(g["sources"]),
            "chars": core_letters,
        }
        rows.append(row)
        campaigns[signature(g["text"])].append(row)

    rows.sort(key=lambda r: (-r["n_accounts"], -r["n_records"], r["text"]))
    with (args.out / "ads_rules.jsonl").open("w", encoding="utf-8") as fh:
        for row in rows:
            fh.write(json.dumps(row, ensure_ascii=False) + "\n")

    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"],
            "examples": [m["text"] for m in members[:5]],
        })
    camp_rows.sort(key=lambda c: (-c["variants"], -c["accounts"]))
    with (args.out / "campaigns.jsonl").open("w", encoding="utf-8") as fh:
        for row in camp_rows:
            fh.write(json.dumps(row, ensure_ascii=False) + "\n")

    kind_dist = collections.Counter(r["kind"] for r in rows)
    keep = [r for r in rows if "low_signal" not in r["quality"]]
    stats = {
        "raw_records": len(raw),
        "unique_texts": len(rows),
        "trainable_texts": len(keep),
        "campaigns": len(camp_rows),
        "dropped": dict(dropped),
        "kind_dist": dict(kind_dist),
        "kind_dist_trainable": dict(collections.Counter(r["kind"] for r in keep)),
        "quality_flags": dict(collections.Counter(
            f for r in rows for f in r["quality"])),
        "sources": dict(collections.Counter(
            s for r in rows for s, c in r["sources"].items())),
        "accounts": sum(r["n_accounts"] for r in rows),
        "time_span": [min(r["first_seen"] for r in rows),
                      max(r["last_seen"] for r in rows)] if rows else [],
    }
    (args.out / "stats.json").write_text(
        json.dumps(stats, ensure_ascii=False, indent=2), encoding="utf-8")

    lines = [
        "# gmeow 封禁广告语料 · 初次清理",
        "",
        f"- 原始记录：{stats['raw_records']}",
        f"- 去重后唯一文本：{stats['unique_texts']}（可训练 {stats['trainable_texts']}）",
        f"- 近重复战役簇：{stats['campaigns']}",
        f"- 丢弃：{stats['dropped'] or '无'}",
        f"- 类别分布：{stats['kind_dist']}",
        f"- 质量标记：{stats['quality_flags']}",
        "",
        "## 账号数最多的 15 条",
        "",
        "| 账号数 | 类别 | 文本 |",
        "|---:|---|---|",
    ]
    for r in rows[:15]:
        text = r["text"].replace("|", "\\|").replace("\n", " ")
        lines.append(f"| {r['n_accounts']} | {r['kind']} | {text[:80]} |")
    lines += ["", "## 最大的 10 个近重复战役簇", "", "| 变体数 | 账号数 | 类别 | 代表文本 |",
              "|---:|---:|---|---|"]
    for c in camp_rows[:10]:
        text = c["canonical"].replace("|", "\\|").replace("\n", " ")
        lines.append(f"| {c['variants']} | {c['accounts']} | {c['kind']} | {text[:70]} |")
    (args.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())

	#!/usr/bin/env python3
	"""Initial cleaning for gmeow ban records -> unique ad texts.

	Input : data/raw/gmeow/bans.jsonl (collector output)
	Output: data/cleaned/gmeow/ads_rules.jsonl one row per unique text
	data/cleaned/gmeow/campaigns.jsonl near-dup groups (contact/URL stripped)
	data/cleaned/gmeow/stats.json
	data/cleaned/gmeow/report.md

	Stages: parse -> normalize -> drop non-text placeholders -> exact dedupe ->
	heuristic kind labeling (7 kinds used by the classifier) -> campaign grouping.
	"""
	from __future__ import annotations

	import argparse
	import collections
	import hashlib
	import json
	import re
	import unicodedata
	from pathlib import Path

	KINDS = ["crypto", "gambling", "porn", "porn_bait", "promo", "scam", "spam_flood"]

	# Regex rules scored like keywords; needed for short lures ("抢了这么多u").
	REGEX_RULES: dict[str, list[tuple[str, int]]] = {
	"crypto": [
	(r"\d+\s*[uU](?![A-Za-z])", 4), # 200U / 10u
	(r"(?:抢\|领\|送\|白嫖\|空投\|白送\|开局)[^。！\n]{0,8}[uU]", 4),
	(r"(?:空投\|白送\|免费领\|零撸\|撸毛)", 4),
	(r"(?:回血\|带单\|跟单\|月化\|翻倍\|策略\|走势\|合约)", 2),
	],
	"gambling": [
	(r"(?:包赢\|稳赚\|必中\|红单\|上岸\|回血)", 3),
	(r"(?:实力平台\|顶级平台\|安全首选\|押金保障)", 3),
	(r"(?:pc28\|pg电子\|pp电子\|jdb\|百家乐\|时时彩)", 5),
	],
	"scam": [
	(r"(?:一单\|每单\|单结\|结算)[^。！\n]{0,6}\d+", 4),
	(r"\d+\s(?:元\|块\|刀\|u\|U)\s(?:一单\|每单\|/单)", 4),
	(r"(?:日赚\|月入\|日入\|躺赚\|日结\|轻松赚)\s\d", 4),
	(r"(?:收资\|资单\|收单\|代收\|跑分\|洗u\|黑u\|白资)", 4),
	(r"(?:解封\|找回\|追款\|查档\|户籍\|定位\|开房记录)", 3),
	(r"(?:兼职\|招聘\|招人\|找人\|做事\|干活\|小白可做\|无本)", 2),
	(r"(?:留资\|资方\|卡商\|号商\|接码\|实名卡)", 4),
	(r"(?:扫码\|二维码\|捡签\|有码)", 2),
	],
	"porn_bait": [
	(r"(?:看\|点\|戳)(?:我)?(?:的)?(?:主页\|头像\|简介\|煮页\|朋友圈\|签名)", 3),
	],
	"promo": [
	(r"(?:加群\|进群\|频道\|客服\|主号\|双向\|机器人)", 2),
	(r"(?:代开\|中转站\|全套业务\|会员代开\|充值折扣\|服务器\|专线)", 2),
	(r"(?:一家\|专业\|源头\|实力\|诚信\|靠谱)", 1),
	],
	}

	# keyword -> weight. Weights are per kind; first hit wins ties by list order.
	RULES: dict[str, list[tuple[str, int]]] = {
	"crypto": [
	("空投", 4), ("usdt", 4), ("泰达", 4), ("钱包", 3), ("助记词", 4),
	("合约", 3), ("eth", 3), ("btc", 3), ("比特币", 3), ("炒币", 4),
	("挖矿", 3), ("交易所", 3), ("量化", 3), ("搬砖", 3), ("u币", 4),
	("波场", 3), ("trx", 2), ("defi", 3), ("币圈", 3), ("usd", 2),
	],
	"gambling": [
	("博彩", 5), ("赌博", 5), ("彩票", 4), ("时时彩", 5), ("快三", 5),
	("棋牌", 4), ("娱乐城", 4), ("上分", 3), ("下分", 3), ("红单", 4),
	("包赔", 3), ("走势", 2), ("pc28", 5), ("pg电子", 4), ("电子游艺", 4),
	("百家乐", 5), ("六合彩", 5), ("足球", 2), ("篮球", 2), ("稳赚", 2),
	("平台", 1), ("押金", 2), ("担保", 2), ("充", 1), ("提现", 2),
	],
	"porn": [
	("楼凤", 5), ("上门", 3), ("全套", 4), ("快餐", 3), ("课表", 4),
	("约课", 4), ("约炮", 5), ("做爱", 5), ("裸聊", 5), ("福利视频", 4),
	("幼女", 5), ("萝莉", 4), ("学生妹", 4), ("人妻", 3), ("少妇", 3),
	("母狗", 5), ("骚逼", 5), ("调教", 3), ("sm圈", 4), ("选妃", 4),
	("外围", 4), ("探花", 4), ("福利姬", 4), ("视频", 1), ("妹妹", 2),
	("生活圈", 3), ("实照", 3), ("真人", 2), ("口嗨", 2), ("上门服务", 3),
	],
	"porn_bait": [
	("看主页", 4), ("点我头像", 4), ("点击头像", 4), ("看简介", 4),
	("简介看", 4), ("煮页", 4), ("主页", 2), ("签名", 2), ("看片", 3),
	("进群看", 3), ("点击进入", 3), ("戳这里", 3), ("扫码", 3),
	],
	"promo": [
	("广告代发", 5), ("代发", 3), ("群发", 4), ("引流", 4), ("涨粉", 4),
	("刷量", 4), ("推广", 3), ("曝光", 2), ("项目招商", 3), ("招商", 2),
	("招代理", 3), ("代理", 1), ("频道", 1), ("进群", 1), ("客服", 2),
	],
	"scam": [
	("兼职", 4), ("日赚", 5), ("日结", 4), ("月入", 4), ("刷单", 5),
	("垫付", 4), ("解封", 5), ("找回", 3), ("追款", 4), ("贷款", 3),
	("接码", 5), ("卖号", 4), ("买号", 4), ("号商", 5), ("社工库", 5),
	("实名", 3), ("接单", 3), ("跑分", 4), ("代收", 3), ("洗钱", 5),
	("洗u", 4), ("黑u", 4), ("假钞", 5), ("空壳", 3), ("盗号", 4),
	("劫持", 4), ("协议号", 4), ("蓝标", 3), ("赚钱", 2), ("做单", 3),
	("一单", 3), ("佣金", 3), ("结算", 2), ("私聊", 2), ("咨询", 2),
	],
	"spam_flood": [
	("加我", 1), ("互粉", 2), ("互关", 2), ("点赞", 2), ("评论", 2),
	],
	}

	PLACEHOLDER = re.compile(
	r"相片內含有廣告\|相片内含有广告\|Image contains Ads\|TG違規內容\|TG违规内容", re.I)
	URL_RE = re.compile(r"(?:https?://\|www\.)\S+\|t\.me/\S+", re.I)
	HANDLE_RE = re.compile(r"@[A-Za-z0-9_]{3,}")
	PHONE_RE = re.compile(r"(?<!\d)1[3-9]\d{9}(?!\d)")
	ZERO_WIDTH = dict.fromkeys(map(ord, "\u200b\u200c\u200d\ufeff\u2060"))


	def _fold_ascii_width(text: str) -> str:
	"""Fold full-width alphanumerics only; Chinese punctuation stays intact."""
	out = []
	for ch in text:
	o = ord(ch)
	if 0xFF21 <= o <= 0xFF3A or 0xFF41 <= o <= 0xFF5A or 0xFF10 <= o <= 0xFF19:
	out.append(chr(o - 0xFEE0))
	elif ch == "\u3000":
	out.append(" ")
	else:
	out.append(ch)
	return "".join(out)


	def normalize(text: str) -> str:
	text = text.translate(ZERO_WIDTH)
	text = _fold_ascii_width(text)
	text = text.replace("\r\n", "\n").replace("\r", "\n")
	lines = []
	for ln in text.split("\n"):
	ln = re.sub(r"[ \t\u00a0]+", " ", ln).strip()
	if ln:
	lines.append(ln)
	return "\n".join(lines)


	def strip_contacts(text: str) -> tuple[str, int, int]:
	urls = len(URL_RE.findall(text))
	handles = len(HANDLE_RE.findall(text))
	core = URL_RE.sub(" ", text)
	core = HANDLE_RE.sub(" ", core)
	core = PHONE_RE.sub(" ", core)
	return core, urls, handles


	def signature(text: str) -> str:
	"""Near-dup key: case/space/punctuation-insensitive, contacts+digits masked."""
	s = text.lower()
	s = URL_RE.sub(" <url> ", s)
	s = HANDLE_RE.sub(" <at> ", s)
	s = PHONE_RE.sub(" <phone> ", s)
	s = re.sub(r"\d+", " <n> ", s)
	s = re.sub(r"[\W_]+", "", s, flags=re.U)
	return hashlib.sha1(s.encode("utf-8")).hexdigest()[:16]


	def _score_blob(blob: str) -> tuple[dict[str, int], dict[str, list[str]]]:
	low = blob.lower()
	scores = {k: 0 for k in KINDS}
	hits: dict[str, list[str]] = {k: [] for k in KINDS}
	for kind, rules in RULES.items():
	for kw, w in rules:
	if kw in low:
	scores[kind] += w
	hits[kind].append(kw)
	for kind, rules in REGEX_RULES.items():
	for pat, w in rules:
	if re.search(pat, blob):
	scores[kind] += w
	hits[kind].append(pat)
	return scores, hits


	def classify(text: str, channel: str \| None = None) -> tuple[str, int, list[str]]:
	scores, hits = _score_blob(text)
	if channel:
	cs, _ = _score_blob(channel)
	for kind, val in cs.items():
	scores[kind] += val // 2
	best = max(scores, key=lambda k: (scores[k], -KINDS.index(k)))
	if scores[best] <= 0:
	return "unknown", 0, []
	if scores[best] < 3 and not (channel and scores[best] >= 2):
	return "unknown", scores[best], hits[best][:8]
	return best, scores[best], hits[best][:8]


	def quality_flags(text: str, urls: int, handles: int) -> list[str]:
	flags: list[str] = []
	core, _, _ = strip_contacts(text)
	compact = re.sub(r"\s+", "", core)
	if len(compact) < 4 and not urls and not handles:
	flags.append("low_signal")
	if PLACEHOLDER.search(text):
	flags.append("placeholder")
	if len(compact) >= 4 and not (urls or handles or PHONE_RE.search(text)) \
	and not re.search(r"联系\|私\|进群\|频道\|头像\|简介\|主页\|客服\|加", text):
	flags.append("no_contact")
	if re.search(r"(.)\1{5,}", compact):
	flags.append("repeated_chars")
	return flags


	def parse_source_msg(info: str) -> tuple[str \| None, str]:
	"""ExReply records are shaped 'Source: <channel>\\nMsg: <message>'."""
	m = re.match(r"Source:\s(.?)\nMsg:\s(.)$", info, re.S)
	if m:
	return m.group(1).strip(), m.group(2).strip()
	return None, info


	def main() -> int:
	ap = argparse.ArgumentParser()
	ap.add_argument("--store", type=Path, default=Path("data/raw/gmeow/bans.jsonl"))
	ap.add_argument("--out", type=Path, default=Path("data/cleaned/gmeow"))
	args = ap.parse_args()
	args.out.mkdir(parents=True, exist_ok=True)

	raw = [json.loads(l) for l in args.store.open(encoding="utf-8") if l.strip()]
	groups: dict[str, dict] = {}
	dropped = collections.Counter()

	for rec in raw:
	channel, msg = parse_source_msg(rec["info"])
	text = normalize(msg)
	if not text:
	dropped["empty"] += 1
	continue
	flags = quality_flags(text, *strip_contacts(text)[1:])
	if "placeholder" in flags:
	dropped["placeholder_image"] += 1
	continue
	key = text
	g = groups.get(key)
	if g is None:
	g = groups[key] = {
	"text": text,
	"channel": channel,
	"reasons": collections.Counter(),
	"tgids": set(),
	"chat_ids": set(),
	"first_seen": rec["recorded_at"],
	"last_seen": rec["recorded_at"],
	"sources": collections.Counter(),
	"n": 0,
	"flags": flags,
	}
	g["n"] += 1
	g["reasons"][rec["reason"]] += 1
	g["tgids"].add(rec["tgid"])
	if rec["chat_id"] is not None:
	g["chat_ids"].add(rec["chat_id"])
	g["sources"][rec["source"]] += 1
	g["first_seen"] = min(g["first_seen"], rec["recorded_at"])
	g["last_seen"] = max(g["last_seen"], rec["recorded_at"])

	rows = []
	campaigns: dict[str, list[dict]] = collections.defaultdict(list)
	for g in groups.values():
	kind, score, hit_words = classify(g["text"], g["channel"])
	core_letters = len(re.sub(r"[\W_]+", "", g["text"], flags=re.U))
	row = {
	"text": g["text"],
	"label": "ad",
	"kind": kind,
	"kind_score": score,
	"kind_hits": hit_words[:8],
	"quality": g["flags"],
	"n_records": g["n"],
	"n_accounts": len(g["tgids"]),
	"n_chats": len(g["chat_ids"]),
	"reasons": dict(g["reasons"]),
	"source_channel": g["channel"],
	"first_seen": g["first_seen"],
	"last_seen": g["last_seen"],
	"sources": dict(g["sources"]),
	"chars": core_letters,
	}
	rows.append(row)
	campaigns[signature(g["text"])].append(row)

	rows.sort(key=lambda r: (-r["n_accounts"], -r["n_records"], r["text"]))
	with (args.out / "ads_rules.jsonl").open("w", encoding="utf-8") as fh:
	for row in rows:
	fh.write(json.dumps(row, ensure_ascii=False) + "\n")

	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"],
	"examples": [m["text"] for m in members[:5]],
	})
	camp_rows.sort(key=lambda c: (-c["variants"], -c["accounts"]))
	with (args.out / "campaigns.jsonl").open("w", encoding="utf-8") as fh:
	for row in camp_rows:
	fh.write(json.dumps(row, ensure_ascii=False) + "\n")

	kind_dist = collections.Counter(r["kind"] for r in rows)
	keep = [r for r in rows if "low_signal" not in r["quality"]]
	stats = {
	"raw_records": len(raw),
	"unique_texts": len(rows),
	"trainable_texts": len(keep),
	"campaigns": len(camp_rows),
	"dropped": dict(dropped),
	"kind_dist": dict(kind_dist),
	"kind_dist_trainable": dict(collections.Counter(r["kind"] for r in keep)),
	"quality_flags": dict(collections.Counter(
	f for r in rows for f in r["quality"])),
	"sources": dict(collections.Counter(
	s for r in rows for s, c in r["sources"].items())),
	"accounts": sum(r["n_accounts"] for r in rows),
	"time_span": [min(r["first_seen"] for r in rows),
	max(r["last_seen"] for r in rows)] if rows else [],
	}
	(args.out / "stats.json").write_text(
	json.dumps(stats, ensure_ascii=False, indent=2), encoding="utf-8")

	lines = [
	"# gmeow 封禁广告语料 · 初次清理",
	"",
	f"- 原始记录：{stats['raw_records']}",
	f"- 去重后唯一文本：{stats['unique_texts']}（可训练 {stats['trainable_texts']}）",
	f"- 近重复战役簇：{stats['campaigns']}",
	f"- 丢弃：{stats['dropped'] or '无'}",
	f"- 类别分布：{stats['kind_dist']}",
	f"- 质量标记：{stats['quality_flags']}",
	"",
	"## 账号数最多的 15 条",
	"",
	"\| 账号数 \| 类别 \| 文本 \|",
	"\|---:\|---\|---\|",
	]
	for r in rows[:15]:
	text = r["text"].replace("\|", "\\\|").replace("\n", " ")
	lines.append(f"\| {r['n_accounts']} \| {r['kind']} \| {text[:80]} \|")
	lines += ["", "## 最大的 10 个近重复战役簇", "", "\| 变体数 \| 账号数 \| 类别 \| 代表文本 \|",
	"\|---:\|---:\|---\|---\|"]
	for c in camp_rows[:10]:
	text = c["canonical"].replace("\|", "\\\|").replace("\n", " ")
	lines.append(f"\| {c['variants']} \| {c['accounts']} \| {c['kind']} \| {text[:70]} \|")
	(args.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())