Copied all files from local

Dockerfile

@@ -0,0 +1,16 @@
+FROM python:3.10
+
+WORKDIR /app
+
+COPY requirements.txt .
+
+RUN pip install -r requirements.txt
+
+COPY . .
+
+EXPOSE 3000
+
+CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "3000"]
+
+
+

api.py

@@ -0,0 +1,75 @@
+from fastapi import FastAPI
+from pydantic import BaseModel
+from model_loader import ModelLoader
+from services.text_filter import TextFilterService
+from services.image_ocr import ImageOCRService
+from typing import Optional
+from fastapi.responses import JSONResponse
+import logging
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    handlers=[
+        logging.StreamHandler()
+    ]
+)
+
+logger = logging.getLogger(__name__)
+
+app = FastAPI()
+logger.info("Starting FastAPI app...")
+
+model_loader = ModelLoader()
+logger.info("ModelLoader initialized.")
+
+text_filter_service = TextFilterService(model_loader)
+logger.info("TextFilterService initialized.")
+
+image_ocr_service = ImageOCRService()
+logger.info("Image OCR image initialized")
+
+
+class InputData(BaseModel):
+    text: Optional[str] = None
+    image_url: Optional[str] = None
+
+@app.post("/filtercomment")
+async def filter_comment(input_data: InputData):
+    logger.info("Received request: %s", input_data)
+    final_text = ""
+    # Case 1: Extract text from image
+    if input_data.image_url:
+        logger.info("Image URL provided: %s", input_data.image_url)
+        try:
+            logger.info("Fetching image from URL...")
+            final_text = image_ocr_service.extract_text(input_data.image_url)
+            logger.info("Generated text: %s", final_text)
+
+        except Exception as e:
+            logger.error("Image processing failed: %s", str(e))
+            return JSONResponse(status_code=400, content={"error": f"Image processing failed: {str(e)}"})
+
+    # Case 2: Use provided text
+    elif input_data.text:
+        logger.info("Text input provided.")
+        final_text = input_data.text
+    else:
+        logger.warning("No input provided.")
+        return JSONResponse(status_code=400, content={"error": "Either 'text' or 'image_url' must be provided."})
+
+    try:
+        logger.info("Processing text through TextFilterService...")
+        results = text_filter_service.process_text(final_text)
+        results["extracted_text"] = final_text
+        logger.info("Text filtering complete. Results: %s", results)
+        return results
+
+    except Exception as e:
+        logger.exception("Text filtering failed.")
+        return JSONResponse(status_code=500, content={"error": f"Text filtering failed: {str(e)}"})
+
+if __name__ == "__main__":
+    import uvicorn
+    logger.info("Starting Uvicorn server...")
+    uvicorn.run(app, host="0.0.0.0", port=3000)

model_loader.py

@@ -0,0 +1,19 @@
+from transformers import AutoTokenizer, AutoModelForSequenceClassification
+import torch
+class ModelLoader:
+    def __init__(self):
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        self._load_models()
+
+    def _load_models(self):
+        self.hf_model = AutoModelForSequenceClassification.from_pretrained("unitary/toxic-bert").to(self.device)
+        self.hf_tokenizer = AutoTokenizer.from_pretrained("unitary/toxic-bert")
+
+        self.identity_model = AutoModelForSequenceClassification.from_pretrained(
+            "Mridul2003/identity-hate-detector"
+        ).to(self.device)
+
+        try:
+            self.identity_tokenizer = AutoTokenizer.from_pretrained("Mridul2003/identity-hate-detector")
+        except Exception:
+            self.identity_tokenizer = self.hf_tokenizer

services/image_ocr.py

@@ -0,0 +1,21 @@
+from PIL import Image
+from io import BytesIO
+import requests
+from transformers import pipeline
+
+class ImageOCRService:
+    def __init__(self):
+        self.pipe = pipeline("image-text-to-text", model="ds4sd/SmolDocling-256M-preview")
+
+    def extract_text(self, image_url: str) -> str:
+        response = requests.get(image_url)
+        image = Image.open(BytesIO(response.content)).convert("RGB")
+
+        result = self.pipe(image, text=[{
+            "role": "user",
+            "content": [
+                {"type": "image", "url": image_url},
+                {"type": "text", "text": "extract text from image"}
+            ]
+        }])
+        return result[0]['generated_text'] if result else ""

services/text_filter.py

@@ -0,0 +1,42 @@
+import torch
+from model_loader import ModelLoader
+
+class TextFilterService:
+    def __init__(self, model_loader: ModelLoader):
+        self.model_loader = model_loader
+
+    def process_text(self, final_text: str) -> dict:
+        hf_tokenizer = self.model_loader.hf_tokenizer
+        hf_model = self.model_loader.hf_model
+        identity_model = self.model_loader.identity_model
+        identity_tokenizer = self.model_loader.identity_tokenizer
+        device = self.model_loader.device
+
+        # Toxic-BERT inference
+        hf_inputs = hf_tokenizer(final_text, return_tensors="pt", padding=True, truncation=True)
+        hf_inputs = {k: v.to(device) for k, v in hf_inputs.items()}
+        with torch.no_grad():
+            hf_outputs = hf_model(**hf_inputs)
+        hf_probs = torch.sigmoid(hf_outputs.logits)[0]
+        hf_labels = hf_model.config.id2label
+        results = {hf_labels.get(i, f"Label {i}"): float(prob) for i, prob in enumerate(hf_probs)}
+
+        # Identity hate classifier
+        identity_inputs = identity_tokenizer(final_text, return_tensors="pt", padding=True, truncation=True)
+        identity_inputs.pop("token_type_ids", None)
+        identity_inputs = {k: v.to(device) for k, v in identity_inputs.items()}
+        with torch.no_grad():
+            identity_outputs = identity_model(**identity_inputs)
+        identity_probs = torch.sigmoid(identity_outputs.logits)
+        identity_prob = identity_probs[0][1].item()
+        not_identity_prob = identity_probs[0][0].item()
+
+        results["identity_hate_custom"] = identity_prob
+        results["not_identity_hate_custom"] = not_identity_prob
+
+        results["safe"] = (
+            all(results.get(label, 0) < 0.5 for label in ['toxic', 'severe_toxic', 'obscene', 'insult', 'identity_hate']) 
+            and identity_prob < 0.5
+        )
+
+        return results