add app

app/api/v1/api.py

@@ -0,0 +1,6 @@
+from fastapi import APIRouter
+from app.api.v1.endpoints import moderation
+
+api_router = APIRouter()
+
+api_router.include_router(moderation.router, prefix="/moderation", tags=["Moderation"])

app/api/v1/endpoints/moderation.py

@@ -0,0 +1,13 @@
+from fastapi import APIRouter
+
+from app.models.schemas import ModerationRequest, ModerationResponse
+from app.services.moderation import moderate_content
+
+router = APIRouter()
+
+
+@router.post("/", response_model=ModerationResponse)
+def testing(request: ModerationRequest):
+    response = moderate_content(request)
+
+    return response

app/core/config.py

@@ -0,0 +1,16 @@
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+class Settings:
+    def __init__(self)->None:
+        self.gemini_apikey = None
+        self.perspective_apikey = None
+
+    def config(self)->None:
+        self.gemini_apikey = os.environ.get("GEMINI_API_KEY_1")
+        self.perspective_apikey = os.environ.get("PERSPECTIVE_API_KEY_1")
+    
+
+settings = Settings()
+settings.config()

app/main.py

@@ -0,0 +1,33 @@
+from fastapi import FastAPI, Response
+from app.api.v1.api import api_router
+from fastapi.middleware.cors import CORSMiddleware
+
+from app.models.schemas import StandardResponse
+
+app = FastAPI(title="Cognisafe API")
+
+origins = ["*"]
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,  # or ["*"] for all origins (not recommended in prod)
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.get("/health")
+async def health_check():
+    return {"status": "healthy"}
+
+@app.get("/helloworld", response_model=StandardResponse)
+def helloworld(response: Response) -> StandardResponse:
+    """
+    Returns helloworld as the standard response
+    """
+    response.status_code = 200
+    response = StandardResponse(error=False, title="Hello World", status=200)
+    return response
+
+
+app.include_router(api_router, prefix="/api/v1")

app/ml_models/classifier.py

@@ -0,0 +1,41 @@
+from transformers import pipeline
+import logging
+
+
+from app.ml_models.classifier_loader import ClassifierLoader
+
+
+logger = logging.getLogger(__name__)
+
+
+class Classifier:
+    def __init__(self, model_name: str = "toxic-bert") -> None:
+        self.model = None
+        self.tokenizer = None
+        self.model_name = model_name
+        self.classifier = None
+
+    def initialize_classifier(self) -> None:
+        loader = ClassifierLoader(self.model_name)
+        self.model = loader.load_model()
+        self.tokenizer = loader.load_tokenizer()
+        self.classifier = pipeline(
+            "text-classification",
+            model=self.model,
+            tokenizer=self.tokenizer,
+            device=-1,
+            top_k=None,
+        )
+
+    def predict_nsfw(self, content: str) -> dict:
+        if self.classifier is None:
+            raise RuntimeError(
+                "Model not initialized. Please call `initialize_classifier()` first."
+            )
+        results = self.classifier(content)
+
+        prediction = {}
+        for result in results[0]:
+            prediction[result["label"]] = result["score"]
+        
+        return prediction

app/ml_models/classifier_loader.py

@@ -0,0 +1,42 @@
+from transformers import AutoTokenizer, AutoModelForSequenceClassification
+from app.ml_models.classifier_path_loader import ClassifierPathLoader
+import logging
+from transformers import AutoTokenizer
+from transformers import AutoModelForSequenceClassification, pipeline
+
+
+logger = logging.getLogger(__name__)
+
+
+class ClassifierLoader:
+    def __init__(self, model_name: str):
+        self.model_name = model_name
+        self.model = None
+        self.tokenizer = None
+
+        path_loader = ClassifierPathLoader()
+        path_loader.set_model(self.model_name)
+        self.model_path = path_loader.get_model_path()
+
+        # If model doesn't exist, download it
+        if not self.model_path.exists():
+            model_name = "unitary/toxic-bert"
+            tokenizer = AutoTokenizer.from_pretrained(model_name)
+            model = AutoModelForSequenceClassification.from_pretrained(model_name)
+
+            tokenizer.save_pretrained(self.model_path)
+            model.save_pretrained(self.model_path)
+
+    def load_model(self):
+        if self.model is None:
+            self.model = AutoModelForSequenceClassification.from_pretrained(
+                self.model_path
+            )
+            logger.info("[✅] Model loaded successfully.")
+        return self.model
+
+    def load_tokenizer(self):
+        if self.tokenizer is None:
+            self.tokenizer = AutoTokenizer.from_pretrained(self.model_path)
+            logger.info("[✅] Tokenizer loaded successfully.")
+        return self.tokenizer

app/ml_models/classifier_path_loader.py

@@ -0,0 +1,20 @@
+from pathlib import Path
+
+path_to_model = {"toxic-bert": Path.cwd() / "app" / "ml_models" / "toxic-bert"}
+
+
+class ClassifierPathLoader:
+    def __init__(self) -> None:
+        self.model_name = None
+        self.model_path = None
+
+    def set_model(self, model_name: str) -> None:
+        if model_name not in path_to_model:
+            raise KeyError(f"Model '{model_name}' not found in path registry.")
+        self.model_name = model_name
+        self.model_path = Path(path_to_model[model_name])
+
+    def get_model_path(self) -> Path:
+        if self.model_path is None:
+            raise RuntimeError("Model not set. Call `set_model()` first.")
+        return self.model_path

app/ml_models/gemini_moderator.py

@@ -0,0 +1,51 @@
+from google import genai
+from pprint import pprint
+from app.core.config import settings
+from app.services.gemini_error_handling import handle_model_error
+
+class GeminiModerator:
+    def __init__(self):
+        self.client = None
+
+    def intitialize_for_cleansing(self) -> None:
+        client = genai.Client(api_key=settings.gemini_apikey)
+        self.client = client
+
+    def initialize_for_misinfo_detection(self) -> None:
+        client = genai.Client(api_key=settings.gemini_apikey)
+        self.client = client
+
+    def check_misinfo(self, content: str) -> str:
+        return self.prompt_model(content, 1) # passing 1 for misinformation detection
+
+    def cleanse(self, content: str) -> str:
+        return self.prompt_model(content, 0) # passing 0 for cleansing hatespeech
+
+    def prompt_model(self, content: str, purpose: int) -> str:
+        prompt_message = [
+            f'You are a content moderation assistant. The following text has been flagged for harmful content. Your task is to clean it by removing or replacing inappropriate words while keeping the meaning intact. Please return the cleaned version of the text. Content: "{content}". I repeat return only the clean version, without any explanation.',
+            f"You are a content moderation assistant. Check if the following content contains any factual inaccuracies. Your task is to identify any statements that contradict established facts or lack evidence. If there is misinformation present, the response must start with 'False' followed by the corrected information or explanation of the inaccuracy. If the content is factually accurate, the response must start with 'True'. Do not flag content as 'False' simply because it is harmful, hateful, or threatening if the claims made are factually correct. Focus solely on the truthfulness of the statements. Content:\"{content}\".",
+        ]
+
+        try:
+            response = self.client.models.generate_content(
+                model="gemini-2.0-flash", contents=prompt_message[purpose]
+            )
+
+            print(response)
+            block_reason = getattr(
+                getattr(response, "prompt_feedback", None), "block_reason", None
+            )
+            if block_reason == "PROHIBITED_CONTENT":
+                return "I have anger issues. I will take a deep breath"
+
+            return response.text
+        except Exception as e:
+            print(f"Something went wrong while prompting: {e}")
+            handle_model_error(e.code, content)
+
+
+gemini_for_cleansing = GeminiModerator()
+gemini_for_cleansing.intitialize_for_cleansing()
+gemini_for_misinfo_detection = GeminiModerator()
+gemini_for_misinfo_detection.initialize_for_misinfo_detection()

app/ml_models/toxic-bert/config.json

@@ -0,0 +1,42 @@
+{
+  "architectures": [
+    "BertForSequenceClassification"
+  ],
+  "attention_probs_dropout_prob": 0.1,
+  "classifier_dropout": null,
+  "gradient_checkpointing": false,
+  "hidden_act": "gelu",
+  "hidden_dropout_prob": 0.1,
+  "hidden_size": 768,
+  "id2label": {
+    "0": "toxic",
+    "1": "severe_toxic",
+    "2": "obscene",
+    "3": "threat",
+    "4": "insult",
+    "5": "identity_hate"
+  },
+  "initializer_range": 0.02,
+  "intermediate_size": 3072,
+  "label2id": {
+    "identity_hate": 5,
+    "insult": 4,
+    "obscene": 2,
+    "severe_toxic": 1,
+    "threat": 3,
+    "toxic": 0
+  },
+  "layer_norm_eps": 1e-12,
+  "max_position_embeddings": 512,
+  "model_type": "bert",
+  "num_attention_heads": 12,
+  "num_hidden_layers": 12,
+  "pad_token_id": 0,
+  "position_embedding_type": "absolute",
+  "problem_type": "multi_label_classification",
+  "torch_dtype": "float32",
+  "transformers_version": "4.51.3",
+  "type_vocab_size": 2,
+  "use_cache": true,
+  "vocab_size": 30522
+}

app/ml_models/toxic-bert/model.safetensors

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:c155addfd79483b0a75955c49b6d42508e26ecd72b9600fb092205d9990df577
+size 437970952

app/ml_models/toxic-bert/special_tokens_map.json

@@ -0,0 +1,7 @@
+{
+  "cls_token": "[CLS]",
+  "mask_token": "[MASK]",
+  "pad_token": "[PAD]",
+  "sep_token": "[SEP]",
+  "unk_token": "[UNK]"
+}

app/ml_models/toxic-bert/tokenizer_config.json

@@ -0,0 +1,58 @@
+{
+  "added_tokens_decoder": {
+    "0": {
+      "content": "[PAD]",
+      "lstrip": false,
+      "normalized": false,
+      "rstrip": false,
+      "single_word": false,
+      "special": true
+    },
+    "100": {
+      "content": "[UNK]",
+      "lstrip": false,
+      "normalized": false,
+      "rstrip": false,
+      "single_word": false,
+      "special": true
+    },
+    "101": {
+      "content": "[CLS]",
+      "lstrip": false,
+      "normalized": false,
+      "rstrip": false,
+      "single_word": false,
+      "special": true
+    },
+    "102": {
+      "content": "[SEP]",
+      "lstrip": false,
+      "normalized": false,
+      "rstrip": false,
+      "single_word": false,
+      "special": true
+    },
+    "103": {
+      "content": "[MASK]",
+      "lstrip": false,
+      "normalized": false,
+      "rstrip": false,
+      "single_word": false,
+      "special": true
+    }
+  },
+  "clean_up_tokenization_spaces": true,
+  "cls_token": "[CLS]",
+  "do_basic_tokenize": true,
+  "do_lower_case": true,
+  "extra_special_tokens": {},
+  "mask_token": "[MASK]",
+  "model_max_length": 512,
+  "never_split": null,
+  "pad_token": "[PAD]",
+  "sep_token": "[SEP]",
+  "strip_accents": null,
+  "tokenize_chinese_chars": true,
+  "tokenizer_class": "BertTokenizer",
+  "unk_token": "[UNK]"
+}

app/models/check_type.py

@@ -0,0 +1,11 @@
+from enum import Enum
+
+
+class CheckType(str, Enum):
+    toxic = "toxic"
+    severe_toxic = "severe_toxic"
+    insult = "insult"
+    obscene = "obscene"
+    threat = "threat"
+    identity_hate = "identity_hate"
+    misinfo = "misinfo"

app/models/moderation_data.py

@@ -0,0 +1,12 @@
+from pydantic import BaseModel
+from typing import List
+from typing import Any, Dict
+from app.models.check_type import CheckType
+
+
+class ModerationData(BaseModel):
+    flagged_for: Dict[CheckType, float]
+    scores: Dict[CheckType, float]
+    original_content: str
+    cleaned_content: str = None
+    corrected_content: Any

app/models/moderation_request.py

@@ -0,0 +1,32 @@
+from pydantic import BaseModel
+from typing import List
+
+from app.models.check_type import CheckType
+from app.services.text_cleaner import clean_text
+
+from app.ml_models.classifier import Classifier
+from app.ml_models.gemini_moderator import gemini_for_cleansing
+from app.ml_models.gemini_moderator import gemini_for_misinfo_detection
+
+
+classifier = Classifier()
+classifier.initialize_classifier()
+
+
+class ModerationRequest(BaseModel):
+    content: str
+    checkFor: List[CheckType]
+    threshold: float = 0.6
+
+    def correct_typos(self) -> None:
+        self.content = clean_text(self.content)
+
+    def classify_moderation(self) -> None:
+        result = classifier.predict_nsfw(self.content)
+        return result
+
+    def cleanse_content(self) -> str:
+        return gemini_for_cleansing.cleanse(self.content)
+
+    def identify_misinfo(self) -> str:
+        return gemini_for_misinfo_detection.check_misinfo(self.content)

app/models/moderation_response.py

@@ -0,0 +1,6 @@
+from app.models.standard_response import StandardResponse
+from app.models.moderation_data import ModerationData
+
+
+class ModerationResponse(StandardResponse):
+    payload: ModerationData  # override with specific structure

app/models/schemas.py

@@ -0,0 +1,5 @@
+from app.models.check_type import CheckType
+from app.models.moderation_data import ModerationData
+from app.models.moderation_request import ModerationRequest
+from app.models.moderation_response import ModerationResponse
+from app.models.standard_response import StandardResponse

app/models/standard_response.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel
+
+
+class StandardResponse(BaseModel):
+    error: bool
+    title: str
+    status: int
+    payload: None  # generic

app/services/gemini_error_handling.py

@@ -0,0 +1,11 @@
+def handle_model_error(code: int, content: str) -> str:
+    response = switch_case(code)
+    return response
+
+
+def switch_case(code: int) -> str:
+    match code:
+        case 503:
+            return "Model is overloaded. Please try again later"
+        case _:
+            return f"Something went wrong while prompting: {code}"

app/services/moderation.py

@@ -0,0 +1,56 @@
+from fastapi.responses import JSONResponse
+from google import genai
+
+# from app.models.moderation_data import ModerationData
+from app.models.moderation_response import ModerationResponse
+from app.models.schemas import ModerationRequest
+from app.models.standard_response import StandardResponse
+from app.models.schemas import ModerationData
+
+
+def to_json_response(data: StandardResponse) -> JSONResponse:
+    return JSONResponse(content=data.model_dump(), status_code=data.status)
+
+
+def moderate_content(request: ModerationRequest) -> ModerationResponse:
+    flagged_for = {}
+    scores = {}
+    corrected_content = None
+    original_content = request.content
+
+    if "misinfo" in request.checkFor:
+        correct_info = request.identify_misinfo()
+        if correct_info.startswith("False"):
+            flagged_for["misinfo"] = 1
+            corrected_content = str(correct_info[6:])
+
+    request.correct_typos()
+    result = request.classify_moderation()
+    result["misinfo"] = flagged_for.get("misinfo", 0)
+
+    for category in request.checkFor:
+        scores[category] = result[category]
+        if result[category] > request.threshold:
+            flagged_for[category] = result[category]
+
+    cleaned_content = request.content
+    if len(flagged_for) > 0:
+        if list(flagged_for.keys()) == ["misinfo"]:
+            # Do nothing
+            pass
+        else:
+            gemini_response = request.cleanse_content()
+            cleaned_content = gemini_response
+
+    payload = ModerationData(
+        flagged_for=flagged_for,
+        cleaned_content=cleaned_content,
+        corrected_content=corrected_content,
+        original_content=original_content,
+        scores=scores,
+    )
+    response = ModerationResponse(
+        error=False, title="Cleaned", status=200, payload=payload
+    )
+
+    return response

app/services/text_cleaner.py

@@ -0,0 +1,22 @@
+# app/services/text_cleaner.py
+from symspellpy.symspellpy import SymSpell
+from pathlib import Path
+
+# Initialize only once when this module is imported
+sym_spell = SymSpell(max_dictionary_edit_distance=2, prefix_length=7)
+
+# Load dictionary
+dict_path = Path.cwd()/"app"/"services"/"frequency_dictionary_en_82_765.txt"
+sym_spell.load_dictionary(dict_path, term_index=0, count_index=1)
+
+# Leetspeak normalizer
+def leetspeak_normalizer(text: str) -> str:
+    leet_map = str.maketrans("014!3$@5#+", "oialesasht")
+    return text.translate(leet_map)
+
+# Combined cleaning function
+def clean_text(text: str) -> str:
+    normalized = leetspeak_normalizer(text)
+    suggestions = sym_spell.lookup_compound(normalized, max_edit_distance=2)
+    corrected = suggestions[0].term if suggestions else normalized
+    return corrected