Spaces:
Running
Running
SooryaKiranKakkarayil
Day 5: Report Agent, Gradio UI, CLI wrapper, Dockerfile, .dockerignore, fix pywin32 marker
e3518f9 | """ | |
| ScamSentry CLI — thin wrapper around the same orchestrator used by app.py. | |
| No duplicated pipeline logic: `python cli.py scan "<text>"` and the Gradio | |
| UI both call agents.orchestrator.run_pipeline() directly, so there's no | |
| risk of the two entry points drifting apart in behavior. | |
| """ | |
| import argparse | |
| import asyncio | |
| import json | |
| import sys | |
| from agents.orchestrator import run_pipeline | |
| def _maybe_parse(value): | |
| """Same defensive parse as app.py / risk_synthesis_gate.py — some | |
| state values are raw JSON text, not parsed dicts, depending on | |
| whether that agent used output_schema.""" | |
| if isinstance(value, str): | |
| try: | |
| return json.loads(value) | |
| except (json.JSONDecodeError, TypeError): | |
| return value | |
| return value | |
| def main(): | |
| parser = argparse.ArgumentParser(prog="scamsentry") | |
| subparsers = parser.add_subparsers(dest="command", required=True) | |
| scan_parser = subparsers.add_parser("scan", help="Scan a message for scam/phishing signals") | |
| scan_parser.add_argument("text", help="The message text to scan") | |
| scan_parser.add_argument("--json", action="store_true", help="Print raw JSON instead of a formatted report") | |
| scan_parser.add_argument("--verbose", action="store_true", help="Print per-agent trace output") | |
| args = parser.parse_args() | |
| if args.command == "scan": | |
| try: | |
| result = asyncio.run( | |
| run_pipeline(text=args.text, verbose=args.verbose) | |
| ) | |
| except Exception as e: | |
| print(f"Scan failed: {e}", file=sys.stderr) | |
| sys.exit(1) | |
| if not result: | |
| print("Pipeline returned no result.", file=sys.stderr) | |
| sys.exit(1) | |
| if args.json: | |
| print(json.dumps(result, indent=2, default=str)) | |
| return | |
| verdict = _maybe_parse(result.get("final_verdict")) or {} | |
| print(f"Risk level: {str(verdict.get('risk_level', 'unknown')).upper()}") | |
| print(f"Confidence: {verdict.get('confidence', 'unknown')}") | |
| if verdict.get("injection_attempt_detected"): | |
| print("Warning: prompt-injection attempt detected in the analyzed content.") | |
| print() | |
| print(result.get("report_text", "(no report generated)")) | |
| if __name__ == "__main__": | |
| main() |