import os import sys import streamlit as st import time sys.path.append(os.path.dirname(os.path.abspath(__file__))) from whisper_util import VoiceProcessor from qwen_utils import ScamAnalyzer # ─── Page config ──────────────────────────────────────────────────────────── st.set_page_config( page_title="Jhol-Check", page_icon="🧐", layout="wide", initial_sidebar_state="collapsed", ) # ─── Design tokens ────────────────────────────────────────────────────────── # Palette: deep navy background, off-white surface, red danger, amber warning, # emerald safe, electric-blue accent. One signature: the animated scan-line # that runs across the evidence card on load — forensic instrument aesthetic. CSS = """ """ st.markdown(CSS, unsafe_allow_html=True) # ─── Helpers ──────────────────────────────────────────────────────────────── TECHNIQUE_CLASSES = { "fear": ("chip-fear", "⚠️ Fear"), "urgency": ("chip-urgency", "⏱ Urgency"), "authority": ("chip-authority", "🏛 Authority"), "greed": ("chip-greed", "💰 Greed"), "scarcity": ("chip-scarcity", "⏳ Scarcity"), "trust exploitation": ("chip-trust", "🤝 Trust Exploit"), "curiosity": ("chip-curiosity", "🔍 Curiosity"), "emotional manipulation": ("chip-emotional", "💔 Emotional Manip."), "social proof": ("chip-social", "👥 Social Proof"), "reciprocity": ("chip-reciprocity","↩ Reciprocity"), "mlm": ("chip-default", "🔄 MLM"), } LABEL_CLASSES = { "authority": ("badge-authority", "ev-authority"), "fear": ("badge-fear", "ev-fear"), "urgency": ("badge-urgency", "ev-urgency"), "credential theft": ("badge-credential", "ev-credential"), "suspicious link": ("badge-link", "ev-suspicious"), "suspicious contact": ("badge-suspicious", "ev-suspicious"), "suspicious upi": ("badge-suspicious", "ev-suspicious"), "advance fee": ("badge-greed", "ev-greed"), "greed": ("badge-greed", "ev-greed"), "isolation": ("badge-credential", "ev-credential"), "extortion": ("badge-fear", "ev-fear"), } RISK_COLORS = { "CRITICAL": ("#ef4444", "metric-scam"), "HIGH": ("#f97316", "metric-warn"), "MEDIUM": ("#f59e0b", "metric-warn"), "LOW": ("#22c55e", "metric-safe"), } ACTION_ICONS = { "do not": "🚫", "report": "📢", "block": "🔒", "verify": "✅", "call": "📞", "visit": "🌐", "never": "❌", "safe": "✅", } def get_action_icon(action_text): low = action_text.lower() for kw, icon in ACTION_ICONS.items(): if kw in low: return icon return "•" def render_techniques(techniques): if not techniques: return '

None detected

' chips = [] for t in techniques: key = t.lower() cls, label = TECHNIQUE_CLASSES.get(key, ("chip-default", t)) chips.append(f'{label}') return f'

{"".join(chips)}

' def render_evidence_card(ev): phrase = ev.get("text", "") label = ev.get("label", "") key = label.lower() badge_cls, card_cls = LABEL_CLASSES.get(key, ("badge-default", "ev-default")) return f"""

"{phrase}"

→ {label}

""" def render_metric(label, value, color_cls, conf=None, bar_color=None): bar_html = "" if conf is not None: pct = min(max(int(conf), 0), 100) bar_html = f"""

""" return f"""

{label}

{value}

{bar_html}

""" def get_risk_color_hex(risk): return RISK_COLORS.get(risk.upper(), ("#8b949e", "metric-neutral"))[0] # ─── Model loading ─────────────────────────────────────────────────────────── @st.cache_resource(show_spinner=False) def load_pipelines(): voice = VoiceProcessor(model_name="openai/whisper-small") analyzer = ScamAnalyzer(repo_id="PrachiSandipkumar/qwen_scam_high_accuracy") return voice, analyzer # ─── Masthead ──────────────────────────────────────────────────────────────── st.markdown(""" """, unsafe_allow_html=True) # ─── Load models ───────────────────────────────────────────────────────────── with st.spinner("Loading AI pipelines…"): voice_engine, analyzer_engine = load_pipelines() # ─── Layout ────────────────────────────────────────────────────────────────── col_left, col_right = st.columns([1, 1.4], gap="large") # ── Session state init ─────────────────────────────────────────────────────── # "txt_input" = the textarea value (Streamlit owns this key) # "pending_example" = text injected by a quick-test button click # "pending_trigger" = True when we want to auto-run analysis on page reload if "pending_example" not in st.session_state: st.session_state["pending_example"] = "" if "pending_trigger" not in st.session_state: st.session_state["pending_trigger"] = False # If a quick-test button was clicked last cycle, push the text into the # textarea's session_state key BEFORE the widget renders, then clear the flag. if st.session_state["pending_example"]: st.session_state["txt_input"] = st.session_state["pending_example"] st.session_state["pending_example"] = "" # ════════════════════════════════════════════════════════════════════════════ # LEFT — INPUT PANEL # ════════════════════════════════════════════════════════════════════════════ with col_left: st.markdown('

Input

', unsafe_allow_html=True) tab_text, tab_voice = st.tabs(["💬 Text Message", "🎙 Call Recording"]) target_text = "" trigger = False with tab_text: st.markdown("
", unsafe_allow_html=True) text_input = st.text_area( label="", placeholder="Paste a suspicious SMS, email, WhatsApp message, or call transcript here…", height=220, key="txt_input", # Streamlit stores the live value in st.session_state["txt_input"] label_visibility="collapsed", ) # Consume the pending_trigger flag set by quick-test buttons if st.session_state.get("pending_trigger", False): st.session_state["pending_trigger"] = False if text_input.strip(): target_text = text_input.strip() trigger = True if st.button("Analyze Message", type="primary", use_container_width=True, key="btn_text"): if text_input.strip(): target_text = text_input.strip() trigger = True else: st.warning("Paste a message first.") with tab_voice: st.markdown("
", unsafe_allow_html=True) st.caption("Upload a recorded scam call. Whisper will transcribe it before analysis.") audio_file = st.file_uploader( label="", type=["mp3", "wav", "m4a", "ogg"], key="audio_upload", label_visibility="collapsed", ) if st.button("Transcribe & Analyze", type="primary", use_container_width=True, key="btn_audio"): if audio_file: tmp = f"/tmp/upload_{audio_file.name}" with open(tmp, "wb") as f: f.write(audio_file.getbuffer()) with st.spinner("Whisper is transcribing the call…"): target_text = voice_engine.process_large_audio(tmp) if os.path.exists(tmp): os.remove(tmp) if target_text.strip(): trigger = True st.info(f"**Transcribed:** {target_text[:200]}{'…' if len(target_text)>200 else ''}") else: st.error("Transcription returned empty. Check the audio file.") else: st.warning("Upload an audio file first.") # ── Quick-test examples ────────────────────────────────────────────────── st.markdown("
", unsafe_allow_html=True) st.markdown('

Quick Tests

', unsafe_allow_html=True) EXAMPLES = [ ("🔴 OTP Scam", "SBI fraud dept here. We detected suspicious Rs.49,000 transaction. Share the OTP you just received to cancel it."), ("🔴 UPI Scam", "Someone sent you Rs.5,000 on Google Pay. Accept payment by entering UPI PIN here: http://gpay-accept.co/receive/5000"), ("🟠 Govt. Impersonation", "CBI officer speaking: Your Aadhaar is linked to money laundering. Pay Rs.50,000 settlement to avoid immediate arrest."), ("🟢 Legit Bank SMS", "HDFC Bank: OTP for NetBanking login is 847291. Valid 10 min. HDFC NEVER asks for OTP. Do NOT share with anyone."), ] cols = st.columns(2) for i, (label, txt) in enumerate(EXAMPLES): with cols[i % 2]: if st.button(label, use_container_width=True, key=f"ex_{i}"): # Write the example into the textarea's session_state key. # Streamlit will re-render the textarea with this value on the # next cycle (which the rerun below triggers immediately). st.session_state["pending_example"] = txt st.session_state["pending_trigger"] = True # auto-run analysis too st.rerun() # ════════════════════════════════════════════════════════════════════════════ # RIGHT — OUTPUT PANEL # ════════════════════════════════════════════════════════════════════════════ with col_right: st.markdown('

Intelligence Report

', unsafe_allow_html=True) if not trigger: st.markdown("""

🛡️

Submit a message or call recording
to generate a fraud intelligence report.

""", unsafe_allow_html=True) else: with st.spinner("Analyzing with fine-tuned Qwen…"): t0 = time.time() report = analyzer_engine.analyze_text(target_text) elapsed = time.time() - t0 scam = report.get("scam", "NO") scam_type = report.get("scam_type", "Other") risk = report.get("risk_level", "LOW").upper() conf = report.get("confidence", 90) techniques= report.get("techniques", []) evidence = report.get("evidence", []) requested = report.get("requested_information", []) reasoning = report.get("reasoning", []) actions = report.get("recommended_action", []) risk_hex, _ = RISK_COLORS.get(risk, ("#8b949e", "metric-neutral")) # ── Verdict banner ────────────────────────────────────────────────── if scam == "YES": st.markdown(f"""

🚨

Threat Detected

{scam_type}

""", unsafe_allow_html=True) else: st.markdown("""

✅

Status

No Fraud Indicators Detected

""", unsafe_allow_html=True) # ── Metric cards ──────────────────────────────────────────────────── m_classification = "SCAM" if scam == "YES" else "CLEAN" m_cls_color = "metric-scam" if scam == "YES" else "metric-safe" st.markdown(f"""

{render_metric("Classification", m_classification, m_cls_color)} {render_metric("Risk Level", risk, _, )} {render_metric("Confidence", f"{conf}%", "metric-info", conf, risk_hex)}

""".replace(_, "metric-warn" if risk in ("HIGH","CRITICAL") else "metric-safe"), unsafe_allow_html=True) # ── Tabs ───────────────────────────────────────────────────────────── tab_ev, tab_reason, tab_json = st.tabs([ "🔬 Evidence & Techniques", "🧠 Reasoning & Actions", "📄 Full JSON", ]) # ── TAB 1: Evidence & Techniques ───────────────────────────────────── with tab_ev: # Message preview st.markdown('

Analyzed Message

', unsafe_allow_html=True) st.markdown(f'

{target_text}

', unsafe_allow_html=True) st.caption(f"Analysis completed in {elapsed:.1f}s") # Techniques st.markdown('

Manipulation Techniques

', unsafe_allow_html=True) st.markdown(render_techniques(techniques), unsafe_allow_html=True) # Evidence extraction st.markdown('

Extracted Evidence Fragments

', unsafe_allow_html=True) if evidence: cards_html = "".join(render_evidence_card(ev) for ev in evidence) st.markdown(cards_html, unsafe_allow_html=True) else: st.markdown('

No specific phrases flagged.

', unsafe_allow_html=True) # Information requested if requested and requested != ["None"]: st.markdown('

Information Being Targeted

', unsafe_allow_html=True) pills = "".join(f'{r}' for r in requested if r != "None") st.markdown(f"

{pills}

", unsafe_allow_html=True) # ── TAB 2: Reasoning & Actions ──────────────────────────────────────── with tab_reason: st.markdown('

Step-by-Step Reasoning

', unsafe_allow_html=True) if reasoning: steps_html = "" for i, step in enumerate(reasoning, 1): steps_html += f"""

{i}

{step}

""" st.markdown(steps_html, unsafe_allow_html=True) else: st.markdown('

No reasoning steps available.

', unsafe_allow_html=True) if actions: st.markdown('

', unsafe_allow_html=True) st.markdown('

Recommended Actions

', unsafe_allow_html=True) items_html = "" for action in actions: icon = get_action_icon(action) items_html += f"""

{icon} {action}

""" st.markdown(items_html, unsafe_allow_html=True) # ── TAB 3: Full JSON ────────────────────────────────────────────────── with tab_json: display_report = {k: v for k, v in report.items() if k != "raw_output"} st.json(display_report) st.caption("raw_output field excluded for readability. Available in the report dict.")