Update app.py

app.py

@@ -1,22 +1,16 @@
-"""
-Enhanced FastAPI: Matchmaking Recommendation API
-Features:
-  - /health: Server status
-  - /recommend/{user_id}: Get top N matches with compatibility scores
-  - /feedback: Record user likes/rejects
-  - Cold-start capable: Any user can get recommendations
-  - CORS enabled for frontend integration
-"""
-
 import os
-import pickle
 import httpx
 import numpy as np
-from datetime import datetime
+import asyncio
+from datetime import datetime, timedelta
 from fastapi import FastAPI, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from sklearn.metrics.pairwise import cosine_similarity
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.preprocessing import StandardScaler
 import logging
+from functools import lru_cache
+import hashlib
 
 # Setup logging
 logging.basicConfig(
@@ -29,15 +23,17 @@ logger = logging.getLogger(__name__)
 SUPABASE_URL = os.getenv('SUPABASE_URL', 'https://nquhiryqtbrtpauuxmsc.supabase.co')
 SERVICE_KEY = os.getenv('SUPABASE_SERVICE_KEY', '')
 HEADERS = {'apikey': SERVICE_KEY, 'Content-Type': 'application/json'}
-MODEL_PATH = os.getenv('MODEL_PATH', 'models/recommendation_model.pkl')
+
+# Cache settings
+CACHE_TTL = 300  # 5 minutes
+PROFILE_BATCH_SIZE = 100  # Fetch profiles in batches
 
 app = FastAPI(
-    title="Matchmaking Recommendation API",
-    version="2.0.0",
-    description="Smart matchmaking using ML-based compatibility scoring"
+    title="Real-time Matchmaking API",
+    version="3.0.0",
+    description="Live ML-based matching without pickle files - instant updates"
 )
 
-# CORS middleware
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],
@@ -46,41 +42,82 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
-# Global model
-model = None
-model_loaded_at = None
+# Global cache
+class MatchmakingCache:
+    def __init__(self):
+        self.profiles = []
+        self.feature_matrix = None
+        self.profile_ids = []
+        self.last_updated = None
+        self.cache_valid = False
+        self.tfidf_vectorizers = {}
+        self.scaler = StandardScaler()
+        
+    def is_expired(self):
+        """Check if cache is expired"""
+        if not self.last_updated:
+            return True
+        return datetime.now() - self.last_updated > timedelta(seconds=CACHE_TTL)
+    
+    def invalidate(self):
+        """Invalidate cache"""
+        self.cache_valid = False
+        logger.info("Cache invalidated")
 
 
-def load_model():
-    """Load trained model from disk"""
-    global model, model_loaded_at
-    
-    if not os.path.exists(MODEL_PATH):
-        logger.error(f"Model file not found at {MODEL_PATH}")
-        raise Exception(f"Model not found at {MODEL_PATH}")
-    
-    try:
-        with open(MODEL_PATH, 'rb') as f:
-            model = pickle.load(f)
-        model_loaded_at = datetime.now()
-        logger.info(f"✓ Model loaded: {model['n_profiles']} profiles, trained at {model['trained_at']}")
-        return True
-    except Exception as e:
-        logger.error(f"✗ Error loading model: {e}")
-        raise
+cache = MatchmakingCache()
 
+# Feature config
+CATEGORICAL_COLS = ['religion', 'marital_status', 'qualification', 'country', 'maslak', 'region_caste']
+TEXT_COLS = ['hobbies', 'personality_traits', 'preferred_partner_criteria']
 
-@app.on_event("startup")
-async def startup():
-    """Load model on startup"""
+
+async def fetch_profiles_paginated(limit=1000):
+    """Fetch profiles with pagination"""
+    all_profiles = []
+    offset = 0
+    
     try:
-        load_model()
+        while offset < limit:
+            try:
+                resp = httpx.get(
+                    f"{SUPABASE_URL}/rest/v1/profiles",
+                    headers=HEADERS,
+                    params={
+                        'select': '*',
+                        'is_active': 'eq.true',
+                        'limit': PROFILE_BATCH_SIZE,
+                        'offset': offset
+                    },
+                    timeout=15
+                )
+                
+                if resp.status_code != 200:
+                    logger.warning(f"Fetch failed at offset {offset}: {resp.status_code}")
+                    break
+                
+                batch = resp.json()
+                if not batch:
+                    break
+                
+                all_profiles.extend(batch)
+                offset += PROFILE_BATCH_SIZE
+                logger.debug(f"Fetched {len(batch)} profiles, total: {len(all_profiles)}")
+                
+            except Exception as e:
+                logger.warning(f"Error fetching batch: {e}")
+                break
+        
+        logger.info(f"✓ Fetched {len(all_profiles)} active profiles")
+        return all_profiles
+        
     except Exception as e:
-        logger.error(f"[STARTUP ERROR] {e}")
+        logger.error(f"✗ Error in pagination: {e}")
+        return all_profiles if all_profiles else []
 
 
 def safe_str(val):
-    """Safe string conversion with None and list handling"""
+    """Safe string conversion"""
     if val is None:
         return ''
     if isinstance(val, list):
@@ -88,184 +125,271 @@ def safe_str(val):
     return str(val).strip().lower()
 
 
-def preprocess_user(user_data):
-    """Extract and clean features from Supabase user dict"""
-    # Age
+def preprocess_profile(profile):
+    """Convert Supabase profile to feature-ready dict"""
     try:
-        age = int(user_data.get('age') or 25)
-        age = max(18, min(100, age))  # Bounds check
+        age = int(profile.get('age') or 25)
+        age = max(18, min(100, age))  # Bounds
     except:
         age = 25
 
-    # Categoricals in order: religion, marital_status, qualification, country, maslak, region_caste
-    cat_cols = ['religion', 'marital_status', 'qualification', 'country', 'maslak', 'region_caste']
-    cat_vals = [safe_str(user_data.get(col)) or 'unknown' for col in cat_cols]
+    cat = {col: safe_str(profile.get(col)) or 'unknown' for col in CATEGORICAL_COLS}
+    text = {col: safe_str(profile.get(col)) for col in TEXT_COLS}
 
-    # Text fields
-    text_fields = {
-        'hobbies': safe_str(user_data.get('hobbies')),
-        'personality_traits': safe_str(user_data.get('personality_traits')),
-        'preferred_partner_criteria': safe_str(user_data.get('preferred_partner_criteria'))
+    return {
+        'id': profile['id'],
+        'full_name': profile.get('full_name', 'Unknown'),
+        'age': age,
+        'gender': safe_str(profile.get('gender')),
+        'city': profile.get('city', ''),
+        'religion': cat['religion'],
+        'marital_status': cat['marital_status'],
+        'qualification': cat['qualification'],
+        'country': cat['country'],
+        'maslak': cat['maslak'],
+        'region_caste': cat['region_caste'],
+        'hobbies': text['hobbies'],
+        'personality_traits': text['personality_traits'],
+        'preferred_partner_criteria': text['preferred_partner_criteria'],
+        'profile_picture_url': profile.get('profile_picture_url'),
+        'is_active': profile.get('is_active', True)
     }
 
-    return age, cat_vals, text_fields
+
+def build_categorical_features(profiles):
+    """Build one-hot encoded categorical features"""
+    cat_features = []
+    
+    for col in CATEGORICAL_COLS:
+        # Get unique values from profiles
+        unique_vals = sorted(set(p[col] for p in profiles if p[col]))
+        
+        # Create mapping
+        val_to_idx = {val: idx for idx, val in enumerate(unique_vals)}
+        
+        # Encode
+        encoded = []
+        for p in profiles:
+            val = p[col]
+            idx = val_to_idx.get(val, 0)
+            # Create one-hot vector
+            one_hot = np.zeros(len(unique_vals))
+            one_hot[idx] = 1
+            encoded.append(one_hot)
+        
+        cat_features.append(np.array(encoded))
+    
+    return np.hstack(cat_features) if cat_features else np.array([])
+
+
+def build_text_features(profiles, vectorizers=None):
+    """Build TF-IDF features for text columns"""
+    text_features = []
+    new_vectorizers = {}
+    
+    for col in TEXT_COLS:
+        corpus = [p[col] for p in profiles]
+        
+        if vectorizers and col in vectorizers:
+            # Use existing vectorizer
+            vec = vectorizers[col].transform(corpus).toarray()
+            new_vectorizers[col] = vectorizers[col]
+        else:
+            # Create new vectorizer
+            tfidf = TfidfVectorizer(
+                max_features=15,
+                stop_words='english',
+                token_pattern=r'(?u)\b\w+\b',
+                min_df=1
+            )
+            vec = tfidf.fit_transform(corpus).toarray()
+            new_vectorizers[col] = tfidf
+        
+        text_features.append(vec)
+    
+    return np.hstack(text_features) if text_features else np.array([]), new_vectorizers
 
 
-def encode_user(age, cat_vals, text_fields, le_dict, scaler, tfidf_dict):
-    """
-    Build feature vector for user using fitted transformers.
-    Order: age (1) + categoricals (6) + tfidf_hobbies (20) + tfidf_traits (20) + tfidf_criteria (20)
-    Total: 1 + 6 + 60 = 67 dimensions
-    """
+def build_feature_matrix(profiles, vectorizers=None):
+    """Build complete feature matrix: age + categoricals + text"""
+    n = len(profiles)
+    
+    if n == 0:
+        return np.array([]), {}, None
+    
     features = []
+    
+    # 1. Age (normalized)
+    ages = np.array([[float(p['age'])] for p in profiles])
+    scaler = StandardScaler()
+    ages_norm = scaler.fit_transform(ages)
+    features.append(ages_norm)
+    
+    # 2. Categorical features (one-hot encoded)
+    cat_feat = build_categorical_features(profiles)
+    if cat_feat.size > 0:
+        features.append(cat_feat)
+    
+    # 3. Text features (TF-IDF)
+    text_feat, new_vectorizers = build_text_features(profiles, vectorizers)
+    if text_feat.size > 0:
+        features.append(text_feat)
+    
+    X = np.hstack(features)
+    
+    logger.info(f"Feature matrix built: {X.shape}")
+    return X, new_vectorizers, scaler
 
-    # 1. Scale age
+
+async def refresh_cache():
+    """Refresh profile cache from Supabase"""
+    global cache
+    
     try:
-        age_scaled = scaler.transform([[float(age)]])[0]
-        features.extend(age_scaled)
+        logger.info("🔄 Refreshing profile cache...")
+        
+        # Fetch profiles
+        raw_profiles = await fetch_profiles_paginated()
+        
+        if not raw_profiles or len(raw_profiles) < 5:
+            logger.warning(f"⚠️ Only {len(raw_profiles)} profiles found")
+            return False
+        
+        # Preprocess
+        profiles = [preprocess_profile(p) for p in raw_profiles]
+        
+        # Build features
+        X, vectorizers, scaler = build_feature_matrix(profiles, cache.tfidf_vectorizers)
+        
+        if X.size == 0:
+            logger.warning("Feature matrix is empty")
+            return False
+        
+        # Update cache
+        cache.profiles = profiles
+        cache.feature_matrix = X
+        cache.profile_ids = [p['id'] for p in profiles]
+        cache.tfidf_vectorizers = vectorizers
+        cache.scaler = scaler
+        cache.last_updated = datetime.now()
+        cache.cache_valid = True
+        
+        logger.info(f"✓ Cache refreshed: {len(profiles)} profiles, {X.shape[1]} features")
+        return True
+        
     except Exception as e:
-        logger.warning(f"Error scaling age: {e}")
-        features.extend([0.5])  # Default: middle of range
-
-    # 2. Encode categoricals (in same order as training)
-    cat_cols = ['religion', 'marital_status', 'qualification', 'country', 'maslak', 'region_caste']
-    for col, val in zip(cat_cols, cat_vals):
-        le = le_dict[col]
-        
-        # Handle unknown categories
-        if val not in le.classes_:
-            if 'unknown' in le.classes_:
-                val = 'unknown'
-            else:
-                val = le.classes_[0]
-        
-        try:
-            encoded = le.transform([val])[0]
-            features.append(float(encoded))
-        except Exception as e:
-            logger.warning(f"Error encoding {col}={val}: {e}")
-            features.append(0.0)
-
-    # 3. TF-IDF for text fields
-    text_cols = ['hobbies', 'personality_traits', 'preferred_partner_criteria']
-    for col in text_cols:
-        try:
-            tfidf = tfidf_dict[col]
-            vec = tfidf.transform([text_fields[col]]).toarray()[0]
-            features.extend(vec)
-        except Exception as e:
-            logger.warning(f"Error TF-IDF encoding {col}: {e}")
-            features.extend([0.0] * 20)  # 20 default features
-
-    return np.array(features).reshape(1, -1)
+        logger.error(f"✗ Cache refresh failed: {e}")
+        return False
+
+
+async def get_cached_profiles():
+    """Get profiles, refresh if needed"""
+    if not cache.cache_valid or cache.is_expired():
+        success = await refresh_cache()
+        if not success:
+            if not cache.profiles:
+                raise Exception("No profiles available")
+    return cache.profiles, cache.feature_matrix
+
+
+@app.on_event("startup")
+async def startup():
+    """Initial cache load"""
+    try:
+        await refresh_cache()
+        logger.info("✓ Startup cache loaded")
+    except Exception as e:
+        logger.warning(f"⚠️ Startup cache load failed: {e}")
 
 
 @app.get("/health")
 async def health():
-    """Health check endpoint"""
+    """Health check"""
     return {
         "status": "running",
-        "model_loaded": model is not None,
-        "model_profiles": model['n_profiles'] if model else 0,
-        "model_trained_at": model['trained_at'] if model else None,
-        "api_loaded_at": model_loaded_at.isoformat() if model_loaded_at else None
+        "cache_valid": cache.cache_valid,
+        "profiles_cached": len(cache.profiles),
+        "last_updated": cache.last_updated.isoformat() if cache.last_updated else None,
+        "cache_expires_in": (
+            max(0, int(CACHE_TTL - (datetime.now() - cache.last_updated).total_seconds()))
+            if cache.last_updated else 0
+        )
     }
 
 
 @app.get("/stats")
 async def stats():
-    """Get model statistics"""
-    if not model:
-        raise HTTPException(503, "Model not loaded")
+    """Model statistics"""
+    profiles, X = await get_cached_profiles()
+    
+    if X is None:
+        return {"error": "Model not ready"}
     
     return {
-        "total_profiles": model['n_profiles'],
-        "feature_dimensions": model['feature_matrix'].shape[1],
-        "model_version": model['version'],
-        "trained_at": model['trained_at'],
-        "categorical_features": len(['religion', 'marital_status', 'qualification', 'country', 'maslak', 'region_caste']),
-        "text_features": len(['hobbies', 'personality_traits', 'preferred_partner_criteria']),
-        "numeric_features": 1  # age
+        "total_profiles": len(profiles),
+        "feature_dimensions": X.shape[1],
+        "categorical_features": len(CATEGORICAL_COLS),
+        "text_features": len(TEXT_COLS),
+        "numeric_features": 1,
+        "cache_age_seconds": (datetime.now() - cache.last_updated).total_seconds() if cache.last_updated else None,
+        "active_profiles": sum(1 for p in profiles if p['is_active'])
     }
 
 
 @app.get("/recommend/{user_id}")
 async def recommend(user_id: str, top_n: int = 10, min_score: float = 0.3):
-    """
-    Get top N compatible matches for a user
-    
-    Args:
-        user_id: UUID of the user
-        top_n: Number of recommendations (default: 10, max: 50)
-        min_score: Minimum compatibility score (0.0-1.0, default: 0.3)
-    
-    Returns:
-        list of matches with compatibility scores
-    """
-    
-    if not model:
-        raise HTTPException(503, "Model not loaded")
+    """Real-time recommendations"""
     
     # Validate params
     top_n = min(max(1, top_n), 50)
     min_score = max(0.0, min(1.0, min_score))
-
+    
     try:
-        # 1. Fetch query user from Supabase
-        logger.info(f"Fetching user {user_id}")
-        resp = httpx.get(
-            f"{SUPABASE_URL}/rest/v1/profiles",
-            headers=HEADERS,
-            params={'id': f'eq.{user_id}', 'select': '*'},
-            timeout=10
-        )
-        
-        if resp.status_code != 200:
-            logger.warning(f"User fetch failed: {resp.status_code}")
-            raise HTTPException(404, "User not found")
+        # Get fresh profiles
+        profiles, X = await get_cached_profiles()
         
-        users = resp.json()
-        if not users:
-            raise HTTPException(404, "User not found")
+        if X is None or len(profiles) < 2:
+            raise HTTPException(503, "Insufficient profiles for matching")
         
-        user = users[0]
+        # Find query user
+        user = None
+        user_idx = None
+        for i, p in enumerate(profiles):
+            if p['id'] == user_id:
+                user = p
+                user_idx = i
+                break
         
-        # 2. Validate gender
-        user_gender = safe_str(user.get('gender'))
-        if user_gender not in ('male', 'female'):
-            raise HTTPException(400, "User gender must be 'male' or 'female'")
+        if user is None:
+            raise HTTPException(404, "User not found")
         
-        opposite_gender = 'female' if user_gender == 'male' else 'male'
+        # Validate gender
+        if user['gender'] not in ('male', 'female'):
+            raise HTTPException(400, "Invalid gender")
         
-        # 3. Encode query user
-        age, cat_vals, text_fields = preprocess_user(user)
-        user_vec = encode_user(
-            age, cat_vals, text_fields,
-            model['label_encoders'],
-            model['scaler'],
-            model['tfidf_vectorizers']
-        )
+        opposite_gender = 'female' if user['gender'] == 'male' else 'male'
         
-        # 4. Get candidates (opposite gender, active)
-        candidates = []
-        for i, p in enumerate(model['profile_data']):
-            if (p.get('gender') == opposite_gender and 
-                p.get('is_active') and 
-                p.get('id') != user_id):  # Exclude self
-                candidates.append((i, p))
+        # Find candidate indices (opposite gender, active, not self)
+        cand_indices = [
+            i for i, p in enumerate(profiles)
+            if p['gender'] == opposite_gender and p['is_active'] and p['id'] != user_id
+        ]
         
-        if len(candidates) < 1:
-            logger.info(f"No candidates found for {user_id}")
-            return {"matches": []}
+        if not cand_indices:
+            return {
+                "user_id": user_id,
+                "matches": [],
+                "total_matches": 0,
+                "query_timestamp": datetime.utcnow().isoformat()
+            }
         
-        # 5. Get candidate feature vectors
-        cand_indices = [c[0] for c in candidates]
-        cand_matrix = model['feature_matrix'][cand_indices]
+        # Compute similarity
+        user_vec = X[user_idx].reshape(1, -1)
+        cand_matrix = X[cand_indices]
         
-        # 6. Compute cosine similarity
         sims = cosine_similarity(user_vec, cand_matrix)[0]
         
-        # 7. Get top N matches above threshold
+        # Get top N
         top_positions = np.argsort(sims)[::-1][:top_n]
         
         matches = []
@@ -275,16 +399,17 @@ async def recommend(user_id: str, top_n: int = 10, min_score: float = 0.3):
             if score < min_score:
                 continue
             
-            candidate = candidates[pos][1]
+            cand_idx = cand_indices[pos]
+            cand = profiles[cand_idx]
+            
             matches.append({
-                "user_id": candidate['id'],
-                "name": candidate.get('full_name', 'Unknown'),
-                "age": candidate.get('age', 0),
-                "city": candidate.get('city', ''),
-                "gender": candidate.get('gender', ''),
-                "religion": candidate.get('religion', ''),
+                "user_id": cand['id'],
+                "name": cand['full_name'],
+                "age": cand['age'],
+                "city": cand['city'],
+                "religion": cand['religion'],
                 "compatibility_score": round(score * 100, 1),
-                "photo_url": candidate.get('profile_picture_url')
+                "photo_url": cand['profile_picture_url']
             })
         
         logger.info(f"Generated {len(matches)} recommendations for {user_id}")
@@ -305,30 +430,18 @@ async def recommend(user_id: str, top_n: int = 10, min_score: float = 0.3):
 
 @app.post("/feedback")
 async def feedback(request: dict):
-    """
-    Record user feedback (like/reject)
-    
-    Args:
-        request: {user_id, target_id, action}
-        action: "like" or "reject"
-    
-    Returns:
-        status confirmation
-    """
-    
+    """Record user feedback (like/reject)"""
     try:
         user_id = request.get('user_id')
         target_id = request.get('target_id')
         action = request.get('action', '').lower()
         
-        # Validate input
         if not user_id or not target_id:
             raise HTTPException(400, "user_id and target_id required")
         
         if action not in ('like', 'reject'):
             raise HTTPException(400, "action must be 'like' or 'reject'")
         
-        # Prepare data for Supabase
         data = {
             'user_id': user_id,
             'matched_user_id': target_id,
@@ -337,9 +450,6 @@ async def feedback(request: dict):
             'created_at': datetime.utcnow().isoformat()
         }
         
-        # Upsert to Supabase matches table
-        logger.info(f"Recording feedback: {user_id} {action} {target_id}")
-        
         resp = httpx.post(
             f"{SUPABASE_URL}/rest/v1/matches",
             headers=HEADERS,
@@ -348,10 +458,10 @@ async def feedback(request: dict):
         )
         
         if resp.status_code in (200, 201, 204):
-            logger.info(f"Feedback recorded successfully")
+            logger.info(f"Feedback recorded: {user_id} {action} {target_id}")
             return {"status": "ok", "message": "Feedback recorded"}
         else:
-            logger.warning(f"Supabase response: {resp.status_code} - {resp.text}")
+            logger.warning(f"Supabase error: {resp.status_code}")
             return {"status": "error", "message": resp.text}
     
     except HTTPException:
@@ -361,19 +471,35 @@ async def feedback(request: dict):
         return {"status": "error", "message": str(e)}
 
 
+@app.post("/refresh-cache")
+async def manual_refresh():
+    """Manually trigger cache refresh"""
+    try:
+        success = await refresh_cache()
+        if success:
+            return {"status": "ok", "message": "Cache refreshed"}
+        else:
+            return {"status": "error", "message": "Refresh failed"}
+    except Exception as e:
+        logger.error(f"Manual refresh error: {e}")
+        return {"status": "error", "message": str(e)}
+
+
 @app.get("/")
 async def root():
     """Root endpoint"""
     return {
-        "name": "Matchmaking Recommendation API",
-        "version": "2.0.0",
+        "name": "Real-time Matchmaking API",
+        "version": "3.0.0",
+        "architecture": "No pickle files - Live from Supabase",
         "endpoints": {
             "health": "/health",
             "stats": "/stats",
-            "recommend": "/recommend/{user_id}",
-            "feedback": "/feedback"
-        },
-        "docs": "/docs"
+            "recommend": "/recommend/{user_id}?top_n=10&min_score=0.3",
+            "feedback": "/feedback",
+            "refresh_cache": "/refresh-cache",
+            "docs": "/docs"
+        }
     }
 
 
@@ -382,5 +508,5 @@ if __name__ == '__main__':
     port = int(os.getenv('PORT', 7860))
     host = os.getenv('HOST', '0.0.0.0')
     
-    logger.info(f"Starting server on {host}:{port}")
+    logger.info(f"Starting Real-time Matchmaking API on {host}:{port}")
     uvicorn.run(app, host=host, port=port, log_level="info")