Update app.py

app.py

@@ -1,15 +1,15 @@
-# ─── flash_attn Mock يجب أن يكون أول شيء تماماً ─────────────────
 import sys
 import types
 import importlib.util
 
+# ─── Mock flash_attn (ضروري للتشغيل على CPU في Hugging Face) ───────
 flash_mock = types.ModuleType("flash_attn")
 flash_mock.__version__ = "2.0.0"
 flash_mock.__spec__ = importlib.util.spec_from_loader("flash_attn", loader=None)
 sys.modules["flash_attn"] = flash_mock
 sys.modules["flash_attn.flash_attn_interface"] = types.ModuleType("flash_attn.flash_attn_interface")
 sys.modules["flash_attn.bert_padding"] = types.ModuleType("flash_attn.bert_padding")
-# ─────────────────────────────────────────────────────────────────
+# ───────────────────────────────────────────────────────────────────
 
 import io
 import time
@@ -17,69 +17,48 @@ import torch
 from PIL import Image
 from transformers import AutoProcessor, AutoModelForCausalLM
 from fastapi import FastAPI, HTTPException, UploadFile, File
+from fastapi.responses import JSONResponse
 from contextlib import asynccontextmanager
 
-# ─── السؤال المباشر للنموذج ───────────────────────────────────────
-VQA_QUESTION = (
-    "Is there a woman or any part of a woman's body in this image? "
-    "Answer yes or no only."
-)
+# السؤال الموجه للنموذج (تم تحسينه ليكون أكثر دقة)
+VQA_QUESTION = "Is there a woman, or any female body part like hair or skin in this image? Answer yes or no."
 
-# ─── تحميل النموذج ───────────────────────────────────────────────
 MODEL_DATA = {}
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     print("📥 Loading Florence-2-large...")
-    start = time.time()
-
     MODEL_DATA["processor"] = AutoProcessor.from_pretrained(
-        "microsoft/Florence-2-large",
+        "microsoft/Florence-2-large", 
         trust_remote_code=True
     )
     MODEL_DATA["model"] = AutoModelForCausalLM.from_pretrained(
-        "microsoft/Florence-2-large",
-        torch_dtype=torch.float32,
-        trust_remote_code=True,
+        "microsoft/Florence-2-large", 
+        trust_remote_code=True, 
         attn_implementation="eager"
     ).eval()
-
-    print(f"✅ Model ready in {time.time()-start:.1f}s")
+    print("✅ Model is ready!")
     yield
     MODEL_DATA.clear()
 
-# ─── FastAPI ──────────────────────────────────────────────────────
-app = FastAPI(
-    title="Female Detection API - VQA",
-    description="Detects women in images using Florence-2-large VQA",
-    version="2.0.0",
-    lifespan=lifespan
-)
-
-@app.get("/health")
-def health():
-    return {"status": "ok", "model_loaded": "model" in MODEL_DATA}
+app = FastAPI(title="Privacy Guard VQA", lifespan=lifespan)
 
-# ─── POST /analyze ────────────────────────────────────────────────
 @app.post("/analyze")
 async def analyze_image(file: UploadFile = File(...)):
-
     if not file.content_type.startswith("image/"):
-        raise HTTPException(status_code=400, detail="الملف المرفوع ليس صورة")
+        raise HTTPException(status_code=400, detail="File is not an image")
 
     try:
+        # قراءة ومعالجة الصورة
         image_bytes = await file.read()
         image = Image.open(io.BytesIO(image_bytes)).convert("RGB")
-    except Exception as e:
-        raise HTTPException(status_code=400, detail=f"خطأ في قراءة الصورة: {str(e)}")
-
-    try:
+        
         processor = MODEL_DATA["processor"]
-        model     = MODEL_DATA["model"]
+        model = MODEL_DATA["model"]
 
-        task   = "<VQA>"
+        # تحضير المدخلات مع مهمة VQA
+        task = "<VQA>"
         prompt = f"{task}{VQA_QUESTION}"
-
         inputs = processor(text=prompt, images=image, return_tensors="pt")
 
         start_time = time.time()
@@ -87,36 +66,41 @@ async def analyze_image(file: UploadFile = File(...)):
             generated_ids = model.generate(
                 input_ids=inputs["input_ids"],
                 pixel_values=inputs["pixel_values"],
-                max_new_tokens=10,
-                num_beams=1,
-                do_sample=False
+                max_new_tokens=50,
+                num_beams=3 # رفع الـ beams يحسن جودة "التفكير" في السؤال
             )
 
-        # ✅ الطريقة الصحيحة: نفس أسلوب باقي المهام
+        # فك التشفير الخام
         generated_text = processor.batch_decode(generated_ids, skip_special_tokens=False)[0]
-        parsed = processor.post_process_generation(
-            generated_text,
-            task=task,
+        
+        # المعالجة البعدية الذكية (هذه هي الخطوة التي كانت تنقصك)
+        parsed_answer = processor.post_process_generation(
+            generated_text, 
+            task=task, 
             image_size=(image.width, image.height)
         )
-
-        # parsed يعيد dict مثل: {"<VQA>": "yes"} أو {"<VQA>": "no"}
-        raw_answer = parsed.get(task, "").strip().lower()
-        elapsed    = round(time.time() - start_time, 2)
-
-        decision = "block" if "yes" in raw_answer else "allow"
+        
+        # استخراج الإجابة النظيفة (yes أو no)
+        clean_answer = parsed_answer[task].strip().lower()
+        
+        # اتخاذ القرار
+        # نستخدم "yes" كشرط للحجب، وأي شيء آخر (أو إذا ارتبك النموذج) نعتبره "allow" 
+        # أو يمكنك عكس المنطق ليكون أكثر تشدداً
+        decision = "block" if "yes" in clean_answer else "allow"
 
         return {
-            "decision":       decision,
-            "vqa_answer":     raw_answer,
-            "question":       VQA_QUESTION,
-            "execution_time": elapsed,
-            "status":         "success"
+            "decision": decision,
+            "vqa_answer": clean_answer,
+            "execution_time": round(time.time() - start_time, 2),
+            "status": "success"
         }
 
     except Exception as e:
-        raise HTTPException(status_code=500, detail=f"خطأ في الاستدلال: {str(e)}")
+        return JSONResponse(status_code=500, content={"error": str(e)})
 
+@app.get("/")
+def home():
+    return {"status": "running", "question_active": VQA_QUESTION}
 
 if __name__ == "__main__":
     import uvicorn