v8

app.py

@@ -1,33 +1,44 @@
-import gradio as gr
-import uuid
 import os
+import uuid
+import shutil
+import gradio as gr
+import numpy as np
+from PIL import Image
 from qdrant_client import QdrantClient
-from qdrant_client.models import PointStruct
+from qdrant_client.models import PointStruct, Distance, VectorParams
 from sentence_transformers import SentenceTransformer
-from PIL import Image
-import numpy as np
 
-# Create uploads folder
+# =====================
+# CONFIG
+# =====================
 UPLOAD_DIR = "uploads"
-os.makedirs(UPLOAD_DIR, exist_ok=True)
-
-# Connect to Qdrant
-COLLECTION = "lost_and_found"
-qclient = QdrantClient(":memory:")  # for demo, replace with your Qdrant server for persistence
+DB_PATH = "./qdrant_db"
+COLLECTION = "items"
 
-# Load CLIP model
-model = SentenceTransformer("sentence-transformers/clip-ViT-B-32-multilingual-v1")
-
-# Ensure collection exists
-qclient.recreate_collection(
-    collection_name=COLLECTION,
-    vectors_config={"size": 512, "distance": "Cosine"}
-)
+os.makedirs(UPLOAD_DIR, exist_ok=True)
 
-# Encode helper
+# =====================
+# INIT
+# =====================
+qclient = QdrantClient(path=DB_PATH)
+if COLLECTION not in [c.name for c in qclient.get_collections().collections]:
+    qclient.recreate_collection(
+        collection_name=COLLECTION,
+        vectors_config=VectorParams(size=512, distance=Distance.COSINE)
+    )
+
+model = SentenceTransformer("clip-ViT-B-32")
+
+# =====================
+# HELPERS
+# =====================
 def encode_data(text=None, image=None):
+    """Encodes either text or image into embedding."""
     if image is not None:
-        img = Image.open(image).convert("RGB")
+        if isinstance(image, str):
+            img = Image.open(image).convert("RGB")
+        else:
+            img = image.convert("RGB")
         emb = model.encode(img, convert_to_numpy=True, normalize_embeddings=True)
     elif text:
         emb = model.encode(text, convert_to_numpy=True, normalize_embeddings=True)
@@ -35,19 +46,22 @@ def encode_data(text=None, image=None):
         raise ValueError("Need text or image")
     return emb.astype(np.float32)
 
-# Add item
 def add_item(mode, text, image, name, phone):
+    """Add new lost/found item."""
     try:
         vector = encode_data(text=text if text else None, image=image if image else None)
+
         payload = {
             "mode": mode,
             "text": text,
             "has_image": image is not None,
         }
 
-        # Save image file if uploaded
         if image is not None:
-            img = Image.open(image).convert("RGB")
+            if isinstance(image, str):
+                img = Image.open(image).convert("RGB")
+            else:
+                img = image.convert("RGB")
             fname = f"{uuid.uuid4().hex}.png"
             fpath = os.path.join(UPLOAD_DIR, fname)
             img.save(fpath)
@@ -65,79 +79,97 @@ def add_item(mode, text, image, name, phone):
     except Exception as e:
         return f"❌ Error: {e}"
 
-# Search items
-def search_items(query_image, query_text, limit, min_score):
+def search_items(text, image, max_results, min_score):
+    """Search lost/found items."""
     try:
-        query_vector = encode_data(
-            text=query_text if query_text else None,
-            image=query_image if query_image else None
-        )
+        vector = encode_data(text=text if text else None, image=image if image else None)
         results = qclient.search(
             collection_name=COLLECTION,
-            query_vector=query_vector.tolist(),
-            limit=limit,
+            query_vector=vector.tolist(),
+            limit=max_results,
+            score_threshold=min_score,
+            with_payload=True
         )
-        out_texts, out_imgs = [], []
+
+        if not results:
+            return "No matches found.", []
+
+        text_results = []
+        gallery = []
+
         for r in results:
-            if r.score < min_score:
-                continue
             pl = r.payload
-            info = f"id:{r.id} | score:{r.score:.4f} | mode:{pl.get('mode','')}"
-            if pl.get("text"):
-                info += f" | text:{pl['text']}"
+            txt = f"🆔 {r.id} | Mode: {pl.get('mode')} | Text: {pl.get('text')}"
             if pl.get("mode") == "found":
-                info += f" | found by: {pl.get('finder_name','?')} ({pl.get('finder_phone','?')})"
-            out_texts.append(info)
+                txt += f" | Found by: {pl.get('finder_name')} ({pl.get('finder_phone')})"
+            txt += f" | Score: {r.score:.3f}"
+            text_results.append(txt)
 
-            if pl.get("image_path"):
-                out_imgs.append(pl["image_path"])
+            if pl.get("image_path") and os.path.exists(pl["image_path"]):
+                gallery.append((pl["image_path"], txt))
+            else:
+                gallery.append((None, txt))
 
-        return "\n".join(out_texts) if out_texts else "No matches.", out_imgs
+        return "\n\n".join(text_results), gallery
     except Exception as e:
         return f"❌ Error: {e}", []
 
-# Clear all images
-def clear_all_images():
+def delete_item(item_id):
+    """Delete one item by ID."""
+    try:
+        qclient.delete(collection_name=COLLECTION, points_selector={"points": [item_id]})
+        return f"🗑️ Deleted item {item_id}"
+    except Exception as e:
+        return f"❌ Error: {e}"
+
+def clear_database():
+    """Clear everything."""
     try:
-        # Clear uploads folder
-        for f in os.listdir(UPLOAD_DIR):
-            os.remove(os.path.join(UPLOAD_DIR, f))
-        # Clear Qdrant
-        qclient.delete(
+        qclient.delete_collection(COLLECTION)
+        qclient.recreate_collection(
             collection_name=COLLECTION,
-            points_selector={"filter": {"must": [{"key": "has_image", "match": {"value": True}}]}}
+            vectors_config=VectorParams(size=512, distance=Distance.COSINE)
         )
-        return "🗑️ All image items cleared!"
+        shutil.rmtree(UPLOAD_DIR, ignore_errors=True)
+        os.makedirs(UPLOAD_DIR, exist_ok=True)
+        return "🧹 Database cleared."
     except Exception as e:
-        return f"❌ Error clearing images: {e}"
+        return f"❌ Error: {e}"
 
-# Gradio UI
+# =====================
+# UI
+# =====================
 with gr.Blocks() as demo:
-    gr.Markdown("# 🔎 Lost & Found System")
-
-    with gr.Tab("➕ Add Item"):
-        mode = gr.Radio(["lost", "found"], label="Mode")
-        text = gr.Textbox(label="Describe the item (optional)")
-        img = gr.Image(type="filepath", label="Upload image (optional)")
-        name = gr.Textbox(label="Finder's Name (only if found)", placeholder="John Doe")
-        phone = gr.Textbox(label="Finder's Phone (only if found)", placeholder="+1234567890")
+    gr.Markdown("# 🔍 Lost & Found System")
+
+    with gr.Tab("➕ Add Lost/Found"):
+        mode = gr.Radio(["lost", "found"], label="Mode", value="lost")
+        text = gr.Textbox(label="Description (optional)")
+        image = gr.Image(type="pil", label="Upload Image")
+        name = gr.Textbox(label="Finder Name (for Found only)")
+        phone = gr.Textbox(label="Finder Phone (for Found only)")
         add_btn = gr.Button("Add Item")
-        add_out = gr.Textbox(label="Add result")
-        add_btn.click(add_item, inputs=[mode, text, img, name, phone], outputs=add_out)
+        add_output = gr.Textbox(label="Status")
+        add_btn.click(add_item, [mode, text, image, name, phone], add_output)
 
-    with gr.Tab("🔍 Search"):
-        query_text = gr.Textbox(label="Search by text (optional)")
-        query_img = gr.Image(type="filepath", label="Search by image (optional)")
+    with gr.Tab("🔎 Search"):
+        s_text = gr.Textbox(label="Search by text (optional)")
+        s_image = gr.Image(type="pil", label="Search by image (optional)")
         max_results = gr.Slider(1, 10, value=5, step=1, label="Max results")
-        score_slider = gr.Slider(0.5, 1.0, value=0.9, step=0.01, label="Min similarity threshold")
+        min_score = gr.Slider(0.5, 1.0, value=0.9, step=0.01, label="Min similarity threshold")
         search_btn = gr.Button("Search")
-        search_out = gr.Textbox(label="Search results (text)")
-        gallery = gr.Gallery(label="Search Results", show_label=True, elem_id="gallery", columns=2, height="auto")
-        search_btn.click(search_items, inputs=[query_img, query_text, max_results, score_slider], outputs=[search_out, gallery])
-
-    with gr.Tab("🗑️ Admin"):
-        clear_btn = gr.Button("Clear All Images")
-        clear_out = gr.Textbox(label="Clear Result")
-        clear_btn.click(clear_all_images, outputs=clear_out)
+        search_text = gr.Textbox(label="Search results (text)")
+        search_gallery = gr.Gallery(label="Search Results").style(grid=[2], height="auto")
+        search_btn.click(search_items, [s_text, s_image, max_results, min_score], [search_text, search_gallery])
+
+    with gr.Tab("⚙️ Admin"):
+        del_id = gr.Textbox(label="Delete Item by ID")
+        del_btn = gr.Button("Delete Item")
+        del_output = gr.Textbox(label="Delete Status")
+        del_btn.click(delete_item, del_id, del_output)
+
+        clear_btn = gr.Button("Clear All Database")
+        clear_output = gr.Textbox(label="Clear Status")
+        clear_btn.click(clear_database, outputs=clear_output)
 
 demo.launch()