Spaces:
Sleeping
Sleeping
| import re | |
| import os | |
| from datetime import datetime | |
| from fpdf import FPDF | |
| from app.models.schemas import ExtractedIntel | |
| from app.core.config import settings | |
| import httpx | |
| import logging | |
| logger = logging.getLogger(__name__) | |
| async def send_guvi_callback(session_id: str, scam_detected: bool, turn_count: int, intel: ExtractedIntel, agent_notes: str = None, duration: int = 0): | |
| """ | |
| Sends the mandatory final result callback to the GUVI evaluation endpoint. | |
| Async version for integration into LangGraph. | |
| """ | |
| url = "https://hackathon.guvi.in/api/updateHoneyPotFinalResult" | |
| payload = { | |
| "sessionId": session_id, | |
| "scamDetected": scam_detected, | |
| "totalMessagesExchanged": turn_count, | |
| "engagementDurationSeconds": duration if 'duration' in locals() else 0, | |
| "extractedIntelligence": { | |
| "phoneNumbers": intel.phone_numbers, | |
| "bankAccounts": intel.bank_details, | |
| "upiIds": intel.upi_ids, | |
| "phishingLinks": intel.phishing_links, | |
| "emailAddresses": intel.emails, | |
| "caseIds": intel.case_ids, | |
| "policyNumbers": intel.policy_numbers, | |
| "orderNumbers": intel.order_numbers | |
| }, | |
| "agentNotes": agent_notes or intel.agent_notes or "Scam engagement in progress." | |
| } | |
| headers = { | |
| "Content-Type": "application/json", | |
| "User-Agent": "Helware-Forensic-Platform/2.0" | |
| } | |
| try: | |
| async with httpx.AsyncClient() as client: | |
| response = await client.post(url, json=payload, headers=headers, timeout=10.0) | |
| if response.status_code == 200: | |
| logger.info(f" Mandatory GUVI callback successful for session {session_id}") | |
| else: | |
| logger.error(f" GUVI callback failed: {response.status_code} - {response.text}") | |
| except Exception as e: | |
| logger.error(f" Critical error during GUVI callback: {e}") | |
| def generate_scam_report(session_id: str, intel: ExtractedIntel, persona_name: str) -> str: | |
| """ | |
| Generates a PDF report for the National Cyber Crime Reporting Portal. | |
| Returns the path to the generated PDF. | |
| """ | |
| pdf = FPDF() | |
| pdf.add_page() | |
| # Title | |
| pdf.set_font("Arial", "B", 16) | |
| pdf.cell(0, 10, "Cyber Crime Incident Report", ln=True, align="C") | |
| pdf.set_font("Arial", "I", 10) | |
| pdf.cell(0, 10, f"Generated on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", ln=True, align="C") | |
| pdf.ln(10) | |
| # Incident Details | |
| pdf.set_font("Arial", "B", 12) | |
| pdf.cell(0, 10, "1. Incident Overview", ln=True) | |
| pdf.set_font("Arial", "", 10) | |
| pdf.multi_cell(0, 10, f"Session ID: {session_id}\nAssigned Persona: {persona_name}\nStatus: Intelligence Gathered") | |
| pdf.ln(5) | |
| # Extracted Intelligence | |
| pdf.set_font("Arial", "B", 12) | |
| pdf.cell(0, 10, "2. Extracted Intelligence (Forensics)", ln=True) | |
| pdf.set_font("Arial", "", 10) | |
| if intel.upi_ids: | |
| pdf.set_font("Arial", "B", 10) | |
| pdf.cell(0, 8, "UPI IDs Identified:", ln=True) | |
| pdf.set_font("Arial", "", 10) | |
| for upi in intel.upi_ids: | |
| pdf.cell(0, 8, f"- {upi}", ln=True) | |
| if intel.bank_details: | |
| pdf.set_font("Arial", "B", 10) | |
| pdf.cell(0, 8, "Bank Account Details:", ln=True) | |
| pdf.set_font("Arial", "", 10) | |
| for bank in intel.bank_details: | |
| pdf.cell(0, 8, f"- {bank}", ln=True) | |
| if intel.phishing_links: | |
| pdf.set_font("Arial", "B", 10) | |
| pdf.cell(0, 8, "Suspicious/Phishing Links:", ln=True) | |
| pdf.set_font("Arial", "", 10) | |
| for link in intel.phishing_links: | |
| pdf.cell(0, 8, f"- {link}", ln=True) | |
| if not any([intel.upi_ids, intel.bank_details, intel.phishing_links]): | |
| pdf.cell(0, 8, "No financial or malicious markers identified in this session yet.", ln=True) | |
| pdf.ln(10) | |
| # Footer/Legal Disclaimer | |
| pdf.set_font("Arial", "I", 8) | |
| pdf.multi_cell(0, 5, "This report is generated by the Agentic Honeypot System for submission to the National Cyber Crime Reporting Portal (cybercrime.gov.in). The data contained herein is for investigative purposes only.") | |
| # Save PDF | |
| reports_dir = settings.REPORTS_DIR | |
| filename = f"report_{session_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf" | |
| file_path = os.path.join(reports_dir, filename) | |
| pdf.output(file_path) | |
| return filename |