# app/api/schemas.py - Pydantic models for API request/response """API request and response schemas.""" from pydantic import BaseModel, Field, ConfigDict from typing import List, Dict, Optional, Any # Request models class AnalyzeRequest(BaseModel): """Request model for message analysis.""" message: str = Field(..., description="The scam message to analyze", min_length=1, max_length=5000) conversation_id: Optional[str] = Field(None, description="Conversation ID for multi-turn tracking") sender_id: Optional[str] = Field(None, description="Optional sender identifier") auto_report: bool = Field(False, description="Auto-report to law enforcement if high risk") class EnforcementReportRequest(BaseModel): """Request for manual law enforcement report.""" conversation_id: str = Field(..., description="Conversation ID to report") additional_notes: Optional[str] = Field(None, description="Additional notes for report") class UPIFreezeRequest(BaseModel): """Request to freeze a UPI ID.""" upi_id: str = Field(..., description="UPI ID to freeze") reason: str = Field(..., description="Reason for freeze request") campaign_id: Optional[str] = Field(None, description="Associated campaign ID") class TelemetryBeacon(BaseModel): """Silent telemetry beacon data - flexible schema.""" fingerprint_id: Optional[str] = Field(None, alias="fp", description="Unique Browser Fingerprint") location: Optional[str] = Field(None, description="IP-based location hint") user_agent: Optional[str] = Field(None, alias="ua", description="Browser User Agent") screen_resolution: Optional[str] = Field(None, alias="res") timezone: Optional[str] = Field(None, alias="tz") language: Optional[str] = Field(None, alias="lang") platform: Optional[str] = None renderer: Optional[str] = None canvas_hash: Optional[str] = None model_config = ConfigDict(populate_by_name=True, extra="allow") # ───────────────────────────────────────────────────────────────────────────── # RESPONSE MODELS # ───────────────────────────────────────────────────────────────────────────── class PersonaInfo(BaseModel): """Persona information in response.""" name: str language: str class HoneypotResponse(BaseModel): """Honeypot response details.""" message: str = Field(..., description="Generated response message") persona: str = Field(..., description="Persona name used") language: str = Field(..., description="Response language") class ExtractedIntelligence(BaseModel): """Extracted intelligence from message.""" phone_numbers: List[str] = [] upi_ids: List[str] = [] bank_accounts: List[str] = [] ifsc_codes: List[str] = [] emails: List[str] = [] urls: List[str] = [] class ThreatIntelligence(BaseModel): """Threat intelligence analysis.""" campaign_id: Optional[str] = None scam_pattern: Optional[str] = None fraud_vector: Optional[str] = None fraud_vector_description: Optional[str] = None related_entities: List[str] = [] severity: Optional[str] = None iocs: Dict[str, List[str]] = {} ttps: List[str] = [] recommended_actions: List[str] = [] class ConversationStrategy(BaseModel): """Conversation strategy information.""" id: str phase: str phase_goal: Optional[str] = None message_count: int trust_level: Optional[str] = None scammer_behavior: Optional[str] = None adaptive_strategy: Optional[str] = None class AnalysisDetails(BaseModel): """Detailed analysis information.""" risk_indicators: List[str] = [] matched_keywords: List[str] = [] scam_category: str class EnforcementAction(BaseModel): """Law enforcement action taken.""" type: str report_id: Optional[str] = None request_id: Optional[str] = None upi_id: Optional[str] = None status: str class Metadata(BaseModel): """Response metadata.""" processing_time_ms: int timestamp: str version: str class AnalyzeResponse(BaseModel): """Complete analysis response.""" status: str is_scam: bool scam_type: str confidence: float threat_level: str risk_score: float = 0.0 risk_explanation: List[str] = [] honeypot_response: HoneypotResponse extracted_intelligence: ExtractedIntelligence aggregated_intelligence: Dict[str, List[str]] = {} threat_intelligence: Dict[str, Any] = {} campaign_cluster: Optional[str] = None # 🔥 Campaign Clustering conversation: ConversationStrategy analysis: AnalysisDetails enforcement_actions: List[EnforcementAction] = [] telemetry: Optional[Dict[str, Any]] = None agentic_steps: Optional[List[str]] = None agent_loop: Optional[List[str]] = None adaptive_policy: Optional[Dict[str, Any]] = None metadata: Metadata class ScamTypeInfo(BaseModel): """Scam type information.""" description: str threat_level: str category: str sample_keywords: List[str] class ScamTypesResponse(BaseModel): """List of scam types.""" total_types: int scam_types: Dict[str, ScamTypeInfo] class PersonaDetail(BaseModel): """Single persona details.""" name: str age: int traits: List[str] language: str sample_response: str class PersonasResponse(BaseModel): """List of personas.""" total_personas: int personas: Dict[str, PersonaDetail] class StatisticsResponse(BaseModel): """System statistics.""" total_conversations: int total_messages: int scams_detected: int intelligence_extracted: int active_conversations: int scam_distribution: Dict[str, int] campaigns: List[Dict[str, Any]] = [] reports_filed: int = 0 class HealthResponse(BaseModel): """Health check response.""" status: str timestamp: str version: str llm_available: bool = False class ConversationDetail(BaseModel): """Conversation details.""" id: str scam_type: Optional[str] persona: Optional[str] phase: str message_count: int created_at: str updated_at: str history: List[Dict[str, Any]] aggregated_intelligence: Dict[str, List[str]] # ───────────────────────────────────────────────────────────────────────────── # GUVI CHALLENGE MANDATORY SCHEMAS # ───────────────────────────────────────────────────────────────────────────── class GUVIMessage(BaseModel): """Mandatory message format from GUVI platform.""" sender: str = Field(..., description="scammer or user") text: str = Field(..., description="Message content") timestamp: Optional[str] = Field(None, description="ISO-8601 format") class GUVIInputRequest(BaseModel): """Mandatory input format for GUVI evaluation.""" sessionId: Optional[str] = Field(None, description="Unique session ID (Primary)") processId: Optional[str] = Field(None, description="Alternative Session ID") session_id: Optional[str] = Field(None, description="Snake case alternative") message: Optional[Any] = Field(None, description="Current message (string, object, or None)") conversationHistory: List[Any] = Field(default=[], description="Previous exchange history") metadata: Optional[Dict[str, Any]] = None # Internal Tracker resolved_session_id: Optional[str] = None model_config = ConfigDict( populate_by_name=True, extra="allow" ) class GUVIIntelligence(BaseModel): """Strict key mapping for extracted intelligence as per spec.""" bankAccounts: List[str] = [] upiIds: List[str] = [] phishingLinks: List[str] = [] phoneNumbers: List[str] = [] suspiciousKeywords: List[str] = [] class GUVIEngagementMetrics(BaseModel): """Metrics required by GUVI.""" engagementDurationSeconds: int totalMessagesExchanged: int class GUVIOutputResponse(BaseModel): """ Mandatory response format for GUVI evaluation. Based on working response from Jan 28, GUVI expects FULL format: - scamDetected, totalMessagesExchanged, extractedIntelligence, agentNotes - Plus status and reply for API response """ status: str = "success" reply: str = Field("", description="AI response to the scammer") scamDetected: bool = False scamConfidence: float = 0.0 riskLevel: str = "LOW" totalMessagesExchanged: int = 1 # GUVI expects this at top level! extractedIntelligence: GUVIIntelligence = Field(default_factory=GUVIIntelligence) engagementMetrics: GUVIEngagementMetrics = Field(default_factory=lambda: GUVIEngagementMetrics(engagementDurationSeconds=0, totalMessagesExchanged=1)) agentNotes: str = "" honeypotResponse: str = "" data: Optional[Dict[str, Any]] = None # 🔥 For Handshake/Init Response model_config = ConfigDict( extra="ignore" ) class GUVIOutputResponseInternal(BaseModel): """Internal version with full fields for callbacks and logging.""" status: str = "success" reply: str = Field(..., description="AI response to the scammer") honeypotResponse: str = "" scamDetected: bool = False scamConfidence: float = 0.0 riskLevel: str = "LOW" extractedIntelligence: GUVIIntelligence = Field(default_factory=GUVIIntelligence) engagementMetrics: GUVIEngagementMetrics = Field(default_factory=lambda: GUVIEngagementMetrics(engagementDurationSeconds=0, totalMessagesExchanged=1)) agentNotes: str = "" data: Optional[Dict[str, Any]] = None # 🔥 Internal support for Handshake class GUVIFinalCallback(BaseModel): """Mandatory final report payload for GUVI evaluation endpoint.""" sessionId: str scamDetected: bool totalMessagesExchanged: int extractedIntelligence: GUVIIntelligence agentNotes: str __all__ = [ "AnalyzeRequest", "AnalyzeResponse", "ScamTypesResponse", "PersonasResponse", "StatisticsResponse", "HealthResponse", "ConversationDetail", "EnforcementReportRequest", "UPIFreezeRequest", "GUVIInputRequest", "GUVIOutputResponse", "GUVIFinalCallback" ]