66 lines
2.3 KiB
Python
66 lines
2.3 KiB
Python
|
|
||
|
|
import os
|
||
|
|
import errno
|
||
|
|
import requests
|
||
|
|
import json
|
||
|
|
import datetime
|
||
|
|
|
||
|
|
import config
|
||
|
|
import blizzard
|
||
|
|
import spec
|
||
|
|
|
||
|
|
def log(*data):
|
||
|
|
print(datetime.datetime.now(), '|', *data)
|
||
|
|
|
||
|
|
class Test(object):
|
||
|
|
def __init__(self):
|
||
|
|
self.http = requests.Session()
|
||
|
|
self.credentials = blizzard.get_credentials(config.pwd)
|
||
|
|
|
||
|
|
def region_oauth(self, region: str) -> dict:
|
||
|
|
api = self.http.post(f"{blizzard.get_bnet_host(region)}/oauth/token", data={'grant_type': 'client_credentials'}, auth=self.credentials)
|
||
|
|
oauth = api.json()
|
||
|
|
log(region, oauth)
|
||
|
|
api.raise_for_status()
|
||
|
|
return oauth
|
||
|
|
|
||
|
|
def api_call(self, url: str, headers=None) -> requests.Response:
|
||
|
|
api = self.http.get(url, 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(self, dst, content):
|
||
|
|
with open(dst, 'w+') as f:
|
||
|
|
f.write(json.dumps(content.json(), indent=4))
|
||
|
|
|
||
|
|
def download_if_modified(self, path, namespace, region, locale, headers=None):
|
||
|
|
oauth = self.region_oauth(region)
|
||
|
|
access_token = oauth['access_token']
|
||
|
|
|
||
|
|
dst = f"{config.pwd}/test.{region}.{locale}.{path.replace('/', '_')}.{namespace}.json"
|
||
|
|
|
||
|
|
url = f"{blizzard.get_api_host(region)}{path}"
|
||
|
|
url += f"?namespace={namespace}-{region}&locale={locale}&access_token={access_token}"
|
||
|
|
|
||
|
|
log('*', url)
|
||
|
|
response = self.api_call(url, headers)
|
||
|
|
|
||
|
|
log('>', response.request.headers)
|
||
|
|
log('<', response.headers)
|
||
|
|
log(response.status_code)
|
||
|
|
return response.headers['Last-Modified']
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
test = Test()
|
||
|
|
lm = test.download_if_modified('/data/wow/achievement/index', 'static', 'eu', 'es_ES')
|
||
|
|
test.download_if_modified('/data/wow/achievement/index', 'static', 'eu', 'es_ES', headers={'If-Modified-Since': lm})
|
||
|
|
test.download_if_modified('/data/wow/achievement/index', 'static', 'eu', 'es_ES', headers={'If-Modified-Since': lm})
|
||
|
|
test.download_if_modified('/data/wow/achievement/index', 'static', 'eu', 'es_ES', headers={'If-Modified-Since': lm})
|