materials/validator.py

import re
from loguru import logger
from src.rag.rag import get_llm
from src.config import settings

# 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_nsfw(text: str) -> bool:
    text_clean = text.strip().lower()
    
    # 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 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:
            continue
        pattern = rf"\b{re.escape(word)}\b"
        if re.search(pattern, text_clean):
            return True
        # Also check obfuscated variations like b**tch, f**k
        obfuscated = word[0] + r"\*+" + word[-1] if len(word) > 2 else ""
        if obfuscated and re.search(rf"\b{obfuscated}\b", text_clean):
            return True
            
    # Check common obfuscated patterns (e.g., f*ck, b*tch, f**k)
    # Match any word that contains asterisks inside it
    if "*" in text_clean:
        # Additional safety check for common swear structures
        for word in ["fuck", "bitch", "shit", "cunt", "asshole", "bastard"]:
            parts = list(word)
            pattern_parts = [parts[0]]
            for char in parts[1:-1]:
                # Allow the character itself, or one or more asterisks
                pattern_parts.append(rf"({re.escape(char)}|\*+)")
            pattern_parts.append(parts[-1])
            pattern = rf"\b{''.join(pattern_parts)}\b"
            if re.search(pattern, text_clean):
                return True
                
    return False

def validate_topics_batch(texts: list[str]) -> list[str]:
    """
    Validates a batch of topics.
    For each topic text:
      Checks against the local English and Arabic NSFW lists.
    Returns:
      A list of results: "ALLOWED" or "not safe for work words".
    """
    logger.info(f"Validating batch of {len(texts)} topics...")
    results = ["ALLOWED"] * len(texts)
    
    for i, text in enumerate(texts):
        if is_local_nsfw(text):
            results[i] = "not safe for work words"
            
    return results


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"
        
    return "ALLOWED"