Intelligent Content Moderation: Safeguarding Your Platform

class ContentModerator:
    """Intelligent Content Moderation System"""
    
    def __init__(self, llm_api):
        self.llm = llm_api
        self.categories = {
            'violence': 'Violent and gory content',
            'adult': 'Pornographic and vulgar content',
            'illegal': 'Illegal and non-compliant information',
            'hate': 'Hate speech and discrimination',
            'privacy': 'Personal privacy information',
            'spam': 'Spam and advertising information'
        }
    
    def moderate(self, content):
        """Moderate content"""
        prompt = f"""Please review the following content and determine if it contains any violations.

Content: {content}

Please analyze the following aspects:
1. Does it contain violent, pornographic, illegal, or other violative content?
2. Does it contain personal privacy or sensitive information?
3. Is it spam, advertising, or meaningless content?
4. Overall content quality score (1-10)

Return in JSON format:
{{
    "safe": true/false,
    "categories": ["violation categories"],
    "severity": "low/medium/high",
    "reasons": ["specific reasons"],
    "score": 1-10,
    "suggestion": "handling suggestion"
}}"""
        
        response = self.llm.generate(prompt, temperature=0.1)
        return json.loads(response)
    
    def batch_moderate(self, contents, parallel=True):
        """Batch moderation"""
        if parallel:
            # Parallel processing
            with ThreadPoolExecutor(max_workers=10) as executor:
                futures = [
                    executor.submit(self.moderate, content) 
                    for content in contents
                ]
                results = [f.result() for f in futures]
        else:
            # Serial processing
            results = [self.moderate(content) for content in contents]
        
        return results
    
    def real_time_filter(self, text_stream):
        """Real-time streaming moderation"""
        buffer = ""
        for chunk in text_stream:
            buffer += chunk
            
            # Check every 50 characters
            if len(buffer) > 50:
                result = self.quick_check(buffer)
                if not result['safe']:
                    # Interrupt immediately
                    return {
                        'blocked': True,
                        'reason': result['reason']
                    }
                buffer = buffer[-25:]  # Retain part for context
        
        # Final full check
        return self.moderate(text_stream.get_full_text())
    
    def custom_rules(self, content, rules):
        """Custom rule moderation"""
        violations = []
        
        for rule in rules:
            if rule['type'] == 'keyword':
                if rule['pattern'] in content.lower():
                    violations.append(rule['action'])
            elif rule['type'] == 'regex':
                if re.search(rule['pattern'], content):
                    violations.append(rule['action'])
            elif rule['type'] == 'ai':
                # Use AI to judge
                check = self.llm.generate(
                    f"Does the content {rule['description']}? {content}"
                )
                if "Yes" in check:
                    violations.append(rule['action'])
        
        return violations