import os import httpx import numpy as np import asyncio import traceback import uuid import math 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 # ── Logging ──────────────────────────────────────────────────────────────── logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # ── Config ───────────────────────────────────────────────────────────────── SUPABASE_URL = os.getenv('SUPABASE_URL', 'https://nquhiryqtbrtpauuxmsc.supabase.co') SERVICE_KEY = os.getenv('SUPABASE_SERVICE_KEY', 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6Im5xdWhpcnlxdGJydHBhdXV4bXNjIiwicm9sZSI6InNlcnZpY2Vfcm9sZSIsImlhdCI6MTc3NjA4MzcxMCwiZXhwIjoyMDkxNjU5NzEwfQ.q7ysbaSfMv15g7PyJRLMwnRpEyia5d_ol3iLZn0Xayo') HEADERS = {'apikey': SERVICE_KEY, 'Content-Type': 'application/json'} CACHE_TTL = 300 PROFILE_BATCH_SIZE = 100 # ── Soft-score weights (must sum to 1.0) ─────────────────────────────────── WEIGHTS = { 'age_compatibility': 0.20, 'personality_traits': 0.25, 'partner_criteria': 0.25, 'hobbies': 0.15, 'categorical_bonus': 0.15, } assert abs(sum(WEIGHTS.values()) - 1.0) < 1e-6, "Weights must sum to 1.0" AGE_IDEAL_GAP = 3 # years — peak of Gaussian AGE_SIGMA = 7 # tolerance in years # ── Marital status compatibility groups ──────────────────────────────────── # Only statuses listed in the same set may match each other. MARITAL_GROUPS: dict[str, set[str]] = { 'single': {'single'}, 'divorced': {'divorced', 'widowed'}, 'widowed': {'divorced', 'widowed'}, 'separated': {'separated'}, } # ── Trait synonym expansion ──────────────────────────────────────────────── TRAIT_SYNONYMS: dict[str, list[str]] = { 'caring': ['caring', 'care', 'compassionate', 'nurturing', 'kind'], 'honest': ['honest', 'truthful', 'sincere', 'transparent'], 'religious': ['religious', 'practicing', 'devout', 'faith', 'deen', 'allah', 'god'], 'family-oriented': ['family', 'family-oriented', 'home', 'homemaker', 'domestic'], 'ambitious': ['ambitious', 'driven', 'goal-oriented', 'career', 'successful'], 'respectful': ['respectful', 'respect', 'humble', 'obedient'], 'educated': ['educated', 'degree', 'professional', 'literate'], 'loyal': ['loyal', 'loyalty', 'faithful', 'committed', 'dedicated'], 'patient': ['patient', 'patience', 'calm', 'composed', 'tolerant'], 'understanding': ['understanding', 'empathetic', 'supportive', 'considerate'], } CATEGORICAL_COLS = ['religion', 'marital_status', 'qualification', 'country', 'maslak', 'region_caste'] TEXT_COLS = ['hobbies', 'personality_traits', 'preferred_partner_criteria'] # ── FastAPI app ──────────────────────────────────────────────────────────── app = FastAPI( title="Real-time Matchmaking API", version="4.0.0", description="Hard-filter + weighted soft-score matchmaking. Religion & marital status are mandatory matches." ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ══════════════════════════════════════════════════════════════════════════ # HELPERS # ══════════════════════════════════════════════════════════════════════════ def safe_str(val) -> str: if val is None: return '' if isinstance(val, list): return ' '.join(str(v) for v in val if v) return str(val).strip().lower() def expand_traits(text: str) -> str: """ Expand trait keywords so 'caring' also pulls in 'compassionate', 'nurturing', etc. Improves recall when users phrase traits differently. """ if not text: return '' lower = text.lower() extras: list[str] = [] for synonyms in TRAIT_SYNONYMS.values(): if any(s in lower for s in synonyms): extras.extend(synonyms) return text + ' ' + ' '.join(extras) if extras else text def text_similarity(text_a: str, text_b: str) -> float: """TF-IDF cosine similarity in [0, 1]. Returns 0 if either text is empty.""" a = text_a.strip() b = text_b.strip() if not a or not b: return 0.0 try: vect = TfidfVectorizer(stop_words='english', min_df=1, token_pattern=r'(?u)\b\w+\b') matrix = vect.fit_transform([a, b]) score = cosine_similarity(matrix[0:1], matrix[1:2])[0][0] return float(np.clip(score, 0.0, 1.0)) except Exception: return 0.0 def age_score(user_age: int, candidate_age: int, user_gender: str) -> float: """ Gaussian age-compatibility score. Males ideally prefer slightly younger; females prefer slightly older. """ ideal_gap = AGE_IDEAL_GAP if user_gender == 'male' else -AGE_IDEAL_GAP gap = (candidate_age - user_age) - ideal_gap return float(math.exp(-(gap ** 2) / (2 * AGE_SIGMA ** 2))) # ── Hard filter functions ────────────────────────────────────────────────── def religion_compatible(user_religion: str, candidate_religion: str) -> bool: """Strict exact match. Missing religion → reject.""" ur = user_religion.strip().lower() cr = candidate_religion.strip().lower() if not ur or not cr: return False return ur == cr def marital_status_compatible(user_status: str, candidate_status: str) -> bool: """ Group-based marital status match (see MARITAL_GROUPS). Missing status → reject (safe default). """ us = user_status.strip().lower() cs = candidate_status.strip().lower() if not us or not cs: return False allowed = MARITAL_GROUPS.get(us) if allowed is None: return us == cs # unknown status: require exact match return cs in allowed # ── Soft score functions ─────────────────────────────────────────────────── def categorical_bonus(user: dict, candidate: dict) -> float: """ Small score boost for shared optional attributes: maslak, country, qualification. Neutral (0.5) when no comparable data exists. """ matches = 0 total = 0 for key in ['maslak', 'country', 'qualification']: u = safe_str(user.get(key)) c = safe_str(candidate.get(key)) if u and c and u != 'unknown' and c != 'unknown': total += 1 if u == c: matches += 1 return matches / total if total > 0 else 0.5 def bidirectional_criteria_score(user: dict, candidate: dict) -> float: """ Score how well each person meets the other's stated criteria. Average of both directions so one-sided mismatches reduce the score. """ user_criteria = expand_traits(safe_str(user.get('preferred_partner_criteria'))) cand_criteria = expand_traits(safe_str(candidate.get('preferred_partner_criteria'))) user_desc = expand_traits( safe_str(user.get('personality_traits')) + ' ' + safe_str(user.get('hobbies')) ) cand_desc = expand_traits( safe_str(candidate.get('personality_traits')) + ' ' + safe_str(candidate.get('hobbies')) ) u2c = text_similarity(user_criteria, cand_desc) if user_criteria.strip() else 0.5 c2u = text_similarity(cand_criteria, user_desc) if cand_criteria.strip() else 0.5 return (u2c + c2u) / 2.0 def score_candidate(user: dict, candidate: dict) -> dict: """ Compute a full score breakdown for one candidate. Returns individual dimension scores + composite. """ user_age = int(user.get('age') or 25) cand_age = int(candidate.get('age') or 25) user_gender = safe_str(user.get('gender')) age_compat = age_score(user_age, cand_age, user_gender) user_traits = expand_traits(safe_str(user.get('personality_traits'))) cand_traits = expand_traits(safe_str(candidate.get('personality_traits'))) trait_sim = text_similarity(user_traits, cand_traits) criteria_sim = bidirectional_criteria_score(user, candidate) hobby_sim = text_similarity( safe_str(user.get('hobbies')), safe_str(candidate.get('hobbies')) ) cat_bonus = categorical_bonus(user, candidate) composite = ( WEIGHTS['age_compatibility'] * age_compat + WEIGHTS['personality_traits'] * trait_sim + WEIGHTS['partner_criteria'] * criteria_sim + WEIGHTS['hobbies'] * hobby_sim + WEIGHTS['categorical_bonus'] * cat_bonus ) return { 'age_compatibility': round(age_compat, 4), 'personality_traits': round(trait_sim, 4), 'partner_criteria': round(criteria_sim, 4), 'hobbies': round(hobby_sim, 4), 'categorical_bonus': round(cat_bonus, 4), 'composite': round(composite, 4), } # ══════════════════════════════════════════════════════════════════════════ # CACHE # ══════════════════════════════════════════════════════════════════════════ class MatchmakingCache: def __init__(self): self.profiles: list[dict] = [] self.last_updated: datetime | None = None self.cache_valid: bool = False self.error_message: str | None = None def is_expired(self) -> bool: if not self.last_updated: return True return datetime.now() - self.last_updated > timedelta(seconds=CACHE_TTL) def invalidate(self): self.cache_valid = False logger.info("Cache invalidated") cache = MatchmakingCache() active_requests: dict = {} # ── Supabase fetch ───────────────────────────────────────────────────────── async def fetch_profiles_paginated(limit: int = 1000) -> list[dict]: all_profiles: list[dict] = [] offset = 0 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 if len(batch) < PROFILE_BATCH_SIZE: break except httpx.TimeoutException: logger.warning(f"Timeout at offset {offset}") break except Exception as e: logger.warning(f"Error at offset {offset}: {e}") break logger.info(f"Fetched {len(all_profiles)} active profiles from Supabase") return all_profiles def preprocess_profile(profile: dict) -> dict: """Normalise a raw Supabase profile into a clean dict.""" try: age = int(profile.get('age') or 25) age = max(18, min(100, age)) except Exception: age = 25 return { 'id': profile['id'], 'full_name': profile.get('full_name', 'Unknown'), 'age': age, 'gender': safe_str(profile.get('gender')), 'city': profile.get('city', ''), 'religion': safe_str(profile.get('religion')), 'marital_status': safe_str(profile.get('marital_status')), 'qualification': safe_str(profile.get('qualification')), 'country': safe_str(profile.get('country')), 'maslak': safe_str(profile.get('maslak')), 'region_caste': safe_str(profile.get('region_caste')), 'hobbies': safe_str(profile.get('hobbies')), 'personality_traits': safe_str(profile.get('personality_traits')), 'preferred_partner_criteria': safe_str(profile.get('preferred_partner_criteria')), 'profile_picture_url': profile.get('profile_picture_url'), 'is_active': profile.get('is_active', True), } async def refresh_cache() -> bool: global cache try: logger.info("Refreshing profile cache...") raw = await fetch_profiles_paginated() if len(raw) < 2: cache.error_message = f"Only {len(raw)} profiles — need at least 2" return False profiles = [] for p in raw: try: profiles.append(preprocess_profile(p)) except Exception as e: logger.warning(f"Preprocessing error for {p.get('id')}: {e}") if not profiles: cache.error_message = "No profiles after preprocessing" return False cache.profiles = profiles cache.last_updated = datetime.now() cache.cache_valid = True cache.error_message = None logger.info(f"Cache refreshed: {len(profiles)} profiles loaded") return True except Exception as e: logger.error(f"Cache refresh failed: {e}\n{traceback.format_exc()}") cache.error_message = str(e) return False async def get_cached_profiles() -> list[dict]: if not cache.cache_valid or cache.is_expired(): logger.info("Cache invalid/expired — refreshing...") success = await refresh_cache() if not success and not cache.profiles: raise HTTPException(503, "Service unavailable: no profiles loaded") if not success: logger.warning("Refresh failed — serving stale cache") return cache.profiles # ══════════════════════════════════════════════════════════════════════════ # STARTUP # ══════════════════════════════════════════════════════════════════════════ @app.on_event("startup") async def startup(): logger.info("=" * 60) logger.info("ML Matchmaking Service v4.0 — Starting") logger.info("=" * 60) success = await refresh_cache() if success: logger.info(f"Startup complete — {len(cache.profiles)} profiles loaded") else: logger.warning("Startup cache load failed — will retry on first request") # ══════════════════════════════════════════════════════════════════════════ # RECOMMENDATION CORE # ══════════════════════════════════════════════════════════════════════════ async def _handle_recommend_request( user_id: str, top_n: int, min_score: float, request_id: str | None, ) -> list[dict]: if request_id is None: request_id = str(uuid.uuid4()) # Race-condition guard active_requests[user_id] = {'timestamp': datetime.now().timestamp(), 'request_id': request_id} logger.info(f"[REQ:{request_id[:8]}] Recommend for user={user_id} top_n={top_n}") try: profiles = await get_cached_profiles() # ── Find requesting user ─────────────────────────────────────────── user = next((p for p in profiles if p['id'] == user_id), None) if user is None: raise HTTPException(404, "User not found in active profiles") user_gender = user['gender'] user_religion = user['religion'] user_marital = user['marital_status'] if user_gender not in ('male', 'female'): raise HTTPException(400, f"Invalid gender: {user_gender!r}") opposite_gender = 'female' if user_gender == 'male' else 'male' # ── LAYER 1: Hard filters ────────────────────────────────────────── rejected = {'gender': 0, 'inactive': 0, 'religion': 0, 'marital_status': 0} candidates: list[dict] = [] for p in profiles: if p['id'] == user_id: continue if p['gender'] != opposite_gender: rejected['gender'] += 1 continue if not p['is_active']: rejected['inactive'] += 1 continue if not religion_compatible(user_religion, p['religion']): rejected['religion'] += 1 continue if not marital_status_compatible(user_marital, p['marital_status']): rejected['marital_status'] += 1 continue candidates.append(p) logger.info( f"Hard filters — total={len(profiles)} passed={len(candidates)} " f"rejected: gender={rejected['gender']} inactive={rejected['inactive']} " f"religion={rejected['religion']} marital={rejected['marital_status']}" ) if not candidates: logger.info(f"No candidates survived hard filters for {user_id}") return [] # Race-condition check after the expensive filter step current = active_requests.get(user_id) if current is None or current['request_id'] != request_id: logger.warning(f"[REQ:{request_id[:8]}] Superseded — discarding") raise HTTPException(409, "Request superseded by newer request") # ── LAYER 2: Soft scoring ────────────────────────────────────────── scored: list[dict] = [] for c in candidates: breakdown = score_candidate(user, c) if breakdown['composite'] < min_score: continue scored.append({ 'id': c['id'], 'full_name': c['full_name'], 'age': c['age'], 'city': c['city'], 'religion': c['religion'], 'marital_status': c['marital_status'], 'profile_picture_url': c['profile_picture_url'], 'score_breakdown': breakdown, 'composite_score': breakdown['composite'], }) # ── LAYER 3: Rank ────────────────────────────────────────────────── scored.sort(key=lambda x: x['composite_score'], reverse=True) results = scored[:top_n] # ── Format response ──────────────────────────────────────────────── formatted = [] for m in results: bd = m['score_breakdown'] formatted.append({ 'user_id': m['id'], 'name': m['full_name'], 'age': m['age'], 'city': m['city'], 'religion': m['religion'], 'marital_status': m['marital_status'], 'compatibility_score': round(m['composite_score'] * 100, 1), 'score': m['composite_score'], 'photo_url': m['profile_picture_url'], 'score_breakdown': { 'age_compatibility': bd['age_compatibility'], 'personality_traits': bd['personality_traits'], 'partner_criteria': bd['partner_criteria'], 'hobbies': bd['hobbies'], 'categorical_bonus': bd['categorical_bonus'], }, 'reasons': _build_reasons(bd), }) # ── Terminal display ─────────────────────────────────────────────── _log_results(user, user_id, user_gender, opposite_gender, top_n, min_score, formatted) logger.info(f"[REQ:{request_id[:8]}] Returning {len(formatted)} matches") return formatted except HTTPException: raise except Exception as e: logger.error(f"[REQ:{request_id[:8]}] Error: {e}\n{traceback.format_exc()}") raise HTTPException(500, f"Recommendation failed: {e}") finally: current = active_requests.get(user_id) if current and current['request_id'] == request_id: active_requests.pop(user_id, None) def _build_reasons(bd: dict) -> list[str]: """Generate human-readable reasons from score breakdown.""" reasons = [] if bd['age_compatibility'] >= 0.75: reasons.append("Great age compatibility") if bd['personality_traits'] >= 0.6: reasons.append("Similar personality traits") if bd['partner_criteria'] >= 0.6: reasons.append("Matches your partner criteria") if bd['hobbies'] >= 0.5: reasons.append("Shared hobbies and interests") if bd['categorical_bonus'] >= 0.67: reasons.append("Shared background (maslak / country)") if not reasons: reasons.append("Compatible profile") return reasons def _log_results(user, user_id, user_gender, opposite_gender, top_n, min_score, formatted): divider = "═" * 72 thin_line = "─" * 72 logger.info("") logger.info(divider) logger.info(" MATCH RESULTS") logger.info(thin_line) logger.info(f" User : {user['full_name']}") logger.info(f" ID : {user_id}") logger.info(f" Gender : {user_gender.upper()} → showing {opposite_gender.upper()} profiles") logger.info(f" Religion : {user.get('religion', '—')}") logger.info(f" Marital : {user.get('marital_status', '—')}") logger.info(f" Matches : {len(formatted)} (top_n={top_n}, min_score={min_score})") logger.info(f" Timestamp : {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") logger.info(thin_line) if formatted: logger.info( f" {'#':<4} {'Name':<22} {'Age':>3} {'Religion':<14} " f"{'Marital':<12} {'Score':>6} Bar" ) logger.info(" " + thin_line) for idx, m in enumerate(formatted, 1): pct = m['compatibility_score'] filled = max(0, min(10, int(round(pct / 10)))) bar = "█" * filled + "░" * (10 - filled) badge = "🟢" if pct >= 80 else "🟡" if pct >= 60 else "🟠" if pct >= 40 else "🔴" bd = m['score_breakdown'] logger.info( f" {idx:<4} {str(m['name']):<22} {m['age']:>3} " f"{str(m['religion']):<14} {str(m['marital_status']):<12} " f"{pct:>5}% {bar} {badge}" ) logger.info( f" age={bd['age_compatibility']:.2f} " f"traits={bd['personality_traits']:.2f} " f"criteria={bd['partner_criteria']:.2f} " f"hobbies={bd['hobbies']:.2f} " f"cat={bd['categorical_bonus']:.2f}" ) logger.info(thin_line) scores = [m['compatibility_score'] for m in formatted] logger.info(f" Best : {formatted[0]['name']} ({max(scores)}%)") logger.info(f" Avg : {sum(scores)/len(scores):.1f}%") else: logger.info(" No matches met the minimum score threshold.") logger.info(divider) logger.info("") # ══════════════════════════════════════════════════════════════════════════ # ENDPOINTS # ══════════════════════════════════════════════════════════════════════════ @app.get("/health") async def health(): return { "status": "running", "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 ), "error": cache.error_message, } @app.get("/stats") async def stats(): try: profiles = await get_cached_profiles() gender_counts: dict[str, int] = {} religion_counts: dict[str, int] = {} for p in profiles: gender_counts[p['gender']] = gender_counts.get(p['gender'], 0) + 1 religion_counts[p['religion']] = religion_counts.get(p['religion'], 0) + 1 return { "total_profiles": len(profiles), "gender_breakdown": gender_counts, "religion_breakdown": religion_counts, "cache_age_seconds": ( (datetime.now() - cache.last_updated).total_seconds() if cache.last_updated else None ), } except Exception as e: raise HTTPException(500, f"Stats failed: {e}") @app.get("/recommend") async def recommend( user_id: str, top_n: int = 10, min_score: float = 0.3, request_id: str = None, ): return await _handle_recommend_request(user_id, top_n, min_score, request_id) @app.get("/recommend/{user_id}") async def recommend_path( user_id: str, top_n: int = 10, min_score: float = 0.3, request_id: str = None, ): return await _handle_recommend_request(user_id, top_n, min_score, request_id) @app.post("/feedback") async def feedback(request: dict): try: user_id = request.get('user_id') target_id = request.get('target_id') action = request.get('action', '').lower() 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'") data = { 'user_id': user_id, 'matched_user_id': target_id, 'is_liked': action == 'like', 'is_skipped': action == 'reject', 'created_at': datetime.utcnow().isoformat(), } resp = httpx.post( f"{SUPABASE_URL}/rest/v1/matches", headers=HEADERS, json=data, timeout=10, ) if resp.status_code in (200, 201, 204): logger.info(f"Feedback recorded: {user_id} {action} {target_id}") return {"status": "ok", "message": "Feedback recorded"} return {"status": "error", "message": resp.text} except HTTPException: raise except Exception as e: logger.error(f"Feedback error: {e}") return {"status": "error", "message": str(e)} @app.post("/refresh-cache") async def manual_refresh(): success = await refresh_cache() if success: return {"status": "ok", "message": f"Cache refreshed — {len(cache.profiles)} profiles"} return {"status": "error", "message": cache.error_message} @app.get("/") async def root(): return { "name": "Real-time Matchmaking API", "version": "4.0.0", "changes_v4": [ "Hard filter: religion must match exactly", "Hard filter: marital status must match (grouped)", "Trait expansion: 'caring' -> compassionate, nurturing, kind, etc.", "Bidirectional partner criteria scoring", "Gaussian age compatibility curve", "Full score breakdown in every response", "Removed TF-IDF matrix cache (no stale vectors)", ], "endpoints": { "health": "/health", "stats": "/stats", "recommend": "/recommend?user_id={uuid}&top_n=10&min_score=0.3", "feedback": "/feedback", "refresh_cache": "/refresh-cache", "docs": "/docs", }, } if __name__ == '__main__': import uvicorn port = int(os.getenv('PORT', 7860)) host = os.getenv('HOST', '0.0.0.0') logger.info(f"Starting Matchmaking API v4.0 on {host}:{port}") uvicorn.run(app, host=host, port=port, log_level="info")