Create app.py

app.py

@@ -0,0 +1,125 @@
+import cv2
+import torch
+import requests
+from PIL import Image
+from fastapi import FastAPI, Header, HTTPException
+from supabase import create_client
+from transformers import AutoImageProcessor, AutoModelForImageClassification
+import os
+
+app = FastAPI()
+
+# ===============================
+# सुरक्षा (API Key Protection)
+# ===============================
+API_TOKEN = os.getenv("API_TOKEN")
+
+def verify_key(x_api_key: str):
+    if API_TOKEN and x_api_key != API_TOKEN:
+        raise HTTPException(status_code=403, detail="Unauthorized")
+
+# ===============================
+# Supabase
+# ===============================
+SUPABASE_URL = os.getenv("SUPABASE_URL")
+SUPABASE_KEY = os.getenv("SUPABASE_KEY")
+
+supabase = create_client(SUPABASE_URL, SUPABASE_KEY)
+
+# ===============================
+# Load Model (only once)
+# ===============================
+model_name = "AdamCodd/vit-base-nsfw-detector"
+
+processor = AutoImageProcessor.from_pretrained(model_name)
+model = AutoModelForImageClassification.from_pretrained(model_name)
+
+# ===============================
+# Image check
+# ===============================
+def check_image(url):
+    try:
+        image = Image.open(requests.get(url, stream=True).raw).convert("RGB")
+        inputs = processor(images=image, return_tensors="pt")
+
+        with torch.no_grad():
+            outputs = model(**inputs)
+
+        probs = torch.softmax(outputs.logits, dim=1)
+        return "explicit" if probs[0][1] > 0.5 else "safe"
+    except:
+        return "safe"
+
+# ===============================
+# Video check
+# ===============================
+def check_video(url, frame_sample_rate=30):
+    try:
+        cap = cv2.VideoCapture(url)
+        frame_count = 0
+
+        while cap.isOpened():
+            ret, frame = cap.read()
+            if not ret:
+                break
+
+            if frame_count % frame_sample_rate == 0:
+                img = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
+                inputs = processor(images=img, return_tensors="pt")
+
+                with torch.no_grad():
+                    outputs = model(**inputs)
+
+                probs = torch.softmax(outputs.logits, dim=1)
+
+                if probs[0][1] > 0.5:
+                    cap.release()
+                    return "explicit"
+
+            frame_count += 1
+
+        cap.release()
+        return "safe"
+
+    except:
+        return "safe"
+
+# ===============================
+# 🔥 MAIN ENDPOINT (like /recommend/all)
+# ===============================
+@app.get("/moderate/all")
+async def moderate_all(x_api_key: str = Header(None)):
+
+    verify_key(x_api_key)
+
+    results = []
+
+    # Images
+    img_posts = supabase.table("posts").select("*").execute().data
+    for post in img_posts:
+        result = check_image(post["image_url"])
+
+        results.append({
+            "id": post["id"],
+            "type": "image",
+            "url": post["image_url"],
+            "result": result
+        })
+
+    # Videos
+    vid_posts = supabase.table("trendz").select("*").execute().data
+    for post in vid_posts:
+        result = check_video(post["video_url"])
+
+        results.append({
+            "id": post["id"],
+            "type": "video",
+            "url": post["video_url"],
+            "result": result
+        })
+
+    return {
+        "status": "success",
+        "count": len(results),
+        "data": results
+    }