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