app.py

"""
ScamSentry — Gradio UI

Thin presentation layer over the same orchestrator engine cli.py uses —
both call agents.orchestrator.run_pipeline() directly, no duplicated
pipeline logic. The "Agent Trajectory" panel just dumps the same state
dict run_pipeline() already returns (intake_result, reputation_signals,
linguistic_signal, injection_guard_result, draft_verdict,
semantic_check_result, final_verdict, report_text) — no new
instrumentation needed.

Note on types: some agents (url_reputation_agent) have no output_schema,
so their state value is the model's raw JSON text, not a parsed dict —
same reason risk_synthesis_gate.py defensively does
`json.loads(x) if isinstance(x, str) else x` before touching intake_result
or semantic_check_result. This file does the same before rendering.
"""

import json
import mimetypes

import gradio as gr

from agents.orchestrator import run_pipeline

TRAJECTORY_KEYS = [
    "intake_result",
    "reputation_signals",
    "linguistic_signal",
    "injection_guard_result",
    "draft_verdict",
    "semantic_check_result",
    "final_verdict",
]


def _maybe_parse(value):
    """Parse a value as JSON if it's a string; return it unchanged
    otherwise (or if it's a string but not valid JSON, e.g. report_text,
    which is deliberately plain prose)."""
    if isinstance(value, str):
        try:
            return json.loads(value)
        except (json.JSONDecodeError, TypeError):
            return value
    return value


def format_verdict_panel(result: dict) -> str:
    verdict = _maybe_parse(result.get("final_verdict")) or {}
    report = result.get("report_text") or "(no report generated)"

    risk = str(verdict.get("risk_level", "unknown")).upper()
    confidence = verdict.get("confidence", "unknown")
    injection = verdict.get("injection_attempt_detected", False)

    lines = [f"### Risk level: {risk}", f"**Confidence:** {confidence}"]
    if injection:
        lines.append("**⚠️ Prompt-injection attempt detected in the analyzed content.**")
    lines.append("")
    lines.append(report)
    return "\n".join(lines)


def format_trajectory(result: dict) -> str:
    trajectory = {k: _maybe_parse(result.get(k)) for k in TRAJECTORY_KEYS if k in result}
    return json.dumps(trajectory, indent=2, default=str)


async def scan(raw_text, image_path):
    if not raw_text and not image_path:
        return "Please paste a message or upload a screenshot.", "{}"

    image_bytes = None
    image_mime = "image/png"
    if image_path:
        with open(image_path, "rb") as f:
            image_bytes = f.read()
        guessed, _ = mimetypes.guess_type(image_path)
        if guessed:
            image_mime = guessed

    try:
        result = await run_pipeline(
            text=raw_text or None,
            image_bytes=image_bytes,
            image_mime=image_mime,
            verbose=False,
        )
    except Exception as e:
        return f"⚠️ Scan failed: {e}", "{}"

    if not result:
        return "⚠️ Pipeline returned no result — check the logs.", "{}"

    return format_verdict_panel(result), format_trajectory(result)


with gr.Blocks(title="ScamSentry") as demo:
    gr.Markdown(
        "# ScamSentry\n"
        "Paste a suspicious message or upload a screenshot to check it for scam/phishing signals."
    )

    with gr.Row():
        with gr.Column():
            text_input = gr.Textbox(
                label="Message text",
                placeholder="Paste the email, SMS, or WhatsApp message here...",
                lines=8,
            )
            image_input = gr.Image(label="Or upload a screenshot", type="filepath")
            scan_button = gr.Button("Scan", variant="primary")

        with gr.Column():
            verdict_output = gr.Markdown(label="Verdict")
            with gr.Accordion("Agent Trajectory (debug view)", open=False):
                trajectory_output = gr.Code(label="Raw agent state", language="json")

    scan_button.click(
        fn=scan,
        inputs=[text_input, image_input],
        outputs=[verdict_output, trajectory_output],
    )

if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860)