E-commerce Content Moderation at Scale: AI-Powered Brand Safety

┌─────────────────────────────────────────────┐
│      User-Generated Content Ingestion       │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐    │
│  │ Reviews  │ │ Listings │ │ Messages │    │
│  └────┬─────┘ └────┬─────┘ └────┬─────┘    │
└───────┼────────────┼────────────┼───────────┘
        │            │            │
        ▼            ▼            ▼
┌─────────────────────────────────────────────┐
│     Content Classification & Routing         │
│  • Review authenticity analysis              │
│  • Product listing validation                │
│  • Message toxicity pre-screening            │
└────────────────┬────────────────────────────┘
                 │
                 ▼
┌─────────────────────────────────────────────┐
│      RAIL Score Safety Evaluation Layer      │
│                                              │
│  ┌──────────────┐  ┌──────────────┐        │
│  │  Toxicity    │  │   Fairness   │        │
│  │  Detection   │  │   Check      │        │
│  └──────────────┘  └──────────────┘        │
│                                              │
│  ┌──────────────┐  ┌──────────────┐        │
│  │Hallucination │  │   Context    │        │
│  │  Detection   │  │Appropriateness│        │
│  └──────────────┘  └──────────────┘        │
│                                              │
│  ┌──────────────┐  ┌──────────────┐        │
│  │ Manipulation │  │  Authenticity│        │
│  │  Detection   │  │   Scoring    │        │
│  └──────────────┘  └──────────────┘        │
└────────────┬───────────────────────────────┘
             │
             ▼
     [ Policy Decision Tree ]
             │
      ┌──────┴──────┐
      ▼             ▼
  [ Auto-Approve ] [ Human Review Queue ]
  [ Auto-Reject  ] [ Seller Education    ]

import os
from rail_score import RailScore
from typing import Dict, List, Any
import re

# Initialize RAIL Score
rail_client = RailScore(api_key=os.environ.get("RAIL_API_KEY"))

class ReviewModerationSystem:
    """
    Multi-dimensional review moderation with RAIL Score
    """

    def __init__(self):
        self.rail = rail_client
        self.auto_approve_threshold = 90
        self.human_review_threshold = 70
        self.auto_reject_threshold = 50

    def moderate_review(self, review: Dict[str, Any]) -> Dict[str, Any]:
        """
        Comprehensive review moderation pipeline
        """

        # Step 1: Extract review context
        context = self._build_review_context(review)

        # Step 2: RAIL Score evaluation
        rail_eval = self.rail.evaluate(
            prompt=context["prompt"],
            response=review["text"],
            categories=[
                "toxicity",
                "fairness",
                "hallucination",
                "context_appropriateness"
            ],
            metadata={
                "content_type": "product_review",
                "reviewer_history": context["reviewer_stats"],
                "product_category": review["product_category"]
            }
        )

        # Step 3: Authenticity analysis
        authenticity_score = self._analyze_authenticity(review, context)

        # Step 4: Pattern detection (coordinated fake reviews)
        pattern_risk = self._detect_fake_review_patterns(review, context)

        # Step 5: Brand safety check
        brand_safety = self._check_brand_safety(review)

        # Step 6: Make moderation decision
        decision = self._make_moderation_decision(
            review=review,
            rail_eval=rail_eval,
            authenticity_score=authenticity_score,
            pattern_risk=pattern_risk,
            brand_safety=brand_safety
        )

        return decision

    def _build_review_context(self, review: Dict) -> Dict:
        """
        Build context for RAIL Score evaluation
        """
        reviewer_stats = self._get_reviewer_history(review["user_id"])
        product_info = self._get_product_info(review["product_id"])

        context_prompt = f"""
        Review Context:

        Product: [PRODUCT_NAME]
        Category: [PRODUCT_CATEGORY]
        Price: $[PRODUCT_PRICE]

        Reviewer Profile:
        - Account age: [DAYS] days
        - Total reviews: [COUNT]
        - Verified purchases: [PERCENTAGE]%
        - Average review length: [WORDS] words
        - Review frequency: [FREQUENCY] per month

        Review Details:
        - Rating: [RATING]/5 stars
        - Verified purchase: [VERIFIED]
        - Review length: [LENGTH] words
        - Time since purchase: [DAYS] days

        Review text:
        [REVIEW_TEXT]
        """

        return {
            "prompt": context_prompt,
            "reviewer_stats": reviewer_stats,
            "product_info": product_info
        }

    def _analyze_authenticity(self, review: Dict, context: Dict) -> float:
        """
        Detect fake review signals
        """
        authenticity_signals = []

        # Signal 1: Suspicious reviewer behavior
        if context["reviewer_stats"]["account_age_days"] < 7:
            authenticity_signals.append(("new_account", -15))

        if context["reviewer_stats"]["total_reviews"] > 50 and \
           context["reviewer_stats"]["verified_purchase_rate"] < 20:
            authenticity_signals.append(("unverified_reviewer", -25))

        if context["reviewer_stats"]["reviews_per_month"] > 30:
            authenticity_signals.append(("review_farm_pattern", -40))

        # Signal 2: Review content analysis
        if self._is_generic_template(review["text"]):
            authenticity_signals.append(("generic_template", -30))

        if self._contains_seller_coaching_phrases(review["text"]):
            authenticity_signals.append(("coached_review", -35))

        # Signal 3: Timing analysis
        if review["days_since_purchase"] < 1 and len(review["text"]) > 200:
            authenticity_signals.append(("suspiciously_fast", -20))

        # Signal 4: Language analysis
        if self._detect_translated_text(review["text"]):
            authenticity_signals.append(("likely_translated", -15))

        # Calculate authenticity score (0-100)
        base_score = 100
        for signal, penalty in authenticity_signals:
            base_score += penalty

        authenticity_score = max(0, min(100, base_score))

        return {
            "score": authenticity_score,
            "signals": authenticity_signals,
            "risk_level": "high" if authenticity_score < 50 else "medium" if authenticity_score < 75 else "low"
        }

    def _detect_fake_review_patterns(self, review: Dict, context: Dict) -> Dict:
        """
        Detect coordinated fake review campaigns
        """
        product_id = review["product_id"]
        timeframe = "last_7_days"

        # Get recent reviews for same product
        recent_reviews = self._get_recent_reviews(product_id, timeframe)

        pattern_risks = []

        # Pattern 1: Review velocity spike
        if len(recent_reviews) > 100 and context["product_info"]["age_days"] < 30:
            pattern_risks.append("suspicious_velocity")

        # Pattern 2: Unusual rating distribution
        five_star_rate = sum(1 for r in recent_reviews if r["rating"] == 5) / len(recent_reviews)
        if five_star_rate > 0.85:
            pattern_risks.append("unnatural_rating_distribution")

        # Pattern 3: Similar review text (copy-paste campaigns)
        similar_reviews = self._find_similar_reviews(review["text"], recent_reviews)
        if len(similar_reviews) > 5:
            pattern_risks.append("duplicate_content_detected")

        # Pattern 4: Same reviewer set across multiple products from same seller
        seller_id = context["product_info"]["seller_id"]
        if self._detect_cross_product_reviewer_overlap(review["user_id"], seller_id) > 0.3:
            pattern_risks.append("coordinated_campaign")

        return {
            "risks_detected": pattern_risks,
            "risk_level": "high" if len(pattern_risks) >= 2 else "medium" if len(pattern_risks) == 1 else "low"
        }

    def _check_brand_safety(self, review: Dict) -> Dict:
        """
        Check for brand safety violations
        """
        violations = []

        # Check for prohibited content
        prohibited_patterns = [
            (r"\b(nazi|hitler|holocaust\s+denial)\b", "hate_speech"),
            (r"\b(n-word|f-slur|r-slur)\b", "slurs"),
            (r"(penis|viagra|cialis).*(enlargement|enhancement)", "adult_content"),
            (r"(contact\s+me|email\s+me|whatsapp).*(@|\d{10})", "solicitation"),
            (r"(counterfeit|fake|replica).*branded", "counterfeiting_admission")
        ]

        for pattern, violation_type in prohibited_patterns:
            if re.search(pattern, review["text"], re.IGNORECASE):
                violations.append(violation_type)

        return {
            "passed": len(violations) == 0,
            "violations": violations
        }

    def _make_moderation_decision(self, review, rail_eval, authenticity_score, pattern_risk, brand_safety):
        """
        Multi-factor moderation decision
        """

        # Critical violations - auto-reject
        if not brand_safety["passed"]:
            return {
                "decision": "reject",
                "reason": "Brand safety violation",
                "details": brand_safety["violations"],
                "notify_seller": False,
                "notify_reviewer": True,
                "message": "Your review violated our community guidelines."
            }

        if rail_eval.toxicity_score < 40:
            return {
                "decision": "reject",
                "reason": "Toxic content detected",
                "details": f"Toxicity score: {rail_eval.toxicity_score}",
                "rail_score": rail_eval.overall_score
            }

        if authenticity_score["score"] < 30:
            return {
                "decision": "reject",
                "reason": "High fake review probability",
                "details": authenticity_score["signals"],
                "flag_reviewer_account": True,
                "investigate_seller": review["rating"] == 5  # 5-star fake reviews benefit seller
            }

        # High-confidence fake review pattern
        if pattern_risk["risk_level"] == "high" and authenticity_score["score"] < 60:
            return {
                "decision": "reject",
                "reason": "Coordinated fake review campaign detected",
                "details": pattern_risk["risks_detected"],
                "escalate_to_trust_and_safety": True,
                "investigate_seller": True
            }

        # High quality, authentic review - auto-approve
        if (rail_eval.overall_score >= self.auto_approve_threshold and
            authenticity_score["score"] >= 75 and
            pattern_risk["risk_level"] == "low"):
            return {
                "decision": "approve",
                "confidence": "high",
                "rail_score": rail_eval.overall_score,
                "authenticity_score": authenticity_score["score"]
            }

        # Borderline cases - human review
        if (self.human_review_threshold <= rail_eval.overall_score < self.auto_approve_threshold or
            50 <= authenticity_score["score"] < 75 or
            pattern_risk["risk_level"] == "medium"):
            return {
                "decision": "human_review_required",
                "priority": "medium" if rail_eval.overall_score > 80 else "high",
                "review_notes": {
                    "rail_score": rail_eval.overall_score,
                    "rail_concerns": rail_eval.concerns,
                    "authenticity_score": authenticity_score["score"],
                    "authenticity_signals": authenticity_score["signals"],
                    "pattern_risks": pattern_risk["risks_detected"]
                },
                "suggested_action": "Likely authentic but needs verification" if authenticity_score["score"] > 60 else "Suspicious - recommend reject"
            }

        # Low scores - reject
        return {
            "decision": "reject",
            "reason": "Failed multiple safety checks",
            "details": {
                "rail_score": rail_eval.overall_score,
                "authenticity_score": authenticity_score["score"]
            }
        }

# Example usage
moderator = ReviewModerationSystem()

sample_review = {
    "user_id": "user_892471",
    "product_id": "prod_445821",
    "rating": 5,
    "text": "This product is amazing! Best purchase ever. Highly recommend to everyone. Five stars!",
    "verified_purchase": False,
    "days_since_purchase": 0,
    "product_category": "electronics"
}

result = moderator.moderate_review(sample_review)
print(result)
# Output: {'decision': 'reject', 'reason': 'High fake review probability', ...}

moderation_signals = {
    "rail_score_safety": {
        "toxicity": rail_eval.toxicity_score,
        "fairness": rail_eval.fairness_score,
        "context": rail_eval.context_appropriateness
    },
    "authenticity_analysis": {
        "reviewer_behavior": authenticity_score,
        "content_patterns": template_detection,
        "timing_analysis": velocity_check
    },
    "pattern_detection": {
        "coordinated_campaigns": cross_product_analysis,
        "duplicate_content": similarity_detection,
        "velocity_anomalies": spike_detection
    },
    "brand_safety": {
        "prohibited_content": keyword_flagging,
        "regulatory_compliance": legal_check
    }
}

# Require multiple signals to align for high-confidence decisions

def minimize_false_positives(moderation_result):
    """
    Conservative rejection, aggressive human review routing
    """
    if moderation_result["decision"] == "reject":
        # Require multiple strong signals to auto-reject
        strong_signals = count_strong_negative_signals(moderation_result)

        if strong_signals < 2:
            # Route to human review instead of auto-reject
            return {
                "decision": "human_review_required",
                "reason": "Single rejection signal - verify before blocking",
                "original_signal": moderation_result
            }

    return moderation_result

def detect_review_campaigns(product_id, timeframe="7_days"):
    """
    Detect coordinated fake review campaigns
    """
    recent_reviews = get_reviews(product_id, timeframe)

    # Signal 1: Reviewer overlap with other products from same seller
    reviewer_overlap = calculate_cross_product_overlap(recent_reviews)

    # Signal 2: Temporal clustering (burst of reviews in short time)
    temporal_clustering = detect_review_bursts(recent_reviews)

    # Signal 3: Content similarity (copy-paste campaigns)
    content_similarity = calculate_pairwise_similarity(recent_reviews)

    # Signal 4: Reviewer account characteristics
    new_account_rate = count_new_accounts(recent_reviews) / len(recent_reviews)

    campaign_score = (
        reviewer_overlap * 0.3 +
        temporal_clustering * 0.25 +
        content_similarity * 0.3 +
        new_account_rate * 0.15
    )

    if campaign_score > 0.7:
        flag_for_investigation(product_id, campaign_score)
        notify_trust_and_safety_team()

def seller_accountability(rejected_review):
    """
    Track fake reviews by product/seller
    """
    if rejected_review["reason"] == "fake_review" and rejected_review["rating"] >= 4:
        # This fake review benefits the seller
        seller_id = get_seller_id(rejected_review["product_id"])

        # Track seller fake review history
        seller_fake_review_count = increment_seller_metric(seller_id, "fake_reviews_detected")

        # Escalation thresholds
        if seller_fake_review_count > 10:
            # Investigate seller for review manipulation
            escalate_to_trust_and_safety({
                "seller_id": seller_id,
                "issue": "suspected_review_manipulation",
                "evidence": get_seller_fake_review_history(seller_id)
            })

        if seller_fake_review_count > 25:
            # Suspend seller account pending investigation
            suspend_seller_account(seller_id, reason="review_manipulation")

def adaptive_detection_improvement():
    """
    Weekly analysis of missed fake reviews
    """
    # Identify fake reviews that initially passed moderation
    # but were later reported/confirmed as fake
    missed_fakes = get_false_negatives(period="last_week")

    for review in missed_fakes:
        # Analyze why it passed initial moderation
        original_scores = get_original_moderation_scores(review)

        # Extract new patterns
        new_patterns = identify_new_tactics(review, original_scores)

        # Update detection models
        update_authenticity_detection(new_patterns)

        # Retroactively re-scan similar reviews
        similar_reviews = find_similar_historical_reviews(review)
        remoderate_batch(similar_reviews)