Versioning changes for Blizzard WoW Game Data APIs https://develop.battle.net/documentation/world-of-warcraft/game-data-apis
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

165 lines
5.0 KiB

import os
import errno
import requests
import json
import datetime
import time
import tempfile
import pytz
import config
import blizzard
import spec
def log(*data):
print(datetime.datetime.now(), '|', *data)
class Updater(object):
RFC2616 = '%a, %d %b %Y %H:%M:%S %Z'
def __init__(self, region: str):
self.region = region
self.http = requests.Session()
self.credentials = blizzard.get_credentials(config.pwd)
self.last_modified = {}
self.oauth = None
def get_oauth(self) -> dict:
if self.oauth is not None:
return self.oauth
api = self.http.post(f"{blizzard.get_bnet_host(self.region)}/oauth/token", data={'grant_type': 'client_credentials'}, auth=self.credentials)
oauth = api.json()
#log(region, oauth)
api.raise_for_status()
self.oauth = oauth
return oauth
def api_call(self, path: str, namespace: str, access_token: str, headers=None) -> requests.Response:
url = f"{blizzard.get_api_host(self.region)}{path}"
qs = {'namespace': f"{namespace}-{self.region}", 'access_token': access_token}
api = self.http.get(url, params=qs, headers=headers) #, headers={'Authorization': f"Authorization: Token {access_token}"})
api.raise_for_status()
return api
def create_dst(self, dst: str):
try:
os.makedirs(os.path.dirname(dst))
except OSError as e:
if e.errno != errno.EEXIST:
raise
def save_raw(self, path: str, raw: requests.Response):
dst = f"{config.raw}/{path.replace('/', '_')}.json"
self.create_dst(dst)
# safe write: tmp write and then mv
fd, tmp_path = tempfile.mkstemp(dir=config.pwd)
try:
with os.fdopen(fd, 'w') as tmp:
json.dump(raw.json(), tmp, indent=2)
finally:
os.replace(tmp_path, dst)
def get_last_modified(self, path: str):
# cached!
if path in self.last_modified:
return self.last_modified[path]
dst = f"{config.raw}/{path.replace('/', '_')}.json"
try:
# mtime to datetime from timestamp, default tz, replace to GMT, format
modified = datetime.datetime.fromtimestamp(os.path.getmtime(dst)).astimezone().replace(tzinfo=pytz.timezone('GMT')).strftime(self.RFC2616)
self.last_modified[path] = modified
return self.last_modified[path]
except FileNotFoundError:
return None
def set_last_modified(self, path: str, modified: str):
dst = f"{config.raw}/{path.replace('/', '_')}.json"
self.last_modified[path] = modified
# same Last-Modified for raw file
epoch = time.mktime(time.strptime(modified, self.RFC2616)) # Tue, 11 May 2021 14:06:37 GMT
os.utime(dst, (epoch, epoch))
def iterate_index(self):
try:
oauth = self.get_oauth()
access_token = oauth['access_token']
except (requests.exceptions.HTTPError, KeyError) as e:
log(type(e), e)
return
for api in spec.apis:
# only indexes
if not 'index' in api or not api['index']:
continue
try:
last_modified = self.get_last_modified(api['path'])
headers = {'If-Modified-Since': last_modified} if last_modified is not None else None
response = self.api_call(api['path'], api['namespace'], access_token, headers=headers)
if response.status_code == 200:
self.save_raw(api['path'], response)
self.set_last_modified(api['path'], response.headers['Last-Modified'])
except requests.exceptions.HTTPError as e:
log(e)
continue
def iterate_links(self):
try:
oauth = self.get_oauth()
access_token = oauth['access_token']
except (requests.exceptions.HTTPError, KeyError) as e:
log(type(e), e)
return
for api in spec.apis:
# no indexes
if 'index' in api and api['index']:
continue
log(api)
continue
try:
last_modified = self.get_last_modified(api['path'])
headers = {'If-Modified-Since': last_modified} if last_modified is not None else None
response = self.api_call(api['path'], api['namespace'], access_token, headers=headers)
log(response.status_code, api)
if response.status_code == 200:
self.save_last_modified(api['path'], response.headers['Last-Modified'])
self.save_raw(api['path'], response)
except requests.exceptions.HTTPError as e:
log(e)
continue
if __name__ == "__main__":
region = blizzard.get_by_key(spec.regions, 'code', 'us')['code']
updater = Updater(region)
updater.iterate_index()
#updater.iterate_links()