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