""" API Client for RewardPilot - with local engine fallback. If the orchestrator is unreachable (sleeping / quota exceeded), all calls are served locally from local_recommender.py. """ import requests, logging, json from typing import Dict, Any, Optional, List logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Import local engine (always available) try: from utils.local_recommender import recommend as _local_recommend, get_analytics as _local_analytics _LOCAL_AVAILABLE = True logger.info(" Local recommendation engine ready") except Exception as e: logger.warning(f" Local engine not available: {e}") _LOCAL_AVAILABLE = False MCC_MAP = { "Groceries": "5411", "Restaurants": "5812", "Dining": "5812", "Fast Food": "5814", "Coffee Shops": "5814", "Wholesale Clubs": "5300", "Gas Stations": "5541", "Airlines": "3000", "Hotels": "7011", "Department Stores": "5311", "Clothing & Apparel": "5651", "Electronics": "5732", "Home Improvement": "5200", "Entertainment": "7841", "Streaming Services": "4899", "Drugstores": "5912", "General Retail": "5999", "Online Shopping": "5999", "Travel": "4722", "Transit": "4111", "Fitness & Gym": "7941", "Automotive": "5533", "Pharmacy & Healthcare": "5912", "Bars/Taverns": "5813", } class RewardPilotClient: def __init__(self, orchestrator_url: str = "http://localhost:8000"): self.orchestrator_url = orchestrator_url.rstrip("/") self.timeout = 35 # ── helpers ──────────────────────────────────────────────────────────── def _category_to_mcc(self, category: str) -> str: return MCC_MAP.get(category, "5999") def _call_orchestrator(self, path: str, method: str = "GET", payload: Dict = None) -> Optional[Dict]: """Try the remote orchestrator; return None on any failure.""" try: url = f"{self.orchestrator_url}{path}" if method == "POST": resp = requests.post(url, json=payload, timeout=self.timeout) else: resp = requests.get(url, timeout=self.timeout) if resp.status_code == 200: data = resp.json() # Quota-exceeded comes back as 200 with {"error":...} sometimes if isinstance(data, dict) and data.get("error"): logger.warning(f"Orchestrator error: {data.get('error')}") return None return data logger.warning(f"Orchestrator HTTP {resp.status_code} for {path}") return None except Exception as e: logger.warning(f"Orchestrator unreachable ({path}): {e}") return None # ── public API (same interface as the original) ───────────────────────── def get_recommendation( self, user_id: str, merchant: str, category: str, amount: float, mcc: Optional[str] = None, ) -> Dict: mcc = mcc or self._category_to_mcc(category) payload = {"user_id": user_id, "merchant": merchant, "mcc": mcc, "amount_usd": amount, "category": category} # 1. Try orchestrator remote = self._call_orchestrator("/recommend", "POST", payload) if remote: rec = remote.get("recommendation", remote) card_name = rec.get("card_name", rec.get("recommended_card", "Unknown")) rewards = float(rec.get("rewards_earned", 0)) return { "success": True, "data": { "recommended_card": card_name, "rewards_earned": rewards, "rewards_rate": rec.get("rewards_rate", "N/A"), "reasoning": rec.get("reasoning", ""), "merchant": merchant, "category": category, "amount": amount, "annual_potential": rec.get("annual_impact", {}).get("potential_savings", rewards * 12), "optimization_score": rec.get("annual_impact", {}).get("optimization_score", 75), "warnings": rec.get("warnings", []), "alternatives": [ {"card": a.get("card_name", ""), "rewards": float(a.get("rewards_earned", 0)), "rate": a.get("rewards_rate", ""), "reason": a.get("reason", "")} for a in rec.get("alternative_options", []) ], "mock_data": False, } } # 2. Fall back to local engine if _LOCAL_AVAILABLE: result = _local_recommend(user_id, merchant, category, mcc, amount) if result.get("error"): return {"success": False, "error": result["error"]} rec = result["recommendation"] return { "success": True, "data": { "recommended_card": rec["card_name"], "rewards_earned": rec["rewards_earned"], "rewards_rate": rec["rewards_rate"], "reasoning": rec["reasoning"], "merchant": merchant, "category": rec["category"], "amount": amount, "annual_potential": rec["annual_impact"]["potential_savings"], "optimization_score": rec["annual_impact"]["optimization_score"], "warnings": rec["warnings"], "alternatives": [ {"card": a["card_name"], "rewards": a["rewards_earned"], "rate": a["rewards_rate"], "reason": a["reason"]} for a in rec["alternative_options"] ], "mock_data": False, "has_optimal_card": rec.get("has_optimal_card", True), "optimal_card_suggestion": rec.get("optimal_card_suggestion"), } } return {"success": False, "error": "No recommendation engine available"} def get_recommendation_sync( self, user_id: str, merchant: str, mcc: str, amount_usd: float, transaction_date: str = "", ) -> Dict: """Legacy sync method used by some tabs.""" category = {v: k for k, v in MCC_MAP.items()}.get(mcc, "General") result = self.get_recommendation(user_id, merchant, category, amount_usd, mcc) if result.get("success"): d = result["data"] return { "merchant": merchant, "amount_usd": amount_usd, "transaction_date": transaction_date, "user_id": user_id, "recommended_card": { "card_name": d["recommended_card"], "reward_rate": d["rewards_rate"], "reward_amount": d["rewards_earned"], "category": d["category"], "reasoning": d["reasoning"], }, "alternative_cards": [ {"card_name": a["card"], "reward_rate": a["rate"], "reward_amount": a["rewards"], "reasoning": a["reason"]} for a in d.get("alternatives", []) ], "total_cards_analyzed": 1 + len(d.get("alternatives", [])), "services_used": ["Local Engine"], } return {"error": True, "message": result.get("error", "Unknown error")} def get_user_analytics(self, user_id: str) -> Dict: # 1. Try orchestrator remote = self._call_orchestrator(f"/analytics/{user_id}") if remote and remote.get("success"): return remote # 2. Local fallback if _LOCAL_AVAILABLE: return _local_analytics(user_id) return {"success": False, "error": "Analytics unavailable"}