# main.py import asyncio import os import shutil import uuid import re import inflect from urllib.parse import urlparse from typing import List from contextlib import asynccontextmanager from collections import OrderedDict from fastapi import FastAPI, UploadFile, File, Form, HTTPException from fastapi.middleware.cors import CORSMiddleware import cloudinary import cloudinary.uploader import cloudinary.api from pinecone import Pinecone, ServerlessSpec # ── Deferred imports ───────────────────────────────────────────── ai = None p = inflect.engine() MAX_CONCURRENT_INFERENCES = int(os.getenv("MAX_CONCURRENT_INFERENCES", "6")) _inference_sem: asyncio.Semaphore _pinecone_pool = OrderedDict() _POOL_MAX = 64 IDX_FACES = "enterprise-faces" IDX_OBJECTS = "enterprise-objects" def _get_pinecone(api_key: str) -> Pinecone: if api_key not in _pinecone_pool: if len(_pinecone_pool) >= _POOL_MAX: _pinecone_pool.popitem(last=False) _pinecone_pool[api_key] = Pinecone(api_key=api_key) _pinecone_pool.move_to_end(api_key) return _pinecone_pool[api_key] def _cld_upload(tmp_path, folder, creds): return cloudinary.uploader.upload( tmp_path, folder=folder, api_key=creds["api_key"], api_secret=creds["api_secret"], cloud_name=creds["cloud_name"], ) def _cld_ping(creds): return cloudinary.api.ping( api_key=creds["api_key"], api_secret=creds["api_secret"], cloud_name=creds["cloud_name"], ) def _cld_root_folders(creds): return cloudinary.api.root_folders( api_key=creds["api_key"], api_secret=creds["api_secret"], cloud_name=creds["cloud_name"], ) def _cld_resources_by_folder(folder, creds, max_results=100, next_cursor=None): kwargs = dict( type="upload", prefix=f"{folder}/", max_results=max_results, 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, creds): return cloudinary.uploader.destroy( public_id, api_key=creds["api_key"], api_secret=creds["api_secret"], cloud_name=creds["cloud_name"], ) def _cld_delete_by_prefix(prefix, creds): """Delete all resources under a folder prefix.""" return cloudinary.api.delete_resources_by_prefix( prefix, api_key=creds["api_key"], api_secret=creds["api_secret"], cloud_name=creds["cloud_name"], ) def _cld_delete_folder(folder, creds): """Delete a Cloudinary folder (must be empty first).""" 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 # Folder may not exist — ignore @asynccontextmanager async def lifespan(app: FastAPI): global ai, _inference_sem from src.models import AIModelManager print("⏳ Loading AI models …") loop = asyncio.get_event_loop() ai = await loop.run_in_executor(None, AIModelManager) _inference_sem = asyncio.Semaphore(MAX_CONCURRENT_INFERENCES) print("✅ Ready!") yield app = FastAPI(lifespan=lifespan) app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"]) os.makedirs("temp_uploads", exist_ok=True) def standardize_category_name(name: str) -> str: clean = re.sub(r'\s+', '_', name.strip().lower()) clean = re.sub(r'[^\w]', '', clean) return p.singular_noun(clean) or clean def sanitize_filename(filename: str) -> str: return re.sub(r'[^\w.\-]', '', re.sub(r'\s+', '_', filename)) def get_cloudinary_creds(env_url: str) -> dict: if not env_url: return {} parsed = urlparse(env_url) return {"api_key": parsed.username, "api_secret": parsed.password, "cloud_name": parsed.hostname} def url_to_public_id(image_url: str, cloud_name: str) -> str: """ Extract Cloudinary public_id from a secure_url. e.g. https://res.cloudinary.com/mycloud/image/upload/v123456/folder/filename.jpg → folder/filename """ try: path = urlparse(image_url).path # /mycloud/image/upload/v123456/folder/filename.jpg # Remove leading /cloudname/image/upload/ and version segment parts = path.strip('/').split('/') # Find 'upload' and skip it + version upload_idx = next(i for i, p in enumerate(parts) if p == 'upload') after_upload = parts[upload_idx + 1:] # Skip version segment (starts with 'v' followed by digits) if after_upload and re.match(r'^v\d+$', after_upload[0]): after_upload = after_upload[1:] # Remove file extension file_with_ext = after_upload[-1] after_upload[-1] = re.sub(r'\.[^.]+$', '', file_with_ext) return '/'.join(after_upload) except Exception: return "" # ══════════════════════════════════════════════════════════════════ # 1. VERIFY KEYS & AUTO-BUILD INDEXES # ══════════════════════════════════════════════════════════════════ @app.post("/api/verify-keys") async def verify_keys(pinecone_key: str = Form(""), cloudinary_url: str = Form("")): if cloudinary_url: try: creds_v = get_cloudinary_creds(cloudinary_url) if not creds_v.get("cloud_name"): raise ValueError("bad url") await asyncio.to_thread(_cld_ping, creds_v) except HTTPException: raise except Exception: raise HTTPException(400, "Invalid Cloudinary Environment URL.") if pinecone_key: try: pc = _get_pinecone(pinecone_key) existing = {idx.name for idx in await asyncio.to_thread(pc.list_indexes)} tasks = [] if IDX_OBJECTS not in existing: tasks.append(asyncio.to_thread(pc.create_index, name=IDX_OBJECTS, dimension=1536, metric="cosine", spec=ServerlessSpec(cloud="aws", region="us-east-1"))) if IDX_FACES not in existing: tasks.append(asyncio.to_thread(pc.create_index, name=IDX_FACES, dimension=512, metric="cosine", spec=ServerlessSpec(cloud="aws", region="us-east-1"))) if tasks: await asyncio.gather(*tasks) except Exception as e: raise HTTPException(400, f"Pinecone Error: {e}") return {"message": "Keys verified and indexes ready!"} # ══════════════════════════════════════════════════════════════════ # 2. UPLOAD # ══════════════════════════════════════════════════════════════════ @app.post("/api/upload") async def upload_new_images( files: List[UploadFile] = File(...), folder_name: str = Form(...), detect_faces: bool = Form(True), user_pinecone_key: str = Form(""), user_cloudinary_url: str = Form("") ): actual_pc_key = user_pinecone_key or os.getenv("DEFAULT_PINECONE_KEY", "") actual_cld_url = user_cloudinary_url or os.getenv("DEFAULT_CLOUDINARY_URL", "") if not actual_pc_key or not actual_cld_url: raise HTTPException(400, "API Keys are missing.") folder = standardize_category_name(folder_name) creds = get_cloudinary_creds(actual_cld_url) if not creds.get("cloud_name"): raise HTTPException(400, "Invalid Cloudinary URL format.") pc = _get_pinecone(actual_pc_key) idx_obj = pc.Index(IDX_OBJECTS) idx_face = pc.Index(IDX_FACES) uploaded_urls = [] for file in files: tmp_path = f"temp_uploads/{uuid.uuid4().hex}_{sanitize_filename(file.filename)}" try: with open(tmp_path, "wb") as buf: shutil.copyfileobj(file.file, buf) res = await asyncio.to_thread(_cld_upload, tmp_path, folder, creds) image_url = res["secure_url"] uploaded_urls.append(image_url) async with _inference_sem: vectors = await ai.process_image_async(tmp_path, is_query=False, detect_faces=detect_faces) face_upserts, object_upserts = [], [] for v in vectors: vec_list = v["vector"].tolist() if hasattr(v["vector"], "tolist") else v["vector"] # ── image_url added to metadata for reverse-lookup (delete by URL) ── record = { "id": str(uuid.uuid4()), "values": vec_list, "metadata": { "url": image_url, "image_url": image_url, # legacy compat "folder": folder, } } (face_upserts if v["type"] == "face" else object_upserts).append(record) upsert_tasks = [] if face_upserts: upsert_tasks.append(asyncio.to_thread(idx_face.upsert, vectors=face_upserts)) if object_upserts: upsert_tasks.append(asyncio.to_thread(idx_obj.upsert, vectors=object_upserts)) if upsert_tasks: await asyncio.gather(*upsert_tasks) except Exception as e: print(f"❌ Upload error: {e}") raise HTTPException(500, f"Upload processing failed: {str(e)}") finally: if os.path.exists(tmp_path): os.remove(tmp_path) return {"message": "Done!", "urls": uploaded_urls} # ══════════════════════════════════════════════════════════════════ # 3. SEARCH # ══════════════════════════════════════════════════════════════════ @app.post("/api/search") async def search_database( file: UploadFile = File(...), detect_faces: bool = Form(True), user_pinecone_key: str = Form(""), user_cloudinary_url: str = Form("") ): actual_pc_key = user_pinecone_key or os.getenv("DEFAULT_PINECONE_KEY", "") if not actual_pc_key: raise HTTPException(400, "Pinecone Key is missing.") tmp_path = f"temp_uploads/query_{uuid.uuid4().hex}_{sanitize_filename(file.filename)}" try: with open(tmp_path, "wb") as buf: shutil.copyfileobj(file.file, buf) async with _inference_sem: vectors = await ai.process_image_async(tmp_path, is_query=True, detect_faces=detect_faces) pc = _get_pinecone(actual_pc_key) idx_obj = pc.Index(IDX_OBJECTS) idx_face = pc.Index(IDX_FACES) async def _query_one(vec_dict: dict): vec_list = vec_dict["vector"].tolist() if hasattr(vec_dict["vector"], "tolist") else vec_dict["vector"] target_idx = idx_face if vec_dict["type"] == "face" else idx_obj try: res = await asyncio.to_thread(target_idx.query, vector=vec_list, top_k=10, include_metadata=True) except Exception as e: if "404" in str(e): raise HTTPException(404, "Pinecone Index not found. Go to Settings → Configuration → Verify Keys.") raise e out = [] for match in res.get("matches", []): score = match["score"] is_face = vec_dict["type"] == "face" if is_face: RAW_THRESHOLD = 0.35 if score < RAW_THRESHOLD: continue ui_score = 0.75 + ((score - RAW_THRESHOLD) / (1.0 - RAW_THRESHOLD)) * 0.24 ui_score = min(0.99, ui_score) else: if score < 0.45: continue ui_score = score caption = "👤 Verified Identity" if is_face else match["metadata"].get("folder", "🎯 Object Match") out.append({ "url": match["metadata"].get("url") or match["metadata"].get("image_url", ""), "score": round(ui_score, 4), "caption": caption, }) return out nested = await asyncio.gather(*[_query_one(v) for v in vectors]) all_results = [r for sub in nested for r in sub] seen = {} for r in all_results: url = r["url"] if url not in seen or r["score"] > seen[url]["score"]: seen[url] = r return {"results": sorted(seen.values(), key=lambda x: x["score"], reverse=True)[:10]} except HTTPException: raise except Exception as e: print(f"❌ Search error: {e}") raise HTTPException(500, str(e)) finally: if os.path.exists(tmp_path): os.remove(tmp_path) # ══════════════════════════════════════════════════════════════════ # 4. CATEGORIES # ══════════════════════════════════════════════════════════════════ @app.post("/api/categories") async def get_categories(user_cloudinary_url: str = Form("")): actual_cld_url = user_cloudinary_url or os.getenv("DEFAULT_CLOUDINARY_URL", "") if not actual_cld_url: return {"categories": []} try: creds = get_cloudinary_creds(actual_cld_url) if not creds.get("cloud_name"): return {"categories": []} result = await asyncio.to_thread(_cld_root_folders, creds) return {"categories": [f["name"] for f in result.get("folders", [])]} except Exception as e: print(f"Category fetch error: {e}") return {"categories": []} # ══════════════════════════════════════════════════════════════════ # 5. CLOUDINARY FOLDER IMAGES (File Explorer) # ══════════════════════════════════════════════════════════════════ @app.post("/api/cloudinary/folder-images") async def get_folder_images( user_cloudinary_url: str = Form(""), folder_name: str = Form(...) ): actual_cld_url = user_cloudinary_url or os.getenv("DEFAULT_CLOUDINARY_URL", "") if not actual_cld_url: raise HTTPException(400, "Cloudinary URL is required.") creds = get_cloudinary_creds(actual_cld_url) if not creds.get("cloud_name"): raise HTTPException(400, "Invalid Cloudinary URL.") try: images = [] next_cursor = None while True: result = await asyncio.to_thread( _cld_resources_by_folder, folder_name, creds, 100, next_cursor ) for r in result.get("resources", []): images.append({ "url": r["secure_url"], "public_id": r["public_id"], "format": r.get("format", ""), "width": r.get("width", 0), "height": r.get("height", 0), "bytes": r.get("bytes", 0), }) next_cursor = result.get("next_cursor") if not next_cursor: break return {"images": images} except Exception as e: print(f"Folder images error: {e}") raise HTTPException(500, str(e)) # ══════════════════════════════════════════════════════════════════ # 6. DELETE SINGLE IMAGE (File Explorer) # Deletes from Cloudinary + removes matching vectors from Pinecone # ══════════════════════════════════════════════════════════════════ @app.post("/api/delete-image") async def delete_image( image_url: str = Form(...), user_cloudinary_url: str = Form(""), user_pinecone_key: str = Form("") ): actual_cld_url = user_cloudinary_url or os.getenv("DEFAULT_CLOUDINARY_URL", "") actual_pc_key = user_pinecone_key or os.getenv("DEFAULT_PINECONE_KEY", "") if not actual_cld_url or not actual_pc_key: raise HTTPException(400, "Both Cloudinary URL and Pinecone Key are required.") creds = get_cloudinary_creds(actual_cld_url) if not creds.get("cloud_name"): raise HTTPException(400, "Invalid Cloudinary URL.") errors = [] # ── 1. Delete from Cloudinary ──────────────────────────────── public_id = url_to_public_id(image_url, creds["cloud_name"]) if public_id: try: await asyncio.to_thread(_cld_delete_resource, public_id, creds) except Exception as e: errors.append(f"Cloudinary deletion failed: {e}") else: errors.append("Could not extract public_id from URL — Cloudinary deletion skipped.") # ── 2. Delete matching vectors from Pinecone by metadata filter ─ # Works only for vectors uploaded AFTER the image_url metadata fix. # Old vectors (without image_url metadata) will not be found — that's expected. try: pc = _get_pinecone(actual_pc_key) idx_obj = pc.Index(IDX_OBJECTS) idx_face = pc.Index(IDX_FACES) for idx in [idx_obj, idx_face]: try: # Query for vectors with this exact image_url in metadata matches = await asyncio.to_thread( idx.query, vector=[0.0] * (1536 if idx == idx_obj else 512), top_k=50, include_metadata=True, filter={"image_url": {"$eq": image_url}} ) ids_to_delete = [m["id"] for m in matches.get("matches", []) if m.get("metadata", {}).get("image_url") == image_url or m.get("metadata", {}).get("url") == image_url] if ids_to_delete: await asyncio.to_thread(idx.delete, ids=ids_to_delete) except Exception as e: # Pinecone filter queries may not be supported on all plan types — log but don't fail print(f"Pinecone vector delete attempt failed (non-critical): {e}") except Exception as e: errors.append(f"Pinecone deletion failed: {e}") if len(errors) == 2: raise HTTPException(500, " | ".join(errors)) return {"message": "Image deleted.", "warnings": errors} # ══════════════════════════════════════════════════════════════════ # 7. RESET DATABASE # Deletes ALL Cloudinary images + wipes + recreates Pinecone indexes # Only operates on the user's OWN keys (never defaults) # ══════════════════════════════════════════════════════════════════ @app.post("/api/reset-database") async def reset_database( user_pinecone_key: str = Form(...), user_cloudinary_url: str = Form(...) ): if not user_pinecone_key or not user_cloudinary_url: raise HTTPException(400, "Your own Pinecone Key and Cloudinary URL are required. This endpoint only operates on personal databases.") # Guard: refuse to operate on server default keys default_pc = os.getenv("DEFAULT_PINECONE_KEY", "") default_cld = os.getenv("DEFAULT_CLOUDINARY_URL", "") if user_pinecone_key == default_pc or user_cloudinary_url == default_cld: raise HTTPException(403, "Cannot reset the shared default database. This operation is only permitted on your personal database.") creds = get_cloudinary_creds(user_cloudinary_url) if not creds.get("cloud_name"): raise HTTPException(400, "Invalid Cloudinary URL.") errors = [] # ── 1. Wipe ALL Cloudinary resources ─────────────────────── try: # Get all root folders and delete resources + folders folders_res = await asyncio.to_thread(_cld_root_folders, creds) folders = [f["name"] for f in folders_res.get("folders", [])] for folder in folders: try: await asyncio.to_thread(_cld_delete_by_prefix, f"{folder}/", creds) await asyncio.to_thread(_cld_delete_folder, folder, creds) except Exception as e: errors.append(f"Cloudinary folder '{folder}' error: {e}") except Exception as e: errors.append(f"Cloudinary wipe error: {e}") # ── 2. Delete and recreate Pinecone indexes ───────────────── try: pc = _get_pinecone(user_pinecone_key) existing = {idx.name for idx in await asyncio.to_thread(pc.list_indexes)} delete_tasks = [] if IDX_OBJECTS in existing: delete_tasks.append(asyncio.to_thread(pc.delete_index, IDX_OBJECTS)) if IDX_FACES in existing: delete_tasks.append(asyncio.to_thread(pc.delete_index, IDX_FACES)) if delete_tasks: await asyncio.gather(*delete_tasks) # Wait a moment for deletion to propagate await asyncio.sleep(3) create_tasks = [ asyncio.to_thread(pc.create_index, name=IDX_OBJECTS, dimension=1536, metric="cosine", spec=ServerlessSpec(cloud="aws", region="us-east-1")), asyncio.to_thread(pc.create_index, name=IDX_FACES, dimension=512, metric="cosine", spec=ServerlessSpec(cloud="aws", region="us-east-1")), ] await asyncio.gather(*create_tasks) # Evict this key from the pool so we get fresh index handles if user_pinecone_key in _pinecone_pool: del _pinecone_pool[user_pinecone_key] except Exception as e: errors.append(f"Pinecone reset error: {e}") if errors: return {"message": "Reset completed with some errors.", "warnings": errors} return {"message": "Database wiped and recreated successfully."} # ══════════════════════════════════════════════════════════════════ # 8. DELETE ACCOUNT # Full wipe: Cloudinary + Pinecone + Supabase settings row # Note: Supabase auth user deletion must be done client-side # via supabase.auth.admin or the user's own session. # ══════════════════════════════════════════════════════════════════ @app.post("/api/delete-account") async def delete_account( user_pinecone_key: str = Form(...), user_cloudinary_url: str = Form(...), user_id: str = Form(...) ): if not user_pinecone_key or not user_cloudinary_url: raise HTTPException(400, "Your own API keys are required.") # Guard against default keys default_pc = os.getenv("DEFAULT_PINECONE_KEY", "") default_cld = os.getenv("DEFAULT_CLOUDINARY_URL", "") if user_pinecone_key == default_pc or user_cloudinary_url == default_cld: raise HTTPException(403, "Cannot delete using shared default keys.") # Reuse reset logic for data wipe creds = get_cloudinary_creds(user_cloudinary_url) # Wipe Cloudinary try: folders_res = await asyncio.to_thread(_cld_root_folders, creds) for f in folders_res.get("folders", []): await asyncio.to_thread(_cld_delete_by_prefix, f"{f['name']}/", creds) await asyncio.to_thread(_cld_delete_folder, f["name"], creds) except Exception as e: print(f"Account delete — Cloudinary error: {e}") # Wipe + delete Pinecone indexes try: pc = _get_pinecone(user_pinecone_key) existing = {idx.name for idx in await asyncio.to_thread(pc.list_indexes)} tasks = [] if IDX_OBJECTS in existing: tasks.append(asyncio.to_thread(pc.delete_index, IDX_OBJECTS)) if IDX_FACES in existing: tasks.append(asyncio.to_thread(pc.delete_index, IDX_FACES)) if tasks: await asyncio.gather(*tasks) if user_pinecone_key in _pinecone_pool: del _pinecone_pool[user_pinecone_key] except Exception as e: print(f"Account delete — Pinecone error: {e}") # Note: Supabase user deletion & settings row deletion is handled # client-side via supabase.auth.signOut() after this endpoint returns. # The Supabase ON DELETE CASCADE on user_settings + user_folders # handles row cleanup automatically when the auth user is deleted. return {"message": "Account data wiped. Please confirm deletion in your Supabase dashboard if needed."} # ══════════════════════════════════════════════════════════════════ # 9. HEALTH CHECK # ══════════════════════════════════════════════════════════════════ @app.get("/api/health") async def health(): return {"status": "ok"}