Create content_moderation.py

content_moderation.py

@@ -0,0 +1,494 @@
+"""
+Advanced Content Moderation System for Helion-V2
+Provides production-grade content filtering and safety checks.
+"""
+
+import re
+import json
+from typing import List, Dict, Tuple, Optional, Set
+from dataclasses import dataclass, asdict
+from datetime import datetime
+import logging
+
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ModerationResult:
+    """Detailed moderation result."""
+    timestamp: str
+    is_approved: bool
+    risk_level: str  # low, medium, high, critical
+    violations: List[str]
+    confidence_scores: Dict[str, float]
+    recommended_action: str
+    sanitized_content: Optional[str] = None
+    metadata: Optional[Dict] = None
+
+
+class ContentFilter:
+    """Multi-layer content filtering system."""
+    
+    def __init__(self, config_path: Optional[str] = None):
+        """
+        Initialize content filter with optional custom configuration.
+        
+        Args:
+            config_path: Path to custom filter configuration JSON
+        """
+        self.config = self._load_config(config_path)
+        self._initialize_filters()
+    
+    def _load_config(self, config_path: Optional[str]) -> Dict:
+        """Load filter configuration."""
+        default_config = {
+            "enable_profanity_filter": True,
+            "enable_toxicity_detection": True,
+            "enable_bias_detection": True,
+            "enable_pii_detection": True,
+            "enable_spam_detection": True,
+            "strictness_level": "medium",  # low, medium, high
+            "blocked_domains": ["example-spam.com"],
+            "allowed_code_patterns": True,
+            "max_repetition_ratio": 0.3
+        }
+        
+        if config_path:
+            try:
+                with open(config_path, 'r') as f:
+                    custom_config = json.load(f)
+                    default_config.update(custom_config)
+            except Exception as e:
+                logger.warning(f"Could not load config from {config_path}: {e}")
+        
+        return default_config
+    
+    def _initialize_filters(self):
+        """Initialize all filter components."""
+        
+        # Profanity and offensive language
+        self.profanity_list = self._load_profanity_list()
+        
+        # Toxic phrases
+        self.toxic_phrases = [
+            "you should kill yourself",
+            "i hope you die",
+            "you deserve to suffer",
+            "stupid idiot moron",
+            "worthless piece of",
+        ]
+        
+        # Bias indicators
+        self.bias_indicators = {
+            "gender": ["all women are", "all men are", "females are", "males are"],
+            "race": ["all [race] are", "typical [race]", "[race] people always"],
+            "religion": ["all [religion] are", "[religion] believers are"],
+            "age": ["all old people", "millennials are all", "boomers are"],
+        }
+        
+        # Spam patterns
+        self.spam_patterns = [
+            r'(?i)(buy now|click here|limited time|act now).{0,50}(http|www)',
+            r'(?i)(viagra|cialis|lottery|prince|inheritance)',
+            r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+',
+        ]
+        
+        # Dangerous instruction patterns
+        self.dangerous_instructions = [
+            r'(?i)how\s+to\s+(make|build|create|construct)\s+(bomb|explosive|poison|weapon)',
+            r'(?i)instructions?\s+(for|to)\s+(kill|murder|harm|torture)',
+            r'(?i)(recipe|guide|tutorial)\s+for\s+(meth|cocaine|heroin)',
+            r'(?i)how\s+to\s+(hack|crack|break\s+into|bypass)',
+        ]
+        
+        # Medical misinformation
+        self.medical_misinfo = [
+            r'(?i)(cancer|covid|hiv).+(cure|treat|prevent).+(bleach|hydrogen\s+peroxide|vitamin\s+c)',
+            r'(?i)vaccines?\s+(cause|lead\s+to|result\s+in)\s+(autism|death|infertility)',
+            r'(?i)essential\s+oils?\s+(cure|treat)\s+(cancer|diabetes|heart\s+disease)',
+        ]
+    
+    def _load_profanity_list(self) -> Set[str]:
+        """Load profanity word list."""
+        # Basic profanity list (expand as needed)
+        return {
+            'fuck', 'shit', 'bitch', 'asshole', 'bastard', 'damn',
+            'cunt', 'piss', 'cock', 'dick', 'pussy', 'slut', 'whore'
+        }
+    
+    def check_profanity(self, text: str) -> Tuple[bool, List[str]]:
+        """
+        Check for profanity in text.
+        
+        Args:
+            text: Text to check
+            
+        Returns:
+            Tuple of (has_profanity, list of found words)
+        """
+        if not self.config["enable_profanity_filter"]:
+            return False, []
+        
+        text_lower = text.lower()
+        words = re.findall(r'\b\w+\b', text_lower)
+        found_profanity = [word for word in words if word in self.profanity_list]
+        
+        return len(found_profanity) > 0, found_profanity
+    
+    def check_toxicity(self, text: str) -> Tuple[bool, float, List[str]]:
+        """
+        Check for toxic content.
+        
+        Args:
+            text: Text to check
+            
+        Returns:
+            Tuple of (is_toxic, toxicity_score, matched_phrases)
+        """
+        if not self.config["enable_toxicity_detection"]:
+            return False, 0.0, []
+        
+        text_lower = text.lower()
+        matched_phrases = []
+        toxicity_score = 0.0
+        
+        for phrase in self.toxic_phrases:
+            if phrase in text_lower:
+                matched_phrases.append(phrase)
+                toxicity_score += 0.3
+        
+        # Check for aggressive language patterns
+        aggressive_patterns = [
+            r'\b(hate|despise|loathe)\s+you\b',
+            r'\byou\s+(are|re)\s+(stupid|dumb|idiot|moron)',
+            r'\bshut\s+up\b',
+            r'\bgo\s+to\s+hell\b',
+        ]
+        
+        for pattern in aggressive_patterns:
+            if re.search(pattern, text_lower):
+                toxicity_score += 0.2
+        
+        is_toxic = toxicity_score > 0.5
+        return is_toxic, min(toxicity_score, 1.0), matched_phrases
+    
+    def check_bias(self, text: str) -> Tuple[bool, Dict[str, List[str]]]:
+        """
+        Check for biased language.
+        
+        Args:
+            text: Text to check
+            
+        Returns:
+            Tuple of (has_bias, dictionary of bias types and matched phrases)
+        """
+        if not self.config["enable_bias_detection"]:
+            return False, {}
+        
+        text_lower = text.lower()
+        bias_found = {}
+        
+        for bias_type, indicators in self.bias_indicators.items():
+            matches = []
+            for indicator in indicators:
+                # Simple pattern matching (can be enhanced with ML)
+                if indicator in text_lower:
+                    matches.append(indicator)
+            
+            if matches:
+                bias_found[bias_type] = matches
+        
+        return len(bias_found) > 0, bias_found
+    
+    def check_pii(self, text: str) -> Tuple[bool, Dict[str, List[str]]]:
+        """
+        Check for personally identifiable information.
+        
+        Args:
+            text: Text to check
+            
+        Returns:
+            Tuple of (has_pii, dictionary of PII types found)
+        """
+        if not self.config["enable_pii_detection"]:
+            return False, {}
+        
+        pii_found = {}
+        
+        # Social Security Number
+        ssn_pattern = r'\b\d{3}-\d{2}-\d{4}\b'
+        ssns = re.findall(ssn_pattern, text)
+        if ssns:
+            pii_found['ssn'] = ssns
+        
+        # Credit Card
+        cc_pattern = r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b'
+        ccs = re.findall(cc_pattern, text)
+        if ccs:
+            pii_found['credit_card'] = ccs
+        
+        # Email
+        email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
+        emails = re.findall(email_pattern, text)
+        if emails:
+            pii_found['email'] = emails
+        
+        # Phone
+        phone_pattern = r'\b(?:\+?1[-.]?)?\(?\d{3}\)?[-.]?\d{3}[-.]?\d{4}\b'
+        phones = re.findall(phone_pattern, text)
+        if phones:
+            pii_found['phone'] = phones
+        
+        # Address (basic)
+        address_pattern = r'\b\d+\s+[A-Za-z]+\s+(?:Street|St|Avenue|Ave|Road|Rd|Boulevard|Blvd)\b'
+        addresses = re.findall(address_pattern, text, re.IGNORECASE)
+        if addresses:
+            pii_found['address'] = addresses
+        
+        return len(pii_found) > 0, pii_found
+    
+    def check_spam(self, text: str) -> Tuple[bool, List[str]]:
+        """
+        Check for spam content.
+        
+        Args:
+            text: Text to check
+            
+        Returns:
+            Tuple of (is_spam, list of matched patterns)
+        """
+        if not self.config["enable_spam_detection"]:
+            return False, []
+        
+        matched_patterns = []
+        
+        for pattern in self.spam_patterns:
+            if re.search(pattern, text):
+                matched_patterns.append(pattern)
+        
+        # Check for blocked domains
+        for domain in self.config["blocked_domains"]:
+            if domain in text.lower():
+                matched_patterns.append(f"Blocked domain: {domain}")
+        
+        return len(matched_patterns) > 0, matched_patterns
+    
+    def check_dangerous_content(self, text: str) -> Tuple[bool, List[str]]:
+        """
+        Check for dangerous instructions or content.
+        
+        Args:
+            text: Text to check
+            
+        Returns:
+            Tuple of (is_dangerous, list of matched categories)
+        """
+        text_lower = text.lower()
+        dangerous_categories = []
+        
+        # Check dangerous instructions
+        for pattern in self.dangerous_instructions:
+            if re.search(pattern, text_lower):
+                dangerous_categories.append("dangerous_instructions")
+                break
+        
+        # Check medical misinformation
+        for pattern in self.medical_misinfo:
+            if re.search(pattern, text_lower):
+                dangerous_categories.append("medical_misinformation")
+                break
+        
+        return len(dangerous_categories) > 0, dangerous_categories
+    
+    def check_repetition(self, text: str) -> Tuple[bool, float]:
+        """
+        Check for excessive repetition (potential spam or model failure).
+        
+        Args:
+            text: Text to check
+            
+        Returns:
+            Tuple of (is_repetitive, repetition_ratio)
+        """
+        words = text.split()
+        if len(words) < 10:
+            return False, 0.0
+        
+        unique_words = len(set(words))
+        total_words = len(words)
+        repetition_ratio = 1.0 - (unique_words / total_words)
+        
+        is_repetitive = repetition_ratio > self.config["max_repetition_ratio"]
+        return is_repetitive, repetition_ratio
+    
+    def moderate_content(self, text: str, context: str = "general") -> ModerationResult:
+        """
+        Perform comprehensive content moderation.
+        
+        Args:
+            text: Text to moderate
+            context: Context of the content (general, chat, code, etc.)
+            
+        Returns:
+            ModerationResult with detailed analysis
+        """
+        violations = []
+        confidence_scores = {}
+        risk_level = "low"
+        
+        # Run all checks
+        has_profanity, profanity_words = self.check_profanity(text)
+        if has_profanity:
+            violations.append(f"Profanity detected: {len(profanity_words)} words")
+            confidence_scores["profanity"] = 0.9
+            risk_level = "medium"
+        
+        is_toxic, toxicity_score, toxic_phrases = self.check_toxicity(text)
+        if is_toxic:
+            violations.append(f"Toxic content detected (score: {toxicity_score:.2f})")
+            confidence_scores["toxicity"] = toxicity_score
+            risk_level = "high"
+        
+        has_bias, bias_types = self.check_bias(text)
+        if has_bias:
+            violations.append(f"Potential bias detected: {', '.join(bias_types.keys())}")
+            confidence_scores["bias"] = 0.7
+            if risk_level == "low":
+                risk_level = "medium"
+        
+        has_pii, pii_types = self.check_pii(text)
+        if has_pii:
+            violations.append(f"PII detected: {', '.join(pii_types.keys())}")
+            confidence_scores["pii"] = 1.0
+            risk_level = "high"
+        
+        is_spam, spam_patterns = self.check_spam(text)
+        if is_spam:
+            violations.append(f"Spam indicators: {len(spam_patterns)}")
+            confidence_scores["spam"] = 0.8
+            if risk_level == "low":
+                risk_level = "medium"
+        
+        is_dangerous, dangerous_categories = self.check_dangerous_content(text)
+        if is_dangerous:
+            violations.append(f"Dangerous content: {', '.join(dangerous_categories)}")
+            confidence_scores["dangerous"] = 0.95
+            risk_level = "critical"
+        
+        is_repetitive, repetition_ratio = self.check_repetition(text)
+        if is_repetitive:
+            violations.append(f"Excessive repetition ({repetition_ratio:.2%})")
+            confidence_scores["repetition"] = repetition_ratio
+        
+        # Determine approval and recommended action
+        is_approved = len(violations) == 0 or (risk_level == "low" and not is_dangerous)
+        
+        if risk_level == "critical":
+            recommended_action = "block"
+        elif risk_level == "high":
+            recommended_action = "review"
+        elif risk_level == "medium":
+            recommended_action = "flag"
+        else:
+            recommended_action = "approve"
+        
+        # Sanitize if needed
+        sanitized_content = None
+        if has_pii:
+            sanitized_content = self._sanitize_pii(text)
+        
+        return ModerationResult(
+            timestamp=datetime.now().isoformat(),
+            is_approved=is_approved,
+            risk_level=risk_level,
+            violations=violations,
+            confidence_scores=confidence_scores,
+            recommended_action=recommended_action,
+            sanitized_content=sanitized_content,
+            metadata={
+                "text_length": len(text),
+                "word_count": len(text.split()),
+                "context": context
+            }
+        )
+    
+    def _sanitize_pii(self, text: str) -> str:
+        """Sanitize text by removing/redacting PII."""
+        sanitized = text
+        
+        # Redact SSN
+        sanitized = re.sub(r'\b\d{3}-\d{2}-\d{4}\b', '[SSN-REDACTED]', sanitized)
+        
+        # Redact credit cards
+        sanitized = re.sub(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', '[CC-REDACTED]', sanitized)
+        
+        # Redact emails
+        sanitized = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '[EMAIL-REDACTED]', sanitized)
+        
+        # Redact phones
+        sanitized = re.sub(r'\b(?:\+?1[-.]?)?\(?\d{3}\)?[-.]?\d{3}[-.]?\d{4}\b', '[PHONE-REDACTED]', sanitized)
+        
+        return sanitized
+    
+    def batch_moderate(self, texts: List[str]) -> List[ModerationResult]:
+        """
+        Moderate multiple texts in batch.
+        
+        Args:
+            texts: List of texts to moderate
+            
+        Returns:
+            List of ModerationResults
+        """
+        return [self.moderate_content(text) for text in texts]
+    
+    def export_results(self, results: List[ModerationResult], filepath: str):
+        """
+        Export moderation results to JSON file.
+        
+        Args:
+            results: List of ModerationResults
+            filepath: Output file path
+        """
+        with open(filepath, 'w') as f:
+            json.dump([asdict(r) for r in results], f, indent=2)
+        
+        logger.info(f"Exported {len(results)} moderation results to {filepath}")
+
+
+# Example usage
+if __name__ == "__main__":
+    # Initialize filter
+    filter_system = ContentFilter()
+    
+    # Test cases
+    test_texts = [
+        "What is the capital of France?",  # Safe
+        "You are a stupid idiot!",  # Toxic
+        "My SSN is 123-45-6789",  # PII
+        "Buy now! Limited time offer! www.spam.com",  # Spam
+        "How to make a bomb at home",  # Dangerous
+    ]
+    
+    print("Content Moderation Results:\n")
+    print("=" * 80)
+    
+    for i, text in enumerate(test_texts, 1):
+        result = filter_system.moderate_content(text)
+        
+        print(f"\nTest {i}: {text[:50]}...")
+        print(f"Approved: {result.is_approved}")
+        print(f"Risk Level: {result.risk_level}")
+        print(f"Violations: {result.violations}")
+        print(f"Recommended Action: {result.recommended_action}")
+        if result.sanitized_content:
+            print(f"Sanitized: {result.sanitized_content[:100]}...")
+        print("-" * 80)
+    
+    # Batch processing example
+    results = filter_system.batch_moderate(test_texts)
+    filter_system.export_results(results, "moderation_results.json")
+    print(f"\n✓ Exported {len(results)} results to moderation_results.json")