from rowers.models import Rower, Workout, TombStone from rowers import utils from rowers.utils import NoTokenError import datetime from django.utils import timezone from datetime import timedelta from rowsandall_app.settings import ( ROJABO_CLIENT_ID, ROJABO_REDIRECT_URI, ROJABO_CLIENT_SECRET, SITE_URL, ROJABO_OAUTH_LOCATION, UPLOAD_SERVICE_URL, UPLOAD_SERVICE_SECRET, ) import gzip import rowers.mytypes as mytypes from rowers.utils import myqueue import requests import base64 from rowers.utils import dologging from json.decoder import JSONDecodeError import django_rq queue = django_rq.get_queue('default') queuelow = django_rq.get_queue('low') queuehigh = django_rq.get_queue('low') requests.packages.urllib3.disable_warnings() oauth_data = { 'client_id': ROJABO_CLIENT_ID, 'client_secret': ROJABO_CLIENT_SECRET, 'redirect_uri': ROJABO_REDIRECT_URI, 'autorization_uri': ROJABO_OAUTH_LOCATION+"oauth/authorize", 'content_type': 'application/json', 'tokenname': 'nktoken', 'refreshtokenname': 'nkrefreshtoken', 'expirydatename': 'nktokenexpirydate', 'bearer_auth': True, 'base_url': ROJABO_OAUTH_LOCATION+"oauth/token", 'scope': 'read', } def get_token(code): # pragma: no cover post_data = { "grant_type": "authorization_code", "code": code, "redirect_uri": oauth_data['redirect_uri'], } auth_string = '{id}:{secret}'.format( id=ROJABO_CLIENT_ID, secret=ROJABO_CLIENT_SECRET ) try: headers = {'Authorization': 'Basic %s' % base64.b64encode(auth_string), 'Content-Type': 'application/x-www-form-urlencoded'} except TypeError: headers = {'Authorization': 'Basic %s' % base64.b64encode( bytes(auth_string, 'utf-8')).decode('utf-8'), 'Content-Type': 'application/x-www-form-urlencoded'} response = requests.post(oauth_data['base_url'], data=post_data, headers=headers) if response.status_code != 200: return (0,response.reason) try: token_json = response.json() thetoken = token_json['access_token'] expires_in = token_json['expires_in'] refresh_token = token_json['refresh_token'] except (KeyError, JSONDecodeError) as e: # pragma: no cover thetoken = 0 expires_in = 0 refresh_token = 0 return [thetoken, expires_in, refresh_token] def rojabo_open(user): r = Rower.objects.get(user=user) if (r.rojabo_token == '') or (r.rojabo_token is None): raise NoTokenError("User has no token") else: if (timezone.now() > r.rojabo_tokenexpirydate): res = rower_rojabo_token_refresh(user) if res is None: # pragma: no cover raise NoTokenError("User has no token") if res[0] is not None: thetoken = res[0] else: # pragma: no cover raise NoTokenError("User has no token") else: thetoken = r.rojabo_token return thetoken def rower_rojabo_token_refresh(user): r = Rower.objects.get(user=user) res = do_refresh_token(r.rojabo_refreshtoken) if res[0]: access_token = res[0] expires_in = res[1] refresh_token = res[2] expirydatetime = timezone.now()+timedelta(seconds=expires_in) r = Rower.objects.get(user=user) r.rojabo_token = access_token r.rojabo_tokenexpirydate = expirydatetime r.rojabo_refreshtoken = refresh_token r.save() return r.rojabo_token else: # pragma: no cover return None def do_refresh_token(refreshtoken): post_data = { "grant_type": "refresh_token", "refresh_token": refreshtoken, "redirect_uri": oauth_data['redirect_uri'], } auth_string = '{id}:{secret}'.format( id=ROJABO_CLIENT_ID, secret=ROJABO_CLIENT_SECRET ) try: headers = {'Authorization': 'Basic %s' % base64.b64encode(auth_string), 'Content-Type': 'application/x-www-form-urlencoded'} except TypeError: headers = {'Authorization': 'Basic %s' % base64.b64encode( bytes(auth_string, 'utf-8')).decode('utf-8'), 'Content-Type': 'application/x-www-form-urlencoded'} response = requests.post(oauth_data['base_url'], data=post_data, headers=headers, verify=False) if response.status_code != 200: return (0,response.reason) try: token_json = response.json() thetoken = token_json['access_token'] expires_in = token_json['expires_in'] refresh_token = token_json['refresh_token'] except (KeyError, JSONDecodeError) as e: # pragma: no cover thetoken = 0 expires_in = 0 refresh_token = 0 return [thetoken, expires_in, refresh_token] aweekago = timezone.now()-timedelta(days=7) today = timezone.now() a_week_from_now = timezone.now()+timedelta(days=7) def stepsconvert(rojabo_steps, startid = 0, warmup = False, cooldown = False): workout_steps = [] for step in rojabo_steps: durationtype = 'Time' durationvalue = 10 if step['duration_type'] == 'seconds': durationvalue = 1000*step['duration_value'] # milliseconds if step['duration_type'] == 'meters': durationtype = 'Distance' durationvalue = step['duration_value']*100 # centimeters elif step['duration_type'] == 'strokes': durationtype = 'Time' try: durationvalue = int(60.*step['duration_value']/step['stroke_rate']) except TypeError: try: durationvalue = step['time']*1000 except KeyError: durationvalue = 1000 intensity = 'Active' if warmup: intensity = 'Warmup' if cooldown: intensity = 'Cooldown' targettype = 'Power' targetvalue = step['target_value'] if targetvalue is None: targettype = 'Cadence' targetvalue = step['stroke_rate'] if targettype == 'Power': targetvalue += 1000 if step['target_type'] == 'rest': targettype = '' intensity = 'Rest' targetvalue = 0 description = step['description'] if step['stroke_rate'] is not None: description = description +' Stroke Rate {cadence} SPM'.format( cadence = step['stroke_rate'] ) newstep = { 'stepId': startid, 'wkt_step_name': step['id'], 'durationType': durationtype, 'durationValue': durationvalue, 'targetType': targettype, 'targetValue': targetvalue, 'intensity': intensity, 'description': description } startid += 1 workout_steps.append(newstep) return workout_steps def get_rojabo_workout_list(user,startdate=aweekago,enddate=a_week_from_now): r = Rower.objects.get(user=user) if (r.rojabo_token == '') or (r.rojabo_token is None): # pragma: no cover s = "Token doesn't exist. Need to authorize" return custom_exception_handler(401, s) elif (timezone.now() > r.rojabo_tokenexpirydate): # pragma: no cover s = "Token expired. Needs to refresh." return custom_exception_handler(401, s) _ = rojabo_open(user) authorizationstring = str('Bearer ' + r.rojabo_token) headers = {'Authorization': authorizationstring, 'Content-Type': 'application/json'} date1 = startdate.strftime('%Y-%m-%d') date2 = enddate.strftime('%Y-%m-%d') url = ROJABO_OAUTH_LOCATION+'api/v2/training_sessions?from={date1}&to={date2}'.format(date1=date1,date2=date2) response = requests.get(url, headers=headers) return response