wow-token-updater.py (5012B)
1 import datetime 2 import time 3 import boto3 4 import json 5 import os 6 import requests 7 8 9 local_region = os.environ['AWS_REGION'] 10 11 dynamo_client = boto3.client('dynamodb', region_name=local_region) 12 tables = { 13 'retail': { 14 'recent': 'wow-token-price-recent', 15 'current': 'wow-token-price', 16 'timestream': 'wow-token-price-history' 17 }, 18 'classic': { 19 'recent': 'wow-token-classic-price-recent', 20 'current': 'wow-token-classic-price', 21 'timestream': 'wow-token-classic-price-history' 22 } 23 } 24 25 26 # a lambda handler to handle eventbridge triggered vents 27 def lambda_handler(event, context): 28 flavors = ['retail', 'classic'] 29 regions = ['us', 'eu', 'kr', 'tw'] 30 timestamp = int(datetime.datetime.utcnow().timestamp()) 31 for flavor in flavors: 32 flavor_handler(flavor, regions, timestamp) 33 34 35 def flavor_handler(flavor: str, regions: list, timestamp: int) -> None: 36 for region in regions: 37 if flavor == 'retail': 38 namespace = 'dynamic' 39 else: 40 namespace = 'dynamic-classic' 41 live_price = get_token_price_from_blizzard(region, namespace) 42 print(f"{flavor} {region} live price {live_price}") 43 print(f'Updating {region.upper()}...') 44 update_token_price(flavor, region, timestamp, live_price) 45 46 47 def update_token_price(flavor: str, region: str, current_time_epoch: int, live_price: int) -> None: 48 stored_price = get_stored_price(flavor, region) 49 print(f'Current live price {live_price}') 50 print(f'Current stored price {stored_price}') 51 if stored_price != live_price: 52 # update the stored price and the recent price 53 print(f"Stored price differs from the live price, updating databases") 54 update_stored_token_price(flavor, region, live_price, current_time_epoch) 55 update_recent_token_price(flavor, region, live_price, current_time_epoch) 56 else: 57 print(f"Price hasn't changed for {flavor} {region.upper()}") 58 59 60 # update the current price records in DynamoDB 61 def update_stored_token_price(flavor: str, region: str, price: int, current_time_epoch: int) -> None: 62 dynamo_client.update_item( 63 TableName=tables[flavor]['current'], 64 Key={ 65 'region': { 66 'S': region 67 } 68 }, 69 UpdateExpression='SET price = :p, current_time = :t', 70 ExpressionAttributeValues={ 71 ':p': { 72 'S': str(price) 73 }, 74 ':t': { 75 'S': str(current_time_epoch) 76 } 77 } 78 # ReturnValues="UPDATED_NEW" 79 ) 80 print(f'Updated {flavor} {region.upper()} price to {price}') 81 82 # add a record to the recent token price table in DynamoDB 83 def update_recent_token_price(flavor: str, region: str, price: int, current_time_epoch: int) -> None: 84 dynamo_client.put_item( 85 TableName=tables[flavor]['recent'], 86 Item={ 87 'region': { 88 'S': region 89 }, 90 'price': { 91 'S': str(price) 92 }, 93 'timestamp': { 94 'N': str(current_time_epoch) 95 }, 96 'expire': { 97 'N': str(current_time_epoch + (60 * 60 * 24 * 31 * 12)) 98 } 99 } 100 # ReturnValues="UPDATED_NEW" 101 ) 102 print(f'Added {region.upper()} price {price} to {tables[flavor]["recent"]} table') 103 104 # get the current stored token price from dynamodb 105 def get_stored_price(flavor: str, region: str) -> int: 106 response = dynamo_client.get_item( 107 TableName=tables[flavor]['current'], 108 Key={ 109 'region': { 110 'S': region 111 } 112 } 113 ) 114 return int(response['Item']['price']['S']) 115 116 117 def get_combined_live_price(game_flavor: str) -> dict: 118 if game_flavor == 'retail': 119 namespace = 'dynamic' 120 else: 121 namespace = 'dynamic-classic' 122 123 return { 124 'us': get_token_price_from_blizzard('us', namespace), 125 'eu': get_token_price_from_blizzard('eu', namespace), 126 'kr': get_token_price_from_blizzard('kr', namespace), 127 'tw': get_token_price_from_blizzard('tw', namespace) 128 } 129 130 131 def get_token_price_from_blizzard(region: str, namespace: str) -> int: 132 api_endpoint = f'https://{region}.api.blizzard.com/data/wow/token/index' 133 params = {'namespace': f'{namespace}-{region}'} 134 headers = {'Authorization': f'Bearer {get_oauth_token()}'} 135 response = requests.get(api_endpoint, params=params, headers=headers) 136 response.raise_for_status() 137 return int(response.json()['price']) 138 139 140 def get_oauth_token() -> str: 141 url = 'https://us.battle.net/oauth/token' 142 payload = { 143 'grant_type': 'client_credentials', 144 'client_id': os.environ.get('BLIZZARD_CLIENT_ID'), 145 'client_secret': os.environ.get('BLIZZARD_CLIENT_SECRET') 146 } 147 response = requests.post(url, data=payload) 148 response.raise_for_status() 149 return json.loads(response.text)['access_token'] 150 151 152 if __name__ == '__main__': 153 lambda_handler(None, None)