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