import streamlit as st import sys import os import time # ── Page Config ─────────────────────────────────────────────────── st.set_page_config( page_title="SHADOW — Kenyan Fraud Intelligence", page_icon="🛡️", layout="wide", initial_sidebar_state="collapsed" ) # ── Styling ─────────────────────────────────────────────────────── st.markdown(""" """, unsafe_allow_html=True) # ── Pipeline Import ─────────────────────────────────────────────── # Works whether run from project root (HF Spaces) or from app/ dir sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) try: from agents.pipeline import ShadowPipeline PIPELINE_AVAILABLE = True except ImportError: PIPELINE_AVAILABLE = False # ── Preset Messages ─────────────────────────────────────────────── PRESETS = { "— Select a demo scenario —": "", "🔴 Safaricom Impersonation": "Habari kutoka Safaricom. Laini yako inatumika na mtu mwingine (double registration). Piga *33*0000* kuzuia hii haraka au akaunti yako itafungwa ndani ya masaa 2.", "🔴 KRA Penalty Threat": "KRA ALERT: Uko na tax arrears ya KES 23,450 kwa iTax system yako. Lipa ndani ya masaa 48 au utashtakiwa. Piga simu 0756XXXXXX sasa.", "🟠 M-Pesa Reversal Scam": "Aki naomba urudishe ile pesa nimekutumia by mistake saa hii. Ni ya fees ya mtoto tafadhali. Tuma haraka 0712XXXXXX.", "🟠 Fuliza Boost Scam": "KAMA ULIPATA FULIZA SEMA THANKS. Inbox nikuboostie fuliza from 0 to 100k in 2 minutes hii January hakuna stress.", "🟡 Betting Jackpot Scam": "Hongera! Wewe ndio mshindi wa 500k SportPesa Weekly Jackpot. Tuma 2,500 ya registration fee kupokea pesa kwa MPESA yako leo.", "🟡 WhatsApp OTP Theft": "Boss nisamehe, nilituma code ya WhatsApp kwa namba yako by mistake. Naomba unitumie hiyo code 6-digits haraka niingie kwa group ya kazi.", "✅ Legitimate M-Pesa": "MPESA Confirmed. You have received Ksh 3,500.00 from JOHN KAMAU 0722XXXXXX on 8/5/26 at 10:23 AM. New M-PESA balance is Ksh 4,120.00.", } # ── Risk Color Helper ───────────────────────────────────────────── def get_risk_color(level: str) -> str: return { "CRITICAL": "#ef4444", "HIGH": "#f97316", "MEDIUM": "#f59e0b", "LOW": "#22c55e" }.get(level, "#64748b") def get_verdict_class(verdict: str) -> str: if verdict == "SCAM": return "verdict-scam" elif verdict == "SUSPICIOUS": return "verdict-suspicious" return "verdict-safe" def get_verdict_emoji(verdict: str) -> str: return {"SCAM": "🚨", "SUSPICIOUS": "⚠️", "SAFE": "✅"}.get(verdict, "❓") def get_trace_dot_color(agent: str, risk_hint: str) -> str: if risk_hint in ["CRITICAL", "HIGH"]: return "#ef4444" elif risk_hint in ["MEDIUM"]: return "#f59e0b" elif agent == "OSINT PRECHECK": return "#8b5cf6" elif agent == "LANGUAGE AGENT": return "#3b82f6" elif agent == "THREAT AGENT": return "#f97316" elif agent == "RISK AGENT": return "#ef4444" elif agent == "ACTION AGENT": return "#22c55e" return "#64748b" # ── Header ──────────────────────────────────────────────────────── st.markdown("""

◈ SHADOW

KENYAN FRAUD INTELLIGENCE SYSTEM

⚡ POWERED BY AMD INSTINCT MI300X + ROCm

""", unsafe_allow_html=True) # ── Layout ──────────────────────────────────────────────────────── left_col, right_col = st.columns([1, 1.3], gap="large") with left_col: st.markdown("#### 📥 Analyze a Message") # Preset selector preset_choice = st.selectbox( "Load a demo scenario", options=list(PRESETS.keys()), label_visibility="collapsed" ) # Pre-fill text area from preset default_text = PRESETS.get(preset_choice, "") message = st.text_area( "Message", value=default_text, height=160, placeholder="Paste a suspicious SMS, WhatsApp message, or notification here...", label_visibility="collapsed" ) analyze_clicked = st.button("🔍 ANALYZE WITH SHADOW", use_container_width=True) # Stats strip st.markdown("
", unsafe_allow_html=True) s1, s2, s3 = st.columns(3) s1.metric("Scam Categories", "11") s2.metric("Languages", "EN / SW / Sheng") s3.metric("Pipeline Agents", "4") st.markdown("---") st.markdown("""

    SHADOW uses a hybrid OSINT + 4-agent LLM pipeline to detect

    Kenyan mobile fraud in real time. Qwen3 inference runs on

    AMD Instinct MI300X via vLLM + ROCm.

""", unsafe_allow_html=True) # ── Analysis Logic ───────────────────────────────────────────────── with right_col: if analyze_clicked: if not message.strip(): st.warning("Please paste a message to analyze.") else: with st.spinner("Shadow is analyzing..."): start = time.time() if PIPELINE_AVAILABLE: try: pipeline = ShadowPipeline() state = pipeline.run(message) action = state.action_data or {} risk = state.risk_data or {} trace = state.execution_trace or [] elapsed = round(time.time() - start, 2) except Exception as e: st.error(f"Pipeline error: {str(e)}") # Safe fallback action = { "verdict": "INCONCLUSIVE", "risk_level": "UNKNOWN", "scam_type": "Error", "dashboard_summary": "An error occurred during analysis.", "confidence": 0.0, "explanation": {"red_flags_found": ["System error"]}, "recommended_actions": [], "do_not_do": [], "safety_tip": {}, "reporting": {} } risk = {"raw_score": 0} trace = [{"agent": "SYSTEM", "step": 1, "summary": "Error running pipeline", "risk_hint": "UNKNOWN"}] elapsed = round(time.time() - start, 2) else: # Fallback demo state if imports fail action = { "verdict": "SUSPICIOUS", "risk_level": "MEDIUM", "scam_type": "Pipeline Offline (Mock)", "dashboard_summary": "This is a fallback response because the pipeline failed to load.", "confidence": 0.50, "explanation": {"red_flags_found": ["Mock execution"]}, "recommended_actions": [{"action": "Check system paths and imports"}], "do_not_do": ["Trust this mock verdict"], "safety_tip": {"english": "System is offline.", "swahili": "Mfumo haupatikani.", "sheng": "System iko chini."}, "reporting": {"should_report": False, "contacts": []} } risk = {"raw_score": 5} trace = [{"agent": "MOCK AGENT", "step": 1, "summary": "Pipeline import failed", "risk_hint": "MEDIUM"}] elapsed = 0.0 # Safe gets with empty defaults to prevent NoneType crashes verdict = action.get("verdict") or "INCONCLUSIVE" risk_level = action.get("risk_level") or "UNKNOWN" scam_type = action.get("scam_type") or "Unknown" summary = action.get("dashboard_summary") or "" confidence = action.get("confidence") if confidence is None: confidence = 0.0 raw_score = risk.get("raw_score") if raw_score is None: raw_score = 0 explanation = action.get("explanation") or {} red_flags = explanation.get("red_flags_found") or [] recommended = action.get("recommended_actions") or [] do_not = action.get("do_not_do") or [] safety_tip = action.get("safety_tip") or {} reporting = action.get("reporting") or {} # ── Verdict Card ────────────────────────────────────── verdict_class = get_verdict_class(verdict) verdict_emoji = get_verdict_emoji(verdict) risk_color = get_risk_color(risk_level) score_pct = min(int((raw_score / 10) * 100), 100) st.markdown(f"""

{verdict_emoji} {verdict}

{summary}

{scam_type} | Confidence: {int(confidence*100)}% | {elapsed}s

""", unsafe_allow_html=True) st.markdown("
", unsafe_allow_html=True) # ── Risk Score Bar ──────────────────────────────────── st.markdown(f"""

⚡ Risk Score

Score: {raw_score}/10 {risk_level}

""", unsafe_allow_html=True) # ── Two columns: Red Flags + Actions ────────────────── c1, c2 = st.columns(2) with c1: flags_html = "".join([f'

⚠ {f}

' for f in red_flags]) or '

None detected

' st.markdown(f"""

🚩 Red Flags

{flags_html}

""", unsafe_allow_html=True) with c2: actions_html = "" for a in recommended: if isinstance(a, dict): action_text = a.get("action", "") if action_text: actions_html += f'

→ {action_text}

' elif isinstance(a, str): actions_html += f'

→ {a}

' donot_html = "".join([f'

✗ {d}

' for d in do_not if isinstance(d, str)]) st.markdown(f"""

✅ What To Do

{actions_html} {donot_html}

""", unsafe_allow_html=True) # ── Execution Trace ─────────────────────────────────── trace_html = """

🧠 Agent Reasoning Timeline

""" if not trace: trace_html += '

No trace available.

' for step in trace: if not isinstance(step, dict): continue agent = step.get("agent") or "SYSTEM" summary_text = step.get("summary") or "" risk_hint = step.get("risk_hint") or "" dot_color = get_trace_dot_color(agent, risk_hint) trace_html += f"""

[{step.get('step', 0)}] {agent}

{summary_text}

""" trace_html += "

" st.markdown(trace_html, unsafe_allow_html=True) # ── Safety Tip ──────────────────────────────────────── if safety_tip: st.markdown(f"""

{safety_tip.get('english', 'Not available')}

{safety_tip.get('swahili', 'Haipatikani')}

SHENG

{safety_tip.get('sheng', 'Haiwezekani')}

""", unsafe_allow_html=True) # ── Reporting ───────────────────────────────────────── if reporting.get("should_report") and reporting.get("contacts"): contacts = reporting.get("contacts", []) contact_parts = [] for c in contacts: if isinstance(c, dict) and 'name' in c and 'value' in c: contact_parts.append(f"{c['name']}: {c['value']}") if contact_parts: contact_str = " | ".join(contact_parts) st.markdown(f"""

📢 Report this: {contact_str}

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

◈
SHADOW IS WATCHING
Paste a message or select a demo scenario
to begin fraud analysis.

""", unsafe_allow_html=True) # ── Footer ───────────────────────────────────────────────────────── st.markdown("""

    SHADOW — AMD Developer Hackathon 2026  | 
    Qwen3 on MI300X via vLLM + ROCm  | 
    Built for Kenya's 54M mobile users  | 
    GitHub

""", unsafe_allow_html=True)