#!/usr/bin/env python3
"""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 (  # noqa: E402
    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))  # deep copy
    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"] = "群成员"
    # the ban list itself is evidence: tell the model what the bot flagged
    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:  # noqa: BLE001
            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:  # noqa: BLE001
                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:  # noqa: BLE001
                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())