refactor: add NSFW and gibbreish words filtering

config.py

@@ -32,6 +32,16 @@ class Settings:
         for origin in os.getenv("CORS_ALLOWED_ORIGINS", "").split(",")
         if origin.strip()
     ]
+    local_arabic_nsfw_words: list[str] = [
+        w.strip()
+        for w in os.getenv("LOCAL_ARABIC_NSFW_WORDS", "").split(",")
+        if w.strip()
+    ]
+    local_nsfw_words: list[str] = [
+        w.strip()
+        for w in os.getenv("LOCAL_NSFW_WORDS", "").split(",")
+        if w.strip()
+    ]
 
 
 @lru_cache()

main.py

@@ -51,6 +51,13 @@ async def lifespan(app: FastAPI):
     except Exception as e:
         logger.warning(f"Embedder failed to load: {e}")
 
+    try:
+        from src.materials.validator import warmup_validation_models
+        warmup_validation_models()
+        logger.info("Validation models loaded and warmed up successfully.")
+    except Exception as e:
+        logger.warning(f"Validation models failed to load: {e}")
+
     from src.rag.batch_workers import start_workers
     start_workers()
 

materials/routes.py

@@ -233,7 +233,7 @@ async def rename_material_endpoint(
     return {"status": "ok"}
 
 
-from src.materials.validator import validate_topic_input
+from src.materials.validator import validate_topic_input_async
 
 @router.post("/topic")
 async def create_topic(
@@ -244,8 +244,8 @@ async def create_topic(
     if not topic_str:
         raise HTTPException(400, "Topic title cannot be empty")
         
-    # NSFW and gibberish validation
-    validation_res = validate_topic_input(topic_str)
+    # NSFW and gibberish validation using batch worker
+    validation_res = await validate_topic_input_async(topic_str)
     if validation_res != "ALLOWED":
         raise HTTPException(400, validation_res)
 

materials/validator.py

@@ -5,12 +5,11 @@ from src.config import settings
 
 logger = logging.getLogger(__name__)
 
-# Basic English NSFW word list to catch obvious cases instantly
-LOCAL_NSFW_WORDS = {
-    "porn", "sex", "nude", "bitch", "fuck", "asshole", "cunt", "dick",
-    "pussy", "nigger", "faggot", "bastard", "slut", "whore", "cock",
-    "boob", "tit", "vagina", "penis", "clitoris"
-}
+# Load English NSFW word list from config settings (kept out of committed code)
+LOCAL_NSFW_WORDS = set(settings.local_nsfw_words)
+
+# Load Arabic NSFW word list from config settings (kept out of committed code)
+LOCAL_ARABIC_NSFW_WORDS = set(settings.local_arabic_nsfw_words)
 
 def is_local_gibberish(text: str) -> bool:
     text_clean = text.strip().lower()
@@ -42,13 +41,25 @@ def is_local_gibberish(text: str) -> bool:
 def is_local_nsfw(text: str) -> bool:
     text_clean = text.strip().lower()
     
-    # Substring checks for high-signal NSFW roots
+    # Check Arabic NSFW words
+    words = text_clean.split()
+    for w in words:
+        if w in LOCAL_ARABIC_NSFW_WORDS:
+            return True
+            
+    # Substring checks for high-signal Arabic NSFW roots
+    arabic_substrings = {"سكس", "بورن", "شرموط", "منيوك", "قحبة", "قحبه", "متناك"}
+    for root in arabic_substrings:
+        if root in text_clean:
+            return True
+            
+    # Substring checks for high-signal English NSFW roots
     high_signal_substrings = {"porn", "nude", "sex", "vagina", "penis", "clitoris"}
     for root in high_signal_substrings:
         if root in text_clean:
             return True
             
-    # Check for direct matches or common word boundary matches
+    # Check for direct matches or common word boundary matches for English
     for word in LOCAL_NSFW_WORDS:
         # If it was already checked as a high-signal substring, skip
         if word in high_signal_substrings:
@@ -78,50 +89,162 @@ def is_local_nsfw(text: str) -> bool:
                 
     return False
 
-def validate_topic_input(topic: str) -> str:
+from transformers import pipeline
+
+_translator = None
+_gibberish_detector = None
+_nsfw_classifier = None
+
+def get_translator():
+    global _translator
+    if _translator is None:
+        logger.info("Loading translation model Helsinki-NLP/opus-mt-ar-en...")
+        _translator = pipeline("translation", model="Helsinki-NLP/opus-mt-ar-en", device="cpu")
+    return _translator
+
+def get_gibberish_detector():
+    global _gibberish_detector
+    if _gibberish_detector is None:
+        logger.info("Loading gibberish detector model madhurjindal/autonlp-Gibberish-Detector-492513457...")
+        _gibberish_detector = pipeline("text-classification", model="madhurjindal/autonlp-Gibberish-Detector-492513457", device="cpu")
+    return _gibberish_detector
+
+def get_nsfw_classifier():
+    global _nsfw_classifier
+    if _nsfw_classifier is None:
+        logger.info("Loading NSFW text classifier model michelleli99/NSFW_text_classifier...")
+        _nsfw_classifier = pipeline("text-classification", model="michelleli99/NSFW_text_classifier", device="cpu")
+    return _nsfw_classifier
+
+def is_arabic(text: str) -> bool:
+    return bool(re.search(r"[\u0600-\u06FF]", text))
+
+def validate_topics_batch(texts: list[str]) -> list[str]:
     """
-    Validates a topic string.
+    Validates a batch of topics.
+    For each topic text:
+      1. Detect if Arabic. If Arabic, translate it to English.
+      2. Run the gibberish detector model.
+      3. Run the NSFW text classifier model.
     Returns:
-        str: "ALLOWED" if the topic is valid, or a error message string if blocked.
+      A list of results: "ALLOWED", "gibberish words", or "not safe for work words".
     """
-    # 1. Quick Local Checks
-    if is_local_nsfw(topic):
-        return "NSFW words, profanity, or slang are not allowed."
+    logger.info(f"Validating batch of {len(texts)} topics...")
+    results = ["ALLOWED"] * len(texts)
+    
+    # 1. Quick local pre-checks
+    for i, text in enumerate(texts):
+        if is_local_nsfw(text):
+            results[i] = "not safe for work words"
+        elif is_local_gibberish(text):
+            results[i] = "gibberish words"
+            
+    # Collect indices of texts that passed local checks and need ML model check
+    pending_indices = [i for i, res in enumerate(results) if res == "ALLOWED"]
+    if not pending_indices:
+        return results
         
+    processed_texts = [texts[i] for i in pending_indices]
+    
+    # Translate Arabic inputs to English
+    arabic_indices = [i for i, idx in enumerate(pending_indices) if is_arabic(processed_texts[i])]
+    if arabic_indices:
+        arabic_texts = [processed_texts[i] for i in arabic_indices]
+        try:
+            translator = get_translator()
+            translations = translator(arabic_texts)
+            for idx, translation in zip(arabic_indices, translations):
+                translated_text = translation.get("translation_text", "").strip()
+                processed_texts[idx] = translated_text
+                logger.info(f"Translated Arabic topic '{texts[pending_indices[idx]]}' to '{translated_text}'")
+                
+                # Check translated text against local English NSFW list
+                if is_local_nsfw(translated_text):
+                    results[pending_indices[idx]] = "not safe for work words"
+        except Exception as e:
+            logger.error(f"Batch translation failed: {e}", exc_info=True)
+
+    # Run Gibberish detector on all processed texts
+    try:
+        gibberish_detector = get_gibberish_detector()
+        gibberish_preds = gibberish_detector(processed_texts)
+        for i, pred in enumerate(gibberish_preds):
+            actual_idx = pending_indices[i]
+            # Only set if it hasn't already been flagged by translation local NSFW check
+            if results[actual_idx] == "ALLOWED":
+                label = pred.get("label", "").lower()
+                if label in ("noise", "word salad"):
+                    results[actual_idx] = "gibberish words"
+    except Exception as e:
+        logger.error(f"Batch gibberish detection failed: {e}", exc_info=True)
+
+    # Run NSFW text classifier on all processed texts
+    try:
+        nsfw_classifier = get_nsfw_classifier()
+        nsfw_preds = nsfw_classifier(processed_texts)
+        for i, pred in enumerate(nsfw_preds):
+            actual_idx = pending_indices[i]
+            label = pred.get("label", "").lower()
+            if "nsfw" in label:
+                # NSFW always takes precedence over gibberish detection
+                results[actual_idx] = "not safe for work words"
+    except Exception as e:
+        logger.error(f"Batch NSFW classification failed: {e}", exc_info=True)
+
+    return results
+
+async def validate_topic_input_async(topic: str) -> str:
+    """
+    Asynchronously validates a topic string by routing it through the validation batch queue.
+    """
+    import uuid
+    from src.rag.batch_workers import validation_queue, validation_job_store
+    from src.rag.schemas import ValidationJob
+
+    job = ValidationJob(job_id=str(uuid.uuid4()), text=topic)
+    validation_job_store[job.job_id] = {"status": "pending", "result": None, "error": None}
+    await validation_queue.put(job)
+    await job.done.wait()
+
+    entry = validation_job_store.pop(job.job_id)
+    if entry["status"] == "error":
+        raise RuntimeError(f"Validation failed: {entry['error']}")
+
+    return entry["result"]
+
+def validate_topic_input(topic: str) -> str:
+    """
+    Synchronous validation fallback (e.g. for non-async contexts).
+    """
+    if is_local_nsfw(topic):
+        return "not safe for work words"
     if is_local_gibberish(topic):
-        return "Topic appears to be gibberish or meaningless text."
+        return "gibberish words"
+        
+    res = validate_topics_batch([topic])
+    return res[0]
 
-    # 2. LLM Check for multi-lingual and advanced cases
+def warmup_validation_models():
+    """
+    Dummy forward passes to warm up translation, gibberish detection, and NSFW classification models.
+    """
+    logger.info("Warming up translation and text classification models...")
     try:
-        llm = get_llm()
-        prompt = (
-            "You are a content filter for an educational application.\n"
-            f"Analyze the topic: \"{topic}\"\n\n"
-            "Determine if it contains:\n"
-            "1. NSFW content, profanity, swearing, slang insults, sexual references, or pornographic terms in ANY language.\n"
-            "2. Gibberish, random sequences of characters/numbers (e.g., \"12321321\", \"asdsaba\", \"aaaabbbb\", \"esaejsaioejasoi\").\n"
-            "3. Completely meaningless or troll input.\n\n"
-            "Respond in one of these two formats:\n"
-            "- If allowed: ALLOWED\n"
-            "- If blocked: BLOCKED: <reason in English>\n"
-            "Do not output any markdown, tags, or extra words. Just the raw text."
-        )
+        translator = get_translator()
+        translator("مرحبا")
+    except Exception as e:
+        logger.warning(f"Translation warmup failed: {e}")
         
-        response = llm.invoke(prompt)
-        result = response.content.strip()
+    try:
+        gibberish = get_gibberish_detector()
+        gibberish("hello")
+    except Exception as e:
+        logger.warning(f"Gibberish detector warmup failed: {e}")
         
-        if result == "ALLOWED":
-            return "ALLOWED"
-        elif result.startswith("BLOCKED:"):
-            reason = result.replace("BLOCKED:", "").strip()
-            return reason or "Topic is not allowed."
-        else:
-            # Fallback if the LLM output structure was unexpected
-            if "blocked" in result.lower():
-                return "Topic contains content that is not allowed."
-            return "ALLOWED"
-            
+    try:
+        nsfw = get_nsfw_classifier()
+        nsfw("hello")
     except Exception as e:
-        logger.error(f"LLM topic validation failed: {e}", exc_info=True)
-        # In case of API failure, fall back to allowing if it passed local checks
-        return "ALLOWED"
+        logger.warning(f"NSFW classifier warmup failed: {e}")
+    logger.info("Validation models warmup complete.")
+

rag/batch_workers.py

@@ -64,6 +64,9 @@ def is_request_in_flight() -> bool:
 
 embedding_queue: asyncio.Queue[EmbeddingJob] = asyncio.Queue()
 
+validation_queue: asyncio.Queue[Any] = asyncio.Queue()
+validation_job_store: dict[str, dict[str, Any]] = {}
+
 
 # ═══════════════════════ Workers ════════════════════════
 
@@ -150,6 +153,72 @@ async def embedding_worker():
             set_request_in_flight(False)
 
 
+async def validation_worker():
+    """
+    Drains up to {BATCH_MAX_SIZE} validation jobs every {BATCH_WINDOW_S * 1000:.0f}ms.
+    Runs batch validation using validate_topics_batch.
+    """
+    from src.materials.validator import validate_topics_batch
+
+    loop = asyncio.get_event_loop()
+
+    while True:
+        # Wait for at least one job
+        first_job = await validation_queue.get()
+        batch = [first_job]
+
+        # Collect up to 7 more within the time window
+        deadline = loop.time() + BATCH_WINDOW_S
+        while len(batch) < BATCH_MAX_SIZE:
+            remaining = deadline - loop.time()
+            if remaining <= 0:
+                break
+            try:
+                job = await asyncio.wait_for(validation_queue.get(), timeout=remaining)
+                batch.append(job)
+            except asyncio.TimeoutError:
+                break
+
+        try:
+            set_request_in_flight(True)
+
+            # Gather all texts from all jobs in the batch
+            all_texts = [job.text for job in batch]
+
+            # Single batch run for the entire batch
+            results = await loop.run_in_executor(
+                None, validate_topics_batch, all_texts
+            )
+
+            # Distribute results back to individual jobs
+            for i, job in enumerate(batch):
+                job_result = results[i]
+
+                if job.job_id not in validation_job_store:
+                    validation_job_store[job.job_id] = {"status": "pending", "result": None, "error": None}
+                
+                validation_job_store[job.job_id].update({
+                    "status": "done",
+                    "result": job_result
+                })
+                job.done.set()
+
+        except Exception as e:
+            logger.error(f"Validation batch failed: {e}", exc_info=True)
+            for job in batch:
+                if job.job_id not in validation_job_store:
+                    validation_job_store[job.job_id] = {"status": "error", "result": None, "error": str(e)}
+                else:
+                    validation_job_store[job.job_id].update({
+                        "status": "error",
+                        "error": str(e)
+                    })
+                if not job.done.is_set():
+                    job.done.set()
+        finally:
+            set_request_in_flight(False)
+
+
 # ═══════════════════════ Warmup Loop ════════════════════════
 
 
@@ -161,6 +230,7 @@ async def _warmup_loop():
     Skipped entirely if a real request is in flight.
     """
     from src.rag.rag import warmup_embedder
+    from src.materials.validator import warmup_validation_models
 
     loop = asyncio.get_event_loop()
 
@@ -171,6 +241,7 @@ async def _warmup_loop():
         t0 = time.monotonic()
         try:
             await loop.run_in_executor(None, warmup_embedder)
+            await loop.run_in_executor(None, warmup_validation_models)
         except Exception as e:
             logger.warning(f"Warmup cycle error (non-fatal): {e}")
             continue
@@ -188,6 +259,7 @@ def start_workers():
     Launch all async worker coroutines. Call once during app startup.
 
     - 1 embedding worker (batched SentenceTransformer inference)
+    - 1 validation worker (batched translation, gibberish, NSFW models validation)
     - 1 warmup loop (keeps OpenMP threads alive)
     """
     global _workers_started
@@ -197,6 +269,7 @@ def start_workers():
 
     # Use only 1 worker to save RAM on this environment
     asyncio.create_task(embedding_worker(), name="embedding_worker_0")
+    asyncio.create_task(validation_worker(), name="validation_worker_0")
     asyncio.create_task(_warmup_loop(), name="warmup_loop")
 
-    logger.info("Embedding batch workers started (1 worker + warmup loop)")
+    logger.info("Batch workers started (embedding worker + validation worker + warmup loop)")

rag/schemas.py

@@ -17,6 +17,15 @@ class EmbeddingJob:
     done: asyncio.Event = field(default_factory=asyncio.Event)
 
 
+@dataclass
+class ValidationJob:
+    """Batch validation of a topic text."""
+    job_id: str
+    text: str
+    done: asyncio.Event = field(default_factory=asyncio.Event)
+
+
+
 # Tutor query and response schemas
 class TutorQuery(BaseModel):
     query: str