| |
| """Real-time Jev cleaning for gmeow ban records. |
| |
| Maps each ban record to the jev request state schema and runs the four |
| questions (is_ad / ad_kind / ad_scope / severity). Unique texts are judged |
| once; results land in an append-only JSONL that doubles as the resume state. |
| |
| Record semantics: |
| Reason=UserName -> the ban was triggered by nickname/bio: text goes to |
| sender.first_name + sender.bio, message.text stays empty |
| Reason=SpamGP -> a group message: text goes to message.text |
| Reason=ExReply -> reply under an external channel post ("Source:"/"Msg:"): |
| Msg goes to message.text, Source becomes chat.title |
| |
| Usage: |
| python scripts/jev_clean_gmeow.py --once |
| python scripts/jev_clean_gmeow.py --watch --interval 20 |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import hashlib |
| import json |
| import re |
| import sys |
| import threading |
| import time |
| import urllib.error |
| import urllib.request |
| from concurrent.futures import ThreadPoolExecutor, as_completed |
| from pathlib import Path |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parent)) |
| from clean_gmeow_ads import ( |
| classify, normalize, parse_source_msg, quality_flags, strip_contacts) |
|
|
| TEMPLATE = Path(".work/teacher/jev_request_template.json") |
| LINK_RE = re.compile(r"https?://|t\.me/|www\.", re.I) |
| HANDLE_RE = re.compile(r"@([A-Za-z0-9_]{3,})") |
|
|
|
|
| def load_env(path: str = ".work/teacher/.env") -> dict: |
| cfg = {} |
| for line in Path(path).read_text(encoding="utf-8").splitlines(): |
| if "=" in line: |
| k, v = line.split("=", 1) |
| cfg[k.strip()] = v.strip() |
| return cfg |
|
|
|
|
| def text_key(text: str) -> str: |
| return hashlib.sha1(text.encode("utf-8")).hexdigest()[:16] |
|
|
|
|
| def make_state(group: dict, template_state: dict) -> dict: |
| """Build the jev state for one deduped text group.""" |
| text = group["text"] |
| reason = group["reason"] |
| channel = group.get("channel") |
| base = json.loads(json.dumps(template_state)) |
| base["chat"] = {"id": next(iter(group["chat_ids"]), None), |
| "title": channel or ""} |
| base["message"].update({ |
| "text": "", |
| "has_link": bool(LINK_RE.search(text)), |
| "has_mention": bool(HANDLE_RE.search(text)), |
| "has_media": reason == "Ad Image", |
| "is_forwarded": reason in ("ExReply", "SpamGP"), |
| "length": len(text), |
| }) |
| sender = base["sender"] |
| sender.update({ |
| "user_id": next(iter(group["tgids"]), None), |
| "username": (HANDLE_RE.search(text).group(1) |
| if reason == "UserName" and HANDLE_RE.search(text) else ""), |
| "first_name": "", |
| "bio": "", |
| "bio_links": [], |
| "is_premium": False, |
| "age_hours": 0, |
| "age_known": False, |
| "msgs_in_group": 0, |
| "prior_ad_hits": group["n"], |
| }) |
| if reason == "UserName": |
| nick, sep, bio = text.partition("||") |
| sender["first_name"] = nick.strip() |
| sender["bio"] = bio.strip() if sep else text |
| else: |
| base["message"]["text"] = text |
| sender["first_name"] = "群成员" |
| |
| base["known_ad_patterns"] = f"反广告机器人已按 reason={reason} 封禁该账号" |
| base["matched_rules"] = [ |
| {"id": i + 1, "name": kw, "category": group["rule_kind"], "note": text[:200]} |
| for i, kw in enumerate(group["rule_hits"][:5]) |
| ] |
| return base |
|
|
|
|
| def request_teacher(cfg: dict, questions: dict, state: dict, retries: int = 8): |
| url = cfg["TEACHER_URL"] |
| payload = json.dumps( |
| {"model": cfg.get("TEACHER_MODEL", "jev-1.13"), |
| "questions": questions, "state": state}, |
| ensure_ascii=False).encode("utf-8") |
| last = None |
| for attempt in range(retries): |
| req = urllib.request.Request(url, data=payload, method="POST", headers={ |
| "Authorization": "Bearer " + cfg["TEACHER_KEY"], |
| "Content-Type": "application/json", |
| "Accept": "application/json", |
| "Accept-Encoding": "identity", |
| "User-Agent": "laya-ad-gmeow/1.0"}) |
| try: |
| with urllib.request.urlopen(req, timeout=180) as resp: |
| return json.loads(resp.read().decode("utf-8", "replace")) |
| except urllib.error.HTTPError as exc: |
| body = exc.read().decode("utf-8", "replace")[:300] |
| last = f"HTTP {exc.code}: {body}" |
| if exc.code == 429: |
| continue |
| if exc.code in (500, 502, 503, 504): |
| time.sleep(2 * (attempt + 1)) |
| continue |
| break |
| except Exception as exc: |
| last = f"{type(exc).__name__}: {exc}" |
| time.sleep(2 * (attempt + 1)) |
| raise RuntimeError(last) |
|
|
|
|
| def parse_answers(resp: dict) -> dict: |
| a = resp.get("answers") or {} |
| out = {} |
| for qid in ("is_ad", "ad_kind", "ad_scope"): |
| v = a.get(qid) or {} |
| out[qid] = v.get("choice") |
| out[qid + "_conf"] = v.get("confidence") |
| sev = a.get("severity") or {} |
| out["severity"] = sev.get("score") |
| out["severity_conf"] = sev.get("confidence") |
| return out |
|
|
|
|
| def build_groups(store: Path) -> dict[str, dict]: |
| groups: dict[str, dict] = {} |
| for line in store.open(encoding="utf-8"): |
| line = line.strip() |
| if not line: |
| continue |
| rec = json.loads(line) |
| channel, msg = parse_source_msg(rec["info"]) |
| text = normalize(msg) |
| if not text: |
| continue |
| flags = quality_flags(text, *strip_contacts(text)[1:]) |
| if "placeholder" in flags: |
| continue |
| key = text_key(text) |
| g = groups.get(key) |
| if g is None: |
| kind, score, hits = classify(text, channel) |
| g = groups[key] = { |
| "key": key, "text": text, "channel": channel, |
| "reason": rec["reason"], "reasons": {}, |
| "tgids": set(), "chat_ids": set(), "n": 0, |
| "first_seen": rec["recorded_at"], "last_seen": rec["recorded_at"], |
| "flags": flags, "rule_kind": kind, "rule_score": score, |
| "rule_hits": hits, |
| } |
| g["n"] += 1 |
| g["reasons"][rec["reason"]] = g["reasons"].get(rec["reason"], 0) + 1 |
| g["reason"] = max(g["reasons"], key=g["reasons"].get) |
| g["tgids"].add(rec["tgid"]) |
| if rec["chat_id"] is not None: |
| g["chat_ids"].add(rec["chat_id"]) |
| g["first_seen"] = min(g["first_seen"], rec["recorded_at"]) |
| g["last_seen"] = max(g["last_seen"], rec["recorded_at"]) |
| return groups |
|
|
|
|
| def load_done(out: Path) -> set[str]: |
| done: set[str] = set() |
| if out.exists(): |
| for line in out.open(encoding="utf-8"): |
| try: |
| done.add(json.loads(line)["key"]) |
| except Exception: |
| continue |
| return done |
|
|
|
|
| def judge_one(cfg, questions, template_state, group) -> dict: |
| state = make_state(group, template_state) |
| t0 = time.time() |
| resp = request_teacher(cfg, questions, state) |
| ans = parse_answers(resp) |
| return { |
| "key": group["key"], |
| "text": group["text"], |
| "label": ans["is_ad"], |
| "kind": ans["ad_kind"] if ans["ad_kind"] != "none" else "none", |
| "scope": ans["ad_scope"], |
| "severity": ans["severity"], |
| "jev_conf": {"is_ad": ans["is_ad_conf"], "ad_kind": ans["ad_kind_conf"], |
| "ad_scope": ans["ad_scope_conf"], "severity": ans["severity_conf"]}, |
| "rule_kind": group["rule_kind"], |
| "rule_score": group["rule_score"], |
| "rule_hits": group["rule_hits"], |
| "agrees": (ans["ad_kind"] == group["rule_kind"]), |
| "quality": group["flags"], |
| "reason": group["reason"], |
| "n_records": group["n"], |
| "n_accounts": len(group["tgids"]), |
| "n_chats": len(group["chat_ids"]), |
| "channel": group["channel"], |
| "first_seen": group["first_seen"], |
| "last_seen": group["last_seen"], |
| "model": resp.get("model"), |
| "latency_s": round(time.time() - t0, 2), |
| "judged_at": time.strftime("%Y-%m-%dT%H:%M:%S"), |
| } |
|
|
|
|
| def run_pass(args, cfg, questions, template_state, groups, done, out, lock) -> int: |
| pending = [g for k, g in groups.items() if k not in done] |
| if args.limit: |
| pending = pending[:args.limit] |
| if not pending: |
| return 0 |
| written = 0 |
| with ThreadPoolExecutor(max_workers=args.workers) as ex: |
| futs = {ex.submit(judge_one, cfg, questions, template_state, g): g |
| for g in pending} |
| for fut in as_completed(futs): |
| g = futs[fut] |
| try: |
| row = fut.result() |
| except Exception as exc: |
| print(f"[err] key={g['key']} {exc}", flush=True) |
| continue |
| with lock: |
| out.parent.mkdir(parents=True, exist_ok=True) |
| with out.open("a", encoding="utf-8") as fh: |
| fh.write(json.dumps(row, ensure_ascii=False) + "\n") |
| done.add(row["key"]) |
| written += 1 |
| if written % 20 == 0: |
| print(f"[jev] judged {written}/{len(pending)}", flush=True) |
| return written |
|
|
|
|
| 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/ads_jev.jsonl")) |
| ap.add_argument("--template", type=Path, default=TEMPLATE) |
| ap.add_argument("--workers", type=int, default=8) |
| ap.add_argument("--limit", type=int, default=0) |
| ap.add_argument("--once", action="store_true") |
| ap.add_argument("--watch", action="store_true") |
| ap.add_argument("--interval", type=int, default=20) |
| ap.add_argument("--duration", type=int, default=0) |
| args = ap.parse_args() |
|
|
| cfg = load_env() |
| tpl = json.loads(args.template.read_text(encoding="utf-8")) |
| questions, template_state = tpl["questions"], tpl["state"] |
| lock = threading.Lock() |
| started = time.time() |
| total = 0 |
| while True: |
| done = load_done(args.out) |
| groups = build_groups(args.store) |
| new = sum(1 for k in groups if k not in done) |
| print(f"[scan] store_groups={len(groups)} done={len(done)} pending={new}", |
| flush=True) |
| total += run_pass(args, cfg, questions, template_state, groups, done, args.out, lock) |
| if args.once or not args.watch: |
| break |
| if args.duration and time.time() - started >= args.duration: |
| break |
| time.sleep(args.interval) |
| print(f"done judged={total} out={args.out}") |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
