bsassoli/smartrec
0
1import requests2import json3import os4 5 6def make_api_request(endpoint, ids):7 """8 Generic function to make API requests with error handling9 10 :param endpoint: The specific API endpoint (e.g. 'get-images-from-id')11 :param ids: List of IDs to query12 :return: JSON response or None if error occurs13 """14 api_key = os.environ.get("RECOMMENDER_KEY")15 if not api_key:16 raise ValueError("API key is not set")17 headers = {18 "x-api-key": api_key,19 "Content-Type": "application/json"20 }21 22 payload = {"ids": ids}23 24 try:25 recommender_url = os.environ.get("RECOMMENDER_URL")26 if not recommender_url: 27 raise ValueError("Recommender URL is not set")28 response = requests.post(f"{recommender_url}/{endpoint}", 29 headers=headers, 30 data=json.dumps(payload))31 32 response.raise_for_status() # Raise an exception for bad status codes33 return response.json()34 35 except requests.exceptions.RequestException as e:36 print(f"Error occurred while calling {endpoint}: {e}")37 return None38 39def get_images_from_id(ids):40 """Get images for specific product IDs"""41 return make_api_request("get-images-from-id", ids)42 43 44def parse_article_data(article_data):45 """Parse the article data and return a dictionary"""46 47 # get the article information48 article_info = article_data['article']49 aricle_id = article_info.get('article_id', 'Unknown')50 article_name = article_info.get('prod_name', 'Unknown')51 article_type = article_info.get('department_name', 'Unknown')52 article_color = article_info.get('perceived_colour_master_name', 'Unknown')53 54 # get the article images55 article_images = get_images_from_id([aricle_id])56 img_url = 'no image'57 if len(article_images) > 0:58 img_url = article_images[0].get('url', 'no image')59 60 return {61 "product_name": article_name,62 "product_type": article_type,63 "product_color": article_color,64 "product_image_url": img_url65 }66 67 68def parse_transaction_data(transaction_data):69 """Parse the transaction data and return a dictionary with article information."""70 71 # Retrieve general transaction details72 transaction_date = transaction_data.get('t_dat', 'Unknown')73 sales_channel_id = transaction_data.get('sales_channel_id', 'Unknown')74 price = transaction_data.get('price', 'Unknown')75 76 # Get the article information77 article_info = transaction_data.get('article', {})78 article_id = article_info.get('article_id', 'Unknown')79 article_name = article_info.get('prod_name', 'Unknown')80 article_type = article_info.get('department_name', 'Unknown')81 article_color = article_info.get('perceived_colour_master_name', 'Unknown')82 article_detail = article_info.get('detail_desc', 'Unknown') 83 84 # Get the article images85 article_images = get_images_from_id([article_id])86 img_url = 'no image'87 if article_images:88 img_url = article_images[0].get('url', 'no image')89 90 return {91 "transaction_date": transaction_date,92 "sales_channel_id": sales_channel_id,93 "price": price,94 "product_name": article_name,95 "product_type": article_type,96 "product_color": article_color,97 "product_image_url": img_url,98 "article_id": article_id99 }100 101 102def parse_recommendations(recommendations, max_recs=2, max_transactions=2, only_with_images=True):103 """Parse the recommendations and return a list of product IDs"""104 105 # get the client information106 customer_info = recommendations[0]['customer'][0]107 customer_id = customer_info.get('customer_id', 'Unknown')108 customer_name = customer_info.get('name', customer_id[:8]) # TODO change to real customer name109 customer_age = customer_info.get('age', 'Unknown')110 111 customer_info = {112 "customer name": customer_name,113 "customer age": customer_age114 }115 116 # get the recommendations117 formatted_recommendations = [parse_article_data(article) for article in recommendations[0]['prediction']]118 119 formatted_transactions = [parse_transaction_data(transaction) for transaction in recommendations[0]['transactions']]120 121 print("formatted_recommendations", len(formatted_recommendations))122 print("formatted_transactions", len(formatted_transactions))123 124 if only_with_images:125 formatted_recommendations = [rec for rec in formatted_recommendations if rec['product_image_url'] != 'no image']126 formatted_transactions = [t for t in formatted_transactions if t['product_image_url'] != 'no image']127 128 if len(formatted_recommendations) > max_recs:129 formatted_recommendations = formatted_recommendations[:max_recs]130 131 if len(formatted_transactions) > max_transactions:132 formatted_transactions = formatted_transactions[:max_transactions]133 134 print("formatted_recommendations", len(formatted_recommendations))135 print("formatted_transactions", len(formatted_transactions))136 return customer_info, formatted_recommendations, formatted_transactions137 138 139def get_recommendations(customer_ids, max_recs=4, max_transactions=2):140 """Get product recommendations for specific customer IDs"""141 if isinstance(customer_ids, str):142 customer_ids = [customer_ids]143 recommendations = make_api_request("get-recommendations", customer_ids)144 if recommendations:145 return parse_recommendations(recommendations, max_recs, max_transactions)146 else:147 raise ValueError("Error occurred while fetching recommendations.")148 return None149 150 151if __name__ == "__main__":152 # Example product IDs153 product_ids = ["0110065002", "0111609001"]154 155 # Example customer ID156 customer_id = ["04a183a27a6877e560e1025216d0a3b40d88668c68366da17edfb18ed89c574c"]157 158 # Demonstrate different API calls159 print("Images:", get_images_from_id(product_ids))160 print("Recommendations:", get_recommendations(customer_id))161 162 163 