Spaces:
Sleeping
Sleeping
| """ | |
| 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 | |