import re from enum import Enum class ScamType(Enum): """Scam type categories""" KRA_IMPERSONATION = 'kra_impersonation', 'KRA Impersonation' MPESA_FRAUD = 'mpesa_fraud', 'M-Pesa Fraud' BANK_IMPERSONATION= 'bank_impersonation','Bank Impersonation' LOTTERY_PRIZE = 'lottery_prize', 'Lottery/Prize Scam' EMERGENCY_SCAM = 'emergency_scam', 'Emergency Scam' LOAN_SCAM = 'loan_scam', 'Loan Scam' INVESTMENT_SCAM = 'investment_scam', 'Investment Scam' ROMANCE_SCAM = 'romance_scam', 'Romance Scam' PHISHING = 'phishing', 'Phishing' OTHER = 'other', 'Other' def __new__(cls, value, label): obj = object.__new__(cls) obj._value_ = value obj.label = label return obj # Priority-ordered rules: first match wins. # Each rule is (set_of_flags_that_trigger_this_type, ScamType) # ALL listed flags must be present for the rule to match, # OR any single flag if the set has only one entry. _SCAM_TYPE_RULES = [ # Single-flag triggers (most specific first) ({"impersonates_KRA"}, ScamType.KRA_IMPERSONATION), ({"impersonates_bank"}, ScamType.BANK_IMPERSONATION), ({"lottery_no_entry"}, ScamType.LOTTERY_PRIZE), ({"unsolicited_prize_notification"}, ScamType.LOTTERY_PRIZE), ({"fake_reversal_trick"}, ScamType.MPESA_FRAUD), ({"exploits_family_emergency"}, ScamType.EMERGENCY_SCAM), ({"advance_fee_fraud"}, ScamType.LOAN_SCAM), ({"job_offer_with_fee"}, ScamType.LOAN_SCAM), ({"offers_unrealistic_reward"}, ScamType.INVESTMENT_SCAM), # Phishing covers any credential/identity harvest ({"requests_OTP"}, ScamType.PHISHING), ({"requests_PIN_or_password"}, ScamType.PHISHING), ({"credential_phishing"}, ScamType.PHISHING), ({"requests_account_details"}, ScamType.PHISHING), # M-Pesa fraud: telecom impersonation + money transfer ({"impersonates_telecom", "requests_money_transfer"}, ScamType.MPESA_FRAUD), # Authority + payment demand without a more specific category ({"impersonates_authority", "demands_immediate_payment"}, ScamType.OTHER), # Police impersonation (distinct from KRA) ({"impersonates_police"}, ScamType.OTHER), ] def classify_scam_type(matched_flags: list) -> str | None: """ Given a list of matched pattern flags, return the most relevant ScamType label string, or None if no scam is detected. """ flag_set = set(matched_flags) if not flag_set: return None for required_flags, scam_type in _SCAM_TYPE_RULES: if required_flags.issubset(flag_set): return scam_type.label # Fallback: something was flagged but didn't match a specific category return ScamType.OTHER.label # Pattern flags mapped to Kenyan-specific keywords and phrases # Each entry: (list of regex patterns, weight 1-3) # Weight 3 = critical indicator, 2 = strong, 1 = moderate PATTERN_RULES = { "impersonates_KRA": ( [r'\bKRA\b', r'kenya revenue authority', r'\bitax\b', r'tax compliance', r'itax', r'kra pin', r'tax discrepanc', r'unfiled returns', r'kra compliance', r'kra enforcement', r'unpaid taxes', r'tax penalty', r'tax audit'], 2 ), "impersonates_police": ( [r'\bDCI\b', r'directorate of criminal', r'\bpolice\b', r'\bofficer\b', r'\binspector\b', r'\bsergeant\b', r'police station', r'criminal investigation', r'police officer', r'law enforcement'], 2 ), "impersonates_bank": ( [r'equity bank', r'\bKCB\b', r'\bNCBA\b', r'co-operative bank', r'co operative bank', r'\bstanbic\b', r'\babsa\b', r'standard chartered', r'family bank', r'bank compliance', r'bank officer', r'bank security'], 2 ), "impersonates_telecom": ( [r'\bsafaricom\b', r'\bairtel\b', r'\btelkom\b', r'm-pesa agent', r'mpesa agent', r'telecom officer', r'network compliance'], 1 ), "impersonates_authority": ( [r'government officer', r'ministry of', r'county government', r'\bntsa\b', r'\bnhif\b', r'\bnssf\b', r'immigration officer', r'judiciary', r'court officer', r'kenya power', r'\bkplc\b', r'nairobi water', r'public health officer', r'national government'], 2 ), "demands_immediate_payment": ( [r'pay now', r'pay immediately', r'send money', r'send.*m-?pesa', r'm-?pesa.*send', r'pay.*today', r'make payment now', r'transfer.*now', r'pay the amount', r'settle.*now', r'settle.*today', r'lipa sasa', r'tuma pesa', r'lipa.*haraka'], 3 ), "threatens_arrest_or_legal": ( [r'\barrest\b', r'\bwarrant\b', r'court case', r'\bDPP\b', r'director of public prosecution', r'\bprosecute\b', r'\bprison\b', r'\bjail\b', r'criminal offense', r'criminal offence', r'taken into custody', r'legal action', r'forward your file', r'kukamatwa', r'kufungwa', r'mahakama'], 3 ), "threatens_account_suspension": ( [r'freeze.*account', r'block.*account', r'suspend.*account', r'close.*account', r'deactivate.*account', r'account.*frozen', r'account.*blocked', r'account.*suspended', r'kuzuia akaunti', r'kufunga akaunti'], 2 ), "creates_false_urgency": ( [r'within \d+ (hour|minute|min)', r'by \d+(am|pm)', r'right now', r'immediately', r'urgent', r'today only', r'expires today', r'no extensions', r'last chance', r'final notice', r'deadline', r'haraka', r'sasa hivi', r'dakika chache'], 2 ), "requests_OTP": ( [r'\bOTP\b', r'one.time password', r'one.time code', r'verification code', r'code sent to you', r'code we sent', r'enter the code', r'confirm.*code', r'namba.*uliyopewa', r'msimbo'], 3 ), "requests_PIN_or_password": ( [r'\bPIN\b', r'\bpassword\b', r'secret code', r'security code', r'account password', r'mpesa pin', r'm-pesa pin', r'banking password', r'usiri', r'nambari ya siri'], 3 ), "requests_ID_number": ( [r'ID number', r'national ID', r'identity number', r'\bpassport\b', r'kra pin number', r'id namba', r'namba ya kitambulisho', r'kitambulisho chako'], 2 ), "requests_account_details": ( [r'account number', r'bank account', r'sort code', r'account details', r'banking details', r'namba ya akaunti'], 2 ), "credential_phishing": ( [r'verify your account', r'confirm your details', r'update your details', r'login details', r'access your account', r'validate your account', r'account verification', r'confirm.*identity'], 2 ), "lottery_no_entry": ( [r'you have won', r'you\'ve won', r'lucky winner', r'\blottery\b', r'\bjackpot\b', r'you are selected', r'selected as.*winner', r'prize.*claim', r'claim.*prize', r'umeshinda', r'mshindi'], 3 ), "advance_fee_fraud": ( [r'processing fee', r'handling fee', r'registration fee', r'small fee', r'administrative fee', r'release.*fee', r'clearance fee', r'ada ya usindikaji', r'malipo ya usajili'], 3 ), "fake_reversal_trick": ( [r'sent.*by mistake', r'wrong.*number', r'wrong account', r'please reverse', r'kindly reverse', r'accidentally.*sent', r'nilituma kwa makosa', r'nirudishie'], 3 ), "offers_unrealistic_reward": ( [r'million(s)? shilling', r'\bmillions\b', r'huge amount', r'large sum', r'windfall', r'unclaimed fund', r'inheritance.*fund', r'you.*inherit'], 2 ), "instructs_secrecy": ( [r"don't tell", r'do not tell', r'keep.*confidential', r'keep.*secret', r'between us', r'private matter', r'tell no one', r'usimwambie', r'siri yetu'], 2 ), "exploits_family_emergency": ( [r'accident', r'hospital', r'\binjured\b', r'\bemergency\b', r'critical condition', r'urgent.*help', r'family.*emergency', r'ajali', r'hospitalini', r'dharura'], 2 ), "job_offer_with_fee": ( [r'job offer', r'vacancy', r'employment.*fee', r'application fee', r'registration.*job', r'nafasi ya kazi', r'kazi.*malipo'], 2 ), "requests_money_transfer": ( [r'm-?pesa', r'paybill.*\d{4,}', r'till.*\d{4,}', r'send.*\d{3,}.*shilling', r'send.*ksh', r'transfer.*ksh', r'tuma.*ksh', r'lipa.*ksh'], 2 ), "unsolicited_prize_notification": ( [r'congratulations.*won', r'you have been selected', r'randomly selected', r'our records show.*won', r'pongezi.*umeshinda'], 3 ), } # Weights that are considered critical — any single match already indicates high risk CRITICAL_FLAGS = { "requests_OTP", "requests_PIN_or_password", "fake_reversal_trick", "lottery_no_entry", "advance_fee_fraud", "unsolicited_prize_notification", "threatens_arrest_or_legal", "exploits_family_emergency", } class PatternAnalyzer: def __init__(self): # Pre-compile all regexes for performance self.compiled_rules = {} for flag, (patterns, weight) in PATTERN_RULES.items(): compiled = [re.compile(p, re.IGNORECASE) for p in patterns] self.compiled_rules[flag] = (compiled, weight) def analyze(self, text: str) -> dict: """ Analyze text for scam patterns. Returns: pattern_score: 0-100 float matched_flags: list of matched pattern flag names is_scam_pattern: bool (True if pattern_score >= 40) """ if not text or not text.strip(): return {"pattern_score": 0.0, "matched_flags": [], "is_scam_pattern": False, "scam_type": None} matched_flags = [] total_weight = 0 has_critical = False for flag, (patterns, weight) in self.compiled_rules.items(): for pattern in patterns: if pattern.search(text): matched_flags.append(flag) total_weight += weight if flag in CRITICAL_FLAGS: has_critical = True break # one match per flag is enough # Score calculation: # Use a reference of 20 weight units = "definitely a scam" # (e.g. 4 strong flags × weight 5, or mix of moderate+critical) # This keeps individual matched flags meaningful rather than diluting # across all 23 possible flags. REFERENCE_WEIGHT = 20.0 base_score = (total_weight / REFERENCE_WEIGHT) * 100 n = len(matched_flags) if n == 0: pattern_score = 0.0 elif n == 1 and not has_critical: # Single weak match: low confidence, cap at 25% pattern_score = min(base_score * 1.0, 25.0) elif has_critical: # Any critical flag raises floor to 60% pattern_score = max(base_score * 1.5, 60.0) else: # Multiple non-critical matches: amplify moderately pattern_score = min(base_score * 1.5, 95.0) pattern_score = min(round(pattern_score, 1), 100.0) is_scam = pattern_score >= 40.0 return { "pattern_score": pattern_score, "matched_flags": matched_flags, "is_scam_pattern": is_scam, "scam_type": classify_scam_type(matched_flags) if is_scam else None, } # Singleton for import _analyzer = None def get_analyzer() -> PatternAnalyzer: global _analyzer if _analyzer is None: _analyzer = PatternAnalyzer() return _analyzer if __name__ == "__main__": analyzer = PatternAnalyzer() tests = [ ("Hello, this is Officer James Mwangi from Kenya Revenue Authority. " "Your unpaid taxes of KSh 45,000 must be paid within 2 hours via M-Pesa " "or we will issue a warrant for your arrest.", "KRA arrest threat scam"), ("Congratulations! You have won KSh 500,000 in our lucky draw. " "To claim your prize, pay a processing fee of KSh 2,000 via paybill 123456.", "Lottery + fee scam"), ("Your M-Pesa OTP is 847291. Please do not share this code with anyone.", "Legitimate OTP message"), ("Hi, I sent money to your number by mistake. Please reverse KSh 3,000 back to me.", "Fake reversal scam"), ("Good morning! This is Equity Bank calling to confirm your appointment " "with our mortgage advisor on Friday at 2 PM.", "Legitimate bank call"), ("This is KRA enforcement. Your iTax account shows three years of unfiled returns. " "We are closing your business unless you pay KSh 85,000 right now. " "Send money to paybill 247247 and send us the transaction code.", "KRA enforcement scam"), ("Your Safaricom bill is ready. Please pay KSh 1,200 via M-Pesa paybill 100100.", "Legit bill payment"), ] print("=" * 70) print("PATTERN ANALYZER TEST") print("=" * 70) for text, desc in tests: result = analyzer.analyze(text) verdict = "SCAM" if result["is_scam_pattern"] else "LEGIT" print(f"\n[{verdict}] {desc}") print(f" Score: {result['pattern_score']}%") if result["matched_flags"]: print(f" Flags: {', '.join(result['matched_flags'])}") else: print(f" Flags: none") print("\n" + "=" * 70)