'
f'{lvl} · detected as {label} · {esc(r.get("language","?"))}'
"
")
out.append(f'
{p}%
')
out.append(f'
'
f'
')
out.append('
0% · hamdecision threshold 55%'
'100% · scam
')
# Timing is honest about which stage it reflects: while streaming, `neural_ms`
# equals `ms`, so we only claim a verdict timing — never a "total".
if ms and neural_ms and neural_ms == ms:
out.append(f'
'
f'verdict {ms} ms on GPU'
f'attribution computing…
')
elif ms:
out.append(f'
'
f'verdict {neural_ms} ms'
f'total {ms} ms on GPU
')
if r.get("green_channel"):
out.append(f'
'
f'🛡 Green Channel cleared — {esc(r.get("green_reason",""))}
')
out.append("
")
return "".join(out)
def h_signals(r):
u, t = r.get("url_signals", {}), r.get("text_signals", {})
chips = []
def chip(on, label, state):
cls = {"hi": "ch-hi", "md": "ch-md", "lo": "ch-lo", "off": "ch-off"}[state]
return f'{label}' if on else f'{label}'
chips.append(chip(u.get("has_url"), "🔗 URL present", "md" if u.get("has_url") else "off"))
chips.append(chip(u.get("suspicious_tld"), "⚠ suspicious TLD", "hi" if u.get("suspicious_tld") else "off"))
chips.append(chip(u.get("has_ip_url"), "🖥 raw IP host", "hi" if u.get("has_ip_url") else "off"))
chips.append(chip(u.get("has_shortened_url"), "🔽 shortener", "hi" if u.get("has_shortened_url") else "off"))
chips.append(chip(u.get("has_legit_domain"), "✓ known domain", "lo" if u.get("has_legit_domain") else "off"))
chips.append(chip(u.get("has_http") and not u.get("has_https"), "⚠ plain HTTP", "md" if (u.get("has_http") and not u.get("has_https")) else "off"))
chips.append(chip(t.get("has_phone"), "☎ phone number", "md" if t.get("has_phone") else "off"))
chips.append(chip(t.get("urgency_count"), "⚡ urgency ×{0}".format(t.get("urgency_count", 0)),
"hi" if (t.get("urgency_count") or 0) >= 3 else "md" if t.get("urgency_count") else "off"))
chips.append(chip(t.get("pct_upper", 0) > 0.2, "⬆ shouty caps", "md" if t.get("pct_upper", 0) > 0.2 else "off"))
chips.append(chip(t.get("has_currency"), "₹ currency", "off"))
rows = []
if u.get("has_url"):
rows.append(f'
🔗 {u.get("num_urls",0)} link(s) detected in the message
neural {r.get("_neural_score",0):.3f}'
f' · rule {r.get("_rule_score",0):.3f}'
f' · ham-rule {r.get("_ham_rule_score",0):.3f}
')
toks = "".join(f'{esc(w)}' for w in r.get("_tokens", []))
if toks:
rows.append(f'
{toks}
')
return ('
Signal breakdown
'
f'
{"".join(chips)}
'
f'
{"".join(rows)}
')
def h_shap(r, ex):
words = list(ex.get("top_spam_words", [])) + list(ex.get("top_ham_words", []))
if not words:
return ('
Why this verdict
'
'
No strong word-level signals for this message.
')
rows = sorted(words, key=lambda w: -abs(w[1]))[:12]
mx = max(abs(w[1]) for w in rows) or 1
bars = []
for w, v in rows:
pct = (abs(v) / mx) * 47
st = (f"left:50%;width:{pct}%;background:var(--hi)" if v > 0
else f"right:50%;width:{pct}%;background:var(--lo)")
bars.append(f'
⚠ Do not act on this message. Never click the link or share '
'your OTP, Aadhaar or bank details. ▸ Block the sender · ▸ Report at '
'cybercrime.gov.in · ▸ National helpline 1930
')
# ─────────────────────────────────────────────────────────────────────────────
# Handlers
# ─────────────────────────────────────────────────────────────────────────────
@gpu
def scan(message, progress=gr.Progress()):
"""
Streaming scan: yields the verdict as soon as the neural pass finishes, then
yields again once SHAP attribution lands.
Judging flow wants the verdict fast (~1-2s) but also wants the word-level
attribution on screen. Blocking on SHAP hides the verdict behind the slowest
step, so this is a generator rather than a single return.
"""
if not message or not message.strip():
yield ('
Type or paste a message to analyse it.
',
"", "", "")
return
t0 = time.perf_counter()
r = _run(message)
neural_ms = int((time.perf_counter() - t0) * 1000)
# First yield: verdict + signals, SHAP panel still loading.
# Pass neural_ms as BOTH args so h_verdict knows SHAP has not run yet.
yield (h_verdict(r, message, neural_ms, neural_ms),
h_signals(r),
SHAP_PENDING,
h_warn(r))
progress(0.5, desc="Generating word-level attribution…")
ex = _explain(message)
total_ms = int((time.perf_counter() - t0) * 1000)
# Second yield: SHAP resolved, timing folded into the verdict card.
yield (h_verdict(r, message, total_ms, neural_ms),
h_signals(r),
h_shap(r, ex),
h_warn(r))
@gpu
def mob_detail(which, custom="", progress=gr.Progress()):
"""Single seeded-message analysis — same streaming treatment as the scanner."""
i = SEED_LABELS.index(which) if which in SEED_LABELS else 0
msg = custom.strip() or SEED_SMS[i]["b"]
head = f'
Analysis · {esc(SEED_SMS[i]["n"])}
'
r = _run(msg)
yield head + h_verdict(r, msg) + h_signals(r) + SHAP_PENDING + h_warn(r)
progress(0.5, desc="Generating word-level attribution…")
ex = _explain(msg)
yield head + h_verdict(r, msg) + h_signals(r) + h_shap(r, ex) + h_warn(r)
def mob_scan_all(progress=gr.Progress()):
"""Bulk scan of the seeded inbox.
Deliberately NOT decorated with @gpu — it always runs on CPU so that walking
16 messages never touches the metered GPU.
"""
rows, threats = [], 0
for i, m in enumerate(SEED_SMS):
progress(i / len(SEED_SMS), desc=f"Scanning {i+1}/{len(SEED_SMS)} · {m['n']}")
try:
res = _run(m["b"])
except Exception:
res = None
spam = bool(res) and res.get("label") == "spam"
threats += spam
if res is None:
tier, colour, soft, brd, tag = "lo", "var(--faint)", "var(--panel2)", "var(--line2)", "—"
else:
# Same three tiers as the detail card: high → SCAM, medium → SUSPICIOUS.
tier, colour, _, _ = risk(res.get("confidence", 0), res.get("risk_level", ""))
if tier == "hi":
soft, brd, tag = "var(--hi-s)", "var(--hi-b)", "SCAM"
elif tier == "md":
soft, brd, tag = "var(--md-s)", "var(--md-b)", "SUSPICIOUS"
else:
soft, brd, tag = "var(--lo-s)", "var(--lo-b)", "SAFE"
hit = "expected scam" if (spam == (m["e"] == "spam")) else "expected safe"
rows.append(
f'
'
f'
{esc(m["n"])}{hit}
'
f'
{esc(m["b"])}
'
f''
f'{tag}
'
)
progress(1, desc="done")
stats = ('
'
f'
{len(SEED_SMS)}Scanned
'
f'
{threats}Flagged scam
'
f'
{len(SEED_SMS)-threats}Safe
'
f'
{round(threats/len(SEED_SMS)*100)}%Flag rate
'
'
')
return stats + f'
{"".join(rows)}
'
# ─────────────────────────────────────────────────────────────────────────────
# JSON API — used by the Android app and any external client
# ─────────────────────────────────────────────────────────────────────────────
@gpu
def predict_api(message: str) -> dict:
return _run(message)
@gpu
def explain_api(message: str) -> dict:
r = _run(message)
ex = _explain(message)
return {"label": r["label"], "confidence": r["confidence"],
"top_spam_words": ex.get("top_spam_words", []),
"top_ham_words": ex.get("top_ham_words", [])}
def check_domain_api(domain: str) -> dict:
from utils.safe_browsing import check_domain_status
st = check_domain_status(domain)
return {"domain": domain, "status": st,
"is_legitimate": st == "known_safe", "is_malicious": st == "known_malicious"}
# ─────────────────────────────────────────────────────────────────────────────
# UI
# ─────────────────────────────────────────────────────────────────────────────
NAV = ('
🛡
'
'ScamShield
'
' model live'
'XLM-RoBERTa · GPU'
'EN · HI · Hinglish
')
HERO = ('
✦ AI for Cybersecurity
'
'
Catch the scam before it catches you
'
'
Paste any SMS — English, Hindi or Hinglish — and ScamShield returns a verdict, '
'the exact signals behind it, and word-level SHAP attribution.
')
FOOT = ('
XLM-RoBERTa · 9 URL + 8 text signals · Google Safe Browsing '
'· SHAP · threshold 0.55 Spam F1 0.94 · multilingual · encrypted Android client '
'ships as an APK on the same API
')
def side_panel():
return (
'
What you are looking at
'
'
'
'
🧠 XLM-RoBERTa reads the message in 100 languages
'
'
🔗 9 URL signals — TLD, shorteners, raw IPs, whitelist
'
'
🛡 Google Safe Browsing escalates known-bad domains
'
'
📊 SHAP shows which words pushed the score
'
'
🎯 Threshold 0.55 tuned to stop flagging Indian bank SMS
'
'
'
'
Mobile app
'
'
The phone on the left is the real UI running this exact pipeline. '
'The Android build reads your inbox (READ_SMS), encrypts every request '
'with AES-256-CBC, and raises a notification when a message is flagged.
'
)
with gr.Blocks(title="ScamShield — Smishing Detector") as demo:
gr.HTML('')
gr.HTML(NAV)
gr.HTML(HERO)
with gr.Tabs():
with gr.TabItem("Scanner"):
with gr.Row():
with gr.Column(scale=7, elem_classes="ta"):
gr.HTML('
Paste a message
')
inp = gr.Textbox(show_label=False, lines=6, max_lines=12,
placeholder="Paste the SMS here…\n\nCtrl + Enter to scan",
elem_classes="ta")
btn = gr.Button("Scan message", variant="primary", elem_classes="scan-btn")
gr.Examples(examples=EXAMPLES, inputs=inp, label="Try an example")
with gr.Column(scale=5):
verdict = gr.HTML()
warn = gr.HTML()
with gr.Row():
with gr.Column():
signals = gr.HTML()
with gr.Column():
shap = gr.HTML()
for e in (btn.click, inp.submit):
e(fn=scan, inputs=inp, outputs=[verdict, signals, shap, warn],
api_name="scan", show_progress="minimal")
with gr.TabItem("Mobile app"):
with gr.Row():
with gr.Column(scale=5, elem_classes="phone"):
gr.HTML('')
gr.HTML('
Browsers cannot read an SMS inbox, so the web '
'demo ships a seeded set. On Android the app reads your real messages.
')
detail = gr.HTML()
with gr.Column(scale=4):
gr.HTML(side_panel())
go_one.click(fn=mob_detail, inputs=[pick, custom], outputs=detail, show_progress="hidden")
go_all.click(fn=mob_scan_all, outputs=stats, show_progress="minimal")
gr.HTML(FOOT)
# ── Hidden JSON endpoints ────────────────────────────────────────────────
# A function only becomes an API route once it is wired to an event, so the
# JSON API is exposed through zero-height components rather than by being
# called directly.
_api_msg = gr.Textbox(visible=False, elem_classes="api-hidden")
_api_json = gr.JSON(visible=False)
_api_dom = gr.Textbox(visible=False)
_api_dom_json = gr.JSON(visible=False)
_api_msg.change(fn=predict_api, inputs=_api_msg, outputs=_api_json,
api_name="predict", show_progress="hidden")
_api_msg.change(fn=explain_api, inputs=_api_msg, outputs=_api_json,
api_name="explain", show_progress="hidden")
_api_dom.change(fn=check_domain_api, inputs=_api_dom, outputs=_api_dom_json,
api_name="check_domain", show_progress="hidden")
THEME = gr.themes.Soft(primary_hue="indigo", neutral_hue="slate", font=["Inter", "system-ui", "sans-serif"])
if __name__ == "__main__":
demo.queue(default_concurrency_limit=2).launch(css=CSS, theme=THEME)