verify_honeypot.py

import asyncio
import sys
import os
import json
from datetime import datetime

# Add the project root to sys.path
sys.path.append(os.getcwd())

from app.agents.orchestrator import HoneypotOrchestrator
from app.config import settings

# ANSI Colors for better visibility
class Colors:
    HEADER = '\033[95m'
    BLUE = '\033[94m'
    CYAN = '\033[96m'
    GREEN = '\033[92m'
    WARNING = '\033[93m'
    FAIL = '\033[91m'
    ENDC = '\033[0m'
    BOLD = '\033[1m'

async def run_test_case(orchestrator, case_name, message):
    print(f"\n{Colors.HEADER}{Colors.BOLD}--- TESTING: {case_name} ---{Colors.ENDC}")
    print(f"{Colors.BLUE}Input Message:{Colors.ENDC} {message}")
    
    start_time = datetime.now()
    try:
        result = await orchestrator.process_message(message=message, conversation_id=f"test_{case_name.lower()}")
        end_time = datetime.now()
        duration = (end_time - start_time).total_seconds()
        
        print(f"{Colors.GREEN}✅ SUCCESS (took {duration:.2f}s){Colors.ENDC}")
        print(f"{Colors.CYAN}Detected Scam:{Colors.ENDC} {result.get('scam_type', 'Unknown')}")
        print(f"{Colors.CYAN}Risk Score:{Colors.ENDC} {result.get('risk_score', 0):.2f}")
        
        intel = result.get('extracted_intelligence', {})
        if intel:
            print(f"{Colors.CYAN}Extracted Intel:{Colors.ENDC} {json.dumps(intel, indent=2)}")
        
        persona = result.get('honeypot_response', {}).get('persona', 'Unknown')
        response = result.get('honeypot_response', {}).get('message', 'No response generated')
        
        print(f"{Colors.CYAN}Active Persona:{Colors.ENDC} {persona}")
        print(f"{Colors.YELLOW}{Colors.BOLD}Honeypot Reply:{Colors.ENDC} {Colors.YELLOW}{response}{Colors.ENDC}")
        
        if result.get('explanation'):
            print(f"{Colors.CYAN}Reasoning:{Colors.ENDC} {result['explanation'][0] if isinstance(result['explanation'], list) else result['explanation']}")

    except Exception as e:
        print(f"{Colors.FAIL}❌ FAILED: {str(e)}{Colors.ENDC}")

async def main():
    print(f"{Colors.HEADER}{Colors.BOLD}🛡️ SENTINEL SCAM HONEYPOT - END-TO-END VERIFICATION{Colors.ENDC}")
    print("="*60)
    
    # Initialize Orchestrator
    orchestrator = HoneypotOrchestrator()
    print("Initializing Agents...")
    await orchestrator.initialize()
    print("All agents ready.\n")
    
    test_cases = [
        {
            "name": "BANKING_KYC_SCAM",
            "message": "Dear customer, your SBI YONO account is blocked today. Please update your KYC immediately at http://sbi-kcy-service.com or visit our nearest branch. Your reference ID is 55421."
        },
        {
            "name": "LOTTERY_PRIZE_SCAM",
            "message": "Congratulations!! You have won 25,00,000 RS from KBC Lucky Draw 2025. To claim your prize money, contact KBC Manager Mr. Amit Sharma on WhatsApp +91-9876543210. Processing fee of 15,000 RS is required."
        },
        {
            "name": "JOB_OFFER_SCAM",
            "message": "Part-time job offer! Earn 3000-8000 daily by simple task in your mobile. No experience needed. Contact us on WhatsApp for more details or join our group. Register now at http://india-jobs-wfh.org"
        }
    ]
    
    for case in test_cases:
        await run_test_case(orchestrator, case["name"], case["message"])
        print("-" * 40)

    print(f"\n{Colors.GREEN}{Colors.BOLD}VERIFICATION COMPLETE{Colors.ENDC}")

if __name__ == "__main__":
    asyncio.run(main())