""" API Client for RewardPilot Orchestrator Service """ import requests import logging from typing import Dict, Any, Optional, Tuple, List logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class RewardPilotClient: """Client for interacting with RewardPilot microservices""" def __init__(self, orchestrator_url: str = "http://localhost:8000"): """ Initialize API client Args: orchestrator_url: Base URL for orchestrator service """ self.orchestrator_url = orchestrator_url.rstrip('/') self.timeout = 10 # seconds def get_recommendation( self, user_id: str, merchant: str, category: str, amount: float, mcc: Optional[str] = None ) -> Dict: """ Get card recommendation for a transaction Args: user_id: User identifier merchant: Merchant name category: Transaction category (e.g., "Groceries", "Dining") amount: Transaction amount in USD mcc: Optional Merchant Category Code Returns: Dictionary with recommendation data """ # Map category to MCC if not provided if not mcc: mcc = self._category_to_mcc(category) payload = { "user_id": user_id, "merchant": merchant, "category": category, "mcc": mcc, "amount_usd": amount } try: logger.info(f"🔗 Calling: {self.orchestrator_url}/recommend") logger.info(f"📦 Payload: {payload}") response = requests.post( f"{self.orchestrator_url}/recommend", json=payload, timeout=self.timeout ) logger.info(f"📡 Response status: {response.status_code}") if response.status_code == 200: data = response.json() logger.info(f"✅ Got recommendation for {merchant}") return { "success": True, "data": data } elif response.status_code == 404: logger.warning(f"⚠️ Endpoint not found (404) - using mock data") return self._get_mock_recommendation(user_id, merchant, category, amount) else: logger.error(f"❌ API error: {response.status_code} - using mock data") return self._get_mock_recommendation(user_id, merchant, category, amount) except requests.exceptions.Timeout: logger.error("❌ Request timeout - using mock data") return self._get_mock_recommendation(user_id, merchant, category, amount) except requests.exceptions.ConnectionError: logger.error("❌ Connection error - using mock data") return self._get_mock_recommendation(user_id, merchant, category, amount) except Exception as e: logger.error(f"❌ Unexpected error: {e} - using mock data") return self._get_mock_recommendation(user_id, merchant, category, amount) def get_user_analytics(self, user_id: str) -> Dict: """ Get analytics for a user Args: user_id: User identifier Returns: Dictionary with analytics data """ logger.info(f"📊 Getting analytics for {user_id} (using mock data)") # Always use mock data since orchestrator doesn't have analytics endpoint return self._get_mock_analytics(user_id) def compare_cards(self, card_ids: List[str]) -> Dict: """ Compare multiple credit cards Args: card_ids: List of card identifiers Returns: Dictionary with comparison data """ payload = { "card_ids": card_ids } try: response = requests.post( f"{self.orchestrator_url}/compare", json=payload, timeout=self.timeout ) if response.status_code == 200: return { "success": True, "data": response.json() } else: return { "success": False, "error": f"API returned status {response.status_code}" } except Exception as e: logger.error(f"❌ Error comparing cards: {e}") return { "success": False, "error": str(e) } def _category_to_mcc(self, category: str) -> str: """Map category name to MCC code""" category_map = { "Groceries": "5411", "Dining": "5812", "Travel": "4511", "Gas": "5541", "Online Shopping": "5999", "Entertainment": "7832", "Pharmacy": "5912", "Department Store": "5311", "Home Improvement": "5211", "Utilities": "4900" } return category_map.get(category, "0000") def _get_mock_recommendation( self, user_id: str, merchant: str, category: str, amount: float ) -> Dict: """ Generate mock recommendation for demo/development This is used when the orchestrator service is unavailable """ # Simple rule-based mock logic card_rules = { "Groceries": { "card": "Amex Gold", "rate": "4x points", "multiplier": 4.0 }, "Dining": { "card": "Capital One Savor", "rate": "4% cashback", "multiplier": 4.0 }, "Travel": { "card": "Chase Sapphire Reserve", "rate": "3x points", "multiplier": 3.0 }, "Gas": { "card": "Costco Visa", "rate": "4% cashback", "multiplier": 4.0 }, "Online Shopping": { "card": "Amazon Prime Card", "rate": "5% cashback", "multiplier": 5.0 } } rule = card_rules.get(category, { "card": "Citi Double Cash", "rate": "2% cashback", "multiplier": 2.0 }) rewards_earned = amount * (rule["multiplier"] / 100) annual_potential = rewards_earned * 12 # Rough estimate logger.warning(f"⚠️ Using mock data for {merchant}") return { "success": True, "data": { "recommended_card": rule["card"], "rewards_earned": round(rewards_earned, 2), "rewards_rate": rule["rate"], "merchant": merchant, "category": category, "amount": amount, "annual_potential": round(annual_potential, 2), "optimization_score": 85, "reasoning": f"Best card for {category.lower()} purchases", "warnings": [], "alternatives": [ { "card": "Citi Double Cash", "rewards": round(amount * 0.02, 2), "rate": "2% cashback" } ], "mock_data": True # Flag to indicate this is mock data } } def _get_mock_analytics(self, user_id: str) -> Dict[str, Any]: """Generate user-specific mock analytics""" # Different data for different users user_profiles = { "u_alice": { "user_id": "u_alice", "total_spending": 4250.75, "total_rewards": 187.50, "optimization_score": 92, "optimized_count": 58, "potential_savings": 125.00, "spending_by_category": { "Groceries": 1200.00, "Restaurants": 850.50, "Gas Stations": 420.25, "Online Shopping": 1200.00, "Entertainment": 580.00 }, "rewards_by_card": { "Amex Gold": 95.50, "Chase Sapphire Reserve": 62.00, "Citi Double Cash": 30.00 }, "monthly_trends": [ {"month": "Aug", "spending": 1200, "rewards": 52}, {"month": "Sep", "spending": 1450, "rewards": 63}, {"month": "Oct", "spending": 1600, "rewards": 72} ] }, "u_bob": { "user_id": "u_bob", "total_spending": 3150.25, "total_rewards": 142.30, "optimization_score": 78, "optimized_count": 42, "potential_savings": 285.50, "spending_by_category": { "Groceries": 800.00, "Restaurants": 1200.00, "Gas Stations": 350.00, "Airlines": 600.00, "Entertainment": 200.25 }, "rewards_by_card": { "Chase Sapphire Reserve": 85.00, "Amex Gold": 42.30, "Capital One Venture": 15.00 }, "monthly_trends": [ {"month": "Aug", "spending": 950, "rewards": 38}, {"month": "Sep", "spending": 1100, "rewards": 52}, {"month": "Oct", "spending": 1100, "rewards": 52} ] }, "u_charlie": { "user_id": "u_charlie", "total_spending": 5420.80, "total_rewards": 245.60, "optimization_score": 85, "optimized_count": 67, "potential_savings": 180.00, "spending_by_category": { "Online Shopping": 2000.00, "Groceries": 1100.00, "Restaurants": 950.00, "Gas Stations": 520.80, "Hotels": 850.00 }, "rewards_by_card": { "Amex Gold": 125.00, "Chase Sapphire Reserve": 95.60, "Citi Double Cash": 25.00 }, "monthly_trends": [ {"month": "Aug", "spending": 1650, "rewards": 75}, {"month": "Sep", "spending": 1850, "rewards": 82}, {"month": "Oct", "spending": 1920, "rewards": 88} ] } } # Get user-specific data or default to alice user_data = user_profiles.get(user_id, user_profiles["u_alice"]) return { "success": True, "data": { **user_data, "mock_data": True } } def health_check(self) -> bool: """ Check if orchestrator service is available Returns: True if service is healthy, False otherwise """ try: response = requests.get( f"{self.orchestrator_url}/health", timeout=5 ) return response.status_code == 200 except: return False # Convenience function def create_client(orchestrator_url: str = "http://localhost:8000") -> RewardPilotClient: """Create and return a RewardPilotClient instance""" return RewardPilotClient(orchestrator_url)