moderator.py

# moderator.py
import re
import unicodedata
from typing import Dict, List, Tuple
from datetime import datetime
import torch
from transformers import pipeline

class SimpleModerator:
    def __init__(self):
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.toxic_model = None
        self.hate_model = None
        
        # Enhanced leet speak mapping
        self.leet_map = {
            "@": "a", "4": "a", "3": "e", "1": "i", "!": "i",
            "0": "o", "$": "s", "5": "s", "7": "t", "+": "t",
            "2": "to", "&": "and", "8": "ate", "6": "g"
        }
        
        # Character normalization
        self.char_normalize = {
            'à': 'a', 'á': 'a', 'â': 'a', 'ã': 'a', 'ä': 'a',
            'è': 'e', 'é': 'e', 'ê': 'e', 'ë': 'e',
            'ì': 'i', 'í': 'i', 'î': 'i', 'ï': 'i',
            'ò': 'o', 'ó': 'o', 'ô': 'o', 'õ': 'o', 'ö': 'o',
            'ù': 'u', 'ú': 'u', 'û': 'u', 'ü': 'u',
            'ý': 'y', 'ÿ': 'y',
            'ç': 'c', 'ñ': 'n'
        }
        
        # DELETE patterns
        self.delete_patterns = [
            "kill yourself", "kill urself", "kys", "commit suicide", "end your life",
            "take your life", "suicide", "kill myself", "kill himself", "kill herself",
            "i will kill you", "ill kill you", "gonna kill you", "murder you",
            "i will murder", "going to kill", "gonna murder", "i will hurt you",
            "should die", "deserve to die", "hope you die", "wish you were dead",
            "should be dead", "drop dead", "go die", "just die",
        ]
        
        # Strong slurs - immediate DELETE
        self.slur_patterns = [
            "nigger", "nigga", "faggot", "fag", "retard", "spic", "kike",
            "chink", "cunt", "whore", "slut"
        ]
        
        # FLAG patterns
        self.flag_patterns = [
            "stupid", "idiot", "dumb", "moron", "loser", "ugly", "fat",
            "worthless", "pathetic", "useless", "failure", "imbecile",
            "disgusting", "terrible", "awful", "despise",
            "shut up", "shut the fuck", "piss off"
        ]
        
        # Words that indicate hate/toxicity
        self.toxic_words = [
            "hate", "hating", "hated", "hater",
            "fuck", "fucking", "fucked", "bitch", "asshole", "ass",
            "damn", "hell", "crap", "shit", "piss"
        ]
        
        print(f"🖥️ Device: {self.device}")
    
    def load_models(self):
        """Load the ML models"""
        print("📥 Loading models...")
        try:
            print("Loading toxic-bert model...")
            self.toxic_model = pipeline(
                "text-classification",
                model="unitary/toxic-bert",
                device=0 if self.device == "cuda" else -1,
                truncation=True,
                max_length=512
            )
            print("✅ Toxic-BERT loaded")
            
            print("Loading dehatebert model...")
            self.hate_model = pipeline(
                "text-classification",
                model="Hate-speech-CNERG/dehatebert-mono-english",
                device=0 if self.device == "cuda" else -1,
                truncation=True,
                max_length=512
            )
            print("✅ DeHateBERT loaded")
            
            print("🎉 Models ready!")
            return True
        except Exception as e:
            print(f"❌ Error loading models: {e}")
            import traceback
            traceback.print_exc()
            return False
    
    def normalize_text(self, text: str) -> str:
        """Enhanced normalization for better pattern matching"""
        text = text.lower()
        
        for char, normal in self.char_normalize.items():
            text = text.replace(char, normal)
        
        for leet, normal in self.leet_map.items():
            text = text.replace(leet, normal)
        
        text = unicodedata.normalize("NFKD", text)
        text = re.sub(r'\s+', ' ', text)
        text = re.sub(r'[^\w\s]', ' ', text)
        text = re.sub(r'\s+', ' ', text)
        text = re.sub(r'(.)\1{2,}', r'\1\1', text)
        
        return text.strip()
    
    def check_patterns(self, text: str) -> Tuple[str, List[str], float]:
        """Check text against patterns and return decision with confidence"""
        normalized = self.normalize_text(text)
        words = normalized.split()
        matched_patterns = []
        max_confidence = 0.0
        
        # Check DELETE patterns (highest priority)
        for pattern in self.delete_patterns:
            if pattern in normalized:
                matched_patterns.append(f"DELETE:{pattern}")
                max_confidence = 1.0
        
        # Check slurs (immediate DELETE)
        for slur in self.slur_patterns:
            if slur in words or slur in normalized:
                matched_patterns.append(f"DELETE:slur:{slur}")
                max_confidence = 1.0
        
        if matched_patterns and max_confidence == 1.0:
            return "DELETE", matched_patterns, 1.0
        
        # Check toxic words (FLAG with high confidence)
        toxic_count = 0
        for word in words:
            if word in self.toxic_words:
                toxic_count += 1
                matched_patterns.append(f"FLAG:toxic_word:{word}")
        
        # Check FLAG patterns
        for pattern in self.flag_patterns:
            if pattern in normalized:
                matched_patterns.append(f"FLAG:{pattern}")
        
        if matched_patterns:
            confidence = min(0.7 + (toxic_count * 0.1), 0.95)
            return "FLAG", matched_patterns, confidence
        
        return "ALLOW", [], 0.0
    
    def get_model_scores(self, text: str) -> Dict:
        """Get model predictions with proper error handling"""
        scores = {
            "toxic_score": 0.0,
            "toxic_label": "unknown",
            "hate_score": 0.0,
            "hate_label": "unknown"
        }
        
        # Try toxic model
        if self.toxic_model is not None:
            try:
                toxic_result = self.toxic_model(text[:512])[0]
                scores["toxic_score"] = float(toxic_result["score"])
                scores["toxic_label"] = toxic_result["label"]
                print(f"Toxic score: {scores['toxic_score']:.3f} ({scores['toxic_label']})")
            except Exception as e:
                print(f"Toxic model error: {e}")
        else:
            print("Toxic model not loaded")
        
        # Try hate model
        if self.hate_model is not None:
            try:
                hate_result = self.hate_model(text[:512])[0]
                hate_score = float(hate_result["score"])
                
                # DeHateBERT returns NON_HATE/HATE labels
                if hate_result["label"] == "NON_HATE":
                    scores["hate_score"] = 1.0 - hate_score
                    scores["hate_label"] = "non_hate"
                else:
                    scores["hate_score"] = hate_score
                    scores["hate_label"] = "hate"
                
                print(f"Hate score: {scores['hate_score']:.3f} ({scores['hate_label']})")
            except Exception as e:
                print(f"Hate model error: {e}")
        else:
            print("Hate model not loaded")
        
        return scores
    
    def moderate(self, text: str) -> Dict:
        """Main moderation function - combines pattern matching and ML models"""
        
        # Pattern matching (primary, fast)
        pattern_decision, matched, pattern_confidence = self.check_patterns(text)
        
        # Model scores (secondary, more nuanced)
        scores = self.get_model_scores(text)
        
        toxic_score = scores["toxic_score"]
        hate_score = scores["hate_score"]
        
        # Combine for final decision
        action = "allow"
        reason = "No issues detected"
        final_confidence = 0.0
        
        # Pattern overrides (highest priority)
        if pattern_decision == "DELETE":
            action = "delete"
            reason = f"Pattern match: {matched[0].replace('DELETE:', '')}"
            final_confidence = 1.0
        elif pattern_decision == "FLAG":
            action = "flag"
            reason = f"Pattern match: {matched[0].replace('FLAG:', '')}"
            final_confidence = pattern_confidence
        
        # Model-based decisions (if no pattern match or pattern is weak)
        elif toxic_score > 0.90:
            action = "delete"
            reason = f"Extreme toxicity detected: {toxic_score:.2f}"
            final_confidence = toxic_score
        elif hate_score > 0.85:
            action = "delete"
            reason = f"Extreme hate speech detected: {hate_score:.2f}"
            final_confidence = hate_score
        elif toxic_score > 0.70:
            action = "flag"
            reason = f"High toxicity: {toxic_score:.2f}"
            final_confidence = toxic_score
        elif hate_score > 0.50:
            action = "flag"
            reason = f"Hate speech indicators: {hate_score:.2f}"
            final_confidence = hate_score
        
        # If both pattern and model agree on FLAG, escalate to DELETE
        if pattern_decision == "FLAG" and (toxic_score > 0.95 or hate_score > 0.90):
            action = "delete"
            reason = f"Pattern + Model agreement: {reason}"
            final_confidence = max(pattern_confidence, toxic_score, hate_score)
        
        normalized_text = self.normalize_text(text)
        
        return {
            "action": action,
            "reason": reason,
            "toxic_score": toxic_score,
            "hate_score": hate_score,
            "pattern_matches": matched,
            "pattern_confidence": pattern_confidence,
            "model_confidence": max(toxic_score, hate_score),
            "final_confidence": final_confidence,
            "normalized_text": normalized_text,
            "timestamp": datetime.now().isoformat()
        }

# Global instance
_moderator_instance = None

def get_moderator():
    """Get or create moderator instance"""
    global _moderator_instance
    if _moderator_instance is None:
        print("🔄 Creating new moderator instance...")
        _moderator_instance = SimpleModerator()
        success = _moderator_instance.load_models()
        if not success:
            print("⚠️ Warning: Models failed to load, using pattern matching only")
    return _moderator_instance