262 lines
7.9 KiB
Python
262 lines
7.9 KiB
Python
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
|