Rename src/cloud_db.py to src/db.py

src/cloud_db.py

@@ -1,246 +0,0 @@
-# src/cloud_db.py — Enterprise Lens V4
-# ════════════════════════════════════════════════════════════════
-# NOTE: In the production FastAPI app (main.py), ALL Pinecone and
-# Cloudinary operations are performed directly — this class is NOT
-# called by main.py. It exists as a standalone utility / SDK wrapper
-# for scripts, notebooks, or future use outside the API.
-#
-# If you use this class, ensure your Pinecone indexes match V4 dims:
-#   enterprise-faces   → 1024-D  (ArcFace-512 + AdaFace-512, fused)
-#   enterprise-objects → 1536-D  (SigLIP-768  + DINOv2-768,  fused)
-# ════════════════════════════════════════════════════════════════
-
-import os
-import uuid
-import cloudinary
-import cloudinary.uploader
-from pinecone import Pinecone, ServerlessSpec
-from dotenv import load_dotenv
-
-load_dotenv()
-
-# ── V4 Index constants — MUST match main.py and models.py ────────
-IDX_FACES         = "enterprise-faces"
-IDX_OBJECTS       = "enterprise-objects"
-IDX_FACES_DIM     = 1024   # ArcFace(512) + AdaFace(512) fused, always 1024
-IDX_OBJECTS_DIM   = 1536   # SigLIP(768)  + DINOv2(768)  fused, always 1536
-
-# V4 face similarity thresholds (fused 1024-D cosine space)
-# These MUST stay in sync with main.py FACE_THRESHOLD_* constants
-FACE_THRESHOLD_HIGH = 0.40   # high-quality face (det_score >= 0.85)
-FACE_THRESHOLD_LOW  = 0.32   # lower-quality face (det_score < 0.85)
-OBJECT_THRESHOLD    = 0.45   # object/scene similarity threshold
-
-
-class CloudDB:
-    """
-    Utility wrapper around Pinecone + Cloudinary for Enterprise Lens V4.
-
-    Index dimensions:
-      enterprise-faces   : 1024-D cosine
-      enterprise-objects : 1536-D cosine
-
-    Face vectors: ArcFace(512) + AdaFace(512) concatenated + L2-normalised
-    Object vectors: SigLIP(768) + DINOv2(768) concatenated + L2-normalised
-    """
-
-    def __init__(self):
-        # ── Cloudinary ────────────────────────────────────────────
-        cloudinary.config(
-            cloud_name = os.getenv("CLOUDINARY_CLOUD_NAME"),
-            api_key    = os.getenv("CLOUDINARY_API_KEY"),
-            api_secret = os.getenv("CLOUDINARY_API_SECRET"),
-        )
-
-        # ── Pinecone ──────────────────────────────────────────────
-        self.pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))
-        self._ensure_indexes()
-        self.index_faces   = self.pc.Index(IDX_FACES)
-        self.index_objects = self.pc.Index(IDX_OBJECTS)
-
-    def _ensure_indexes(self):
-        """
-        Create Pinecone indexes at correct V4 dimensions if they don't exist.
-        Safe to call multiple times — skips existing indexes.
-        """
-        existing = {idx.name for idx in self.pc.list_indexes()}
-
-        if IDX_FACES not in existing:
-            print(f"📦 Creating {IDX_FACES} at {IDX_FACES_DIM}-D...")
-            self.pc.create_index(
-                name      = IDX_FACES,
-                dimension = IDX_FACES_DIM,   # 1024-D — ArcFace+AdaFace
-                metric    = "cosine",
-                spec      = ServerlessSpec(cloud="aws", region="us-east-1"),
-            )
-            print(f"   ✅ {IDX_FACES} created at {IDX_FACES_DIM}-D")
-        else:
-            # Validate existing index has correct dimension
-            desc = self.pc.describe_index(IDX_FACES)
-            actual_dim = desc.dimension
-            if actual_dim != IDX_FACES_DIM:
-                raise ValueError(
-                    f"❌ {IDX_FACES} exists at {actual_dim}-D but V4 needs "
-                    f"{IDX_FACES_DIM}-D. Go to Settings → Danger Zone → "
-                    f"Reset Database to recreate at correct dimensions."
-                )
-
-        if IDX_OBJECTS not in existing:
-            print(f"📦 Creating {IDX_OBJECTS} at {IDX_OBJECTS_DIM}-D...")
-            self.pc.create_index(
-                name      = IDX_OBJECTS,
-                dimension = IDX_OBJECTS_DIM,   # 1536-D — SigLIP+DINOv2
-                metric    = "cosine",
-                spec      = ServerlessSpec(cloud="aws", region="us-east-1"),
-            )
-            print(f"   ✅ {IDX_OBJECTS} created at {IDX_OBJECTS_DIM}-D")
-        else:
-            desc = self.pc.describe_index(IDX_OBJECTS)
-            actual_dim = desc.dimension
-            if actual_dim != IDX_OBJECTS_DIM:
-                raise ValueError(
-                    f"❌ {IDX_OBJECTS} exists at {actual_dim}-D but V4 needs "
-                    f"{IDX_OBJECTS_DIM}-D. Go to Settings → Danger Zone → "
-                    f"Reset Database to recreate at correct dimensions."
-                )
-
-    # ── Upload image to Cloudinary ────────────────────────────────
-    def upload_image(self, file_path: str, folder_name: str = "visual_search") -> str:
-        """Upload image to Cloudinary, return secure_url."""
-        response = cloudinary.uploader.upload(file_path, folder=folder_name)
-        return response["secure_url"]
-
-    # ── Store vector in correct Pinecone index ────────────────────
-    def add_vector(self, data_dict: dict, image_url: str, image_id: str = None):
-        """
-        Upsert one vector into the correct Pinecone index.
-
-        data_dict keys:
-          type       : "face" or "object"
-          vector     : np.ndarray or list — must match index dimension
-          face_crop  : str  (base64 JPEG thumbnail, face only)
-          det_score  : float (InsightFace detection confidence, face only)
-          face_quality: float (alias for det_score)
-          face_width_px: int (face bounding box width in pixels)
-          face_idx   : int (face index within the source image)
-          bbox       : list [x, y, w, h]
-          folder     : str (Cloudinary folder / category name)
-        """
-        vec_id   = image_id or str(uuid.uuid4())
-        vec_list = (data_dict["vector"].tolist()
-                    if hasattr(data_dict["vector"], "tolist")
-                    else list(data_dict["vector"]))
-
-        if data_dict["type"] == "face":
-            # ── V4 face metadata — full set required for UI ───────
-            payload = [{
-                "id":     vec_id,
-                "values": vec_list,
-                "metadata": {
-                    "image_url":     image_url,
-                    "url":           image_url,          # alias for compatibility
-                    "folder":        data_dict.get("folder", ""),
-                    "face_idx":      data_dict.get("face_idx", 0),
-                    "bbox":          str(data_dict.get("bbox", [])),
-                    "face_crop":     data_dict.get("face_crop", ""),    # base64 thumb
-                    "det_score":     data_dict.get("det_score", 1.0),
-                    "face_quality":  data_dict.get("face_quality",
-                                     data_dict.get("det_score", 1.0)),
-                    "face_width_px": data_dict.get("face_width_px", 0),
-                },
-            }]
-            self.index_faces.upsert(vectors=payload)
-
-        else:
-            # ── V4 object metadata ────────────────────────────────
-            payload = [{
-                "id":     vec_id,
-                "values": vec_list,
-                "metadata": {
-                    "image_url": image_url,
-                    "url":       image_url,
-                    "folder":    data_dict.get("folder", ""),
-                },
-            }]
-            self.index_objects.upsert(vectors=payload)
-
-    # ── Search ────────────────────────────────────────────────────
-    def search(self, query_dict: dict, top_k: int = 10,
-               min_score: float = None) -> list:
-        """
-        Search the correct Pinecone index for one query vector.
-
-        For face vectors: uses adaptive threshold based on det_score.
-        For object vectors: uses OBJECT_THRESHOLD (default 0.45).
-
-        Returns list of dicts: {url, score, caption, [face_crop, folder]}
-        """
-        vec_list = (query_dict["vector"].tolist()
-                    if hasattr(query_dict["vector"], "tolist")
-                    else list(query_dict["vector"]))
-        results  = []
-
-        if query_dict["type"] == "face":
-            # ── V4 face search ────────────────────────────────────
-            # Adaptive threshold: high-quality faces are stricter
-            det_score = query_dict.get("det_score", 1.0)
-            threshold = (FACE_THRESHOLD_HIGH if det_score >= 0.85
-                         else FACE_THRESHOLD_LOW)
-            if min_score is not None:
-                threshold = min_score
-
-            response = self.index_faces.query(
-                vector=vec_list, top_k=top_k * 3,   # over-fetch, filter below
-                include_metadata=True,
-            )
-
-            # Deduplicate by image_url — keep best score per image
-            image_map = {}
-            for match in response.get("matches", []):
-                raw = match["score"]
-                if raw < threshold:
-                    continue
-                url = (match["metadata"].get("url") or
-                       match["metadata"].get("image_url", ""))
-                if not url:
-                    continue
-                if url not in image_map or raw > image_map[url]["raw"]:
-                    image_map[url] = {
-                        "raw":       raw,
-                        "face_crop": match["metadata"].get("face_crop", ""),
-                        "folder":    match["metadata"].get("folder", ""),
-                    }
-
-            # Remap raw cosine → UI percentage (75%–99%)
-            for url, d in image_map.items():
-                lo = FACE_THRESHOLD_LOW
-                ui = round(min(0.99, 0.75 + ((d["raw"] - lo) / (1.0 - lo)) * 0.24), 4)
-                results.append({
-                    "url":       url,
-                    "score":     ui,
-                    "raw_score": round(d["raw"], 4),
-                    "face_crop": d["face_crop"],
-                    "folder":    d["folder"],
-                    "caption":   "👤 Verified Identity Match",
-                })
-
-            results = sorted(results, key=lambda x: x["score"], reverse=True)[:top_k]
-
-        else:
-            # ── V4 object search ──────────────────────────────────
-            threshold = min_score if min_score is not None else OBJECT_THRESHOLD
-            response  = self.index_objects.query(
-                vector=vec_list, top_k=top_k, include_metadata=True)
-
-            for match in response.get("matches", []):
-                if match["score"] < threshold:
-                    continue
-                results.append({
-                    "url":     (match["metadata"].get("url") or
-                                match["metadata"].get("image_url", "")),
-                    "score":   round(match["score"], 4),
-                    "folder":  match["metadata"].get("folder", ""),
-                    "caption": "🎯 Visual & Semantic Match",
-                })
-
-        return results

src/db.py

@@ -0,0 +1,356 @@
+"""
+src/db.py — All external service I/O lives here.
+
+Two responsibilities:
+  1. PineconePool  — LRU cache of Pinecone client objects (avoids per-request TCP overhead)
+  2. Cloudinary helpers — thin wrappers that always inject credentials explicitly
+                          (stateless; no global cloudinary.config() call needed)
+
+Search helpers (search_faces, search_objects) live here too so main.py
+stays thin and the query logic is testable in isolation.
+"""
+
+import io
+from collections import OrderedDict
+from typing import Optional
+
+import cloudinary
+import cloudinary.api
+import cloudinary.uploader
+from pinecone import Pinecone, ServerlessSpec
+
+from .config import (
+    PINECONE_POOL_MAX,
+    IDX_FACES, IDX_OBJECTS,
+    IDX_FACES_DIM, IDX_OBJECTS_DIM,
+    FACE_THRESHOLD_HIGH, FACE_THRESHOLD_LOW, FACE_DET_SCORE_HQ_SPLIT,
+    FACE_TOP_K_FETCH,
+    OBJECT_SCORE_THRESHOLD, OBJECT_TOP_K,
+)
+from .utils import to_list, face_ui_score
+
+
+# ════════════════════════════════════════════════════════════════════
+#  PINECONE LRU CONNECTION POOL
+# ════════════════════════════════════════════════════════════════════
+
+class PineconePool:
+    """
+    LRU cache of Pinecone() client objects keyed by API key.
+
+    Why pool?  Creating a Pinecone() client opens a TCP connection and
+    authenticates — adds ~200 ms on first use.  Re-using clients across
+    requests eliminates that overhead for repeat users.
+
+    Implementation:  OrderedDict + move_to_end = O(1) LRU without any
+    extra dependency.  On overflow, popitem(last=False) evicts the
+    least-recently-used entry.
+    """
+
+    def __init__(self, maxsize: int = PINECONE_POOL_MAX):
+        self._pool: OrderedDict[str, Pinecone] = OrderedDict()
+        self._maxsize = maxsize
+
+    def get(self, api_key: str) -> Pinecone:
+        if api_key not in self._pool:
+            if len(self._pool) >= self._maxsize:
+                self._pool.popitem(last=False)
+            self._pool[api_key] = Pinecone(api_key=api_key)
+        self._pool.move_to_end(api_key)
+        return self._pool[api_key]
+
+
+# Module-level singleton — shared across all requests
+pinecone_pool = PineconePool()
+
+
+# ════════════════════════════════════════════════════════════════════
+#  CLOUDINARY WRAPPERS
+#  All functions accept a `creds` dict — no global state.
+#  Call via asyncio.to_thread() from async endpoints.
+# ════════════════════════════════════════════════════════════════════
+
+def cld_upload(file_obj: io.BytesIO, folder: str, creds: dict) -> dict:
+    """Upload an in-memory file to Cloudinary. Returns the full API response."""
+    return cloudinary.uploader.upload(
+        file_obj, folder=folder,
+        api_key=creds["api_key"], api_secret=creds["api_secret"],
+        cloud_name=creds["cloud_name"],
+    )
+
+
+def cld_ping(creds: dict) -> dict:
+    """Lightweight connectivity check — verifies credentials are valid."""
+    return cloudinary.api.ping(
+        api_key=creds["api_key"], api_secret=creds["api_secret"],
+        cloud_name=creds["cloud_name"],
+    )
+
+
+def cld_root_folders(creds: dict) -> dict:
+    """List all top-level folders in the account."""
+    return cloudinary.api.root_folders(
+        api_key=creds["api_key"], api_secret=creds["api_secret"],
+        cloud_name=creds["cloud_name"],
+    )
+
+
+def cld_list_folder_images(
+    folder: str,
+    creds: dict,
+    next_cursor: Optional[str] = None,
+    max_results: int = 100,
+) -> dict:
+    """Paginated listing of resources under a folder prefix."""
+    kwargs = dict(
+        type="upload",
+        prefix=f"{folder}/",
+        max_results=min(max_results, 100),   # Cloudinary hard-caps at 100
+        api_key=creds["api_key"],
+        api_secret=creds["api_secret"],
+        cloud_name=creds["cloud_name"],
+    )
+    if next_cursor:
+        kwargs["next_cursor"] = next_cursor
+    return cloudinary.api.resources(**kwargs)
+
+
+def cld_delete_resource(public_id: str, creds: dict) -> dict:
+    """Delete a single resource by public_id."""
+    return cloudinary.uploader.destroy(
+        public_id,
+        api_key=creds["api_key"], api_secret=creds["api_secret"],
+        cloud_name=creds["cloud_name"],
+    )
+
+
+def cld_delete_folder_resources(folder: str, creds: dict) -> dict:
+    """Delete all resources under a folder prefix (not the folder record itself)."""
+    return cloudinary.api.delete_resources_by_prefix(
+        f"{folder}/",
+        api_key=creds["api_key"], api_secret=creds["api_secret"],
+        cloud_name=creds["cloud_name"],
+    )
+
+
+def cld_remove_folder(folder: str, creds: dict):
+    """Delete the folder record itself. Silently ignores errors (already gone)."""
+    try:
+        return cloudinary.api.delete_folder(
+            folder,
+            api_key=creds["api_key"], api_secret=creds["api_secret"],
+            cloud_name=creds["cloud_name"],
+        )
+    except Exception:
+        pass
+
+
+def cld_delete_all_paginated(creds: dict) -> int:
+    """
+    Delete ALL resources in a Cloudinary account, paginating in batches of 100.
+    Returns the total number of resources deleted.
+    Used only by the destructive /api/reset-database and /api/delete-account endpoints.
+    """
+    deleted = 0
+    while True:
+        try:
+            res = cloudinary.api.resources(
+                type="upload", max_results=100,
+                api_key=creds["api_key"], api_secret=creds["api_secret"],
+                cloud_name=creds["cloud_name"],
+            )
+            resources = res.get("resources", [])
+            if not resources:
+                break
+            public_ids = [r["public_id"] for r in resources]
+            cloudinary.api.delete_resources(
+                public_ids,
+                api_key=creds["api_key"], api_secret=creds["api_secret"],
+                cloud_name=creds["cloud_name"],
+            )
+            deleted += len(public_ids)
+            if not res.get("next_cursor"):
+                break
+        except Exception:
+            break
+    return deleted
+
+
+# ════════════════════════════════════════════════════════════════════
+#  PINECONE INDEX MANAGEMENT
+# ════════════════════════════════════════════════════════════════════
+
+def ensure_indexes(pc: Pinecone) -> list[str]:
+    """
+    Create enterprise-faces and enterprise-objects indexes if they don't exist.
+    Returns a list of index names that were freshly created (empty list if both existed).
+    """
+    existing       = {idx.name for idx in pc.list_indexes()}
+    created        = []
+    spec           = ServerlessSpec(cloud="aws", region="us-east-1")
+
+    if IDX_OBJECTS not in existing:
+        pc.create_index(name=IDX_OBJECTS, dimension=IDX_OBJECTS_DIM, metric="cosine", spec=spec)
+        created.append(IDX_OBJECTS)
+
+    if IDX_FACES not in existing:
+        pc.create_index(name=IDX_FACES, dimension=IDX_FACES_DIM, metric="cosine", spec=spec)
+        created.append(IDX_FACES)
+
+    return created
+
+
+def delete_and_recreate_indexes(pc: Pinecone):
+    """
+    Destroy both indexes and recreate them empty.
+    Used by /api/reset-database. Caller is responsible for ensuring
+    this is not called on the shared demo database.
+    """
+    import asyncio, time
+    existing = {idx.name for idx in pc.list_indexes()}
+    for name in [IDX_OBJECTS, IDX_FACES]:
+        if name in existing:
+            pc.delete_index(name)
+    time.sleep(3)   # Pinecone needs a moment to fully delete before recreating
+    spec = ServerlessSpec(cloud="aws", region="us-east-1")
+    pc.create_index(name=IDX_OBJECTS, dimension=IDX_OBJECTS_DIM, metric="cosine", spec=spec)
+    pc.create_index(name=IDX_FACES,   dimension=IDX_FACES_DIM,   metric="cosine", spec=spec)
+
+
+# ════════════════════════════════════════════════════════════════════
+#  PINECONE SEARCH HELPERS
+# ════════════════════════════════════════════════════════════════════
+
+def search_faces(idx_face, vec: list, det_score: float) -> dict:
+    """
+    Query the faces index for one face vector.
+
+    Threshold selection:
+      - High-quality face (det_score >= FACE_DET_SCORE_HQ_SPLIT) → FACE_THRESHOLD_HIGH
+      - Lower-quality face                                        → FACE_THRESHOLD_LOW
+    Adaptive thresholding prevents low-quality query faces from flooding results
+    with false positives while still matching when confidence warrants it.
+
+    Returns: { image_url → { raw_score, face_crop, folder, face_width_px } }
+    De-duplicated per image URL — only the highest-scoring face match per image
+    is retained (multiple face vectors can be stored per source image during upload).
+    """
+    threshold = (
+        FACE_THRESHOLD_HIGH
+        if det_score >= FACE_DET_SCORE_HQ_SPLIT
+        else FACE_THRESHOLD_LOW
+    )
+    response  = idx_face.query(vector=vec, top_k=FACE_TOP_K_FETCH, include_metadata=True)
+    image_map = {}
+
+    for match in response.get("matches", []):
+        raw = match["score"]
+        if raw < threshold:
+            continue
+        url = match["metadata"].get("url", "")
+        if not url:
+            continue
+        if url not in image_map or raw > image_map[url]["raw_score"]:
+            image_map[url] = {
+                "raw_score":     raw,
+                "face_crop":     match["metadata"].get("face_crop",     ""),
+                "folder":        match["metadata"].get("folder",        ""),
+                "face_width_px": int(match["metadata"].get("face_width_px", 0)),
+            }
+    return image_map
+
+
+def search_objects(idx_obj, vec: list) -> list:
+    """
+    Query the objects index for one object vector.
+
+    Returns list of { url, score, folder } dicts for matches above
+    OBJECT_SCORE_THRESHOLD, sorted by score descending.
+    """
+    response = idx_obj.query(vector=vec, top_k=OBJECT_TOP_K, include_metadata=True)
+    results  = []
+    for match in response.get("matches", []):
+        if match["score"] < OBJECT_SCORE_THRESHOLD:
+            continue
+        url = match["metadata"].get("url", "")
+        if url:
+            results.append({
+                "url":    url,
+                "score":  round(match["score"], 4),
+                "folder": match["metadata"].get("folder", ""),
+            })
+    return sorted(results, key=lambda x: x["score"], reverse=True)
+
+
+def merge_face_results(raw_groups: list[dict]) -> list[dict]:
+    """
+    Merge per-face search results from multiple query faces into one ranked list.
+
+    Algorithm:
+      1. Build a global image_url → {best_score, matched_face_indices, ...} map.
+      2. An image is included if ANY query face matched it.
+      3. Score = best raw score across all matching faces, remapped to UI score.
+      4. Multi-face boost: +5 % per additional matched face (capped at 0.99).
+         Rationale: an image containing 3 of your 3 searched people should rank
+         above one containing only 1 of 3.
+
+    Args:
+        raw_groups: list of dicts, each containing '_image_map' from search_faces()
+
+    Returns: sorted list of merged result dicts (best first), max 20 items.
+    """
+    global_map: dict[str, dict] = {}
+
+    for gi, group in enumerate(raw_groups):
+        for url, d in group["_image_map"].items():
+            raw = d["raw_score"]
+            if url not in global_map:
+                global_map[url] = {
+                    "raw_score":     raw,
+                    "face_crop":     d["face_crop"],
+                    "folder":        d["folder"],
+                    "face_width_px": d["face_width_px"],
+                    "matched_faces": [gi],
+                }
+            else:
+                entry = global_map[url]
+                entry["matched_faces"].append(gi)
+                if raw > entry["raw_score"]:
+                    entry["raw_score"]     = raw
+                    entry["face_crop"]     = d["face_crop"]
+                    entry["face_width_px"] = d["face_width_px"]
+
+    results = []
+    for url, d in global_map.items():
+        n = len(d["matched_faces"])
+        results.append({
+            "url":           url,
+            "score":         face_ui_score(d["raw_score"], n),
+            "raw_score":     round(d["raw_score"], 4),
+            "face_crop":     d["face_crop"],
+            "folder":        d["folder"],
+            "face_width_px": d["face_width_px"],
+            "matched_faces": d["matched_faces"],
+            "caption":       f"👥 {n} faces matched" if n > 1 else "👤 Verified Identity",
+        })
+
+    return sorted(results, key=lambda x: x["score"], reverse=True)[:20]
+
+
+def merge_object_results(nested: list[list[dict]]) -> list[dict]:
+    """
+    Merge object search results from multiple crops into one ranked list.
+
+    When YOLO produces N crops, each gets its own embedding and Pinecone query.
+    The same image URL may appear in multiple crop results; only the highest
+    score per URL is kept.
+
+    Returns: sorted list, max OBJECT_TOP_K items.
+    """
+    seen: dict[str, dict] = {}
+    for results in nested:
+        for r in results:
+            url = r["url"]
+            if url and (url not in seen or r["score"] > seen[url]["score"]):
+                seen[url] = r
+    return sorted(seen.values(), key=lambda x: x["score"], reverse=True)[:OBJECT_TOP_K]