""" Indian Multi-Modal Scam Dataset Loader Loads the curated dataset of India-specific scam patterns and provides keyword matching, pattern detection, and similarity scoring utilities for integration with the main scam detection pipeline. """ import json import os import re from typing import List, Dict, Optional DATASET_DIR = os.path.join(os.path.dirname(__file__), "samples") class IndianScamDataset: """Loads and queries the Indian multi-modal scam dataset.""" def __init__(self): self.text_scams: List[Dict] = [] self.url_scams: List[Dict] = [] self.benign_samples: List[Dict] = [] self.scam_keywords: Dict[str, List[str]] = {} # category -> keywords self.benign_keywords: List[str] = [] self.url_indicators: Dict[str, List[str]] = {} # indicator -> urls self._load_all() def _load_json(self, filename: str) -> List[Dict]: filepath = os.path.join(DATASET_DIR, filename) if os.path.exists(filepath): with open(filepath, 'r', encoding='utf-8') as f: return json.load(f) return [] def _load_all(self): """Load all dataset files and build lookup indexes.""" self.text_scams = self._load_json("text_scams.json") self.url_scams = self._load_json("url_scams.json") self.benign_samples = self._load_json("benign_samples.json") # Build keyword index by category for sample in self.text_scams: cat = sample.get("category", "unknown") if cat not in self.scam_keywords: self.scam_keywords[cat] = [] for kw in sample.get("keywords", []): if kw.lower() not in [k.lower() for k in self.scam_keywords[cat]]: self.scam_keywords[cat].append(kw) # Build benign keyword index for sample in self.benign_samples: for kw in sample.get("keywords", []): if kw.lower() not in [k.lower() for k in self.benign_keywords]: self.benign_keywords.append(kw) # Build URL indicator index for sample in self.url_scams: for ind in sample.get("indicators", []): if ind not in self.url_indicators: self.url_indicators[ind] = [] self.url_indicators[ind].append(sample.get("url", "")) def get_all_scam_keywords(self) -> List[str]: """Return a flat list of all unique scam keywords.""" all_kw = set() for keywords in self.scam_keywords.values(): for kw in keywords: all_kw.add(kw.lower()) return list(all_kw) def match_text_keywords(self, text: str) -> Dict: """ Match input text against the dataset's keyword patterns. Returns matched categories, keyword hits, and a boost score. """ text_lower = text.lower() matches = {} total_hits = 0 for category, keywords in self.scam_keywords.items(): hits = [] for kw in keywords: if kw.lower() in text_lower: hits.append(kw) if hits: matches[category] = hits total_hits += len(hits) # Calculate a boost score (0-30) based on keyword density boost = min(30, total_hits * 5) return { "matched_categories": matches, "total_keyword_hits": total_hits, "boost_score": boost, "top_category": max(matches, key=lambda c: len(matches[c])) if matches else None } def match_benign_patterns(self, text: str) -> Dict: """ Match text against known benign/safe patterns (e.g. real bank SMS). Returns matched keywords and a reduction score. """ text_lower = text.lower() hits = [kw for kw in self.benign_keywords if kw.lower() in text_lower] # Reduce risk score if strongly matching benign patterns reduction = min(40, len(hits) * 10) return { "matched_keywords": hits, "is_likely_benign": len(hits) >= 2, "reduction_score": reduction } def match_url_indicators(self, url: str) -> Dict: """ Analyze a URL against known Indian phishing indicators. Returns detected indicators and a risk boost score. """ url_lower = url.lower() detected = [] # Check for IP-based domain if re.search(r'https?://\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', url_lower): detected.append("ip_address_domain") # Check for suspicious TLDs suspicious_tlds = ['.xyz', '.tk', '.ml', '.ga', '.cf', '.gq', '.buzz', '.top'] for tld in suspicious_tlds: if tld in url_lower: detected.append("suspicious_tld") break # Check for Indian bank/service impersonation indian_brands = ['sbi', 'hdfc', 'icici', 'axis', 'paytm', 'phonepe', 'gpay', 'upi', 'npci', 'irctc', 'aadhaar', 'uidai', 'epfo', 'gst', 'incometax', 'flipkart', 'jio', 'airtel'] for brand in indian_brands: if brand in url_lower and not any(legit in url_lower for legit in [f'{brand}.co.in', f'{brand}.com', f'{brand}.org.in', f'{brand}.gov.in']): detected.append(f"{brand}_impersonation") # Check for subdomain spoofing (e.g., sbi.co.in.phishingsite.com) if re.search(r'\.gov\.in\.[a-z]+\.', url_lower) or re.search(r'\.co\.in\.[a-z]+\.', url_lower): detected.append("subdomain_spoofing") # Check for free hosting free_hosts = ['herokuapp', 'netlify', 'vercel', 'blogspot', 'wordpress.com', 'wixsite', 'weebly', 'github.io'] for host in free_hosts: if host in url_lower: detected.append("free_hosting") break # Check for scam keywords in URL path scam_url_keywords = ['verify', 'claim', 'refund', 'cashback', 'winner', 'kyc', 'update', 'reactivate', 'login', 'otp', 'reward'] for kw in scam_url_keywords: if kw in url_lower: detected.append(f"url_keyword_{kw}") boost = min(40, len(detected) * 8) return { "detected_indicators": detected, "indicator_count": len(detected), "boost_score": boost, "is_likely_phishing": len(detected) >= 2 } def get_similar_scam_examples(self, text: str, top_n: int = 3) -> List[Dict]: """ Find the most similar scam examples from the dataset by keyword overlap. Useful for explainability — showing users similar known scam patterns. """ text_lower = text.lower() scored = [] for sample in self.text_scams: keywords = sample.get("keywords", []) overlap = sum(1 for kw in keywords if kw.lower() in text_lower) if overlap > 0: scored.append({ "id": sample["id"], "category": sample["category"], "text_preview": sample["text"][:100] + "...", "keyword_overlap": overlap, "reference_risk_score": sample["risk_score"] }) scored.sort(key=lambda x: x["keyword_overlap"], reverse=True) return scored[:top_n] def get_dataset_stats(self) -> Dict: """Return summary statistics of the dataset.""" categories = {} for sample in self.text_scams: cat = sample.get("category", "unknown") categories[cat] = categories.get(cat, 0) + 1 return { "total_text_scams": len(self.text_scams), "total_url_scams": len(self.url_scams), "total_benign": len(self.benign_samples), "total_samples": len(self.text_scams) + len(self.url_scams) + len(self.benign_samples), "categories": categories, "total_unique_keywords": len(self.get_all_scam_keywords()), "url_indicator_types": len(self.url_indicators) } # Singleton instance _dataset_instance: Optional[IndianScamDataset] = None def get_dataset() -> IndianScamDataset: """Get or create the singleton dataset instance.""" global _dataset_instance if _dataset_instance is None: _dataset_instance = IndianScamDataset() return _dataset_instance