train.py

"""
Robust matrimonial recommendation engine.

Pipeline
--------
1. Hard filters  – eliminate incompatible candidates entirely
2. Soft scoring  – rank surviving candidates by weighted similarity
3. Return top-N  – with full score breakdown for transparency / debugging

Hard filter rules (all must pass):
  • opposite_gender
  • is_active
  • not self
  • religion  – exact match
  • marital_status – grouped match (see MARITAL_GROUPS)

Soft score dimensions (weights sum to 1.0):
  • age_compatibility   0.20  – Gaussian, peaks at ideal age gap
  • personality_traits  0.25  – cosine similarity on sentence embeddings
  • partner_criteria     0.25  – bidirectional keyword + embedding match
  • hobbies             0.15  – cosine similarity
  • categorical_bonus   0.15  – maslak, country, qualification
"""

import os
import math
import logging
import httpx
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from sklearn.feature_extraction.text import TfidfVectorizer

logging.basicConfig(level=logging.INFO, format="%(levelname)s  %(message)s")
logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

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"}

TOP_N = 20          # maximum recommendations returned
AGE_IDEAL_GAP = 3   # years – peak of the Gaussian for age compatibility
AGE_SIGMA = 7       # standard deviation in years; larger = more forgiving

# 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"

# Marital status groups: candidates in the same group may match each other.
# Adjust to your product rules.
MARITAL_GROUPS = {
    "single":   {"single"},
    "divorced": {"divorced", "widowed"},   # divorced ↔ widowed allowed
    "widowed":  {"divorced", "widowed"},
    "separated":{"separated"},
}

# Personality / trait keywords used to expand text matching
TRAIT_SYNONYMS = {
    "caring":      ["caring", "care", "compassionate", "nurturing", "kind"],
    "honest":      ["honest", "truthful", "sincere", "transparent"],
    "religious":   ["religious", "practicing", "devout", "practicing muslim",
                    "practicing christian", "allah", "god", "faith", "deen"],
    "family-oriented": ["family", "family-oriented", "family oriented",
                        "home", "homemaker", "domestic"],
    "ambitious":   ["ambitious", "driven", "goal-oriented", "career", "successful"],
    "respectful":  ["respectful", "respect", "humble", "obedient"],
    "educated":    ["educated", "degree", "professional", "literate"],
}

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def safe_str(val) -> str:
    """Normalise any DB value to a lowercase string."""
    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 that 'caring' also pulls in 'compassionate',
    'nurturing', etc.  This improves recall when users describe traits
    differently.
    """
    words = text.lower().split()
    extras = []
    for word in words:
        for canonical, synonyms in TRAIT_SYNONYMS.items():
            if word in synonyms or canonical in text.lower():
                extras.extend(synonyms)
                break
    return text + " " + " ".join(extras)


def text_similarity(text_a: str, text_b: str) -> float:
    """
    TF-IDF cosine similarity between two text blobs.
    Returns a value in [0, 1].
    """
    if not text_a.strip() or not text_b.strip():
        return 0.0
    try:
        vect = TfidfVectorizer(stop_words="english", min_df=1)
        matrix = vect.fit_transform([text_a, text_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, gender: str) -> float:
    """
    Gaussian age-compatibility score.

    Convention (adjustable):
      • Male users prefer slightly younger females  → ideal gap = +AGE_IDEAL_GAP
      • Female users prefer slightly older males    → ideal gap = -AGE_IDEAL_GAP
    Gap is defined as (candidate_age – user_age).
    """
    ideal_gap = AGE_IDEAL_GAP if gender == "male" else -AGE_IDEAL_GAP
    gap = candidate_age - user_age - ideal_gap
    return float(math.exp(-(gap ** 2) / (2 * AGE_SIGMA ** 2)))


def marital_status_compatible(user_status: str, candidate_status: str) -> bool:
    """
    Returns True if the two marital statuses are compatible according to
    MARITAL_GROUPS.  Falls back to exact match for unrecognised values.
    """
    us = user_status.strip().lower()
    cs = candidate_status.strip().lower()
    if not us or not cs:
        return False          # missing data → reject (safe default)
    allowed = MARITAL_GROUPS.get(us)
    if allowed is None:
        return us == cs       # unknown status: require exact match
    return cs in allowed


def religion_compatible(user_religion: str, candidate_religion: str) -> bool:
    """Strict exact match on religion (case-insensitive)."""
    ur = user_religion.strip().lower()
    cr = candidate_religion.strip().lower()
    if not ur or not cr:
        return False          # missing religion → reject
    return ur == cr


def categorical_bonus(user: dict, candidate: dict) -> float:
    """
    Small score boost for shared optional attributes:
    maslak, country, qualification level.
    Score is in [0, 1].
    """
    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:
            total += 1
            if u == c:
                matches += 1

    return matches / total if total > 0 else 0.5   # neutral when no data


def bidirectional_criteria_score(user: dict, candidate: dict) -> float:
    """
    Score how well the user meets the candidate's criteria AND
    how well the candidate meets the user's criteria.

    Returns the average of both directions so a one-sided mismatch
    still reduces the score.
    """
    user_criteria = expand_traits(safe_str(user.get("preferred_partner_criteria")))
    cand_criteria = expand_traits(safe_str(candidate.get("preferred_partner_criteria")))

    # Build text descriptions of each person for comparison
    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"))
    )

    # user → candidate: does candidate match user's criteria?
    u2c = text_similarity(user_criteria, cand_desc) if user_criteria else 0.5

    # candidate → user: does user match candidate's criteria?
    c2u = text_similarity(cand_criteria, user_desc) if cand_criteria else 0.5

    return (u2c + c2u) / 2.0


# ---------------------------------------------------------------------------
# Supabase helpers
# ---------------------------------------------------------------------------

def fetch_all_active_profiles() -> list[dict]:
    """Page through all active profiles (100 per request)."""
    profiles = []
    offset = 0

    while True:
        resp = httpx.get(
            f"{SUPABASE_URL}/rest/v1/profiles",
            headers=HEADERS,
            params={
                "select": "*",
                "is_active": "eq.true",
                "limit": 100,
                "offset": offset,
            },
            timeout=15,
        )
        resp.raise_for_status()
        batch = resp.json()
        if not batch:
            break
        profiles.extend(batch)
        offset += 100
        if len(batch) < 100:
            break

    logger.info("Fetched %d active profiles", len(profiles))
    return profiles


def fetch_profile(user_id: str) -> dict | None:
    """Return a single profile or None."""
    resp = httpx.get(
        f"{SUPABASE_URL}/rest/v1/profiles",
        headers=HEADERS,
        params={"id": f"eq.{user_id}", "select": "*"},
        timeout=10,
    )
    resp.raise_for_status()
    data = resp.json()
    return data[0] if data else None


# ---------------------------------------------------------------------------
# Core recommendation logic
# ---------------------------------------------------------------------------

def score_candidate(user: dict, candidate: dict) -> dict:
    """
    Compute a detailed score breakdown for one candidate.
    Returns a dict with individual dimension scores and a composite.
    """
    user_age = int(user.get("age") or 25)
    cand_age = int(candidate.get("age") or 25)
    user_gender = safe_str(user.get("gender"))

    # ── Dimension scores ──────────────────────────────────────────────────
    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)

    user_hobbies = safe_str(user.get("hobbies"))
    cand_hobbies = safe_str(candidate.get("hobbies"))
    hobby_sim = text_similarity(user_hobbies, cand_hobbies)

    cat_bonus = categorical_bonus(user, candidate)

    # ── Weighted composite ────────────────────────────────────────────────
    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 {
        "candidate_id":        candidate["id"],
        "candidate_name":      candidate.get("full_name", "Unknown"),
        "age":                 cand_age,
        "religion":            candidate.get("religion"),
        "marital_status":      candidate.get("marital_status"),
        "scores": {
            "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_score": round(composite, 4),
    }


def recommend(user_id: str, all_profiles: list[dict] | None = None) -> list[dict]:
    """
    Return up to TOP_N recommendations for user_id.

    Parameters
    ----------
    user_id      : the requester's profile ID
    all_profiles : pre-fetched profiles (optional, for caching)

    Returns
    -------
    List of score dicts sorted by composite_score descending.
    """
    # 1. Load user
    user = fetch_profile(user_id)
    if not user:
        raise ValueError(f"User {user_id!r} not found")

    user_gender = safe_str(user.get("gender"))
    user_religion = safe_str(user.get("religion"))
    user_marital = safe_str(user.get("marital_status"))

    if user_gender not in ("male", "female"):
        raise ValueError(f"Invalid gender for user {user_id!r}: {user_gender!r}")

    opposite_gender = "female" if user_gender == "male" else "male"

    # 2. Load pool
    profiles = all_profiles if all_profiles is not None else fetch_all_active_profiles()

    # 3. Hard filters
    candidates = []
    reject_counts = {
        "self":           0,
        "gender":         0,
        "inactive":       0,
        "religion":       0,
        "marital_status": 0,
    }

    for profile in profiles:
        pid = profile.get("id")

        if pid == user_id:
            reject_counts["self"] += 1
            continue

        if safe_str(profile.get("gender")) != opposite_gender:
            reject_counts["gender"] += 1
            continue

        if not profile.get("is_active"):
            reject_counts["inactive"] += 1
            continue

        if not religion_compatible(user_religion, safe_str(profile.get("religion"))):
            reject_counts["religion"] += 1
            continue

        if not marital_status_compatible(user_marital, safe_str(profile.get("marital_status"))):
            reject_counts["marital_status"] += 1
            continue

        candidates.append(profile)

    logger.info(
        "Hard filter summary — total=%d  passed=%d  rejected: %s",
        len(profiles), len(candidates), reject_counts,
    )

    if not candidates:
        logger.warning("No candidates survived hard filters for user %s", user_id)
        return []

    # 4. Soft scoring
    scored = [score_candidate(user, c) for c in candidates]

    # 5. Rank and return top-N
    scored.sort(key=lambda x: x["composite_score"], reverse=True)
    results = scored[:TOP_N]

    logger.info(
        "Top-%d recommendations for %s | best=%.3f  worst=%.3f",
        len(results), user_id,
        results[0]["composite_score"] if results else 0,
        results[-1]["composite_score"] if results else 0,
    )

    return results


# ---------------------------------------------------------------------------
# Debug / test runner
# ---------------------------------------------------------------------------

async def test_recommend():
    """Step-by-step diagnostic that mirrors the original debug script."""
    print("\n" + "=" * 70)
    print("DEBUGGING RECOMMEND ENDPOINT")
    print("=" * 70)

    # Resolve a test user dynamically
    print("\n0️⃣  Resolving a test user...")
    try:
        resp = httpx.get(
            f"{SUPABASE_URL}/rest/v1/profiles",
            headers=HEADERS,
            params={"select": "id", "is_active": "eq.true", "limit": 1},
            timeout=10,
        )
        data = resp.json()
        if not data:
            print("   ✗ No active users found"); return
        user_id = data[0]["id"]
        print(f"   ✓ Using user_id: {user_id}")
    except Exception as e:
        print(f"   ✗ Could not resolve test user: {e}"); return

    # Run the full pipeline
    try:
        results = recommend(user_id)
    except Exception as e:
        print(f"\n✗ recommend() raised: {e}")
        import traceback; traceback.print_exc()
        return

    print(f"\n✅  recommend() returned {len(results)} result(s)")

    if results:
        print("\nTop 5 results:")
        print(f"  {'Name':<25} {'Age':>4}  {'Score':>7}  Breakdown")
        print("  " + "-" * 70)
        for r in results[:5]:
            s = r["scores"]
            breakdown = (
                f"age={s['age_compatibility']:.2f}  "
                f"traits={s['personality_traits']:.2f}  "
                f"criteria={s['partner_criteria']:.2f}  "
                f"hobbies={s['hobbies']:.2f}  "
                f"cat={s['categorical_bonus']:.2f}"
            )
            print(f"  {r['candidate_name']:<25} {r['age']:>4}  {r['composite_score']:>7.4f}  {breakdown}")


if __name__ == "__main__":
    import asyncio
    asyncio.run(test_recommend())