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))