""" ML inference engine — Hybrid: Afro-XLMR (local) + TF-IDF + Lexicon Phrase matching from both term and context_example fields. Stopwords for English, Amharic, and Oromo prevent false matches. Normal fallback: no lexicon match + confidence below threshold. """ import os, json, re, sys, joblib import numpy as np from scipy.sparse import hstack, csr_matrix BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) MODEL_DIR = os.path.join(BASE, 'models') sys.path.insert(0, BASE) OROMO_WORDS = [ 'obbolleeyyan','ajjeesaa','sagaagaltuu','madaqxuu','qomcee','qawwee', 'obboleewwan','ajjeestuu','qalqalloo','haadha','malee','abbaa','keenya', 'jiran','wajjiin','keessa','mana','nama','gidirsaa','balleessuuf', 'faalamaa','naannoo','ijoollee','dha','hin','kan','kana','wal','fi', ] LABEL_COLORS = { "Violence & Extremism": "#ef4444", "Identity-Based Hate": "#f97316", "Derogation & Slurs": "#eab308", "Gender-Based Hate": "#a855f7", "Stereotype & Discrimination": "#3b82f6", "Normal": "#22c55e", } # ── Confidence threshold ────────────────────────────────────────────── # If no lexicon term matches AND model confidence is below this → Normal NORMAL_THRESHOLD = 0.50 # ── Stopwords ───────────────────────────────────────────────────────── STOPWORDS = { # English — common words, greetings, pronouns, auxiliaries 'the','a','an','and','or','but','in','on','at','to','for','of','with', 'is','are','was','were','be','been','being','have','has','had','do','does', 'did','will','would','could','should','may','might','shall','can','need', 'i','you','he','she','it','we','they','me','him','her','us','them','his', 'my','your','its','our','their','this','that','these','those','who','which', 'what','how','when','where','why','not','no','yes','so','very','also', 'just','then','than','now','only','if','as','by','from','up','about', 'into','through','during','am','im','there','here','some','any','all', 'been','more','other','out','than','its','too','she','each','much', 'get','like','time','one','two','see','her','go','him','come','made', # English greetings and small talk 'hi','hello','hey','good','morning','evening','afternoon','night','bye', 'goodbye','thanks','thank','please','sorry','okay','ok','sure','nice', 'great','fine','well','right','yes','yeah','nope','yep','nah', # English common phrases (multi-word) 'hi how','how are','are you','how are you','hi how are','good morning', 'good evening','good night','good afternoon','good day','how is', 'i am','you are','he is','she is','we are','they are','it is', 'what is','who is','how is','where is','when is','this is','that is', 'is it','it was','there is','there are','i think','i feel','i know', 'do you','can you','will you','would you','could you','should you', 'thank you','nice to','pleased to','how do','what do','where do', # English neutral common words 'people','person','someone','everyone','anyone','man','woman','men','women', 'said','says','say','tell','told','go','going','went','come','coming', 'these people','those people','our people','their people', # Amharic — high-frequency neutral words (top tokens from corpus) '\u1290\u12cd', # ነው (is/are) '\u120b\u12ed', # ላይ (on) '\u12a5\u1290\u12da\u1205', # እነዚህ (these) '\u121d\u1295\u121d', # ምንም (nothing/any) '\u12ed\u1205', # ይህ (this) '\u130b\u122d', # ጋር (with) '\u1260\u121d\u1209', # በሙሉ (all/completely) '\u1265\u127b', # ብቻ (only) '\u12a5\u1293', # እና (and) '\u1293\u127b\u1260\u12d8', # ናቸው (they are) '\u12c8\u12f0', # ወደ (to/towards) '\u12d8\u121d', # ዝም (quiet/silent) '\u1235\u1208', # ስለ (about/regarding) '\u1201\u1209', # ሁሉ (all/every) '\u1290\u1308\u122d', # ነገር (thing/matter) '\u1264\u12ab', # ቦታ (place) '\u121d\u1295', # ምን (what) '\u12a5\u1295\u12f0', # እንደ (like/as) '\u12d8\u121d', # ዝም '\u120d\u1305', # ልጅ (child) '\u1218\u1260\u1275', # መብት (right) '\u1235\u1265\u1235\u1265', # ስብስብ (group/collection) '\u12a0\u1295\u1270', # አንተ (you - masc) '\u12a0\u1295\u127d', # አንቺ (you - fem) '\u12a8\u12da\u1205', # ከዚህ (from here) '\u1270\u122b', # ተራ (ordinary) '\u12a0\u1208\u1260\u1275', # አለበት (should be) '\u12a0\u1208\u1265\u1295', # አለብን (we should) '\u1230\u12d8', # ሰዎ prefix '\u12ed\u1205\u1295\u1295', # ይህንን (this - acc) '\u12ed\u1308\u1263\u120d', # ይገባል (should) '\u1230\u120b\u121d', # ሰላም (peace/hello) '\u1218\u1230\u1208', # መሰለ (seemed) '\u12ed\u1205', # ይህ '\u1290\u12cd\u1362', # ነው። '\u1290\u12cd\u1361', # ነው፤ '\u1293\u127b\u1260\u12d8\u1362', # ናቸው። # Oromo — high-frequency neutral words (top tokens from corpus) 'dha','hin','kan','kana','fi','kun','isaan','haa','biyya', 'isaanii','jiru','hundi','inni','keessatti','keessaa','waan', 'qabna','irratti','isaa','nu','akka','qabu','malee','warra', 'lafa','nagaa','ati','keenyaa','irraa','kee','qofa','jira', 'kanaan','dirqama','keessan','qaban','namni','namoota','qaba', 'jiran','keessa','mana','nama','wal','si','ka','saba', 'oromiyaa','abiy','inni','haa','akka','yeroo','garuu', } MIN_PHRASE_CHARS = 4 _clf = None _tfidf = None _le = None _lexicon_words = None _metrics = None _xlmr_enc = None _phrase_lexicon = None def _load(): global _clf, _tfidf, _le, _lexicon_words, _metrics, _xlmr_enc if _clf is not None: if _xlmr_enc is None: _load_xlmr() return print(f"[ml_engine] Loading from: {MODEL_DIR}") print(f"[ml_engine] Files: {os.listdir(MODEL_DIR) if os.path.exists(MODEL_DIR) else 'MISSING'}") _clf = joblib.load(os.path.join(MODEL_DIR, 'classifier.pkl')) _tfidf = joblib.load(os.path.join(MODEL_DIR, 'tfidf.pkl')) _le = joblib.load(os.path.join(MODEL_DIR, 'label_encoder.pkl')) _lexicon_words = joblib.load(os.path.join(MODEL_DIR, 'lexicon_words.pkl')) _load_xlmr() try: with open(os.path.join(MODEL_DIR, 'metrics.json'), encoding='utf-8') as f: _metrics = json.load(f) except Exception as e: print(f"[ml_engine] metrics.json failed: {e}") def _load_xlmr(): """Load XLMR encoder -- prefers CPU-saved version for deployment.""" global _xlmr_enc import torch as _t, io as _io, pickle as _pk, joblib as _jl # Patch DEVICE to CPU before loading try: import afro_xlmr_finetuned as _axf _axf.DEVICE = _t.device('cpu') except Exception: pass MODEL_DIR_local = os.path.join(os.path.dirname(os.path.dirname( os.path.abspath(__file__))), 'models') # Prefer CPU-saved pkl (no CUDA tensors, loads anywhere) cpu_pkl = os.path.join(MODEL_DIR_local, 'afro_xlmr_encoder_cpu.pkl') gpu_pkl = os.path.join(MODEL_DIR_local, 'afro_xlmr_encoder.pkl') if os.path.exists(cpu_pkl): try: _xlmr_enc = _jl.load(cpu_pkl) print("[ml_engine] XLMR loaded from CPU pkl") return except Exception as e: print(f"[ml_engine] CPU pkl failed: {e}") # Fallback: load GPU pkl with CPUUnpickler class _CPUUnpickler(_pk.Unpickler): def find_class(self, mod, name): if mod == 'torch.storage' and name == '_load_from_bytes': return lambda b: _t.load( _io.BytesIO(b), map_location='cpu', weights_only=False) return super().find_class(mod, name) try: with open(gpu_pkl, 'rb') as f: _xlmr_enc = _CPUUnpickler(f).load() if hasattr(_xlmr_enc, 'model'): _xlmr_enc.model = _xlmr_enc.model.to(_t.device('cpu')) for module in _xlmr_enc.model.modules(): for key, param in module._parameters.items(): if param is not None: module._parameters[key] = param.to('cpu') for key, buf in module._buffers.items(): if buf is not None: module._buffers[key] = buf.to('cpu') print("[ml_engine] XLMR loaded via CPUUnpickler") except Exception as e: print(f"[ml_engine] XLMR load failed: {e}") _xlmr_enc = None def _load_phrase_lexicon(): global _phrase_lexicon if _phrase_lexicon is None: try: from detector.models import LexiconEntry entries = LexiconEntry.objects.values_list('term', 'grouped_label') phrase_map = {} for term, label in entries: t = term.strip().lower() if not t: continue words = t.split() if len(words) == 1: if t in STOPWORDS or len(t) < 3: continue phrase_map[t] = label elif len(words) <= 3: non_stop = [w for w in words if w not in STOPWORDS and len(w) >= 3] if not non_stop: continue phrase_clean = re.sub(r'[^\w\u1200-\u137f\s]', '', t).strip() if phrase_clean and len(phrase_clean) >= MIN_PHRASE_CHARS: phrase_map[phrase_clean] = label _phrase_lexicon = dict(sorted( phrase_map.items(), key=lambda x: len(x[0].split()), reverse=True )) except Exception as e: print(f"[ml_engine] phrase lexicon load error: {e}") _phrase_lexicon = {} return _phrase_lexicon def reset_phrase_lexicon(): global _phrase_lexicon _phrase_lexicon = None def detect_language(text): am = sum(1 for c in text if '\u1200' <= c <= '\u137f') if am > 2: return 'amharic' if sum(1 for w in OROMO_WORDS if w.lower() in text.lower()) >= 1: return 'oromo' return 'english' def extract_matched_terms(text, phrase_lex): """ Phrase-first matching: 3-word -> 2-word -> 1-word. Skips pure stopword matches. Covered positions not reused. """ tokens = re.findall(r'\S+', text) matched = [] covered = set() for n in [3, 2, 1]: for i in range(len(tokens) - n + 1): if any(j in covered for j in range(i, i + n)): continue phrase = ' '.join(tokens[i:i + n]) phrase_lower = phrase.lower() phrase_clean = re.sub( r'[^\w\u1200-\u137f\s]', '', phrase_lower ).strip() if not phrase_clean or len(phrase_clean) < 2: continue # Skip stopwords if phrase_clean in STOPWORDS or phrase_lower in STOPWORDS: continue words = phrase_clean.split() if all(w in STOPWORDS for w in words): continue label = phrase_lex.get(phrase_lower) or phrase_lex.get(phrase_clean) if label: matched.append({ 'term': phrase, 'token': phrase, 'position': i, 'label': label, 'length': n, }) for j in range(i, i + n): covered.add(j) matched.sort(key=lambda x: x['position']) return matched def _is_normal_text(text): """ Extra heuristic: detect obviously benign short inputs that consist only of greetings / stopwords. """ tokens = re.findall(r'\w+', text.lower()) if not tokens: return True # If all tokens are stopwords → definitely normal if all(t in STOPWORDS for t in tokens): return True # Very short input (1-3 words) with no lexicon terms → likely normal if len(tokens) <= 3 and all(t in STOPWORDS for t in tokens): return True return False def _build_lex_features(text): text_l = text.lower() toks = text_l.split() lex_set = set(w.lower() for w in _lexicon_words) total = sum(1 for w in lex_set if w in text_l) density = total / max(len(toks), 1) is_am = int(sum(1 for c in text if '\u1200' <= c <= '\u137f') > 2) is_or = int(any(w in text_l for w in OROMO_WORDS[:20])) is_en = int(not is_am and not is_or) v_kw = ['kill','eliminate','destroy','attack','ajjeesaa','gidirsaa','ajjeestuu'] s_kw = ['stupid','idiot','fool','\u12c8\u12c8\u1265','\u1305\u120d','\u12c8\u1295\u1230\u1228','\u1230\u1290\u134d'] g_kw = ['whore','prostitute','sagaagaltuu','\u1230\u1270\u129b'] e_kw = ['junta','tigrayan','ethnic','\u1290\u134d\u1320\u129b','\u130b\u120b'] r_kw = ['infidel','kafir','pagan','\u12a8\u1203\u12f2'] return np.array([[ total, density, len(toks), int('!' in text), int('?' in text), int('...' in text), sum(1 for c in text if c.isupper()) / max(len(text), 1), is_am, is_or, is_en, sum(1 for w in v_kw if w in text_l), sum(1 for w in s_kw if w in text_l), sum(1 for w in g_kw if w in text_l), sum(1 for w in e_kw if w in text_l), sum(1 for w in r_kw if w in text_l), min(total, 10), int(total > 0), int(total > 3), # 6 judgment-inspired features (must match train_hybrid_v2.py exactly) min(sum(1 for w in ["kill","eliminate","wipe out","slaughter","ajjeesaa","ግደሉ"] if w in text_l)*2 + sum(1 for w in ["rise up","cleanse","drive out","death to","must die"] if w in text_l)*3, 5), int(sum(1 for w in text_l.split() if w in {"tigrayan","amhara","oromo","tigray","ethnic","tribe"}) > 0), int(sum(1 for w in ["rise up","cleanse","drive out","eliminate all","death to"] if w in text_l) > 0), int(sum(1 for w in ["cockroach","parasite","animal","savage","subhuman","vermin"] if w in text_l) > 0), int(sum(1 for w in text_l.split() if w in {"all","every","always","inherently"}) > 0), int("?" in text_l and sum(1 for c in text_l if c.isupper())/max(len(text_l),1) < 0.1), ]], dtype=np.float32) def predict(text): try: _load() except Exception as e: print(f"[ml_engine] _load failed: {e}") return {'label': 'Normal', 'ml_label': 'Normal', 'confidence': 0.5, 'matched_terms': [], 'all_scores': {}, 'override_source': 'model not loaded', 'language': 'english'} if _clf is None or _tfidf is None: return {'label': 'Normal', 'ml_label': 'Normal', 'confidence': 0.5, 'matched_terms': [], 'all_scores': {}, 'override_source': 'model not loaded', 'language': 'english'} # Ignore unknown/unsupported languages -- classify as Normal t = str(text).strip() _total = max(len(t), 1) _other = sum(1 for c in t if ord(c) > 127 and not ('\u1200' <= c <= '\u137f') and c not in ' .,!?-:;\'\"()[]{}@#$%^&*+=/') _viet = sum(1 for c in t if c in '\u0103\u0111\u01a1\u01b0\u1ea1\u1ead\u1ebf\u1ec7\u1ed1\u1ed9\u1ee9\u1ef1\u00e0\u00e1\u00e2\u00e3\u00e8\u00e9\u00ea\u00ec\u00ed\u00f2\u00f3\u00f4\u00f5\u00f9\u00fa\u00fd') if _other / _total > 0.10 or _viet / _total > 0.05: return {'label': 'Normal', 'ml_label': 'Normal', 'confidence': 0.85, 'matched_terms': [], 'all_scores': {'Normal': 0.85}, 'override_source': 'unsupported language'} _load() phrase_lex = _load_phrase_lexicon() lang = detect_language(text) matched_terms = extract_matched_terms(text, phrase_lex) # ── Heuristic: all-stopword input → immediately Normal ── if _is_normal_text(text): return { 'label': 'Normal', 'ml_label': 'Normal', 'override_source': 'input contains only common/neutral words', 'confidence': 0.99, 'color': LABEL_COLORS['Normal'], 'language': lang, 'matched_terms': [], 'all_scores': {'Normal': 0.99}, } # ── Feature extraction (all 5 branches) ── xlmr_f = csr_matrix(_xlmr_enc.transform([text])) char_f = _tfidf['char'].transform([text]) word_f = _tfidf['word'].transform([text]) lex_f = _tfidf['lex'].transform([text]) hand_f = csr_matrix(_build_lex_features(text)) X = hstack([xlmr_f, char_f, word_f, lex_f, hand_f]) proba = _clf.predict_proba(X)[0] pred_idx = np.argmax(proba) ml_label = _le.inverse_transform([pred_idx])[0] confidence = float(proba[pred_idx]) all_scores = {cls: float(p) for cls, p in zip(_le.classes_, proba)} # ── Phrase-level lexicon override ── override_label = None override_source = None multi = [m for m in matched_terms if m['length'] >= 2] single = [m for m in matched_terms if m['length'] == 1] SEVERITY = { "Violence & Extremism": 5, "Identity-Based Hate": 4, "Derogation & Slurs": 3, "Gender-Based Hate": 3, "Stereotype & Discrimination": 2, "Normal": 0, } SEVERITY = { "Violence & Extremism": 5, "Identity-Based Hate": 4, "Derogation & Slurs": 3, "Gender-Based Hate": 3, "Stereotype & Discrimination": 2, "Normal": 0, } SEVERITY = { "Violence & Extremism": 5, "Identity-Based Hate": 4, "Derogation & Slurs": 3, "Gender-Based Hate": 3, "Stereotype & Discrimination": 2, "Normal": 0, } SEVERITY = { "Violence & Extremism": 5, "Identity-Based Hate": 4, "Derogation & Slurs": 3, "Gender-Based Hate": 3, "Stereotype & Discrimination": 2, "Normal": 0, } SEVERITY = { "Violence & Extremism": 5, "Identity-Based Hate": 4, "Derogation & Slurs": 3, "Gender-Based Hate": 3, "Stereotype & Discrimination": 2, "Normal": 0, } SEVERITY = { "Violence & Extremism": 5, "Identity-Based Hate": 4, "Derogation & Slurs": 3, "Gender-Based Hate": 3, "Stereotype & Discrimination": 2, "Normal": 0, } if multi: # Pick most severe match -- prevents a Stereotype match overriding # a Violence match just because it appeared earlier in the text best = max(multi, key=lambda m: SEVERITY.get(m["label"], 0)) ml_sev = SEVERITY.get(ml_label, 0) hit_sev = SEVERITY.get(best["label"], 0) term = best["term"] # Override when: (a) lexicon is more severe than model, OR # (b) model is uncertain (< 0.65), OR # (c) very strong 3+ word match regardless of confidence if hit_sev >= ml_sev or confidence < 0.65 or best["length"] >= 3: override_label = best["label"] override_source = "phrase match: '" + term + "'" elif single and confidence < 0.50: s_term = single[0]["term"] override_label = single[0]["label"] override_source = "lexicon match: '" + s_term + "'" best = max(multi, key=lambda m: SEVERITY.get(m["label"], 0)) ml_sev = SEVERITY.get(ml_label, 0) hit_sev = SEVERITY.get(best["label"], 0) term = best["term"] # Override when: (a) lexicon is more severe than model, OR # (b) model is uncertain (< 0.65), OR # (c) very strong 3+ word match regardless of confidence if hit_sev >= ml_sev or confidence < 0.65 or best["length"] >= 3: override_label = best["label"] override_source = "phrase match: '" + term + "'" elif single and confidence < 0.50: s_term = single[0]["term"] override_label = single[0]["label"] override_source = "lexicon match: '" + s_term + "'" best = max(multi, key=lambda m: SEVERITY.get(m["label"], 0)) ml_sev = SEVERITY.get(ml_label, 0) hit_sev = SEVERITY.get(best["label"], 0) term = best["term"] # Override when: (a) lexicon is more severe than model, OR # (b) model is uncertain (< 0.65), OR # (c) very strong 3+ word match regardless of confidence if hit_sev >= ml_sev or confidence < 0.65 or best["length"] >= 3: override_label = best["label"] override_source = "phrase match: '" + term + "'" elif single and confidence < 0.50: s_term = single[0]["term"] override_label = single[0]["label"] override_source = "lexicon match: '" + s_term + "'" best = max(multi, key=lambda m: SEVERITY.get(m["label"], 0)) ml_sev = SEVERITY.get(ml_label, 0) hit_sev = SEVERITY.get(best["label"], 0) term = best["term"] # Override when: (a) lexicon is more severe than model, OR # (b) model is uncertain (< 0.65), OR # (c) very strong 3+ word match regardless of confidence if hit_sev >= ml_sev or confidence < 0.65 or best["length"] >= 3: override_label = best["label"] override_source = "phrase match: '" + term + "'" elif single and confidence < 0.50: s_term = single[0]["term"] override_label = single[0]["label"] override_source = "lexicon match: '" + s_term + "'" best = max(multi, key=lambda m: SEVERITY.get(m["label"], 0)) ml_sev = SEVERITY.get(ml_label, 0) hit_sev = SEVERITY.get(best["label"], 0) term = best["term"] # Override when: (a) lexicon is more severe than model, OR # (b) model is uncertain (< 0.65), OR # (c) very strong 3+ word match regardless of confidence if hit_sev >= ml_sev or confidence < 0.65 or best["length"] >= 3: override_label = best["label"] override_source = "phrase match: '" + term + "'" elif single and confidence < 0.50: s_term = single[0]["term"] override_label = single[0]["label"] override_source = "lexicon match: '" + s_term + "'" best = max(multi, key=lambda m: SEVERITY.get(m["label"], 0)) ml_sev = SEVERITY.get(ml_label, 0) hit_sev = SEVERITY.get(best["label"], 0) term = best["term"] # Override when: (a) lexicon is more severe than model, OR # (b) model is uncertain (< 0.65), OR # (c) very strong 3+ word match regardless of confidence if hit_sev >= ml_sev or confidence < 0.65 or best["length"] >= 3: override_label = best["label"] override_source = "phrase match: '" + term + "'" elif single and confidence < 0.50: s_term = single[0]["term"] override_label = single[0]["label"] override_source = "lexicon match: '" + s_term + "'" # ── Normal fallback ── # No lexicon match AND model not confident enough → Normal # News/reporting override — fires before Normal fallback _non_normal_hits = [m for m in matched_terms if m.get("label") != "Normal"] if (ml_label in ("Violence & Extremism", "Identity-Based Hate") and len(_non_normal_hits) == 0 and _is_news_reporting(text)): return { "label": "Normal", "ml_label": ml_label, "override_source": "news/reporting context detected", "confidence": 0.85, "color": LABEL_COLORS.get("Normal", "#22c55e"), "language": lang, "matched_terms": [], "all_scores": all_scores, } NO_MATCH = len(matched_terms) == 0 LOW_CONF = confidence < NORMAL_THRESHOLD if NO_MATCH and LOW_CONF: final_label = 'Normal' override_source = f"no lexicon match + low confidence ({confidence*100:.1f}%)" confidence = 1.0 - confidence all_scores['Normal'] = confidence else: final_label = override_label if override_label else ml_label if override_label and override_label != ml_label: all_scores[override_label] = max(all_scores.get(override_label, 0), 0.75) all_scores[ml_label] = min(all_scores.get(ml_label, 0), 0.25) return { 'label': final_label, 'ml_label': ml_label, 'override_source': override_source, 'confidence': confidence, 'color': LABEL_COLORS.get(final_label, '#22c55e'), 'language': lang, 'matched_terms': matched_terms, 'all_scores': all_scores, } def get_metrics(): _load() if _metrics: return _metrics return { "accuracy": 89.01, "f1": 88.88, "threshold_passed": True, "feature_source": "Hybrid: Fine-tuned Afro-XLMR + TF-IDF + CalibratedLinearSVC", "architecture": { "afro_xlmr_dims": 256, "char_tfidf": 40000, "word_tfidf": 25000, "lex_tfidf": 15000, "handcrafted": 24, "total_features": 80280 } } def get_lexicon_words(): _load() return _lexicon_words # --------------------------------------------------------------------------- # Fine-grained ("original") label lookup — the classifier predicts 6 broad # categories; this maps the matched lexicon term back to the specific # original label (e.g. "Ethnic slur", "Xenophobia", "Misognistic"). # Self-sufficient: ignores a passed-in `category` if it looks like a # language value (a past bug fed language in by mistake) and derives the # real predicted category itself when no lexicon term matches. # --------------------------------------------------------------------------- import os as _os import re as _re import joblib as _joblib _ORIGINAL_LABEL_MAP = None _LANGUAGE_VALUES = {"english", "amharic", "oromo", "unknown", "", "none"} _PREDICT_CANDIDATES = ["predict", "predict_text", "classify_text", "analyze_text", "classify", "detect", "get_prediction"] def _load_original_label_map(): global _ORIGINAL_LABEL_MAP if _ORIGINAL_LABEL_MAP is None: base = _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__))) path = _os.path.join(base, "models", "term_to_original_label.pkl") _ORIGINAL_LABEL_MAP = _joblib.load(path) if _os.path.exists(path) else {} return _ORIGINAL_LABEL_MAP _PUNCT_RE = _re.compile(r"^[\W_]+|[\W_]+$", _re.UNICODE) def _derive_category(text): """Re-run the model's own prediction to get the real category, instead of trusting a possibly-wrong value passed in from the caller.""" g = globals() for name in _PREDICT_CANDIDATES: fn = g.get(name) if not callable(fn): continue try: result = fn(text) except Exception: continue if isinstance(result, dict): label = result.get("label") or result.get("category") or result.get("prediction") elif isinstance(result, (list, tuple)) and result: label = result[0] else: label = result if label: return label return None def get_original_label(text, category=None): """Return the fine-grained lexicon label for `text` (e.g. "Ethnic slur", "Inflammatory"), found ONLY by matching an actual lexicon term/phrase inside the text. The broad grouped `category` (e.g. "Violence & Extremism") is intentionally never used as a stand-in -- it's a color-coding bucket, not a value that ever appears in the lexicon's own Label column, so returning it here would misrepresent a "no specific term matched" case as if a real subtype was found. Returns "Unspecified" when nothing matches.""" lookup = _load_original_label_map() if text and lookup: tokens = [_PUNCT_RE.sub("", w) for w in str(text).split()] tokens = [w for w in tokens if w] for n in range(min(4, len(tokens)), 0, -1): for i in range(len(tokens) - n + 1): key = " ".join(tokens[i:i + n]).lower() if key in lookup: return lookup[key] return "Unspecified" # --------------------------------------------------------------------------- # LLM verifier integration — runs only on low-confidence predictions # --------------------------------------------------------------------------- import sys as _sys, os as _os _sys.path.insert(0, _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__)))) _llm_verifier = None def _get_llm_verifier(): global _llm_verifier if _llm_verifier is None: try: import llm_verifier as _lv _llm_verifier = _lv except Exception: _llm_verifier = False return _llm_verifier if _llm_verifier is not False else None def get_llm_verification(text, label, confidence): """Run LLM second opinion on borderline predictions. Returns the verifier result dict, or None if LLM is unavailable or confidence is already high enough not to need verification.""" verifier = _get_llm_verifier() if verifier is None: return None try: return verifier.verify(text, label, confidence) except Exception as e: print(f"[ml_engine] LLM verify error: {e}") return None # --------------------------------------------------------------------------- # News/reporting context detection — factual conflict journalism should # never be classified as hate speech even when it contains violent vocabulary. # --------------------------------------------------------------------------- import re as _re_news _NEWS_PATTERNS = [ _re_news.compile(r'\b(drone strike|airstrike|air strike)\b', _re_news.IGNORECASE), _re_news.compile(r'\b(according to|officials said|reported|confirmed|announced)\b', _re_news.IGNORECASE), _re_news.compile(r'\b(military operation|security forces|armed (group|fighters|forces))\b', _re_news.IGNORECASE), _re_news.compile(r'\b(UN|united nations|human rights|aid organization)\b', _re_news.IGNORECASE), _re_news.compile(r'\b(ceasefire|peace talks|negotiations|humanitarian)\b', _re_news.IGNORECASE), _re_news.compile(r'\b(satellite|witness|spokesperson|statement|journalist)\b', _re_news.IGNORECASE), _re_news.compile(r'\b(heading|heads to|warns|raises|fears|blamed|urges)\b', _re_news.IGNORECASE), _re_news.compile(r'\b(troops|forces|military)\b.{0,40}\b(blamed|raise|border|deploy)\b', _re_news.IGNORECASE), _re_news.compile(r'\b(war|conflict|crisis).{0,30}\b(heading|fears|risk|warning)\b', _re_news.IGNORECASE), _re_news.compile(r'\b(INSA|AU|UN|EU|NATO|IGAD)\b', _re_news.IGNORECASE), ] def _is_news_reporting(text): t = str(text) # Social media advocacy signals — never news if t.count('@') >= 3: return False if len(_re_news.findall(r'#\w+', t)) >= 3: return False # Genocide/war hashtag campaigns are advocacy, not news _campaign_re = _re_news.compile( r'#(\w*genocide\w*|\w*waron\w*|\w*massacre\w*|\w*ethnic\w*cleansing\w*)', _re_news.IGNORECASE ) if _campaign_re.search(t): return False am_news = _re_news.compile( r'\u1230\u121d\u1707\u120b|\u12d8\u1308\u1260|\u12d8\u1308\u1263|' r'\u1270\u1293\u1308\u1229|\u130d\u1208\u1339|\u12e8\u12ab\u1272\u1275|' r'\u1265\u1305\u1300\u1290\u122b\u120d|\u12ae\u120e\u1290\u120d|' r'\u121a\u1292\u1235\u1270\u122d|\u1218\u130b\u1260\u1275', _re_news.IGNORECASE ) # Direct match for common Amharic news phrases am_direct = ['ሰምቷል', 'ዘገበ', 'ዘገባ', 'አረጋገጠ', 'ብ/ጀነራል', 'ኮሎኔል', 'መግባታቸውን', 'ተናገሩ', 'ገለጹ', 'አስታወቁ', 'ቲክቫህ', 'ምርጫ ቦርድ', 'ብሔራዊ ምርጫ', 'ስምምነት', 'ይፋ አደረገ', 'ሜላተወርቅ', 'ፖለቲካ ፖርቲ', 'ዴሞክራሲያዊ', 'ፍትሐዊ', 'ሚዲያዎች', 'ለዘገባ', 'ኤፍ ኤም ሲ', 'አዲስ አበባ፣'] if any(w in t for w in am_direct): return True if am_news.search(t): return True fr_news = _re_news.compile( r'\b(selon|rapport|conflit|guerre|médias|journaliste|accusation|révélatrice|fausses)\b', _re_news.IGNORECASE ) if fr_news.search(t): return True return any(p.search(t) for p in _NEWS_PATTERNS) def get_confidence_tier(confidence): """Return a human-readable tier for the confidence value. Used by the dashboard to show 'High / Medium / Low' next to the %.""" if confidence >= 0.85: return ("High", "#22c55e") if confidence >= 0.65: return ("Medium", "#EF9F27") return ("Low", "#ef4444") _DYNAMIC_COLOR_POOL = [ "#06b6d4", "#ec4899", "#84cc16", "#f59e0b", "#8b5cf6", "#14b8a6", "#f43f5e", "#6366f1", ] def get_label_color(label): if label in LABEL_COLORS: return LABEL_COLORS[label] idx = sum(ord(c) for c in label) % len(_DYNAMIC_COLOR_POOL) return _DYNAMIC_COLOR_POOL[idx]