Update app.py

app.py

@@ -1,6 +1,16 @@
+# ─── flash_attn Mock يجب أن يكون أول شيء ───────────────────────
 import sys
 import types
 import importlib.util
+
+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
 import torch
@@ -10,77 +20,106 @@ from fastapi import FastAPI, HTTPException, UploadFile, File
 from fastapi.responses import JSONResponse
 from contextlib import asynccontextmanager
 
-# --- Mock flash_attn (نفس خدعتك للـ CPU) ---
-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")
-
-# --- القائمة المخصصة للجندر (تجنب حجب الرجال) ---
-# أضفنا وصف "female" و "woman" لتقييد البحث
-FORBIDDEN_LABELS = "woman, girl, lady, female hair, female skin, female arm, female hand, female leg, female chest"
+# ─── القائمة الأنثوية ────────────────────────────────────────────
+FORBIDDEN_LABELS = (
+    "woman, girl, lady, female, female hair, female skin, "
+    "female arm, female hand, female leg, female chest, "
+    "woman face, girl face"
+)
 
+# ─── تحميل النموذج ───────────────────────────────────────────────
 MODEL_DATA = {}
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    print("📥 Loading Florence-2-large for Female Detection...")
-    MODEL_DATA["processor"] = AutoProcessor.from_pretrained("microsoft/Florence-2-large", trust_remote_code=True)
+    print("📥 Loading Florence-2-large...")
+    start = time.time()
+    
+    MODEL_DATA["processor"] = AutoProcessor.from_pretrained(
+        "microsoft/Florence-2-large",
+        trust_remote_code=True
+    )
     MODEL_DATA["model"] = AutoModelForCausalLM.from_pretrained(
-        "microsoft/Florence-2-large", trust_remote_code=True, attn_implementation="eager"
+        "microsoft/Florence-2-large",
+        torch_dtype=torch.float32,   # ← ضروري على CPU
+        trust_remote_code=True,
+        attn_implementation="eager"
     ).eval()
+    
+    print(f"✅ Model ready in {time.time()-start:.1f}s")
     yield
+    MODEL_DATA.clear()
 
-app = FastAPI(lifespan=lifespan)
+# ─── FastAPI ─────────────────────────────────────────────────────
+app = FastAPI(
+    title="Female Detection API",
+    description="Detects female content in images using Florence-2-large",
+    version="1.0.0",
+    lifespan=lifespan
+)
+
+@app.get("/health")
+def health():
+    return {"status": "ok", "model_loaded": "model" in MODEL_DATA}
 
 @app.post("/analyze")
 async def analyze_image(file: UploadFile = File(...)):
+    # التحقق من نوع الملف
+    if not file.content_type.startswith("image/"):
+        raise HTTPException(status_code=400, detail="الملف ليس صورة")
+
     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"]
+
         task = "<OPEN_VOCABULARY_DETECTION>"
         prompt = f"{task}{FORBIDDEN_LABELS}"
-        
-        inputs = MODEL_DATA["processor"](text=prompt, images=image, return_tensors="pt")
-        
+
+        inputs = processor(text=prompt, images=image, return_tensors="pt")
+
         start_time = time.time()
         with torch.no_grad():
-            generated_ids = MODEL_DATA["model"].generate(
+            generated_ids = model.generate(
                 input_ids=inputs["input_ids"],
                 pixel_values=inputs["pixel_values"],
                 max_new_tokens=512,
-                num_beams=1
+                num_beams=1,
+                do_sample=False
             )
-        
-        generated_text = MODEL_DATA["processor"].batch_decode(generated_ids, skip_special_tokens=False)[0]
-        parsed_result = MODEL_DATA["processor"].post_process_generation(
-            generated_text, 
-            task=task, 
+
+        generated_text = processor.batch_decode(generated_ids, skip_special_tokens=False)[0]
+        parsed_result = processor.post_process_generation(
+            generated_text,
+            task=task,
             image_size=(image.width, image.height)
         )
-        
-        # استخراج النتائج
+
         results = parsed_result.get(task, {})
-        detected_objects = results.get('bboxes', [])
-        detected_labels = results.get('labels', []) # المسميات التي وجدها النموذج
-        
-        # منطق القرار: نحجب فقط إذا وجد شيئاً من القائمة الأنثوية
-        decision = "allow"
-        if len(detected_objects) > 0:
-            decision = "block"
-            
+        bboxes  = results.get("bboxes", [])
+        labels  = results.get("labels", [])
+        elapsed = round(time.time() - start_time, 2)
+
+        # ─── منطق القرار ─────────────────────────────────────────
+        decision = "block" if len(bboxes) > 0 else "allow"
+
         return {
-            "decision": decision,
-            "detected_count": len(detected_objects),
-            "found": detected_labels, 
-            "execution_time": round(time.time() - start_time, 2),
+            "decision": decision,           # allow | block
+            "detected_count": len(bboxes),
+            "found": labels,                # ما وجده النموذج فعلاً
+            "bboxes": bboxes,               # إحداثيات المناطق المكتشفة
+            "execution_time": elapsed,
             "status": "success"
         }
+
     except Exception as e:
-        return JSONResponse(status_code=500, content={"error": str(e)})
+        raise HTTPException(status_code=500, detail=f"خطأ في الاستدلال: {str(e)}")
+
 
 if __name__ == "__main__":
     import uvicorn