""" Text normalizer — runs before any model or keyword check. Handles: 1. Common English typos and phonetic misspellings 2. Kinyarwanda typos and variant spellings 3. SMS/WhatsApp abbreviations common in Rwandan groups 4. Mixed-case normalization 5. Repeated characters (e.g. "baaad" → "bad", "noooo" → "no") 6. Missing spaces after punctuation 7. Number/letter substitutions (l33t-style: "m3dicine" → "medicine") This runs on the RAW text before language detection and translation, so the models always receive cleaner input. """ from __future__ import annotations import re import logging logger = logging.getLogger(__name__) _URL_ONLY_RE = re.compile(r"^\s*(https?://\S+|www\.\S+)\s*$", re.I) # ============================================================================= # ENGLISH TYPO / PHONETIC CORRECTIONS # Key = regex pattern, Value = replacement # Ordered from most specific to most general # ============================================================================= _EN_CORRECTIONS: list[tuple[re.Pattern, str]] = [ # ── Medical / pharmacy ──────────────────────────────────────────────── (re.compile(r"\bmed[ei]c[ie]ne\b", re.I), "medicine"), (re.compile(r"\bmedecine\b", re.I), "medicine"), (re.compile(r"\bmedeicne\b", re.I), "medicine"), (re.compile(r"\bmedcine\b", re.I), "medicine"), (re.compile(r"\bpharma[cs]y\b", re.I), "pharmacy"), (re.compile(r"\bpharmcy\b", re.I), "pharmacy"), (re.compile(r"\bpharmacie\b", re.I), "pharmacy"), (re.compile(r"\bpharmaice\b", re.I), "pharmacy"), (re.compile(r"\bpharmaci[ae]\b", re.I), "pharmacy"), (re.compile(r"\bprescri[pb]tion\b", re.I), "prescription"), (re.compile(r"\bhos[pb]ital\b", re.I), "hospital"), (re.compile(r"\bhospitla\b", re.I), "hospital"), (re.compile(r"\bdocotr\b", re.I), "doctor"), (re.compile(r"\bdokter\b", re.I), "doctor"), (re.compile(r"\bnurce\b", re.I), "nurse"), (re.compile(r"\binjec[st]ion\b", re.I), "injection"), # ── Violence / crime ────────────────────────────────────────────────── (re.compile(r"\braped?\b", re.I), "raped"), (re.compile(r"\brapd\b", re.I), "raped"), (re.compile(r"\bkiled\b", re.I), "killed"), (re.compile(r"\bkill[dt]\b", re.I), "killed"), (re.compile(r"\bmurder[dt]\b", re.I), "murdered"), (re.compile(r"\bassult(?:ed)?\b", re.I), "assaulted"), (re.compile(r"\battac?ked\b", re.I), "attacked"), (re.compile(r"\brobery\b", re.I), "robbery"), (re.compile(r"\bthret(?:en)?(?:ed)?\b", re.I), "threatened"), (re.compile(r"\bharas(?:s)?(?:ed|ment)?\b", re.I), "harassed"), # ── Food / poisoning ────────────────────────────────────────────────── (re.compile(r"\bpoisend?\b", re.I), "poisoned"), (re.compile(r"\bpoisining\b", re.I), "poisoning"), (re.compile(r"\bvomitting\b", re.I), "vomiting"), (re.compile(r"\bdiarhea\b", re.I), "diarrhea"), (re.compile(r"\bdiarhoea\b", re.I), "diarrhoea"), (re.compile(r"\bcontaminted\b", re.I), "contaminated"), (re.compile(r"\bcontamineted\b", re.I), "contaminated"), (re.compile(r"\bexpird\b", re.I), "expired"), (re.compile(r"\bexpireed\b", re.I), "expired"), (re.compile(r"\brottened?\b", re.I), "rotten"), # ── Service / banking ───────────────────────────────────────────────── (re.compile(r"\bfraud[ulent]*\b", re.I), "fraud"), (re.compile(r"\bfrode\b", re.I), "fraud"), (re.compile(r"\bscamed\b", re.I), "scammed"), (re.compile(r"\bscamm?ed\b", re.I), "scammed"), (re.compile(r"\bovercharg[ei]d\b", re.I), "overcharged"), (re.compile(r"\bovercharg\b", re.I), "overcharge"), (re.compile(r"\brefund[dt]\b", re.I), "refunded"), (re.compile(r"\baccaunt\b", re.I), "account"), (re.compile(r"\baccount\b", re.I), "account"), (re.compile(r"\btransac[st]ion\b", re.I), "transaction"), (re.compile(r"\bwithdrawl\b", re.I), "withdrawal"), (re.compile(r"\bwithdrawal\b", re.I), "withdrawal"), (re.compile(r"\bdeposit[te]\b", re.I), "deposit"), (re.compile(r"\bpasword\b", re.I), "password"), (re.compile(r"\bpasswrod\b", re.I), "password"), # ── Telecom ─────────────────────────────────────────────────────────── (re.compile(r"\binternet\b", re.I), "internet"), (re.compile(r"\binternt\b", re.I), "internet"), (re.compile(r"\bconection\b", re.I), "connection"), (re.compile(r"\bconnetion\b", re.I), "connection"), (re.compile(r"\bnetowrk\b", re.I), "network"), (re.compile(r"\bsignal\b", re.I), "signal"), (re.compile(r"\bsingal\b", re.I), "signal"), (re.compile(r"\bsubscri[pb]tion\b", re.I), "subscription"), (re.compile(r"\bbundle[sd]?\b", re.I), "bundle"), (re.compile(r"\bbundel\b", re.I), "bundle"), # ── General service words ───────────────────────────────────────────── (re.compile(r"\bservise\b", re.I), "service"), (re.compile(r"\bservis\b", re.I), "service"), (re.compile(r"\bcustommer\b", re.I), "customer"), (re.compile(r"\bcustmer\b", re.I), "customer"), (re.compile(r"\bcompalint\b", re.I), "complaint"), (re.compile(r"\bcomplaint\b", re.I), "complaint"), (re.compile(r"\bproblem\b", re.I), "problem"), (re.compile(r"\bprobem\b", re.I), "problem"), (re.compile(r"\bproblm\b", re.I), "problem"), (re.compile(r"\bissue[sd]?\b", re.I), "issue"), (re.compile(r"\bmaneger\b", re.I), "manager"), (re.compile(r"\bmanagr\b", re.I), "manager"), (re.compile(r"\brecei[vp]t\b", re.I), "receipt"), (re.compile(r"\breciept\b", re.I), "receipt"), (re.compile(r"\bpayement\b", re.I), "payment"), (re.compile(r"\bpayemnt\b", re.I), "payment"), (re.compile(r"\bpayment\b", re.I), "payment"), (re.compile(r"\bdelivry\b", re.I), "delivery"), (re.compile(r"\bdeliveri\b", re.I), "delivery"), (re.compile(r"\bwaitting\b", re.I), "waiting"), (re.compile(r"\bwaitig\b", re.I), "waiting"), (re.compile(r"\bappointement\b", re.I), "appointment"), (re.compile(r"\bappointment\b", re.I), "appointment"), ] # ============================================================================= # KINYARWANDA TYPO / VARIANT CORRECTIONS # ============================================================================= _RW_CORRECTIONS: list[tuple[re.Pattern, str]] = [ # Common complaint words with variant spellings (re.compile(r"\bnab[iy]\b", re.I), "nabi"), # bad (re.compile(r"\bbib[iy]\b", re.I), "bibi"), # bad/terrible (re.compile(r"\bikibaz[ao]\b", re.I), "ikibazo"), # problem (re.compile(r"\bingoran[ae]\b", re.I), "ingorane"), # problem/difficulty (re.compile(r"\bmurako[sz]e\b", re.I), "murakoze"), # thank you (re.compile(r"\bmuraho\b", re.I), "muraho"), # hello (re.compile(r"\bamahoro\b", re.I), "amahoro"), # peace/hello (re.compile(r"\bserivisi\b", re.I), "serivisi"), # service (re.compile(r"\bservis[iy]\b", re.I), "serivisi"), # service (typo) (re.compile(r"\bamafarang[ae]\b", re.I), "amafaranga"), # money (re.compile(r"\bamafarang\b", re.I), "amafaranga"), # money (truncated) (re.compile(r"\bubuzim[ae]\b", re.I), "ubuzima"), # health (re.compile(r"\bibitar[ao]\b", re.I), "ibitaro"), # hospital (re.compile(r"\bumugand[ae]\b", re.I), "umuganda"), # community work (re.compile(r"\bkigal[iy]\b", re.I), "kigali"), # Kigali (re.compile(r"\brwand[ae]\b", re.I), "rwanda"), # Rwanda ] # ============================================================================= # SMS / WHATSAPP ABBREVIATIONS (common in Rwandan WhatsApp groups) # ============================================================================= _ABBREV: list[tuple[re.Pattern, str]] = [ # English abbreviations (re.compile(r"\bu\b", re.I), "you"), (re.compile(r"\bur\b", re.I), "your"), (re.compile(r"\br\b", re.I), "are"), (re.compile(r"\bpls\b", re.I), "please"), (re.compile(r"\bplz\b", re.I), "please"), (re.compile(r"\bplease\b", re.I), "please"), (re.compile(r"\bthx\b", re.I), "thanks"), (re.compile(r"\bthnx\b", re.I), "thanks"), (re.compile(r"\btnx\b", re.I), "thanks"), (re.compile(r"\bbtw\b", re.I), "by the way"), (re.compile(r"\basap\b", re.I), "as soon as possible"), (re.compile(r"\bfyi\b", re.I), "for your information"), (re.compile(r"\bomg\b", re.I), "oh my god"), (re.compile(r"\bsmh\b", re.I), "shaking my head"), (re.compile(r"\bwtf\b", re.I), "what the"), (re.compile(r"\bimo\b", re.I), "in my opinion"), (re.compile(r"\bimho\b", re.I), "in my opinion"), (re.compile(r"\bngl\b", re.I), "not going to lie"), (re.compile(r"\bnot gonna lie\b", re.I), "not going to lie"), (re.compile(r"\bw/o\b", re.I), "without"), (re.compile(r"\bw/\b", re.I), "with"), (re.compile(r"\b(\d+)k\b", re.I), r"\1000"), # 50k → 50000 (re.compile(r"\brwf\b", re.I), "RWF"), (re.compile(r"\bfrw\b", re.I), "RWF"), # Kinyarwanda abbreviations / SMS style (re.compile(r"\bnk\b", re.I), "nka"), # like/as (re.compile(r"\bnd\b", re.I), "ndi"), # I am (re.compile(r"\bknd\b", re.I), "kandi"), # and/also (re.compile(r"\bnb[iy]\b", re.I), "nabi"), # bad (re.compile(r"\bbb[iy]\b", re.I), "bibi"), # bad (re.compile(r"\bmrk\b", re.I), "murakoze"), # thank you (re.compile(r"\bmrh\b", re.I), "muraho"), # hello ] # ============================================================================= # NUMBER/LETTER SUBSTITUTIONS (l33t speak, common in informal messages) # ============================================================================= _LEET: list[tuple[re.Pattern, str]] = [ (re.compile(r"0"), "o"), (re.compile(r"1"), "i"), (re.compile(r"3"), "e"), (re.compile(r"4"), "a"), (re.compile(r"5"), "s"), (re.compile(r"@"), "a"), ] # ============================================================================= # REPEATED CHARACTER COLLAPSE # "baaad" → "bad", "noooo" → "no", "helpppp" → "help" # But preserve intentional doubles like "good", "food", "been" # ============================================================================= _REPEAT_RE = re.compile(r"(.)\1{2,}") # 3+ of same char → keep 1 def _collapse_repeats(text: str) -> str: return _REPEAT_RE.sub(r"\1\1", text) # keep max 2 (handles "good", "food") # ============================================================================= # MAIN NORMALIZE FUNCTION # ============================================================================= def normalize(text: str) -> str: """ Clean and normalize text before language detection and model inference. Steps: 1. Collapse excessive repeated characters 2. Fix number/letter substitutions in obvious leet-speak words 3. Expand SMS abbreviations 4. Fix Kinyarwanda typos 5. Fix English typos 6. Fix spacing issues Returns the normalized text. If normalization fails, returns original. """ if not text or not text.strip(): return text if _URL_ONLY_RE.match(text): return text.strip() try: original = text # Step 1: Collapse repeats ("baaad" → "baad", "noooo" → "no") text = _collapse_repeats(text) # Step 2: Leet-speak — only apply inside words that look like leet # (contains digits mixed with letters) to avoid mangling phone numbers def _fix_leet(word: str) -> str: if re.search(r"[a-zA-Z]", word) and re.search(r"[0-9@]", word): for pat, rep in _LEET: word = pat.sub(rep, word) return word text = re.sub(r"\b\w+\b", lambda m: _fix_leet(m.group()), text) # Step 3: SMS abbreviations for pat, rep in _ABBREV: text = pat.sub(rep, text) # Step 4: Kinyarwanda corrections for pat, rep in _RW_CORRECTIONS: text = pat.sub(rep, text) # Step 5: English corrections for pat, rep in _EN_CORRECTIONS: text = pat.sub(rep, text) # Step 6: Fix spacing text = re.sub(r"\s+", " ", text).strip() # Add space after punctuation if missing (e.g. "bad.The" → "bad. The") text = re.sub(r"([.!?,;:])([A-Za-z])", r"\1 \2", text) if text != original: logger.debug("Normalized: %r → %r", original[:80], text[:80]) return text except Exception as e: logger.warning("Normalization failed: %s", e) return text