data/spam.py

from __future__ import annotations
import random

try:
    from .taxonomy import SPAM_LABELS
except ImportError:
    from taxonomy import SPAM_LABELS
DIFRAUD = "redasers/difraud"
SMS_SPAM = "ucirvine/sms_spam"
_IDX = {lbl: i for i, lbl in enumerate(SPAM_LABELS)}
_N = len(SPAM_LABELS)
DIFRAUD_DECEPTIVE_LABELS = {
    "phishing": ("phishing", "spam"),
    "job_scams": ("scam", "spam"),
    "sms": ("spam",),
    "fake_news": (),
    "twitter_rumours": (),
    "political_statements": (),
    "product_reviews": (),
}


def _vecs(positives):
    positives = list(positives)
    labels = [0.0] * _N
    if not positives:
        return (labels, [1.0] * _N)
    mask = [0.0] * _N
    for name in positives:
        idx = _IDX[name]
        labels[idx] = 1.0
        mask[idx] = 1.0
    return (labels, mask)


def _get_text(row, *keys):
    for key in keys:
        val = row.get(key)
        if isinstance(val, str) and val.strip():
            return val
    return ""


def _get_label(row):
    for key in ("label", "labels", "is_spam", "class", "target"):
        val = row.get(key)
        if val is None:
            continue
        if isinstance(val, str):
            low = val.strip().lower()
            if low in ("spam", "deceptive", "fraud", "1", "true", "phishing", "scam"):
                return 1
            if low in ("ham", "truthful", "legit", "0", "false"):
                return 0
            continue
        try:
            return 1 if float(val) >= 0.5 else 0
        except (TypeError, ValueError):
            continue
    return None


def _iter_difraud(streaming, max_rows):
    from datasets import load_dataset

    for domain, deceptive_labels in DIFRAUD_DECEPTIVE_LABELS.items():
        try:
            ds = load_dataset(
                "json",
                data_files=f"hf://datasets/{DIFRAUD}/{domain}/train.jsonl",
                split="train",
                streaming=streaming,
            )
        except Exception as err:
            print(f"[spam] {DIFRAUD}:{domain} unavailable ({err}); skipping domain.")
            continue
        emitted = 0
        for ex in ds:
            if max_rows is not None and emitted >= max_rows:
                break
            text = _get_text(ex, "text", "content", "body", "message")
            label = _get_label(ex)
            if not text or label is None:
                continue
            if label == 0:
                yield (text, ())
                emitted += 1
            elif deceptive_labels:
                yield (text, deceptive_labels)
                emitted += 1


def _iter_sms_spam(streaming, max_rows):
    from datasets import load_dataset

    for repo in (SMS_SPAM, "sms_spam"):
        try:
            ds = load_dataset(repo, split="train", streaming=streaming)
            break
        except Exception as err:
            last = err
    else:
        print(f"[spam] {SMS_SPAM} unavailable ({last}); skipping SMS spam.")
        return
    for i, ex in enumerate(ds):
        if max_rows is not None and i >= max_rows:
            break
        text = _get_text(ex, "sms", "text", "message")
        label = _get_label(ex)
        if not text or label is None:
            continue
        yield (text, ("spam",) if label == 1 else ())


def load_spam_examples(max_rows=None, streaming=True):
    iters = [_iter_difraud(streaming, max_rows), _iter_sms_spam(streaming, max_rows)]
    live = list(iters)
    while live:
        for it in list(live):
            try:
                text, positives = next(it)
            except StopIteration:
                live.remove(it)
                continue
            labels, mask = _vecs(positives)
            yield {"text": text, "labels": labels, "mask": mask}


def _scam_prize(f):
    return f"CONGRATULATIONS {f.first_name().upper()}! You have WON {f.currency_symbol()}{f.random_int(50000, 5000000):,} in the {f.company()} International Lottery. To claim your prize send your full name and bank details to {f.email()} within 48 hours."


def _scam_crypto(f):
    return f"URGENT investment opportunity: turn {f.currency_symbol()}250 into {f.currency_symbol()}{f.random_int(5000, 90000):,} in 7 days with our AI crypto bot. Guaranteed returns, zero risk. DM {f.user_name()} or wallet bc1{f.bothify('?' * 30).lower()} to start now."


def _scam_giftcard(f):
    return f"Hi it's {f.first_name()} from HR. I'm in a meeting and need you to buy {f.random_int(3, 10)} {f.random_element(('Apple', 'Amazon', 'Google Play'))} gift cards ({f.currency_symbol()}100 each) for a client. Scratch the codes and text them to {f.phone_number()} asap, I'll reimburse you."


def _phish_login(f):
    return f"[{f.company()}] Security alert: unusual sign-in to your account was detected. Verify your identity within 24h or your account will be suspended: http://{f.domain_word()}-secure-{f.random_int(10, 99)}.{f.tld()}/verify?id={f.uuid4()}"


def _phish_bank(f):
    return f"Dear customer, your {f.company()} debit card ending {f.random_int(1000, 9999)} has been temporarily locked. Reconfirm your card number and PIN here to restore access: https://{f.domain_word()}.{f.tld()}/unlock"


def _phish_delivery(f):
    return f"Your parcel {f.bothify('??########').upper()} could not be delivered due to an unpaid fee of {f.currency_symbol()}{f.random_int(1, 4)}.99. Pay now to reschedule: http://{f.domain_word()}-{f.random_int(1, 99)}.{f.tld()}/pay"


def _spam_promo(f):
    return f"LAST CHANCE! {f.random_int(50, 90)}% OFF everything at {f.company()} this weekend only. Free shipping on orders over {f.currency_symbol()}25. Shop now: {f.url()} Reply STOP to opt out."


def _ham_short(f):
    return f.random_element(
        (
            lambda: (
                f"Hey {f.first_name()}, are we still on for lunch at {f.time()} tomorrow?"
            ),
            lambda: (
                f"Thanks for sending the report — I'll review it and get back to you by {f.day_of_week()}."
            ),
            lambda: (
                f"Reminder: the {f.company()} team standup moved to {f.time()} in room {f.random_int(100, 400)}."
            ),
            lambda: (
                "Got it, the package arrived this morning. Appreciate the quick turnaround!"
            ),
            lambda: (
                f"Can you forward me the slides from {f.first_name()}'s presentation when you get a sec?"
            ),
        )
    )()


_RECIPES = [
    (("scam", "spam"), [_scam_prize, _scam_crypto, _scam_giftcard]),
    (("phishing", "spam"), [_phish_login, _phish_bank, _phish_delivery]),
    (("spam",), [_spam_promo]),
    ((), [_ham_short]),
]


def synth_spam(n, seed=0):
    from faker import Faker

    faker = Faker()
    Faker.seed(seed)
    random.seed(seed)
    weighted = (
        [_RECIPES[0]] * 3 + [_RECIPES[1]] * 3 + [_RECIPES[2]] * 1 + [_RECIPES[3]] * 7
    )
    for _ in range(n):
        positives, builders = random.choice(weighted)
        text = random.choice(builders)(faker)
        labels = [0.0] * _N
        for name in positives:
            labels[_IDX[name]] = 1.0
        yield {"text": text, "labels": labels, "mask": [1.0] * _N}


def licenses():
    return [
        (DIFRAUD, "MIT", False),
        (SMS_SPAM, "CC-BY-4.0", True),
        ("Faker", "MIT", False),
    ]