| """
|
| SQLAlchemy ORM Models.
|
|
|
| Defines database models for:
|
| - Conversation: Honeypot conversation sessions
|
| - Message: Individual messages in conversations
|
| - ExtractedIntelligence: Financial intelligence extracted from conversations
|
| """
|
|
|
| from datetime import datetime
|
| from typing import List, Optional
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| class Conversation:
|
| """
|
| Conversation model representing a honeypot session.
|
|
|
| Attributes:
|
| id: Primary key
|
| session_id: Unique session UUID
|
| language: Detected language (en, hi, hinglish)
|
| persona: Active persona name
|
| scam_detected: Whether scam was detected
|
| confidence: Detection confidence score
|
| turn_count: Number of conversation turns
|
| created_at: Session start timestamp
|
| updated_at: Last update timestamp
|
| """
|
|
|
| def __init__(
|
| self,
|
| session_id: str,
|
| language: str = "en",
|
| persona: Optional[str] = None,
|
| scam_detected: bool = False,
|
| confidence: float = 0.0,
|
| turn_count: int = 0,
|
| ) -> None:
|
| """Initialize Conversation model."""
|
| self.id: Optional[int] = None
|
| self.session_id = session_id
|
| self.language = language
|
| self.persona = persona
|
| self.scam_detected = scam_detected
|
| self.confidence = confidence
|
| self.turn_count = turn_count
|
| self.created_at = datetime.utcnow()
|
| self.updated_at = datetime.utcnow()
|
|
|
| def to_dict(self) -> dict:
|
| """Convert model to dictionary."""
|
| return {
|
| "id": self.id,
|
| "session_id": self.session_id,
|
| "language": self.language,
|
| "persona": self.persona,
|
| "scam_detected": self.scam_detected,
|
| "confidence": self.confidence,
|
| "turn_count": self.turn_count,
|
| "created_at": self.created_at.isoformat() if self.created_at else None,
|
| "updated_at": self.updated_at.isoformat() if self.updated_at else None,
|
| }
|
|
|
|
|
| class Message:
|
| """
|
| Message model representing a single conversation message.
|
|
|
| Attributes:
|
| id: Primary key
|
| conversation_id: Foreign key to Conversation
|
| turn_number: Turn number in conversation
|
| sender: Message sender (scammer/agent)
|
| message: Message content
|
| timestamp: Message timestamp
|
| """
|
|
|
| def __init__(
|
| self,
|
| conversation_id: int,
|
| turn_number: int,
|
| sender: str,
|
| message: str,
|
| ) -> None:
|
| """Initialize Message model."""
|
| self.id: Optional[int] = None
|
| self.conversation_id = conversation_id
|
| self.turn_number = turn_number
|
| self.sender = sender
|
| self.message = message
|
| self.timestamp = datetime.utcnow()
|
|
|
| def to_dict(self) -> dict:
|
| """Convert model to dictionary."""
|
| return {
|
| "id": self.id,
|
| "conversation_id": self.conversation_id,
|
| "turn_number": self.turn_number,
|
| "sender": self.sender,
|
| "message": self.message,
|
| "timestamp": self.timestamp.isoformat() if self.timestamp else None,
|
| }
|
|
|
|
|
| class ExtractedIntelligence:
|
| """
|
| ExtractedIntelligence model for storing financial intelligence.
|
|
|
| Attributes:
|
| id: Primary key
|
| conversation_id: Foreign key to Conversation
|
| upi_ids: List of extracted UPI IDs
|
| bank_accounts: List of extracted bank account numbers
|
| ifsc_codes: List of extracted IFSC codes
|
| phone_numbers: List of extracted phone numbers
|
| phishing_links: List of extracted phishing URLs
|
| extraction_confidence: Overall extraction confidence
|
| created_at: Extraction timestamp
|
| """
|
|
|
| def __init__(
|
| self,
|
| conversation_id: int,
|
| upi_ids: Optional[List[str]] = None,
|
| bank_accounts: Optional[List[str]] = None,
|
| ifsc_codes: Optional[List[str]] = None,
|
| phone_numbers: Optional[List[str]] = None,
|
| phishing_links: Optional[List[str]] = None,
|
| extraction_confidence: float = 0.0,
|
| ) -> None:
|
| """Initialize ExtractedIntelligence model."""
|
| self.id: Optional[int] = None
|
| self.conversation_id = conversation_id
|
| self.upi_ids = upi_ids or []
|
| self.bank_accounts = bank_accounts or []
|
| self.ifsc_codes = ifsc_codes or []
|
| self.phone_numbers = phone_numbers or []
|
| self.phishing_links = phishing_links or []
|
| self.extraction_confidence = extraction_confidence
|
| self.created_at = datetime.utcnow()
|
|
|
| def to_dict(self) -> dict:
|
| """Convert model to dictionary."""
|
| return {
|
| "id": self.id,
|
| "conversation_id": self.conversation_id,
|
| "upi_ids": self.upi_ids,
|
| "bank_accounts": self.bank_accounts,
|
| "ifsc_codes": self.ifsc_codes,
|
| "phone_numbers": self.phone_numbers,
|
| "phishing_links": self.phishing_links,
|
| "extraction_confidence": self.extraction_confidence,
|
| "created_at": self.created_at.isoformat() if self.created_at else None,
|
| }
|
|
|
| def has_intelligence(self) -> bool:
|
| """Check if any intelligence was extracted."""
|
| return any([
|
| self.upi_ids,
|
| self.bank_accounts,
|
| self.ifsc_codes,
|
| self.phone_numbers,
|
| self.phishing_links,
|
| ])
|
|
|
