modifymain

main.py

@@ -21,12 +21,14 @@ from fastapi import FastAPI, File, UploadFile, Form, Request, Depends
 from fastapi.responses import HTMLResponse, RedirectResponse
 from fastapi.templating import Jinja2Templates
 from fastapi.staticfiles import StaticFiles
+import requests
+import uvicorn
 
 sys.path.append(os.path.abspath(os.path.dirname(__file__)))
 from models.densenet.preprocess.preprocessingwangchan import get_tokenizer, get_transforms
 from models.densenet.train_densenet_only import DenseNet121Classifier
 from models.densenet.train_text_only import TextClassifier
-import requests
+
 
 HF_MODEL_URL = "https://huggingface.co/qqqqqqat/densenet_wangchan/resolve/main/best_fusion_densenet.pth"
 LOCAL_MODEL_PATH = "models/densenet/best_fusion_densenet.pth"
@@ -48,10 +50,13 @@ FUSION_LABELMAP_PATH = "models/densenet/label_map_fusion_densenet.json"
 FUSION_WEIGHTS_PATH  = "models/densenet/best_fusion_densenet.pth"
 with open(FUSION_LABELMAP_PATH, "r", encoding="utf-8") as f:
     label_map = json.load(f)
+
 class_names = [label for label, _ in sorted(label_map.items(), key=lambda x: x[1])]
 NUM_CLASSES = len(class_names)
 device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
 print(f"🧠 Using device: {device}")
+
+
 class FusionDenseNetText(nn.Module):
     def __init__(self, num_classes, dropout=0.3):
         super().__init__()
@@ -67,9 +72,11 @@ class FusionDenseNetText(nn.Module):
         fused_in   = torch.cat([logits_img, logits_txt], dim=1)
         fused_out  = self.fusion(fused_in)
         return fused_out, logits_img, logits_txt
+
+
 print("🔄 Loading AI model...")
 fusion_model = FusionDenseNetText(num_classes=NUM_CLASSES).to(device)
-# ดาวน์โหลดก่อน
+
 download_model_if_needed()
 
 fusion_model = FusionDenseNetText(num_classes=NUM_CLASSES).to(device)
@@ -77,38 +84,55 @@ fusion_model.load_state_dict(torch.load(LOCAL_MODEL_PATH, map_location=device))
 fusion_model.eval()
 print("✅ AI Model loaded successfully!")
 fusion_model.eval()
+
 tokenizer = get_tokenizer()
 transform = get_transforms((224, 224))
+
+
 def _find_last_conv2d(mod: torch.nn.Module):
     last = None
     for m in mod.modules():
         if isinstance(m, torch.nn.Conv2d): last = m
     return last
+
+
 def compute_gradcam_overlay(img_pil, image_tensor, target_class_idx):
     img_branch = fusion_model.image_model
     target_layer = _find_last_conv2d(img_branch)
     if target_layer is None: return None
+
     activations, gradients = [], []
+
     def fwd_hook(_m, _i, o): activations.append(o)
     def bwd_hook(_m, gin, gout): gradients.append(gout[0])
+
     h1 = target_layer.register_forward_hook(fwd_hook)
     h2 = target_layer.register_full_backward_hook(bwd_hook)
+
     try:
         img_branch.zero_grad()
         logits_img = img_branch(image_tensor)
         score = logits_img[0, target_class_idx]
         score.backward()
+
         act = activations[-1].detach()[0]
         grad = gradients[-1].detach()[0]
+
         weights = torch.mean(grad, dim=(1, 2))
         cam = torch.relu(torch.sum(weights[:, None, None] * act, dim=0))
+
         cam -= cam.min(); cam /= (cam.max() + 1e-8)
+
         cam_img = Image.fromarray((cam.cpu().numpy() * 255).astype(np.uint8)).resize(img_pil.size, Image.BILINEAR)
+
         cam_np = np.asarray(cam_img).astype(np.float32) / 255.0
         heatmap = cm.get_cmap("jet")(cam_np)[:, :, :3]
+
         img_np = np.asarray(img_pil.convert("RGB")).astype(np.float32) / 255.0
         overlay = (0.6 * img_np + 0.4 * heatmap)
+
         return np.clip(overlay * 255, 0, 255).astype(np.uint8)
+    
     finally:
         h1.remove(); h2.remove(); img_branch.zero_grad()
 
@@ -118,129 +142,191 @@ app.mount("/static", StaticFiles(directory="static"), name="static")
 templates = Jinja2Templates(directory="templates")
 os.makedirs("uploads", exist_ok=True)
 
-EXPIRATION_MINUTES = 10 
-results_cache = {}       
-cache_lock = threading.Lock() 
+
+# ===== Cache Cleanup Thread =====
+EXPIRATION_MINUTES = 10
+results_cache = {}
+cache_lock = threading.Lock()
+
 
 def cleanup_expired_cache():
-    """
-    ฟังก์ชันนี้จะทำงานใน Background Thread เพื่อตรวจสอบและลบ Cache ที่หมดอายุ
-    """
     while True:
-        with cache_lock: # ล็อคเพื่อความปลอดภัยในการเข้าถึง cache
-            # สร้าง list ของ key ที่จะลบ เพื่อไม่ให้แก้ไข dict ขณะวน loop
+        with cache_lock:
             expired_keys = []
             current_time = time.time()
             for key, value in results_cache.items():
                 if current_time - value["created_at"] > EXPIRATION_MINUTES * 60:
                     expired_keys.append(key)
-            
-            # ลบ key ที่หมดอายุ
+
             for key in expired_keys:
                 del results_cache[key]
                 print(f"🧹 Cache expired and removed for key: {key}")
-        
-        time.sleep(60) # ตรวจสอบทุกๆ 60 วินาที
+
+        time.sleep(60)
+
 
 @app.on_event("startup")
 async def startup_event():
-    """
-    เริ่ม Background Thread สำหรับทำความสะอาด Cache เมื่อแอปเริ่มทำงาน
-    """
     cleanup_thread = threading.Thread(target=cleanup_expired_cache, daemon=True)
     cleanup_thread.start()
     print("🗑️  Cache cleanup task started.")
 
-SYMPTOM_MAP = {
-    "noSymptoms": "ไม่มีอาการ", "drinkAlcohol": "ดื่มเหล้า", "smoking": "สูบบุหรี่",
-    "chewBetelNut": "เคี้ยวหมาก", "eatSpicyFood": "กินเผ็ดแสบ", "wipeOff": "เช็ดออกได้",
-    "alwaysHurts": "เจ็บเมื่อโดนแผล"
-}
+
+# ===============================
+#      >>> TURNSTILE VERIFY <<<
+# ===============================
+
+TURNSTILE_SECRET = "0x4AAAAAACEfyIeAjGlYCXeasGsMxTuTlHU"   ### (เพิ่ม)
 def process_with_ai_model(image_path: str, prompt_text: str):
-    try:
-        image_pil = Image.open(image_path)
-        image_pil = ImageOps.exif_transpose(image_pil)
-        image_pil = image_pil.convert("RGB")
-        image_tensor = transform(image_pil).unsqueeze(0).to(device)
-        enc = tokenizer(prompt_text, return_tensors="pt", padding="max_length",
-                        truncation=True, max_length=128)
-        ids, mask = enc["input_ids"].to(device), enc["attention_mask"].to(device)
-        with torch.no_grad():
-            fused_logits, _, _ = fusion_model(image_tensor, ids, mask)
-            probs_fused = torch.softmax(fused_logits, dim=1)[0].cpu().numpy()
-        pred_idx   = int(np.argmax(probs_fused))
-        pred_label = class_names[pred_idx]
-        confidence = float(probs_fused[pred_idx]) * 100
-        gradcam_overlay_np = compute_gradcam_overlay(image_pil, image_tensor, pred_idx)
-        def image_to_base64(img):
-            buffered = BytesIO()
-            img.save(buffered, format="JPEG")
-            return base64.b64encode(buffered.getvalue()).decode('utf-8')
-        original_b64 = image_to_base64(image_pil)
-        if gradcam_overlay_np is not None:
-            gradcam_pil = Image.fromarray(gradcam_overlay_np)
-            gradcam_b64 = image_to_base64(gradcam_pil)
-        else:
-            gradcam_b64 = original_b64
-        return original_b64, gradcam_b64, pred_label, f"{confidence:.2f}"
-    except Exception as e:
-        print(f"❌ Error during AI processing: {e}")
-        return None, None, "Error", "0.00"
-
-@app.get("/", response_class=RedirectResponse)
-async def root():
-    return RedirectResponse(url="/detect")
-@app.get("/detect", response_class=HTMLResponse)
-async def show_upload_form(request: Request):
-    return templates.TemplateResponse("detect.html", {"request": request})
+    """ประมวลผลภาพด้วย DenseNet + Text model และสร้างผลลัพธ์"""
+
+    # --- Load image ---
+    img_pil = Image.open(image_path).convert("RGB")
+    img_tensor = transform(img_pil).unsqueeze(0).to(device)
+
+    # --- Tokenize text prompt ---
+    encoding = tokenizer(
+        prompt_text,
+        padding="max_length",
+        truncation=True,
+        max_length=128,
+        return_tensors="pt"
+    )
+    input_ids = encoding["input_ids"].to(device)
+    attention_mask = encoding["attention_mask"].to(device)
+
+    # --- Forward pass ---
+    with torch.no_grad():
+        fused_out, logits_img, logits_txt = fusion_model(
+            img_tensor, input_ids, attention_mask
+        )
+        probs = torch.softmax(fused_out, dim=1)[0]
+        pred_idx = torch.argmax(probs).item()
+
+    predicted_label = class_names[pred_idx]
+    confidence = probs[pred_idx].item()
+
+    name_out = f"{predicted_label} ({confidence:.2f})"
+
+    # --- Compute GradCAM ---
+    overlay_np = compute_gradcam_overlay(img_pil, img_tensor, pred_idx)
+    if overlay_np is not None:
+        overlay_img = Image.fromarray(overlay_np)
+        buffered = BytesIO()
+        overlay_img.save(buffered, format="PNG")
+        gradcam_b64 = base64.b64encode(buffered.getvalue()).decode()
+    else:
+        gradcam_b64 = None
+
+    # --- Encode original image ---
+    buffer2 = BytesIO()
+    img_pil.save(buffer2, format="PNG")
+    image_b64 = base64.b64encode(buffer2.getvalue()).decode()
+
+    return image_b64, gradcam_b64, predicted_label, confidence
+
 
 @app.post("/uploaded")
 async def handle_upload(
     request: Request,
     file: UploadFile = File(...),
     checkboxes: List[str] = Form([]),
-    symptom_text: str = Form("")
+    symptom_text: str = Form(""),
+    cf_turnstile_response: str = Form("")   ### (เพิ่ม)
 ):
+
+    # ---------- (เพิ่ม) VERIFY TURNSTILE ---------
+    
+    if not cf_turnstile_response:
+        return templates.TemplateResponse(
+            "detect.html",
+            {"request": request, "error": "Turnstile token missing"}
+        )
+
+    verify_resp = requests.post(
+        "https://challenges.cloudflare.com/turnstile/v0/siteverify",
+        data={
+            "secret": TURNSTILE_SECRET,
+            "response": cf_turnstile_response
+        }
+    )
+
+    cf_result = verify_resp.json()
+
+    if not cf_result.get("success"):
+        return templates.TemplateResponse(
+            "detect.html",
+            {"request": request, "error": "การยืนยันความปลอดภัยไม่สำเร็จ กรุณาลองใหม่อีกครั้ง"}
+        )
+
+    # ------------------------------------------------
+
+
+    # >>> โค้ดเดิมของคุณทั้งหมดด้านล่าง (ไม่แก้) <<<
+
     temp_filepath = os.path.join("uploads", f"{uuid.uuid4()}_{file.filename}")
     with open(temp_filepath, "wb") as buffer:
         shutil.copyfileobj(file.file, buffer)
+
+    SYMPTOM_MAP = {
+        "noSymptoms": "ไม่มีอาการ", "drinkAlcohol": "ดื่มเหล้า", "smoking": "สูบบุหรี่",
+        "chewBetelNut": "เคี้ยวหมาก", "eatSpicyFood": "กินเผ็ดแสบ", "wipeOff": "เช็ดออกได้",
+        "alwaysHurts": "เจ็บเมื่อโดนแผล"
+    }
+
     final_prompt_parts = []
     selected_symptoms_thai = {SYMPTOM_MAP.get(cb) for cb in checkboxes if SYMPTOM_MAP.get(cb)}
+
     if "ไม่มีอาการ" in selected_symptoms_thai:
         symptoms_group   = {"เจ็บเมื่อโดนแผล", "กินเผ็ดแสบ"}
         lifestyles_group = {"ดื่มเหล้า", "สูบบุหรี่", "เคี้ยวหมาก"}
         patterns_group   = {"เช็ดออกได้"}
         special_group    = {"ไม่มีอาการ"}
+
         final_selected = (selected_symptoms_thai - symptoms_group) | \
                          (selected_symptoms_thai & (lifestyles_group | patterns_group | special_group))
+
         final_prompt_parts.append(" ".join(sorted(list(final_selected))))
+
     elif selected_symptoms_thai:
         final_prompt_parts.append(" ".join(sorted(list(selected_symptoms_thai))))
+
     if symptom_text and symptom_text.strip():
         final_prompt_parts.append(symptom_text.strip())
+
     final_prompt = "; ".join(final_prompt_parts) if final_prompt_parts else "ไม่มีอาการ"
+
     image_b64, gradcam_b64, name_out, eva_output = process_with_ai_model(
         image_path=temp_filepath, prompt_text=final_prompt
     )
+
     os.remove(temp_filepath)
+
     result_id = str(uuid.uuid4())
+
     result_data = {
-        "image_b64_data": image_b64, "gradcam_b64_data": gradcam_b64,
-        "name_out": name_out, "eva_output": eva_output,
+        "image_b64_data": image_b64,
+        "gradcam_b64_data": gradcam_b64,
+        "name_out": name_out,
+        "eva_output": eva_output,
     }
+
     with cache_lock:
         results_cache[result_id] = {
             "data": result_data,
-            "created_at": time.time() 
+            "created_at": time.time()
         }
 
-    results_url = request.url_for('show_results', result_id=result_id)
+    results_url = request.url_for("show_results", result_id=result_id)
     return RedirectResponse(url=results_url, status_code=303)
 
+
+
 @app.get("/results/{result_id}", response_class=HTMLResponse)
 async def show_results(request: Request, result_id: str):
     with cache_lock:
         cached_item = results_cache.get(result_id)
+
     if not cached_item or (time.time() - cached_item["created_at"] > EXPIRATION_MINUTES * 60):
         if cached_item:
             with cache_lock:
@@ -250,6 +336,8 @@ async def show_results(request: Request, result_id: str):
     context = {"request": request, **cached_item["data"]}
     return templates.TemplateResponse("detect.html", context)
 
+
+
 if __name__ == "__main__":
-    port = int(os.environ.get("PORT", 8000))   # ใช้ PORT ของ hosting ถ้ามี
-    uvicorn.run(app, host="0.0.0.0", port=port)
+    port = int(os.environ.get("PORT", 8000))
+    uvicorn.run(app, host="0.0.0.0", port=port)