""" data/recommender.py — Rule-Based Worker Recommendation Engine for FixItGo Given problem text + optional area, returns the best matching workers: 1. Use AI recommender to detect category 2. Filter workers by that category 3. Sort by: area match bonus (local workers first) + rating DESC + total reviews 4. Return top N with explanation strings. Usage: from data.recommender import recommend_workers results = recommend_workers("tap is leaking badly", "Adyar") # → list of dicts with worker info + reason string """ from data.ai_recommender import recommend_service from data.database import get_workers def recommend_workers(problem_text: str, user_area: str = "", top_n: int = 3) -> dict: """ Recommend the best workers for a given problem. Args: problem_text : Free-text description of the problem. user_area : User's locality (optional, for proximity scoring). top_n : Number of workers to return (default 3). Returns: { 'ai_result' : dict from recommend_service(), 'workers' : list of top worker dicts with 'reason' key added, 'message' : human-readable summary string, } """ ai = recommend_service(problem_text) category = ai.get("category") if not category: return { "ai_result": ai, "workers": [], "message": "Could not detect service category. Please describe the problem in more detail.", } # ── Fetch workers for the suggested category ────────────────────────── workers = get_workers(category=category, sort="rating") if not workers: return { "ai_result": ai, "workers": [], "message": f"No {category} workers found in the database yet.", } # ── Score each worker ───────────────────────────────────────────────── area_lower = user_area.lower().strip() scored = [] for w in workers: score = 0.0 # Rating weight (max 5 → contribute up to 50 points) score += float(w.get("rating") or 0) * 10 # Review volume bonus (log scale, up to 10 pts) import math reviews = int(w.get("total_reviews") or 0) if reviews > 0: score += min(10, math.log(reviews + 1, 2)) # Area proximity bonus (up to 20 pts) worker_area = (w.get("area") or "").lower() if area_lower and area_lower in worker_area: score += 20 # exact area match elif area_lower and any(part in worker_area for part in area_lower.split()): score += 8 # partial match scored.append((score, w)) # Best score first scored.sort(key=lambda x: x[0], reverse=True) top_workers = scored[:top_n] # ── Build result list with explanation ──────────────────────────────── result_workers = [] for rank, (score, w) in enumerate(top_workers, start=1): reasons = [] rating = float(w.get("rating") or 0) if rating >= 4.7: reasons.append("⭐ Top rated") elif rating >= 4.0: reasons.append("👍 Highly rated") else: reasons.append("✅ Verified professional") # Area match reason if area_lower and area_lower in (w.get("area") or "").lower(): reasons.append(f"📍 Near {user_area}") elif w.get("area"): reasons.append(f"📍 Based in {w['area']}") # Jobs done jobs = int(w.get("total_jobs") or 0) if jobs >= 30: reasons.append(f"🔨 {jobs}+ jobs done") w_copy = dict(w) w_copy["reason"] = " · ".join(reasons) w_copy["rank"] = rank result_workers.append(w_copy) urgency = ai.get("urgency", "Normal") icon = ai.get("icon", "🔧") price = ai.get("price_range", "") msg = ( f"We detected this as a **{category}** problem ({icon}). " f"Urgency: **{urgency}**. Estimated price: **{price}**. " f"Here are the top {len(result_workers)} recommended workers:" ) return { "ai_result": ai, "workers": result_workers, "message": msg, }