notd5a/de-scam-berta-v3-hybrid-detector-v0.2.4

                Predicted Benign    Predicted Spam
True Benign             28,817               359
True Spam                  380            10,695

                Predicted Benign    Predicted Spam
True Benign             27,854             1,322
True Spam                1,066            10,009

Input Message
    |
    ├── len <= 60 chars ──────────► CharCNN (100%)
    |
    ├── 60 < len < 120 chars ────► Sigmoid ensemble blend
    |                                 prob = (1-w)*cnn + w*deberta
    |                                 w = sigmoid((len - 90) / 10)
    |
    └── len >= 120 chars ─────────► DeBERTa (100%)

Input Text
    └─► DeBERTa-v3-base encoder (gradient checkpointed)
            ├─► [CLS] embedding (768d)
            └─► Attention-weighted pooling (768d)
                                                    ┐
23 Engineered Features                              │
    └─► Linear(23→128) + LayerNorm + GELU           ├─► concat (1664d)
                                                    ┘
        └─► Linear(1664→256) + LayerNorm + GELU
            └─► Residual block (256d → 256d)
                └─► Linear(256→2) → spam logits

Character IDs (max 160 chars)
    └─► Embedding(92, 64)
        ├─► Conv1d(64, 128, kernel=2) + BN + ReLU → MaxPool
        ├─► Conv1d(64, 128, kernel=3) + BN + ReLU → MaxPool
        └─► Conv1d(64, 128, kernel=5) + BN + ReLU → MaxPool
                                                            ┐
            Concat (384d)                                   │
                                                            ├─► concat (448d)
23 Features → Linear(23→64) + LayerNorm + GELU (64d)       │
                                                            ┘
    └─► Linear(448→128) + LayerNorm + GELU + Dropout
        └─► Linear(128→2) → spam logits

import re
import math
import json
import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
from collections import Counter
from transformers import AutoTokenizer, AutoModel
from huggingface_hub import hf_hub_download
import joblib

# ── Feature extraction (must match training) ────────────────────────────────

URGENCY_WORDS = {
    "urgent", "immediately", "expires", "verify", "confirm", "suspended",
    "locked", "alert", "action required", "limited time", "click here",
    "act now", "final notice", "winner", "prize", "claim", "free",
    "blocked", "deactivated", "unusual activity",
}
URL_PATTERN      = re.compile(r'(https?://|www\.)\S+|\w+\.(com|net|org|io|co|uk)', re.I)
SHORTENED        = {"bit.ly","tinyurl.com","goo.gl","t.co","ow.ly","smsg.io","rb.gy"}
PHONE_PATTERN    = re.compile(r'(\+?\d[\d\s\-().]{7,}\d)')
EMAIL_PATTERN    = re.compile(r'[\w.+-]+@[\w-]+\.[a-z]{2,}', re.I)
CURRENCY_PATTERN = re.compile(r'[$\xa3\u20ac\u20b9\xa5]|(usd|gbp|eur|inr)', re.I)
LEET_MAP         = str.maketrans("013457@!", "oieastai")
OBFUSCATED_URL   = re.compile(
    r"(https?(?:clue|[a-z]{4,}[a-z0-9]{2,})\b)"
    r"|(?:h\s*t\s*t\s*p)"
    r"|(?:www\s*\.\s*\w)"
    r"|(?:\w+\s*\.\s*(?:com|net|org|xyz|info|co)\b)", re.I)
SPACED_WORD      = re.compile(r"\b(?:\w\s){3,}\w\b")


def extract_features(text):
    """Extract all 23 features for a single message."""
    words   = text.split()
    letters = [c for c in text if c.isalpha()]
    chars   = list(text)
    n       = len(chars)

    original = [
        len(text), len(words),
        sum(len(w) for w in words) / max(len(words), 1),
        sum(1 for c in letters if c.isupper()) / max(len(letters), 1),
        sum(1 for c in text if c.isdigit()) / max(len(text), 1),
        sum(1 for c in text if not c.isalnum() and not c.isspace()) / max(len(text), 1),
        text.count('!'), text.count('?'),
        int(bool(URL_PATTERN.search(text))), len(URL_PATTERN.findall(text)),
        int(any(d in text.lower() for d in SHORTENED)),
        int(bool([m for m in PHONE_PATTERN.findall(text) if len(re.sub(r'\D','',m)) >= 7])),
        int(bool(EMAIL_PATTERN.search(text))), int(bool(CURRENCY_PATTERN.search(text))),
        sum(1 for w in URGENCY_WORDS if w in text.lower()),
    ]

    non_ascii = sum(1 for c in chars if ord(c) > 127)
    counts = Counter(text.lower())
    entropy = -sum((c/n)*math.log2(c/n) for c in counts.values() if c > 0) if n > 0 else 0.0
    translated = text.translate(LEET_MAP)
    leet = sum(1 for a, b in zip(text, translated) if a != b)
    mdr, cr = 0, 0
    for c in chars:
        if c.isdigit(): cr += 1; mdr = max(mdr, cr)
        else: cr = 0
    reps = sum(1 for i in range(1, n) if chars[i] == chars[i-1]) if n > 1 else 0

    new = [
        non_ascii / max(n, 1), entropy,
        len(SPACED_WORD.findall(text)), leet / max(n, 1), mdr,
        reps / max(n-1, 1),
        len(set(w.lower() for w in words)) / max(len(words), 1),
        int(bool(OBFUSCATED_URL.search(text))),
    ]
    return original + new


# ── Model definition ─────────────────────────────────────────────────────────

class AttentionPooling(nn.Module):
    def __init__(self, hidden_size):
        super().__init__()
        self.attention = nn.Sequential(
            nn.Linear(hidden_size, hidden_size), nn.Tanh(),
            nn.Linear(hidden_size, 1, bias=False),
        )

    def forward(self, hidden_states, attention_mask):
        scores = self.attention(hidden_states).squeeze(-1)
        scores = scores.masked_fill(attention_mask == 0, float("-inf"))
        weights = torch.softmax(scores, dim=-1).unsqueeze(-1)
        return (hidden_states * weights).sum(dim=1)


class DeBERTaSingleHead(nn.Module):
    def __init__(self, model_name, num_extra_features=23, num_labels=2, dropout=0.1):
        super().__init__()
        self.deberta = AutoModel.from_pretrained(model_name)
        H = self.deberta.config.hidden_size
        self.attn_pool = AttentionPooling(H)
        feat_dim = 128
        self.feature_proj = nn.Sequential(
            nn.Linear(num_extra_features, feat_dim), nn.LayerNorm(feat_dim),
            nn.GELU(), nn.Dropout(dropout),
        )
        combined_dim = 2 * H + feat_dim  # 768*2 + 128 = 1664
        self.fc1 = nn.Linear(combined_dim, 256)
        self.ln1 = nn.LayerNorm(256)
        self.residual_block = nn.Sequential(
            nn.Linear(256, 256), nn.LayerNorm(256),
            nn.GELU(), nn.Dropout(dropout),
            nn.Linear(256, 256), nn.LayerNorm(256),
        )
        self.dropout = nn.Dropout(dropout)
        self.output_head = nn.Linear(256, num_labels)

    def forward(self, input_ids, attention_mask, extra_features):
        out = self.deberta(input_ids=input_ids, attention_mask=attention_mask)
        hidden = out.last_hidden_state
        cls_emb  = hidden[:, 0, :]
        attn_emb = self.attn_pool(hidden, attention_mask)
        feat     = self.feature_proj(extra_features)
        x = torch.cat([cls_emb, attn_emb, feat], dim=1)
        x = F.gelu(self.ln1(self.fc1(x)))
        x = x + self.residual_block(x)
        return self.output_head(self.dropout(x))


# ── Load model ───────────────────────────────────────────────────────────────

model_id  = "notd5a/deberta-v3-malicious-sms-mms-detector"
device    = "cuda" if torch.cuda.is_available() else "cpu"
tokenizer = AutoTokenizer.from_pretrained(model_id)
scaler    = joblib.load(hf_hub_download(model_id, "scaler.pkl"))

model = DeBERTaSingleHead(model_id)
state = torch.load(hf_hub_download(model_id, "pytorch_model.pt"), map_location=device)
model.load_state_dict(state)
model.float().to(device).eval()

with open(hf_hub_download(model_id, "threshold.json")) as f:
    thresholds = json.load(f)
SPAM_THRESHOLD = thresholds["optimal_threshold"]


# ── Predict ──────────────────────────────────────────────────────────────────

def predict(texts):
    if isinstance(texts, str):
        texts = [texts]

    enc = tokenizer(texts, max_length=128, padding="max_length",
                    truncation=True, return_tensors="pt")
    raw_feats = np.array([extract_features(t) for t in texts], dtype=np.float32)
    scaled = torch.tensor(scaler.transform(raw_feats), dtype=torch.float32).to(device)

    with torch.no_grad():
        logits = model(
            enc["input_ids"].to(device),
            enc["attention_mask"].to(device),
            scaled,
        )
        spam_probs = torch.softmax(logits, dim=1)[:, 1].cpu().numpy()

    return [{
        "text":             t,
        "prediction":       "spam" if sp >= SPAM_THRESHOLD else "benign",
        "is_spam":          bool(sp >= SPAM_THRESHOLD),
        "spam_probability": round(float(sp), 4),
    } for t, sp in zip(texts, spam_probs)]


# ── Example ──────────────────────────────────────────────────────────────────

results = predict([
    "Your account has been suspended. Verify immediately: http://bit.ly/abc123",
    "Hey, are you free for lunch tomorrow?",
    "Flat 30% OFF on all ethnic wear! Shop now at bit.ly/sale2026",
])
for r in results:
    flag = "SPAM" if r["is_spam"] else "benign"
    print(f"  [{flag}] (spam: {r['spam_probability']:.3f}) {r['text'][:80]}")

# Clone the repo
git lfs install
git clone https://huggingface.co/notd5a/deberta-v3-malicious-sms-mms-detector
cd deberta-v3-malicious-sms-mms-detector

# Install dependencies
pip install torch transformers scikit-learn joblib sentencepiece

# Run hybrid inference (repo root = DeBERTa dir, charcnn/ = CharCNN dir)
python hybrid_router_inference.py \
    --deberta_dir . \
    --short_dir charcnn \
    --text "Your account has been suspended. Verify at bit.ly/xyz"

# With JSON output
python hybrid_router_inference.py \
    --deberta_dir . \
    --short_dir charcnn \
    --text "Your account has been suspended" \
    --json

# With explainability (token importance + feature contributions)
python hybrid_router_inference.py \
    --deberta_dir . \
    --short_dir charcnn \
    --text "Your account has been suspended. Verify at bit.ly/xyz" \
    --explain

# Batch inference on CSV
python hybrid_router_inference.py \
    --deberta_dir . \
    --short_dir charcnn \
    --input test_messages.csv \
    --output predictions.csv

from hybrid_router_inference import HybridDetector

# From a cloned HuggingFace repo:
detector = HybridDetector.load(deberta_dir=".", short_dir="charcnn")

# Or from local training directories:
detector = HybridDetector.load(
    deberta_dir="model_output_v2.4",
    short_dir="cnn_model_v3",
)

# Single classification
result = detector.classify("Win a free iPhone! Click here now!")
# {
#     "text": "Win a free iPhone! Click here now!",
#     "prediction": "spam",
#     "is_spam": True,
#     "spam_probability": 0.9812,
#     "model_used": "charcnn",
#     "routing_reason": "Routed to CharCNN (length 34 <= 60)"
# }

# With explainability
result = detector.classify(
    "Your Chase account has been locked. Verify: chase-secure.com/verify",
    explain=True,
)
# Adds: token_importance, feature_contributions, explanation

# Batch prediction
results = detector.predict([
    "Hey, are you coming to dinner tonight?",
    "URGENT: Your bank account has been compromised. Act now!",
    "Your package is on the way! Track: amzn.to/3xK9",
])