Batch EN detox inference and safer gibberish early-exit

app/services/moderation_service.py

@@ -121,6 +121,79 @@ def _blacklist_early_result(profanity_hits, insult_hits, dil, cleaned_text):
     return verdict["decision"], verdict["reason"], verdict["risk_level"], dil, cleaned_text, detail
 
 
+def _is_obvious_gibberish(cleaned_text: str) -> bool:
+    letters = re.sub(r"[^a-z]", "", cleaned_text.lower())
+    if not letters:
+        return True
+
+    tokens = [t for t in cleaned_text.split() if t]
+    if len(tokens) <= 1:
+        return True
+
+    vowel_ratio = sum(1 for c in letters if c in "aeiou") / max(len(letters), 1)
+    unique_ratio = len(set(letters)) / max(len(letters), 1)
+    if len(letters) >= 7 and vowel_ratio < 0.18:
+        return True
+    if len(letters) >= 10 and unique_ratio < 0.25:
+        return True
+
+    return False
+
+
+def _split_en_detox_result(raw_res, expected_len: int):
+    if isinstance(raw_res, list):
+        parsed = []
+        for item in raw_res:
+            if isinstance(item, dict):
+                parsed.append({k: float(v) for k, v in item.items()})
+        if len(parsed) == expected_len:
+            return parsed
+
+    if isinstance(raw_res, dict):
+        out = [dict() for _ in range(expected_len)]
+        for key, value in raw_res.items():
+            if isinstance(value, (list, tuple)) and len(value) == expected_len:
+                for i in range(expected_len):
+                    out[i][key] = float(value[i])
+            elif hasattr(value, "__len__") and not isinstance(value, (str, bytes)):
+                try:
+                    if len(value) == expected_len:
+                        for i in range(expected_len):
+                            out[i][key] = float(value[i])
+                        continue
+                except TypeError:
+                    pass
+                for i in range(expected_len):
+                    out[i][key] = float(value)
+            else:
+                for i in range(expected_len):
+                    out[i][key] = float(value)
+        if all(out[i] for i in range(expected_len)):
+            return out
+
+    return []
+
+
+def _predict_en_detox_batched(text_list, state, batch_size: int = 8):
+    if not text_list:
+        return []
+
+    scores = []
+    effective_batch = max(1, int(batch_size))
+    for i in range(0, len(text_list), effective_batch):
+        chunk = text_list[i : i + effective_batch]
+        raw_chunk = state["D_EN"].predict(chunk)
+        parsed = _split_en_detox_result(raw_chunk, len(chunk))
+
+        # Fallback for unexpected return shape from Detoxify implementation.
+        if not parsed:
+            parsed = [{k: float(v) for k, v in state["D_EN"].predict(text).items()} for text in chunk]
+
+        scores.extend(parsed)
+
+    return scores
+
+
 def _tr_off_scores_batched(text_list, state, batch_size: int = 8):
     if not text_list:
         return []
@@ -212,13 +285,14 @@ def run_moderation(text: str, platform_dil: str = "tr"):
         return _blacklist_early_result(profanity_hits, insult_hits, dil, temiz)
 
     if dil == "en":
+        gb = None
         if state["GB_PIPE"] is not None:
             gb_raw = state["GB_PIPE"](temiz)[0]
             gb = {
                 "label": str(gb_raw.get("label", "")),
                 "score": float(gb_raw.get("score", 0.0)),
             }
-            if gb.get("label", "").lower() == "noise" and gb.get("score", 0.0) > 0.98:
+            if gb.get("label", "").lower() == "noise" and gb.get("score", 0.0) > 0.98 and _is_obvious_gibberish(temiz):
                 early_exit_count += 1
                 _log_pipeline_counts(early_exit_count, detoxify_call_count)
                 return (
@@ -272,8 +346,8 @@ def run_moderation_batch(texts, platform_dil: str = "tr", batch_size: int = 8):
     detoxify_call_count = 0
 
     results = [None] * len(texts)
-    tr_pending = []
-    tr_pending_texts = []
+    pending = []
+    pending_texts = []
 
     for idx, text in enumerate(texts):
         temiz = clean_text_nfkc(text)
@@ -298,62 +372,70 @@ def run_moderation_batch(texts, platform_dil: str = "tr", batch_size: int = 8):
 
         if dil == "en":
             gb = None
+            gb_is_noise = False
             if state["GB_PIPE"] is not None:
                 gb_raw = state["GB_PIPE"](temiz)[0]
                 gb = {
                     "label": str(gb_raw.get("label", "")),
                     "score": float(gb_raw.get("score", 0.0)),
                 }
-                if gb.get("label", "").lower() == "noise" and gb.get("score", 0.0) > 0.98:
-                    early_exit_count += 1
-                    results[idx] = (
-                        "🗑️ GIBBERISH/SPAM",
-                        "Metin anlamsız karakter dizileri içeriyor.",
-                        "LOW",
-                        "en",
-                        temiz,
-                        {"gibberish": gb, "action": "MONITOR", "detox": {}},
-                    )
-                    continue
-
-            raw_res = state["D_EN"].predict(temiz)
-            detoxify_call_count += 1
-            res = {k: float(v) for k, v in raw_res.items()}
-            tox_score = float(res.get("toxicity", 0.0))
-            ins_score = float(res.get("insult", 0.0))
-            identity_attack = float(res.get("identity_attack", 0.0))
-            detail = {
-                "detox": res,
-                "insult": ins_score,
-                "toxicity": tox_score,
-                "identity_attack": identity_attack,
-                "hits": profanity_hits,
-                "insult_hits": insult_hits,
-                "gibberish": gb if state["GB_PIPE"] is not None else None,
-            }
-            verdict = calculate_verdict(
-                profanity_hits,
-                insult_hits,
-                {
-                    "off_score": 0.0,
-                    "detox_toxicity": tox_score,
-                },
-            )
-            detail.update({"action": ACTION_MAP.get(verdict["risk_level"], "MONITOR")})
-            results[idx] = (verdict["decision"], verdict["reason"], verdict["risk_level"], dil, temiz, detail)
+                gb_is_noise = gb.get("label", "").lower() == "noise" and gb.get("score", 0.0) > 0.98
+
+            if gb_is_noise and _is_obvious_gibberish(temiz):
+                early_exit_count += 1
+                results[idx] = (
+                    "🗑️ GIBBERISH/SPAM",
+                    "Metin anlamsız karakter dizileri içeriyor.",
+                    "LOW",
+                    "en",
+                    temiz,
+                    {"gibberish": gb, "action": "MONITOR", "detox": {}},
+                )
+                continue
+
+            pending.append((idx, temiz, profanity_hits, insult_hits, gb))
+            pending_texts.append(temiz)
             continue
 
-        tr_pending.append((idx, temiz, profanity_hits, insult_hits))
-        tr_pending_texts.append(temiz)
-
-    if tr_pending_texts:
-        off_scores = _tr_off_scores_batched(tr_pending_texts, state, batch_size=batch_size)
-        for pending, off_score in zip(tr_pending, off_scores):
-            idx, temiz, profanity_hits, insult_hits = pending
-            result, detoxify_called = _tr_result_with_off_score(temiz, profanity_hits, insult_hits, off_score, state, dil)
-            if detoxify_called:
-                detoxify_call_count += 1
-            results[idx] = result
+        pending.append((idx, temiz, profanity_hits, insult_hits, None))
+        pending_texts.append(temiz)
+
+    if pending_texts:
+        if dil == "en":
+            en_scores = _predict_en_detox_batched(pending_texts, state, batch_size=batch_size)
+            detoxify_call_count += len(pending_texts)
+            for pending_item, res in zip(pending, en_scores):
+                idx, temiz, profanity_hits, insult_hits, gb = pending_item
+                tox_score = float(res.get("toxicity", 0.0))
+                ins_score = float(res.get("insult", 0.0))
+                identity_attack = float(res.get("identity_attack", 0.0))
+                detail = {
+                    "detox": res,
+                    "insult": ins_score,
+                    "toxicity": tox_score,
+                    "identity_attack": identity_attack,
+                    "hits": profanity_hits,
+                    "insult_hits": insult_hits,
+                    "gibberish": gb if state["GB_PIPE"] is not None else None,
+                }
+                verdict = calculate_verdict(
+                    profanity_hits,
+                    insult_hits,
+                    {
+                        "off_score": 0.0,
+                        "detox_toxicity": tox_score,
+                    },
+                )
+                detail.update({"action": ACTION_MAP.get(verdict["risk_level"], "MONITOR")})
+                results[idx] = (verdict["decision"], verdict["reason"], verdict["risk_level"], dil, temiz, detail)
+        else:
+            off_scores = _tr_off_scores_batched(pending_texts, state, batch_size=batch_size)
+            for pending_item, off_score in zip(pending, off_scores):
+                idx, temiz, profanity_hits, insult_hits, _ = pending_item
+                result, detoxify_called = _tr_result_with_off_score(temiz, profanity_hits, insult_hits, off_score, state, dil)
+                if detoxify_called:
+                    detoxify_call_count += 1
+                results[idx] = result
 
     _log_pipeline_counts(early_exit_count, detoxify_call_count)