import json import secrets import string from datetime import datetime, timedelta import requests class ShopifyService: @staticmethod def send_get_request(url, access_token): headers = { "X-Shopify-Access-Token": access_token, "Content-Type": "application/json", } try: response = requests.get(url, headers=headers) if response.status_code == 200: return json.loads(response.text) else: print( f"Error: Received status code {response.status_code}. Response Text: {response.text}" ) return None except requests.RequestException as e: print("Error:", e) return None @staticmethod def generate_random_code(size=12, chars=string.ascii_letters + string.digits): return "".join(secrets.choice(chars) for _ in range(size)) # example store nane: storyit-discounts-test-store @staticmethod def create_single_use_discount_code(store_name, access_token, price_rule_id): headers = { "X-Shopify-Access-Token": access_token, "Content-Type": "application/json", } discount_code = ShopifyService.generate_random_code() data = {"discount_code": {"code": discount_code}} url = f"https://{store_name}.myshopify.com/admin/api/2023-10/price_rules/{price_rule_id}/discount_codes.json" try: response = requests.post(url, data=json.dumps(data), headers=headers) if response.status_code == 200 or response.status_code == 201: print("Response:", response.text) return discount_code else: print( f"Error: Received status code {response.status_code}. Response Text: {response.text}" ) return None except requests.RequestException as e: print("Error:", e) return None @staticmethod def create_price_rule( store_name, access_token, offer_name, amount, discount_type, product_id, collection_id, min_purchase_amount=0, prerequisite_collection_id=None, prerequisite_quantity=None, ): headers = { "X-Shopify-Access-Token": access_token, "Content-Type": "application/json", } # Convert our discount type into shopify terms ("value type") value_type = "percentage" if discount_type == "percent" else "fixed_amount" # Format the start and end dates starts_at = datetime.utcnow().isoformat()[:-7] + "Z" # now ends_at = (datetime.utcnow() + timedelta(days=30)).isoformat()[ :-7 ] + "Z" # one month from now data = { "price_rule": { "title": f"StoryIt: {offer_name} (do not edit)", "value_type": value_type, "value": f'-{amount if value_type == "fixed_amount" else amount}', "customer_selection": "all", "usage_limit": 1, # single-use "starts_at": starts_at, "ends_at": ends_at, "target_type": "line_item", "target_selection": ( "entitled" if product_id or collection_id else "all" ), "allocation_method": ( "each" if product_id or prerequisite_collection_id else "across" ), "allocation_limit": 1 if product_id else None, } } # Add entitled product IDs if product_id is provided if product_id: data["price_rule"]["entitled_product_ids"] = [product_id] # Add entitled collection IDs if collection_id is provided if collection_id: data["price_rule"]["entitled_collection_ids"] = [collection_id] # Add min purchase amount if provided if ( min_purchase_amount and min_purchase_amount > 0 and not prerequisite_collection_id ): data["price_rule"]["prerequisite_subtotal_range"] = { "greater_than_or_equal_to": str(min_purchase_amount) } # Add prerequisite collection ID if provided if prerequisite_collection_id: data["price_rule"]["prerequisite_collection_ids"] = [ prerequisite_collection_id ] # Add prerequisite quantity if provided if prerequisite_quantity: data["price_rule"]["prerequisite_to_entitlement_quantity_ratio"] = { "prerequisite_quantity": prerequisite_quantity, "entitled_quantity": 1, } url = f"https://{store_name}.myshopify.com/admin/api/2023-10/price_rules.json" try: response = requests.post( url, json=data, headers=headers ) # Use json argument to automatically serialize data if response.status_code in {200, 201}: print("Response:", response.text) return response.json()["price_rule"]["id"] else: print( f"Error: Received status code {response.status_code}. Response Text: {response.text}" ) return None except requests.RequestException as e: print("Error:", e) return None @staticmethod def send_cashback( store_name, email, cashback_percentage, access_token ): headers = { "X-Shopify-Access-Token": access_token, "Content-Type": "application/json", } orders_url = f"https://{store_name}.myshopify.com/admin/api/2023-10/orders.json?status=any" orders_data = ShopifyService.send_get_request( orders_url, access_token ) # Getting the order data order = None if orders_data: # checking if the order data is valid for order in orders_data[ "orders" ]: # use the email to get the latest order ID of that customer print(order.get("contact_email")) contact_email = order.get( "contact_email", "No email provided" ) # Get the contact email if contact_email == email: order = order break if order: # this checks if we found the order of the designated email order_id = order.get("id") # Getting the id of the order print(order_id) transactions_url = f"https://{store_name}.myshopify.com/admin/api/2023-10/orders/{order_id}/transactions.json" # Fetch the transaction data of that order transactions_data = ShopifyService.send_get_request( transactions_url, access_token ) # Getting the order data if transactions_data: for transaction in transactions_data["transactions"]: # looping through transactions to find the one with the right order_id (a little confiusing as we make the API call with the order_id) if (transaction.get("order_id") == order_id and transaction.get("parent_id")): # Check if parent_id is there (because authorization transactions dont have a parent_id) parent_id = transaction.get("parent_id") # get the parent ID cashback_amount = float(transaction.get("amount")) * float( cashback_percentage / 100 ) # calculate the cashback amount (not sure what variable is used to get the amount they originally paid?) url = f"https://{store_name}.myshopify.com/admin/api/2023-10/orders/{order_id}/refunds.json" headers = { "X-Shopify-Access-Token": access_token, "Content-Type": "application/json", } # Dynamic refund data based on function parameters refund_data = { "refund": { "currency": "USD", "notify": True, "note": "Storyit Offer Completion", "transactions": [ { "parent_id": parent_id, # Use dynamic value "amount": cashback_amount, # Use dynamic value "kind": "refund", "gateway": "bogus", } ], } } refund_payload = json.dumps(refund_data) print(str(refund_payload)) try: response = requests.post( url, headers=headers, data=refund_payload ) if response.status_code == 201: print("Refund created successfully.") return json.loads(response.text)["refund"]["id"] else: print( f"Error: Received status code {response.status_code}. Response Text: {response.text}" ) return None except requests.RequestException as e: print("Error:", e) return None else: print(f"Could not find latest order for {email}") return None else: print(f"Could not find order data") return None if __name__ == '__main__': email = "conall@storyit.us" store_name = "storyit-notifications-test" access_token = "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" cashback_percentage = 5 refund_id = ShopifyService.send_cashback(store_name, email, cashback_percentage, access_token) print(refund_id)