""" Synthetic & Hugging Face content case generator for ContentGuardEnv. Generates realistic social media posts with associated metadata, user account history, and ground-truth moderation decisions. Integrates natively with Hugging Face datasets. By default, external dataset loading is disabled to keep startup/test paths deterministic. Enable with CONTENT_GUARD_USE_HF_DATA=1. """ import os import random import copy from typing import Any, Dict, Tuple import logging logger = logging.getLogger(__name__) HF_DATASET = None _HF_DATA_ATTEMPTED = False _HF_DATA_ENABLED = os.getenv("CONTENT_GUARD_USE_HF_DATA", "0").strip().lower() in {"1", "true", "yes", "on"} try: from datasets import load_dataset as _hf_load_dataset except Exception: _hf_load_dataset = None def _maybe_load_hf_data() -> None: """Lazy one-time dataset initialization to avoid import-time blocking.""" global HF_DATASET, _HF_DATA_ATTEMPTED if _HF_DATA_ATTEMPTED: return _HF_DATA_ATTEMPTED = True if not _HF_DATA_ENABLED: logger.info("Hugging Face dataset disabled (CONTENT_GUARD_USE_HF_DATA=0). Using synthetic cases.") return if _hf_load_dataset is None: logger.warning("datasets package unavailable; using synthetic cases.") return try: logger.info("Initializing Hugging Face Datasets connection...") # Keep dataset bounded for memory and faster cold starts. HF_DATASET = _hf_load_dataset("SetFit/toxic_conversations", split="train[:1000]") logger.info(f"Loaded {len(HF_DATASET)} Hugging Face evaluation records.") except Exception as e: logger.warning(f"Could not load Hugging Face dataset (falling back to synthetic): {str(e)}") HF_DATASET = None # ── Post templates (Fallback / Structure) ────────────────────────────────────── POSTS = [ { "violation": "hate_speech", "severity": 5, "action": "remove", "ruling": "upheld", "policy_references": ["Hate Speech Policy", "Community Standards Section 12"], "content_variants": [ "People from [COUNTRY] are all criminals and shouldn't be allowed here. Send them back where they came from.", ], "platform": "social_feed", "has_image": False, "account_age_days_range": (1, 90), "prior_violations_range": (0, 3), "appeal_message": "I was just expressing my political opinion about immigration policy. This is free speech.", }, { "violation": "spam", "severity": 2, "action": "restrict_reach", "ruling": "upheld", "policy_references": ["Spam and Fake Engagement Policy"], "content_variants": [ "💰 MAKE $5000/WEEK FROM HOME!! No experience needed! DM me NOW for the secret link! Limited spots!! 💰💰💰", ], "platform": "comment_section", "has_image": False, "account_age_days_range": (1, 30), "prior_violations_range": (1, 8), "appeal_message": "This is a legitimate business promotion. I'm just trying to make a living.", }, { "violation": "safe", "severity": 1, "action": "no_action", "ruling": "overturned", "policy_references": [], "content_variants": [ "Unpopular opinion: pineapple on pizza is actually really good and anyone who disagrees is wrong 🍍🍕", ], "platform": "public_post", "has_image": False, "account_age_days_range": (100, 3000), "prior_violations_range": (0, 1), "appeal_message": "Thank you, I knew this was wrongly removed. This is normal discussion.", } ] PLATFORMS = ["social_feed", "public_post", "comment_section", "private_group", "marketplace", "stories"] COUNTRIES = ["Brazil", "India", "Germany", "Nigeria", "Indonesia", "USA", "France"] DEVICE_TYPES = ["mobile_ios", "mobile_android", "desktop_web", "desktop_app"] def get_real_hf_case() -> Tuple[str, str, int, str]: """Pulls a random record from the real Hugging Face dataset.""" if HF_DATASET is None or len(HF_DATASET) == 0: raise ValueError("HF dataset is not initialized.") record = HF_DATASET[random.randrange(len(HF_DATASET))] # SetFit/toxic_conversations uses 'text' and 'label' (0=safe, 1=toxic) content = record.get("text", "") is_toxic = record.get("label", 0) == 1 if is_toxic: # Heavily simplify the map for the demo violation_type = random.choice(["hate_speech", "harassment", "violence", "misinformation"]) severity = random.randint(3, 5) action = "remove" if severity == 5 else "add_warning_label" else: violation_type = "safe" severity = 1 action = "no_action" return content, violation_type, severity, action def generate_case(task_id: str) -> Tuple[Dict[str, Any], Dict[str, Any]]: """Generate a content moderation case leveraging real HF data if available.""" # 1. Base Meta structure template = copy.deepcopy(random.choice(POSTS)) _maybe_load_hf_data() # 2. Content & Ground Truth assignment if HF_DATASET is not None and len(HF_DATASET) > 0: # Real AI Training Data Path content, violation, severity, action = get_real_hf_case() template["violation"] = violation template["severity"] = severity template["action"] = action if violation == "safe": template["ruling"] = "overturned" template["policy_references"] = [] else: template["ruling"] = "upheld" template["policy_references"] = [f"{violation.capitalize()} Policy"] else: # Synthetic fallback content = random.choice(template["content_variants"]) account_age = random.randint(1, 3000) prior_violations = random.randint(0, 5) case: Dict[str, Any] = { "post_id": f"POST-{random.randint(100000, 999999)}", "content": content[:800], # Cap length for UI aesthetics "platform": random.choice(PLATFORMS), "has_media_attachment": random.random() > 0.5, "language": "en", "user_account": { "account_age_days": account_age, "follower_count": random.randint(0, 50000), "is_verified": random.random() < 0.05, "prior_violations": prior_violations, "account_type": random.choice(["personal", "page", "business"]), "region": random.choice(COUNTRIES), "trust_score": round(random.uniform(0.1, 1.0), 2), # New metric for agent evaluation }, "engagement": { "likes": random.randint(0, 5000), "shares": random.randint(0, 2000), "comments": random.randint(0, 800), "reports_received": random.randint(1, 150), "community_impact_score": round(random.uniform(1.0, 10.0), 1), # Creativity: Novel metric "viral_potential": random.choice(["Low", "Medium", "High", "Critical"]), # Predictive moderation signal }, "flags": { "auto_detected": random.random() < 0.7, "user_reported": random.random() < 0.9, "repeat_offender": prior_violations >= 3, }, "device": random.choice(DEVICE_TYPES), } if task_id in ("medium", "hard"): case["detected_violation"] = template["violation"] if task_id == "hard": case["action_taken"] = template["action"] case["user_appeal"] = random.choice([ "I believe this was moderated in error. Please restore my post.", "My political views are being silenced! This doesn't violate any rules.", "I see others posting the same thing, why was mine removed?", "It was a joke for my friends, please help." ]) # Ground truth if task_id == "easy": ground_truth = {"violation": template["violation"]} elif task_id == "medium": ground_truth = { "action": template["action"], "severity": template["severity"], "violation": template["violation"], } else: ground_truth = { "ruling": template["ruling"], "policy_references": template["policy_references"], "violation": template["violation"], "action": template["action"], } return case, ground_truth