1pip install openai google-api-python-client requests python-dotenv

1import openai
2import os
3from dotenv import load_dotenv
4from typing import Dict
5
6load_dotenv()
7openai.api_key = os.getenv("OPENAI_API_KEY")
8
9class ContentModerator:
10    """AI-powered content moderation system"""
11    
12    def __init__(self):
13        self.openai_key = os.getenv("OPENAI_API_KEY")
14    
15    def check_content_openai(self, text: str) -> Dict:
16        """
17        Use OpenAI Moderation API to check content
18        Categories: hate, harassment, self-harm, sexual, violence
19        """
20        try:
21            response = openai.moderations.create(input=text)
22            result = response.results[0]
23            
24            return {
25                "flagged": result.flagged,
26                "categories": result.categories.model_dump(),
27                "category_scores": result.category_scores.model_dump(),
28                "highest_score_category": max(
29                    result.category_scores.model_dump().items(),
30                    key=lambda x: x[1]
31                )[0] if result.flagged else None
32            }
33        except Exception as e:
34            print(f"OpenAI moderation error: {e}")
35            return {"error": str(e)}
36    
37    def calculate_risk_score(self, moderation_result: Dict) -> int:
38        """Calculate overall risk score 0-100"""
39        if "error" in moderation_result:
40            return 0  # Fail open - let human review
41        
42        if not moderation_result["flagged"]:
43            return 10  # Low risk
44        
45        # Get highest category score (0.0 to 1.0)
46        scores = moderation_result["category_scores"]
47        max_score = max(scores.values())
48        
49        # Convert to 0-100 scale
50        risk_score = int(max_score * 100)
51        
52        # Weight certain categories higher
53        categories = moderation_result["categories"]
54        if categories.get("hate"):
55            risk_score = min(risk_score + 20, 100)
56        if categories.get("violence"):
57            risk_score = min(risk_score + 15, 100)
58        
59        return risk_score
60    
61    def make_decision(self, risk_score: int, categories: Dict) -> Dict:
62        """Decide what action to take based on risk"""
63        
64        if risk_score < 30:
65            return {
66                "action": "approve",
67                "reason": "Low risk content",
68                "requires_human_review": False
69            }
70        elif risk_score < 80:
71            return {
72                "action": "review",
73                "reason": "Moderate risk - needs human review",
74                "requires_human_review": True,
75                "priority": "medium" if risk_score < 60 else "high"
76            }
77        else:
78            return {
79                "action": "reject",
80                "reason": "High risk content - auto-rejected",
81                "requires_human_review": True,
82                "priority": "urgent"
83            }
84    
85    def moderate_text(self, text: str) -> Dict:
86        """Complete moderation pipeline for text content"""
87        
88        # Step 1: Check with OpenAI
89        moderation_result = self.check_content_openai(text)
90        
91        # Step 2: Calculate risk score
92        risk_score = self.calculate_risk_score(moderation_result)
93        
94        # Step 3: Make decision
95        decision = self.make_decision(
96            risk_score,
97            moderation_result.get("categories", {})
98        )
99        
100        # Step 4: Return complete analysis
101        return {
102            "text_preview": text[:100] + "..." if len(text) > 100 else text,
103            "risk_score": risk_score,
104            "flagged": moderation_result.get("flagged", False),
105            "categories": moderation_result.get("categories", {}),
106            "category_scores": moderation_result.get("category_scores", {}),
107            "decision": decision,
108            "timestamp": datetime.now().isoformat()
109        }
110
111# Test the moderator
112if __name__ == "__main__":
113    moderator = ContentModerator()
114    
115    # Test cases
116    test_texts = [
117        "Great product! Highly recommend.",
118        "This is spam! Click here to win $1000!!! BUY NOW!!!",
119        "[Toxic content example removed for article]",
120        "I respectfully disagree with this policy and here's why..."
121    ]
122    
123    for text in test_texts:
124        print(f"\nTesting: {text[:50]}...")
125        result = moderator.moderate_text(text)
126        print(f"Risk Score: {result['risk_score']}")
127        print(f"Action: {result['decision']['action']}")
128        print(f"Flagged: {result['flagged']}")

1import re
2from typing import List
3
4class SpamDetector:
5    """Detect spam content using patterns and heuristics"""
6    
7    def __init__(self):
8        # Common spam indicators
9        self.spam_keywords = [
10            "click here", "buy now", "limited time", "act now",
11            "free money", "winner", "prize", "congratulations",
12            "weight loss", "viagra", "casino", "lottery",
13            "meet singles", "hot girls", "penis enlargement"
14        ]
15        
16        self.suspicious_patterns = [
17            r'\b\d{10,}\b',  # Long numbers (phone numbers, etc.)
18            r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+',  # URLs
19            r'[A-Z]{10,}',  # Excessive caps
20            r'(.)\1{4,}',  # Character repetition (!!!!! or aaaaa)
21            r'\$\d+',  # Dollar amounts
22        ]
23    
24    def calculate_spam_score(self, text: str) -> Dict:
25        """Calculate spam probability 0-100"""
26        text_lower = text.lower()
27        score = 0
28        flags = []
29        
30        # Check spam keywords
31        keyword_count = sum(1 for keyword in self.spam_keywords if keyword in text_lower)
32        if keyword_count > 0:
33            score += min(keyword_count * 20, 60)
34            flags.append(f"Contains {keyword_count} spam keywords")
35        
36        # Check suspicious patterns
37        for pattern in self.suspicious_patterns:
38            if re.search(pattern, text):
39                score += 10
40                flags.append(f"Matches spam pattern: {pattern[:20]}")
41        
42        # Check excessive caps (>50% uppercase)
43        if len(text) > 10:
44            caps_ratio = sum(1 for c in text if c.isupper()) / len(text)
45            if caps_ratio > 0.5:
46                score += 25
47                flags.append("Excessive capitalization")
48        
49        # Check excessive punctuation
50        punctuation_count = sum(1 for c in text if c in "!?.")
51        if punctuation_count > len(text) / 10:  # More than 10% punctuation
52            score += 15
53            flags.append("Excessive punctuation")
54        
55        # Check very short promotional messages
56        if len(text) < 50 and any(word in text_lower for word in ["buy", "click", "win", "free"]):
57            score += 20
58            flags.append("Short promotional message")
59        
60        return {
61            "spam_score": min(score, 100),
62            "is_likely_spam": score > 50,
63            "flags": flags
64        }
65
66# Integrate into ContentModerator
67class ContentModerator:
68    # ... previous code ...
69    
70    def __init__(self):
71        self.openai_key = os.getenv("OPENAI_API_KEY")
72        self.spam_detector = SpamDetector()
73    
74    def moderate_text(self, text: str) -> Dict:
75        # ... previous OpenAI check ...
76        
77        # Add spam detection
78        spam_result = self.spam_detector.calculate_spam_score(text)
79        
80        # Adjust risk score if spam detected
81        if spam_result["is_likely_spam"]:
82            risk_score = max(risk_score, 70)  # Minimum 70 for spam
83        
84        # ... rest of function ...
85        
86        return {
87            # ... previous fields ...
88            "spam_analysis": spam_result,
89            # ... rest of response ...
90        }

1from googleapiclient import discovery
2import json
3
4class ToxicityDetector:
5    """Use Google Perspective API for toxicity detection"""
6    
7    def __init__(self, api_key: str):
8        self.client = discovery.build(
9            "commentanalyzer",
10            "v1alpha1",
11            developerKey=api_key,
12            discoveryServiceUrl="https://commentanalyzer.googleapis.com/$discovery/rest?version=v1alpha1",
13            static_discovery=False
14        )
15    
16    def analyze_toxicity(self, text: str) -> Dict:
17        """Analyze text for various types of toxicity"""
18        
19        analyze_request = {
20            'comment': {'text': text},
21            'requestedAttributes': {
22                'TOXICITY': {},
23                'SEVERE_TOXICITY': {},
24                'IDENTITY_ATTACK': {},
25                'INSULT': {},
26                'PROFANITY': {},
27                'THREAT': {}
28            },
29            'languages': ['en']
30        }
31        
32        try:
33            response = self.client.comments().analyze(body=analyze_request).execute()
34            
35            scores = {}
36            for attribute in response['attributeScores']:
37                score = response['attributeScores'][attribute]['summaryScore']['value']
38                scores[attribute.lower()] = score
39            
40            return {
41                "toxicity_scores": scores,
42                "is_toxic": scores.get("toxicity", 0) > 0.7,
43                "max_toxicity": max(scores.values()),
44                "toxic_attributes": [
45                    attr for attr, score in scores.items() if score > 0.7
46                ]
47            }
48        except Exception as e:
49            print(f"Perspective API error: {e}")
50            return {"error": str(e)}
51
52# Add to ContentModerator
53class ContentModerator:
54    def __init__(self):
55        self.openai_key = os.getenv("OPENAI_API_KEY")
56        self.spam_detector = SpamDetector()
57        self.toxicity_detector = ToxicityDetector(os.getenv("PERSPECTIVE_API_KEY"))
58    
59    def moderate_text(self, text: str) -> Dict:
60        # ... previous checks ...
61        
62        # Add toxicity detection
63        toxicity_result = self.toxicity_detector.analyze_toxicity(text)
64        
65        # Adjust risk score for severe toxicity
66        if toxicity_result.get("max_toxicity", 0) > 0.9:
67            risk_score = max(risk_score, 95)
68        elif toxicity_result.get("is_toxic", False):
69            risk_score = max(risk_score, 75)
70        
71        # ... rest of function ...

1class CustomRulesEngine:
2    """Custom rules specific to your community"""
3    
4    def __init__(self):
5        # Your custom blocked words/phrases
6        self.blocked_keywords = [
7            "competitor_name",
8            "external_platform",
9            "buy my course"
10        ]
11        
12        # Regex patterns for complex rules
13        self.blocked_patterns = [
14            r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',  # Email addresses
15            r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',  # Phone numbers
16            r'whatsapp|telegram|discord',  # Off-platform coordination
17        ]
18        
19        # Context-aware rules
20        self.require_review_if = [
21            ("refund", "scam"),  # If mentions both
22            ("admin", "complaint"),
23            ("lawsuit", "lawyer")
24        ]
25    
26    def check_custom_rules(self, text: str) -> Dict:
27        """Apply custom community rules"""
28        text_lower = text.lower()
29        violations = []
30        requires_review = False
31        
32        # Check blocked keywords
33        for keyword in self.blocked_keywords:
34            if keyword.lower() in text_lower:
35                violations.append(f"Contains blocked keyword: {keyword}")
36        
37        # Check blocked patterns
38        for pattern in self.blocked_patterns:
39            if re.search(pattern, text, re.IGNORECASE):
40                violations.append(f"Matches blocked pattern")
41        
42        # Check multi-word context rules
43        for word1, word2 in self.require_review_if:
44            if word1 in text_lower and word2 in text_lower:
45                requires_review = True
46                violations.append(f"Contains sensitive combination: {word1} + {word2}")
47        
48        return {
49            "custom_violations": violations,
50            "violation_count": len(violations),
51            "requires_manual_review": requires_review or len(violations) > 0
52        }

1import base64
2from PIL import Image
3import io
4
5class ImageModerator:
6    """Moderate image content"""
7    
8    def __init__(self):
9        self.openai_key = os.getenv("OPENAI_API_KEY")
10    
11    def analyze_image(self, image_path: str) -> Dict:
12        """Analyze image for inappropriate content"""
13        
14        # Convert image to base64
15        with open(image_path, "rb") as image_file:
16            base64_image = base64.b64encode(image_file.read()).decode('utf-8')
17        
18        try:
19            response = openai.chat.completions.create(
20                model="gpt-4o-mini",
21                messages=[
22                    {
23                        "role": "user",
24                        "content": [
25                            {
26                                "type": "text",
27                                "text": """Analyze this image for content moderation. Check for:
28                                1. Nudity or sexual content
29                                2. Violence or gore
30                                3. Hate symbols or extremism
31                                4. Spam or promotional content
32                                5. Personal information (faces, addresses, etc.)
33                                
34                                Return JSON:
35                                {
36                                    "is_appropriate": true/false,
37                                    "violations": ["list of issues"],
38                                    "risk_level": "low/medium/high",
39                                    "description": "brief description of image"
40                                }"""
41                            },
42                            {
43                                "type": "image_url",
44                                "image_url": {
45                                    "url": f"data:image/jpeg;base64,{base64_image}"
46                                }
47                            }
48                        ]
49                    }
50                ],
51                max_tokens=300
52            )
53            
54            result_text = response.choices[0].message.content
55            # Parse JSON from response
56            import json
57            result = json.loads(result_text)
58            
59            return result
60            
61        except Exception as e:
62            print(f"Image moderation error: {e}")
63            return {"error": str(e)}

1from datetime import datetime
2import json
3
4class ModerationPipeline:
5    """Complete content moderation pipeline"""
6    
7    def __init__(self):
8        self.text_moderator = ContentModerator()
9        self.image_moderator = ImageModerator()
10    
11    def moderate_post(self, content: Dict) -> Dict:
12        """
13        Moderate a complete post (text + images + metadata)
14        
15        Args:
16            content: {
17                "text": "post text",
18                "images": ["path1.jpg", "path2.jpg"],
19                "author_id": "user123",
20                "metadata": {...}
21            }
22        """
23        results = {
24            "post_id": content.get("id", "unknown"),
25            "timestamp": datetime.now().isoformat(),
26            "text_moderation": None,
27            "image_moderation": [],
28            "final_decision": None
29        }
30        
31        # Moderate text
32        if content.get("text"):
33            results["text_moderation"] = self.text_moderator.moderate_text(content["text"])
34        
35        # Moderate images
36        if content.get("images"):
37            for img_path in content["images"]:
38                img_result = self.image_moderator.analyze_image(img_path)
39                results["image_moderation"].append(img_result)
40        
41        # Calculate overall decision
42        text_risk = results["text_moderation"]["risk_score"] if results["text_moderation"] else 0
43        image_risks = [img.get("risk_level", "low") for img in results["image_moderation"]]
44        
45        max_image_risk = 0
46        if "high" in image_risks:
47            max_image_risk = 90
48        elif "medium" in image_risks:
49            max_image_risk = 60
50        
51        overall_risk = max(text_risk, max_image_risk)
52        
53        # Final decision
54        if overall_risk < 30:
55            decision = "auto_approve"
56        elif overall_risk < 80:
57            decision = "human_review"
58        else:
59            decision = "auto_reject"
60        
61        results["final_decision"] = {
62            "action": decision,
63            "overall_risk_score": overall_risk,
64            "requires_review": decision == "human_review"
65        }
66        
67        return results
68    
69    def log_decision(self, results: Dict):
70        """Log moderation decision for analytics"""
71        # In production: send to database or logging service
72        with open("moderation_log.jsonl", "a") as f:
73            f.write(json.dumps(results) + "\n")
74
75# Usage example
76if __name__ == "__main__":
77    pipeline = ModerationPipeline()
78    
79    test_post = {
80        "id": "post_123",
81        "text": "Check out this great product!",
82        "images": ["product.jpg"],
83        "author_id": "user_456"
84    }
85    
86    result = pipeline.moderate_post(test_post)
87    print(json.dumps(result, indent=2))
88    
89    print(f"\nDecision: {result['final_decision']['action']}")
90    print(f"Overall Risk: {result['final_decision']['overall_risk_score']}")

1from flask import Flask, request, jsonify
2from moderation import ModerationPipeline
3
4app = Flask(__name__)
5pipeline = ModerationPipeline()
6
7@app.route('/moderate', methods=['POST'])
8def moderate_content():
9    """
10    Endpoint to moderate content
11    
12    POST /moderate
13    {
14        "text": "content to moderate",
15        "images": ["url1", "url2"],
16        "author_id": "user123"
17    }
18    """
19    try:
20        data = request.json
21        result = pipeline.moderate_post(data)
22        
23        return jsonify({
24            "success": True,
25            "result": result
26        }), 200
27        
28    except Exception as e:
29        return jsonify({
30            "success": False,
31            "error": str(e)
32        }), 500
33
34@app.route('/health', methods=['GET'])
35def health_check():
36    return jsonify({"status": "healthy"}), 200
37
38if __name__ == '__main__':
39    app.run(host='0.0.0.0', port=5000)

1python api.py

1curl -X POST http://localhost:5000/moderate \
2  -H "Content-Type: application/json" \
3  -d '{"text": "Great post!", "author_id": "user123"}'

1function moderate_comment($comment) {
2    $response = wp_remote_post('http://your-server:5000/moderate', array(
3        'body' => json_encode(array(
4            'text' => $comment['comment_content'],
5            'author_id' => $comment['user_id']
6        )),
7        'headers' => array('Content-Type' => 'application/json')
8    ));
9    
10    $result = json_decode(wp_remote_retrieve_body($response), true);
11    
12    if ($result['result']['final_decision']['action'] == 'auto_reject') {
13        wp_die('Your comment was rejected by our moderation system.');
14    }
15}

1import discord
2import requests
3
4@bot.event
5async def on_message(message):
6    # Moderate message
7    response = requests.post('http://localhost:5000/moderate', json={
8        'text': message.content,
9        'author_id': str(message.author.id)
10    })
11    
12    result = response.json()['result']
13    
14    if result['final_decision']['action'] == 'auto_reject':
15        await message.delete()
16        await message.author.send("Your message was removed for violating community guidelines.")

1async function moderatePost(text, authorId) {
2    const response = await fetch('http://your-server:5000/moderate', {
3        method: 'POST',
4        headers: {'Content-Type': 'application/json'},
5        body: JSON.stringify({
6            text: text,
7            author_id: authorId
8        })
9    });
10    
11    const result = await response.json();
12    
13    if (result.result.final_decision.action === 'auto_reject') {
14        alert('Your post violates community guidelines and cannot be published.');
15        return false;
16    }
17    
18    return true;
19}

1import pandas as pd
2from collections import Counter
3
4class ModerationAnalytics:
5    """Analyze moderation logs and generate reports"""
6    
7    def load_logs(self, log_file: str = "moderation_log.jsonl") -> pd.DataFrame:
8        """Load moderation logs into DataFrame"""
9        import json
10        
11        logs = []
12        with open(log_file, 'r') as f:
13            for line in f:
14                logs.append(json.loads(line))
15        
16        return pd.DataFrame(logs)
17    
18    def generate_report(self, df: pd.DataFrame) -> Dict:
19        """Generate moderation statistics report"""
20        
21        # Extract decisions
22        decisions = df['final_decision'].apply(lambda x: x['action'])
23        
24        # Count by decision type
25        decision_counts = Counter(decisions)
26        
27        # Average risk scores
28        avg_risk = df['final_decision'].apply(lambda x: x['overall_risk_score']).mean()
29        
30        # Most common violations
31        violations = []
32        for record in df['text_moderation']:
33            if record and record.get('categories'):
34                violations.extend([k for k, v in record['categories'].items() if v])
35        
36        violation_counts = Counter(violations)
37        
38        return {
39            "total_posts_moderated": len(df),
40            "auto_approved": decision_counts.get("auto_approve", 0),
41            "auto_rejected": decision_counts.get("auto_reject", 0),
42            "human_reviews_needed": decision_counts.get("human_review", 0),
43            "average_risk_score": round(avg_risk, 2),
44            "approval_rate": round(decision_counts.get("auto_approve", 0) / len(df) * 100, 2),
45            "most_common_violations": dict(violation_counts.most_common(5)),
46            "moderation_efficiency": f"{(decision_counts.get('auto_approve', 0) + decision_counts.get('auto_reject', 0)) / len(df) * 100:.1f}%"
47        }

1rejection_messages = {
2    "spam": "Your post appears to be spam or promotional content.",
3    "hate": "Your post contains hate speech or discriminatory language.",
4    "violence": "Your post contains violent or graphic content.",
5    "sexual": "Your post contains inappropriate sexual content."
6}

1def create_appeal(post_id, user_id, reason):
2    # Flag for expedited human review
3    # Track appeal success rate to improve models
4    pass

1def adjust_for_user_trust(risk_score, user_trust_score):
2    """Trusted users get benefit of doubt"""
3    if user_trust_score > 90:  # Highly trusted
4        risk_score *= 0.7  # Reduce risk by 30%
5    elif user_trust_score < 20:  # New or problematic
6        risk_score *= 1.3  # Increase risk by 30%
7    
8    return min(risk_score, 100)