1import re
2import math
3import json
4import torch
5import numpy as np
6from collections import Counter
7from transformers import AutoTokenizer, AutoModel
8from huggingface_hub import hf_hub_download
9import joblib
10
11# --- Feature extraction functions (must match training) ---
12
13URGENCY_WORDS = {
14 "urgent", "immediately", "expires", "verify", "confirm", "suspended",
15 "locked", "alert", "action required", "limited time", "click here",
16 "act now", "final notice", "winner", "prize", "claim", "free",
17 "blocked", "deactivated", "unusual activity"
18}
19
20URL_PATTERN = re.compile(r'(https?://|www\.)\S+|\w+\.(com|net|org|io|co|uk)', re.I)
21SHORTENED_DOMAINS = {"bit.ly","tinyurl.com","goo.gl","t.co","ow.ly","smsg.io","rb.gy"}
22PHONE_PATTERN = re.compile(r'(\+?\d[\d\s\-().]{7,}\d)')
23EMAIL_PATTERN = re.compile(r'[\w.+-]+@[\w-]+\.[a-z]{2,}', re.I)
24CURRENCY_PATTERN = re.compile(r'[$£€₹¥]|(usd|gbp|eur|inr)', re.I)
25
26LEET_MAP = str.maketrans("013457@!", "oieastai")
27OBFUSCATED_URL = re.compile(
28 r"(https?(?:clue|[a-z]{4,}[a-z0-9]{2,})\b)"
29 r"|(?:h\s*t\s*t\s*p)"
30 r"|(?:www\s*\.\s*\w)"
31 r"|(?:\w+\s*\.\s*(?:com|net|org|xyz|info|co)\b)", re.I)
32SPACED_WORD = re.compile(r"\b(?:\w\s){3,}\w\b")
33
34
35def extract_features(text):
36 words = text.split()
37 letters = [c for c in text if c.isalpha()]
38 chars = list(text)
39 n = len(chars)
40
41 original = [
42 len(text), # char_count
43 len(words), # word_count
44 sum(len(w) for w in words) / max(len(words), 1),
45 sum(1 for c in letters if c.isupper()) / max(len(letters), 1),
46 sum(1 for c in text if c.isdigit()) / max(len(text), 1),
47 sum(1 for c in text if not c.isalnum() and not c.isspace()) / max(len(text), 1),
48 text.count('!'), # exclamation_count
49 text.count('?'), # question_mark_count
50 int(bool(URL_PATTERN.search(text))),
51 len(URL_PATTERN.findall(text)),
52 int(any(d in text.lower() for d in SHORTENED_DOMAINS)),
53 int(bool([m for m in PHONE_PATTERN.findall(text) if len(re.sub(r'\D','',m)) >= 7])),
54 int(bool(EMAIL_PATTERN.search(text))),
55 int(bool(CURRENCY_PATTERN.search(text))),
56 sum(1 for w in URGENCY_WORDS if w in text.lower()),
57 ]
58
59 counts = Counter(text.lower())
60 entropy = -sum((c/n) * math.log2(c/n) for c in counts.values() if c > 0) if n > 0 else 0.0
61 translated = text.translate(LEET_MAP)
62 leet_changes = sum(1 for a, b in zip(text, translated) if a != b)
63 max_drun, cur = 0, 0
64 for c in chars:
65 if c.isdigit():
66 cur += 1
67 max_drun = max(max_drun, cur)
68 else:
69 cur = 0
70
71 repeats = sum(1 for i in range(1, n) if chars[i] == chars[i-1]) if n > 1 else 0
72
73 new_features = [
74 sum(1 for c in chars if ord(c) > 127) / max(n, 1), # unicode_ratio
75 entropy, # char_entropy
76 len(SPACED_WORD.findall(text)), # suspicious_spacing
77 leet_changes / max(n, 1), # leet_ratio
78 max_drun, # max_digit_run
79 repeats / max(n - 1, 1) if n > 1 else 0.0, # repeated_char_ratio
80 len(set(w.lower() for w in words)) / max(len(words), 1), # vocab_richness
81 int(bool(OBFUSCATED_URL.search(text))), # has_obfuscated_url
82 ]
83
84 return original + new_features
85
86
87# --- Model loading and inference ---
88
89model_id = "notd5a/deberta-v3-malicious-sms-mms-detector"
90device = "cuda" if torch.cuda.is_available() else "cpu"
91
92tokenizer = AutoTokenizer.from_pretrained(model_id)
93scaler = joblib.load(hf_hub_download(model_id, "scaler.pkl"))
94
95with open(hf_hub_download(model_id, "threshold.json")) as f:
96 THRESHOLD = json.load(f)["threshold"]
97
98
99class AttentionPooling(torch.nn.Module):
100 def __init__(self, hidden_size):
101 super().__init__()
102 self.attention = torch.nn.Sequential(
103 torch.nn.Linear(hidden_size, hidden_size),
104 torch.nn.Tanh(),
105 torch.nn.Linear(hidden_size, 1, bias=False),
106 )
107
108 def forward(self, hidden_states, attention_mask):
109 scores = self.attention(hidden_states).squeeze(-1)
110 scores = scores.masked_fill(attention_mask == 0, float("-inf"))
111 weights = torch.softmax(scores, dim=-1).unsqueeze(-1)
112 return (hidden_states * weights).sum(dim=1)
113
114
115class DeBERTaWithFeaturesV2(torch.nn.Module):
116 def __init__(self, model_name, num_extra_features=23, num_labels=2, dropout=0.1):
117 super().__init__()
118 self.deberta = AutoModel.from_pretrained(model_name)
119 H = self.deberta.config.hidden_size
120 self.attn_pool = AttentionPooling(H)
121 feat_dim = 128
122 self.feature_proj = torch.nn.Sequential(
123 torch.nn.Linear(num_extra_features, feat_dim),
124 torch.nn.LayerNorm(feat_dim),
125 torch.nn.GELU(),
126 torch.nn.Dropout(dropout),
127 )
128 combined_dim = 2 * H + feat_dim
129 bottleneck = 256
130 self.fc1 = torch.nn.Linear(combined_dim, bottleneck)
131 self.ln1 = torch.nn.LayerNorm(bottleneck)
132 self.residual_block = torch.nn.Sequential(
133 torch.nn.Linear(bottleneck, bottleneck),
134 torch.nn.LayerNorm(bottleneck),
135 torch.nn.GELU(),
136 torch.nn.Dropout(dropout),
137 torch.nn.Linear(bottleneck, bottleneck),
138 torch.nn.LayerNorm(bottleneck),
139 )
140 self.dropout = torch.nn.Dropout(dropout)
141 self.output_head = torch.nn.Linear(bottleneck, num_labels)
142
143 def forward(self, input_ids, attention_mask, extra_features):
144 out = self.deberta(input_ids=input_ids, attention_mask=attention_mask)
145 hidden = out.last_hidden_state
146 cls_emb = hidden[:, 0, :]
147 attn_emb = self.attn_pool(hidden, attention_mask)
148 feat = self.feature_proj(extra_features)
149 combined = torch.cat([cls_emb, attn_emb, feat], dim=1)
150 x = torch.nn.functional.gelu(self.ln1(self.fc1(combined)))
151 x = x + self.residual_block(x)
152 return self.output_head(self.dropout(x))
153
154
155model = DeBERTaWithFeaturesV2(model_id)
156state_dict = torch.load(hf_hub_download(model_id, "pytorch_model.pt"), map_location=device)
157model.load_state_dict(state_dict)
158model.to(device).eval()
159
160
161def predict(texts):
162 if isinstance(texts, str):
163 texts = [texts]
164
165 enc = tokenizer(
166 texts,
167 max_length=256,
168 padding="max_length",
169 truncation=True,
170 return_tensors="pt"
171 )
172
173 raw_feats = np.array([extract_features(t) for t in texts], dtype=np.float32)
174 scaled_feats = torch.tensor(scaler.transform(raw_feats), dtype=torch.float32).to(device)
175
176 with torch.no_grad():
177 logits = model(
178 enc["input_ids"].to(device),
179 enc["attention_mask"].to(device),
180 scaled_feats
181 )
182 probs = torch.softmax(logits, dim=1)[:, 1].cpu().numpy()
183
184 return [
185 {
186 "text": t,
187 "prob_spam": round(float(p), 4),
188 "label": int(p >= THRESHOLD),
189 "prediction": "spam" if p >= THRESHOLD else "benign"
190 }
191 for t, p in zip(texts, probs)
192 ]
193
194
195# --- Example usage ---
196results = predict([
197 "Your account has been suspended. Verify immediately: http://bit.ly/abc123",
198 "Hey, are you free for lunch tomorrow?",
199 "Y ou've got mail: new messa ge w7",
200])
201
202for r in results:
203 print(r)